test: Use CCheckQueue local thread pool

hebasto · hebasto · commit dba30695fc42 · 2020-09-24T06:55:33.000+03:00
diff --git a/src/test/checkqueue_tests.cpp b/src/test/checkqueue_tests.cpp
@@ -148,10 +148,7 @@ typedef CCheckQueue<FrozenCleanupCheck> FrozenCleanup_Queue;
 static void Correct_Queue_range(std::vector<size_t> range)
 {
     auto small_queue = MakeUnique<Correct_Queue>(QUEUE_BATCH_SIZE);
-    boost::thread_group tg;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-       tg.create_thread([&]{small_queue->Thread();});
-    }
+    small_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
     // Make vChecks here to save on malloc (this test can be slow...)
     std::vector<FakeCheckCheckCompletion> vChecks;
     for (const size_t i : range) {
@@ -168,8 +165,7 @@ static void Correct_Queue_range(std::vector<size_t> range)
             BOOST_REQUIRE_EQUAL(FakeCheckCheckCompletion::n_calls, i);
         }
     }
-    tg.interrupt_all();
-    tg.join_all();
+    small_queue->StopWorkerThreads();
 }
 
 /** Test that 0 checks is correct
@@ -212,11 +208,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Correct_Random)
 BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)
 {
     auto fail_queue = MakeUnique<Failing_Queue>(QUEUE_BATCH_SIZE);
-
-    boost::thread_group tg;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-       tg.create_thread([&]{fail_queue->Thread();});
-    }
+    fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
 
     for (size_t i = 0; i < 1001; ++i) {
         CCheckQueueControl<FailingCheck> control(fail_queue.get());
@@ -237,18 +229,14 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)
             BOOST_REQUIRE(success);
         }
     }
-    tg.interrupt_all();
-    tg.join_all();
+    fail_queue->StopWorkerThreads();
 }
 // Test that a block validation which fails does not interfere with
 // future blocks, ie, the bad state is cleared.
 BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)
 {
     auto fail_queue = MakeUnique<Failing_Queue>(QUEUE_BATCH_SIZE);
-    boost::thread_group tg;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-       tg.create_thread([&]{fail_queue->Thread();});
-    }
+    fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
 
     for (auto times = 0; times < 10; ++times) {
         for (const bool end_fails : {true, false}) {
@@ -263,8 +251,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)
             BOOST_REQUIRE(r != end_fails);
         }
     }
-    tg.interrupt_all();
-    tg.join_all();
+    fail_queue->StopWorkerThreads();
 }
 
 // Test that unique checks are actually all called individually, rather than
@@ -273,11 +260,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)
 BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)
 {
     auto queue = MakeUnique<Unique_Queue>(QUEUE_BATCH_SIZE);
-    boost::thread_group tg;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-       tg.create_thread([&]{queue->Thread();});
-
-    }
+    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
 
     size_t COUNT = 100000;
     size_t total = COUNT;
@@ -300,8 +283,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)
         }
         BOOST_REQUIRE(r);
     }
-    tg.interrupt_all();
-    tg.join_all();
+    queue->StopWorkerThreads();
 }
 
 
@@ -313,10 +295,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)
 BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)
 {
     auto queue = MakeUnique<Memory_Queue>(QUEUE_BATCH_SIZE);
-    boost::thread_group tg;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-       tg.create_thread([&]{queue->Thread();});
-    }
+    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
     for (size_t i = 0; i < 1000; ++i) {
         size_t total = i;
         {
@@ -335,20 +314,16 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)
         }
         BOOST_REQUIRE_EQUAL(MemoryCheck::fake_allocated_memory, 0U);
     }
-    tg.interrupt_all();
-    tg.join_all();
+    queue->StopWorkerThreads();
 }
 
 // Test that a new verification cannot occur until all checks
 // have been destructed
 BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)
 {
     auto queue = MakeUnique<FrozenCleanup_Queue>(QUEUE_BATCH_SIZE);
-    boost::thread_group tg;
     bool fails = false;
-    for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {
-        tg.create_thread([&]{queue->Thread();});
-    }
+    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
     std::thread t0([&]() {
         CCheckQueueControl<FrozenCleanupCheck> control(queue.get());
         std::vector<FrozenCleanupCheck> vChecks(1);
@@ -378,9 +353,8 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)
     FrozenCleanupCheck::cv.notify_one();
     // Wait for control to finish
     t0.join();
-    tg.interrupt_all();
-    tg.join_all();
     BOOST_REQUIRE(!fails);
+    queue->StopWorkerThreads();
 }
 
 
@@ -445,4 +419,3 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
     }
 }
 BOOST_AUTO_TEST_SUITE_END()
-
diff --git a/src/test/transaction_tests.cpp b/src/test/transaction_tests.cpp
@@ -427,12 +427,10 @@ BOOST_AUTO_TEST_CASE(test_big_witness_transaction)
 
     // check all inputs concurrently, with the cache
     PrecomputedTransactionData txdata(tx);
-    boost::thread_group threadGroup;
     CCheckQueue<CScriptCheck> scriptcheckqueue(128);
     CCheckQueueControl<CScriptCheck> control(&scriptcheckqueue);
 
-    for (int i=0; i<20; i++)
-        threadGroup.create_thread(std::bind(&CCheckQueue<CScriptCheck>::Thread, std::ref(scriptcheckqueue)));
+    scriptcheckqueue.StartWorkerThreads(20);
 
     std::vector<Coin> coins;
     for(uint32_t i = 0; i < mtx.vin.size(); i++) {
@@ -454,9 +452,7 @@ BOOST_AUTO_TEST_CASE(test_big_witness_transaction)
 
     bool controlCheck = control.Wait();
     assert(controlCheck);
-
-    threadGroup.interrupt_all();
-    threadGroup.join_all();
+    scriptcheckqueue.StopWorkerThreads();
 }
 
 SignatureData CombineSignatures(const CMutableTransaction& input1, const CMutableTransaction& input2, const CTransactionRef tx)

Original file line number	Diff line number	Diff line change
`@@ -148,10 +148,7 @@ typedef CCheckQueue<FrozenCleanupCheck> FrozenCleanup_Queue;`
`148`	`148`	`static void Correct_Queue_range(std::vector<size_t> range)`
`149`	`149`	`{`
`150`	`150`	`auto small_queue = MakeUnique<Correct_Queue>(QUEUE_BATCH_SIZE);`
`151`		`- boost::thread_group tg;`
`152`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`153`		`- tg.create_thread([&]{small_queue->Thread();});`
`154`		`- }`
	`151`	`+ small_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`155`	`152`	`// Make vChecks here to save on malloc (this test can be slow...)`
`156`	`153`	`std::vector<FakeCheckCheckCompletion> vChecks;`
`157`	`154`	`for (const size_t i : range) {`
`@@ -168,8 +165,7 @@ static void Correct_Queue_range(std::vector<size_t> range)`
`168`	`165`	`BOOST_REQUIRE_EQUAL(FakeCheckCheckCompletion::n_calls, i);`
`169`	`166`	`}`
`170`	`167`	`}`
`171`		`- tg.interrupt_all();`
`172`		`- tg.join_all();`
	`168`	`+ small_queue->StopWorkerThreads();`
`173`	`169`	`}`
`174`	`170`
`175`	`171`	`/** Test that 0 checks is correct`
`@@ -212,11 +208,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Correct_Random)`
`212`	`208`	`BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)`
`213`	`209`	`{`
`214`	`210`	`auto fail_queue = MakeUnique<Failing_Queue>(QUEUE_BATCH_SIZE);`
`215`		`-`
`216`		`- boost::thread_group tg;`
`217`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`218`		`- tg.create_thread([&]{fail_queue->Thread();});`
`219`		`- }`
	`211`	`+ fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`220`	`212`
`221`	`213`	`for (size_t i = 0; i < 1001; ++i) {`
`222`	`214`	`CCheckQueueControl<FailingCheck> control(fail_queue.get());`
`@@ -237,18 +229,14 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)`
`237`	`229`	`BOOST_REQUIRE(success);`
`238`	`230`	`}`
`239`	`231`	`}`
`240`		`- tg.interrupt_all();`
`241`		`- tg.join_all();`
	`232`	`+ fail_queue->StopWorkerThreads();`
`242`	`233`	`}`
`243`	`234`	`// Test that a block validation which fails does not interfere with`
`244`	`235`	`// future blocks, ie, the bad state is cleared.`
`245`	`236`	`BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)`
`246`	`237`	`{`
`247`	`238`	`auto fail_queue = MakeUnique<Failing_Queue>(QUEUE_BATCH_SIZE);`
`248`		`- boost::thread_group tg;`
`249`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`250`		`- tg.create_thread([&]{fail_queue->Thread();});`
`251`		`- }`
	`239`	`+ fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`252`	`240`
`253`	`241`	`for (auto times = 0; times < 10; ++times) {`
`254`	`242`	`for (const bool end_fails : {true, false}) {`
`@@ -263,8 +251,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)`
`263`	`251`	`BOOST_REQUIRE(r != end_fails);`
`264`	`252`	`}`
`265`	`253`	`}`
`266`		`- tg.interrupt_all();`
`267`		`- tg.join_all();`
	`254`	`+ fail_queue->StopWorkerThreads();`
`268`	`255`	`}`
`269`	`256`
`270`	`257`	`// Test that unique checks are actually all called individually, rather than`
`@@ -273,11 +260,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)`
`273`	`260`	`BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)`
`274`	`261`	`{`
`275`	`262`	`auto queue = MakeUnique<Unique_Queue>(QUEUE_BATCH_SIZE);`
`276`		`- boost::thread_group tg;`
`277`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`278`		`- tg.create_thread([&]{queue->Thread();});`
`279`		`-`
`280`		`- }`
	`263`	`+ queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`281`	`264`
`282`	`265`	`size_t COUNT = 100000;`
`283`	`266`	`size_t total = COUNT;`
`@@ -300,8 +283,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)`
`300`	`283`	`}`
`301`	`284`	`BOOST_REQUIRE(r);`
`302`	`285`	`}`
`303`		`- tg.interrupt_all();`
`304`		`- tg.join_all();`
	`286`	`+ queue->StopWorkerThreads();`
`305`	`287`	`}`
`306`	`288`
`307`	`289`
`@@ -313,10 +295,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)`
`313`	`295`	`BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)`
`314`	`296`	`{`
`315`	`297`	`auto queue = MakeUnique<Memory_Queue>(QUEUE_BATCH_SIZE);`
`316`		`- boost::thread_group tg;`
`317`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`318`		`- tg.create_thread([&]{queue->Thread();});`
`319`		`- }`
	`298`	`+ queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`320`	`299`	`for (size_t i = 0; i < 1000; ++i) {`
`321`	`300`	`size_t total = i;`
`322`	`301`	`{`
`@@ -335,20 +314,16 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)`
`335`	`314`	`}`
`336`	`315`	`BOOST_REQUIRE_EQUAL(MemoryCheck::fake_allocated_memory, 0U);`
`337`	`316`	`}`
`338`		`- tg.interrupt_all();`
`339`		`- tg.join_all();`
	`317`	`+ queue->StopWorkerThreads();`
`340`	`318`	`}`
`341`	`319`
`342`	`320`	`// Test that a new verification cannot occur until all checks`
`343`	`321`	`// have been destructed`
`344`	`322`	`BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)`
`345`	`323`	`{`
`346`	`324`	`auto queue = MakeUnique<FrozenCleanup_Queue>(QUEUE_BATCH_SIZE);`
`347`		`- boost::thread_group tg;`
`348`	`325`	`bool fails = false;`
`349`		`- for (auto x = 0; x < SCRIPT_CHECK_THREADS; ++x) {`
`350`		`- tg.create_thread([&]{queue->Thread();});`
`351`		`- }`
	`326`	`+ queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);`
`352`	`327`	`std::thread t0([&]() {`
`353`	`328`	`CCheckQueueControl<FrozenCleanupCheck> control(queue.get());`
`354`	`329`	`std::vector<FrozenCleanupCheck> vChecks(1);`
`@@ -378,9 +353,8 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)`
`378`	`353`	`FrozenCleanupCheck::cv.notify_one();`
`379`	`354`	`// Wait for control to finish`
`380`	`355`	`t0.join();`
`381`		`- tg.interrupt_all();`
`382`		`- tg.join_all();`
`383`	`356`	`BOOST_REQUIRE(!fails);`
	`357`	`+ queue->StopWorkerThreads();`
`384`	`358`	`}`
`385`	`359`
`386`	`360`
`@@ -445,4 +419,3 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)`
`445`	`419`	`}`
`446`	`420`	`}`
`447`	`421`	`BOOST_AUTO_TEST_SUITE_END()`
`448`		`-`