adapt functions

Author: tnagler

inst/include/RcppThread/ThreadPool.hpp

@@ -36,6 +36,13 @@ class ThreadPool
     ThreadPool& operator=(const ThreadPool&) = delete;
     ThreadPool& operator=(ThreadPool&& other) = delete;
 
+    //! @brief returns a reference to the global thread pool instance.
+    static ThreadPool& globalInstance()
+    {
+        static ThreadPool instance_;
+        return instance_;
+    }
+
     template<class F, class... Args>
     void push(F&& f, Args&&... args);
 

inst/include/RcppThread/parallelFor.hpp

@@ -13,11 +13,9 @@ namespace RcppThread {
 
 //! computes an index-based for loop in parallel batches.
 //! @param begin first index of the loop.
-//! @param size the loop runs in the range `[begin, begin + size)`.
+//! @param end the loop runs in the range `[begin, end)`.
 //! @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:
@@ -33,29 +31,45 @@ namespace RcppThread {
 //!     x[i] = i;
 //! });
 //! ```
-//! The function sets up a `ThreadPool` object to do the scheduling. If you
-//! want to run multiple parallel for loops, consider creating a `ThreadPool`
-//! yourself and using `ThreadPool::forEach()`.
+//! The function dispatches to a global thread pool, so it can safely be nested
+//! or called multiple times with almost no overhead.
 //!
 //! **Caution**: if the iterations are not independent from another,
 //! the tasks need to be synchronized manually (e.g., using mutexes).
 template<class F>
-inline void parallelFor(ptrdiff_t begin, ptrdiff_t size, F&& f,
-                        size_t nThreads = std::thread::hardware_concurrency(),
-                        size_t nBatches = 0)
+inline void
+parallelFor(int begin,
+            int end,
+            F&& f,
+            size_t nThreads = std::thread::hardware_concurrency(),
+            size_t nBatches = 0)
 {
-    ThreadPool pool(nThreads);
-    pool.parallelFor(begin, size, f, nBatches);
-    pool.join();
+    if (end < begin)
+        throw std::runtime_error("can only run forward loops");
+    if (end == begin)
+        return;
+
+    nThreads = std::thread::hardware_concurrency();
+    auto batches = createBatches(begin, end - begin, nThreads, nBatches);
+    tpool::FinishLine finishLine{ batches.size() };
+    auto doBatch = [f, &finishLine](const Batch& b) {
+        for (ptrdiff_t i = b.begin; i < b.end; i++)
+            f(i);
+        finishLine.cross();
+    };
+    for (const auto& batch : batches)
+        ThreadPool::globalInstance().push(doBatch, batch);
+    finishLine.wait();
 }
 
 //! 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 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 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:
@@ -71,21 +85,23 @@ inline void parallelFor(ptrdiff_t begin, ptrdiff_t size, F&& f,
 //!     xx *= 2;
 //! });
 //! ```
-//! The function sets up a `ThreadPool` object to do the scheduling. If you
-//! want to run multiple parallel for loops, consider creating a `ThreadPool`
-//! yourself and using `ThreadPool::forEach()`.
+//! The function dispatches to a global thread pool, so it can safely be nested
+//! or called multiple times with almost no overhead.
 //!
 //! **Caution**: if the iterations are not independent from another,
 //! the tasks need to be synchronized manually (e.g., using mutexes).
 template<class I, class F>
-inline void parallelForEach(I& items, F&& f,
-    size_t nThreads = std::thread::hardware_concurrency(),
-    size_t nBatches = 0)
+inline void
+parallelForEach(I& items,
+                F&& f,
+                size_t nThreads = std::thread::hardware_concurrency(),
+                size_t nBatches = 0)
 {
-    ThreadPool pool(nThreads);
-    pool.parallelForEach(items, f, nBatches);
-    pool.join();
+    // loop ranges ranges indicate iterator offset
+    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)); });
 }
-
-
 }

inst/include/RcppThread/tpool.hpp

@@ -54,8 +54,9 @@ class FinishLine
     //! indicates that a runner has crossed the finish line.
     void cross() noexcept
     {
-        if (--runners_ <= 0)
+        if (--runners_ <= 0) {
             cv_.notify_all();
+        }
     }
 
     //! waits for all active runners to cross the finish line.
@@ -188,18 +189,24 @@ class TaskQueue
     //! currently locked; enlarges the queue if full.
     bool try_push(Task&& task)
     {
+        // must hold lock in case there are multiple producers, abort if already
+        // taken, so we can check out next queue
+        std::unique_lock<std::mutex> lk(mutex_, std::try_to_lock);
+        if (!lk)
+            return false;
+
         auto b = bottom_.load(m_relaxed);
         auto t = top_.load(m_acquire);
         RingBuffer<Task>* buf_ptr = buffer_.load(m_relaxed);
 
         if (buf_ptr->capacity() < (b - t) + 1) {
-            // capacity reached, create buffer with double size
             old_buffers_.emplace_back(
               exchange(buf_ptr, buf_ptr->enlarge(b, t)));
             buffer_.store(buf_ptr, m_relaxed);
         }
 
         buf_ptr->store(b, std::move(task));
+
         std::atomic_thread_fence(m_release);
         bottom_.store(b + 1, m_relaxed);
 
@@ -209,10 +216,6 @@ class TaskQueue
     //! pops a task from the top of the queue; returns false if lost race.
     bool try_pop(Task& task)
     {
-        std::unique_lock<std::mutex> lk(mutex_, std::try_to_lock);
-        if (!lk)
-            return false;
-
         auto t = top_.load(m_acquire);
         std::atomic_thread_fence(m_seq_cst);
         auto b = bottom_.load(m_acquire);

tests/tests.cpp

@@ -162,15 +162,21 @@ void testThreadPoolNestedParallelFor()
             x[i][j] *= 2;
         });
     });
-    pool.join();
+    pool.wait();
 
     size_t count_wrong = 0;
     for (auto xx : x) {
         for (auto xxx : xx)
             count_wrong += xxx != 2;
     }
-    if (count_wrong > 0)
+    if (count_wrong > 0) {
+        for (auto xx : x) {
+            for (auto xxx : xx)
+                std::cout << xxx;
+            std::cout << std::endl;
+        }
         throw std::runtime_error("nested parallelFor gives wrong result");
+    }
 }
 
 // [[Rcpp::export]]
@@ -212,15 +218,21 @@ void testThreadPoolNestedParallelForEach()
             xxx *= 2;
         });
     });
-    pool.join();
+    pool.wait();
 
     size_t count_wrong = 0;
     for (auto xx : x) {
         for (auto xxx : xx)
             count_wrong += xxx != 2;
     }
-    if (count_wrong > 0)
-        throw std::runtime_error("nested parallelForEach gives wrong result");
+    if (count_wrong > 0) {
+        for (auto xx : x) {
+            for (auto xxx : xx)
+                std::cout << xxx;
+            std::cout << std::endl;
+        }
+        throw std::runtime_error("nested parallelFor gives wrong result");
+    }
 }
 
 // [[Rcpp::export]]
@@ -293,8 +305,14 @@ void testNestedParallelFor()
         for (auto xxx : xx)
             count_wrong += xxx != 2;
     }
-    if (count_wrong > 0)
+    if (count_wrong > 0) {
+        for (auto xx : x) {
+            for (auto xxx : xx)
+                std::cout << xxx;
+            std::cout << std::endl;
+        }
         throw std::runtime_error("nested parallelFor gives wrong result");
+    }
 }
 
 // [[Rcpp::export]]
@@ -391,7 +409,7 @@ void testProgressCounter()
 {
     RcppThread::ProgressCounter cntr(20, 1);
     RcppThread::parallelFor(0, 20, [&] (int i) {
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
         cntr++;
     });
 }
@@ -402,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(100));
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
         ++bar;
     });
 }

tests/testthat/tests.R

@@ -34,52 +34,52 @@ test_that("parallelFor works", {
     expect_silent(testThreadPoolParallelFor())
 })
 
-# test_that("nested parallelFor works", {
-#     expect_silent(testThreadPoolNestedParallelFor())
-# })
-
-# test_that("parallelForEach works", {
-#     expect_silent(testThreadPoolParallelForEach())
-# })
-#
-# test_that("nested parallelForEach works", {
-#     expect_silent(testThreadPoolNestedParallelForEach())
-# })
-#
-# test_that("works single threaded", {
-#     expect_silent(testThreadPoolSingleThreaded())
-# })
-#
-# test_that("destructible without join", {
-#     expect_silent(testThreadPoolDestructWOJoin())
-# })
-#
-#
-# ## -------------------------------------------------------
-# context("Parallel for functions")
-# test_that("parallelFor works", {
-#     expect_silent(testParallelFor())
-# })
-#
-# test_that("nested parallelFor works", {
-#     expect_silent(testNestedParallelFor())
-# })
-#
-# test_that("parallelForEach works", {
-#     expect_silent(testParallelForEach())
-# })
-#
-# test_that("nested parallelForEach works", {
-#     expect_silent(testNestedParallelForEach())
-# })
-#
-#
-# ## ------------------------------------------------------
-# context("Progress tracking")
-# test_that("ProgressCounter works", {
-#     expect_output(testProgressCounter(), "100% \\(done\\)")
-# })
-#
-# test_that("ProgressBar works", {
-#     expect_output(testProgressBar(),"100% \\(done\\)")
-# })
+test_that("nested parallelFor works", {
+    expect_silent(testThreadPoolNestedParallelFor())
+})
+
+test_that("parallelForEach works", {
+    expect_silent(testThreadPoolParallelForEach())
+})
+
+test_that("nested parallelForEach works", {
+    expect_silent(testThreadPoolNestedParallelForEach())
+})
+
+test_that("works single threaded", {
+    expect_silent(testThreadPoolSingleThreaded())
+})
+
+test_that("destructible without join", {
+    expect_silent(testThreadPoolDestructWOJoin())
+})
+
+
+## -------------------------------------------------------
+context("Parallel for functions")
+test_that("parallelFor works", {
+    expect_silent(testParallelFor())
+})
+
+test_that("nested parallelFor works", {
+    expect_silent(testNestedParallelFor())
+})
+
+test_that("parallelForEach works", {
+    expect_silent(testParallelForEach())
+})
+
+test_that("nested parallelForEach works", {
+    expect_silent(testNestedParallelForEach())
+})
+
+
+## ------------------------------------------------------
+context("Progress tracking")
+test_that("ProgressCounter works", {
+    expect_output(testProgressCounter(), "100% \\(done\\)")
+})
+
+test_that("ProgressBar works", {
+    expect_output(testProgressBar(),"100% \\(done\\)")
+})