fix synchronization

Author: tnagler

inst/include/RcppThread/ThreadPool.hpp

@@ -22,6 +22,16 @@
 
 namespace RcppThread {
 
+namespace util {
+void
+waitAndSync(tpool::FinishLine& finishLine)
+{
+    finishLine.wait_for(std::chrono::milliseconds(50));
+    Rcout << "";
+    checkUserInterrupt();
+}
+}
+
 //! Implemenation of the thread pool pattern based on `Thread`.
 class ThreadPool
 {
@@ -234,19 +244,8 @@ ThreadPool::parallelForEach(I& items, F&& f, size_t nBatches)
 inline void
 ThreadPool::wait()
 {
-    while (!taskManager_.empty()) {
-        finishLine_.wait_for(std::chrono::milliseconds(50));
-        if (isInterrupted()) {
-            taskManager_.stop();
-            break;
-        }
-        if (taskManager_.empty())
-            break;
-        Rcout << "";
-        std::this_thread::yield();
-    }
-    Rcout << "";
-    checkUserInterrupt();
+    while (!taskManager_.empty())
+        util::waitAndSync(finishLine_);
 }
 
 //! waits for all jobs to finish and joins all threads.

inst/include/RcppThread/parallelFor.hpp

@@ -59,17 +59,14 @@ parallelFor(int begin,
     };
     for (const auto& batch : batches)
         ThreadPool::globalInstance().push(doBatch, batch);
-    finishLine.wait();
+    util::waitAndSync(finishLine);
 }
 
 //! computes a range-based for loop in parallel batches.
 //! @param items an object allowing for `items.size()` and whose elements
 //!   are accessed by the `[]` operator.
 //! @param f a function (the 'loop body').
-//! @param nThreads the number of threads to use; the default uses the
-//! number
-//!   of cores in the machine;  if `nThreads = 0`, all work will be done in
-//!   the main thread.
+//! @param nThreads deprecated; loop is run on global thread pool.
 //! @param nBatches the number of batches to create; the default (0)
 //!   triggers a heuristic to automatically determine the number of batches.
 //! @details Consider the following code:
@@ -101,7 +98,6 @@ parallelForEach(I& items,
     const auto begin_it = std::begin(items);
     const auto end_it = std::end(items);
     auto size = std::distance(begin_it, end_it);
-    parallelFor(
-      0, size, [f, &items, &begin_it](int i) { f(*(begin_it + i)); });
+    parallelFor(0, size, [f, &items, &begin_it](int i) { f(*(begin_it + i)); });
 }
 }

tests/tests.cpp

@@ -379,10 +379,10 @@ void testNestedParallelForEach()
 // {
 //     ThreadPool pool(2);
 //     auto dummy = [] {
-//         std::this_thread::sleep_for(std::chrono::milliseconds(500));
+//         std::this_thread::sleep_for(std::chrono::milliseconds(100));
 //         checkUserInterrupt();
 //     };
-//     for (size_t i = 0; i < 10; i++)
+//     for (size_t i = 0; i < 20; i++)
 //         pool.push(dummy);
 //     pool.join();
 //     std::this_thread::sleep_for(std::chrono::milliseconds(5000));
@@ -391,15 +391,15 @@ void testNestedParallelForEach()
 // // [[Rcpp::export]]
 // void testThreadPoolInterruptWait()
 // {
-//     ThreadPool pool(0);
+//     ThreadPool pool(2);
 //     auto dummy = [] {
-//         std::this_thread::sleep_for(std::chrono::milliseconds(500));
+//         std::this_thread::sleep_for(std::chrono::milliseconds(100));
 //         checkUserInterrupt();
 //     };
 //     for (size_t i = 0; i < 20; i++) {
 //         pool.push(dummy);
 //     }
-//     pool.join();
+//     pool.wait();
 //     std::this_thread::sleep_for(std::chrono::milliseconds(3000));
 // }
 
@@ -409,7 +409,7 @@ void testProgressCounter()
 {
     RcppThread::ProgressCounter cntr(20, 1);
     RcppThread::parallelFor(0, 20, [&] (int i) {
-        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
         cntr++;
     });
 }
@@ -420,7 +420,7 @@ void testProgressBar()
     // 20 iterations in loop, update progress every 1 sec
     RcppThread::ProgressBar bar(20, 1);
     RcppThread::parallelFor(0, 20, [&] (int i) {
-        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
         ++bar;
     });
 }