refactor (#52)

* try locks on pop

* new tpool

* run less tests

* fix compiler warnings

* test runs :/

* compiler warning

* just pushing

* try own check

* tryfix

* meh

* only thread

* push10

* couts

* stupid

* more

* enable push

* dumdum

* now

* more

* more

* nowait

* no push lock

* waitk

* msgs

* stopped check

* cleanup

* other tests

* thread localized steals

* Rcout on wait

* no Rcout

* progress

* no pushreturn

* mtx name

* no pushreturn

* wait on destruct

* more progress

* more msgs

* time stamps

* fix

* no pushreturn

* away

* wtf

* run nothing

* detach on destruct

* fix

* destruct print

* more

* no print on wait

* time on destruct

* delete print

* more

* actually use multiple queues

* destruction order

* more order

* time print

* finish print

* fix

* no globals

* print too

* clear on stop

* fix

* parfor

* don't wait on join

* after cv

* before cv

* ptr to taskmanager

* don't stop on mngr dtor

* don't clear either

* reorder

* manually destroy cv

* wrap in pointer

* learned: windows hangs when destructing static cv

* no global instance in parallel for

* 5 test runs by default

* with print

* fix nBatches docs

* get future before push

* get after join

* copy function on push

* check future result

* leftover

* prints

* run 5x

* get future?

* get first

* no print

* run 10

* other OSs

* fix2

* bench pdfs on gitignore

* clean up prints

* other sync

* some cleanup

* finish line -> todo list

* more clean up

* more clean up

* even more

* update NEWS

* more clean up

* even more

* clean up unit tests

* pull tpool

* restrict to 10 runs

Author: tnagler

.Rbuildignore

@@ -13,4 +13,5 @@ revdep
 ^\.github$
 ^CRAN-RELEASE$
 .vscode/
-new-benchmarks.R
+new-benchmarks.R
+bench*

.github/workflows/R-CMD-check.yaml

@@ -80,6 +80,8 @@ jobs:
           _R_CHECK_CRAN_INCOMING_: false
         run: rcmdcheck::rcmdcheck(args = c("--no-manual", "--as-cran"), error_on = "warning", check_dir = "check")
         shell: Rscript {0}
+        # run: cd .. && R CMD build RcppThread && R CMD check RcppThread_1.1.0.tar.gz 
+        # shell: bash
 
       - name: Show testthat output
         if: always()

.gitignore

@@ -5,4 +5,5 @@ inst/doc
 revdep
 __pycache__
 .vscode/
-new-benchmarks.R
+new-benchmarks.R
+bench*

NEWS.md

@@ -5,7 +5,7 @@
 * Add classes `ProgressCounter` and `ProgressBar` for tracking progress in long-
   running loops (#49).
 
-* Increased speed for short running tasks due to lock-free queue (#50, #51).
+* Increased speed due to lock-free queue (#51).
 
 
 # RcppThread 1.0.0

inst/include/RcppThread/Progress.hpp

@@ -181,7 +181,7 @@ class ProgressBar : public ProgressPrinter {
     std::string makeBar(size_t pct, size_t numBars = 40) {
         std::ostringstream msg;
         msg << "[";
-        int i = 0;
+        size_t i = 0;
         for (; i < pct / 100.0 * numBars; i++)
             msg << "=";
         for (; i < numBars; i++)

inst/include/RcppThread/ThreadPool.hpp

@@ -22,16 +22,6 @@
 
 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
 {
@@ -46,13 +36,6 @@ 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);
 
@@ -76,21 +59,13 @@ class ThreadPool
     void clear();
 
   private:
-    void startWorker();
     void joinWorkers();
+    void execute(std::function<void()>& task);
 
-    template<class Task>
-    void executeSafely(Task& task);
-
-    bool allJobsDone();
-    void waitForEvents();
-    void rethrowExceptions();
-
-    std::vector<std::thread> workers_;
-    size_t nWorkers_;
-    // variables for synchronization between workers
+    // variables for synchronization between workers (destructed last)
     tpool::detail::TaskManager taskManager_;
-    tpool::FinishLine finishLine_{ 0 };
+    tpool::TodoList todoList_;
+    std::vector<std::thread> workers_;
 };
 
 //! constructs a thread pool with as many workers as there are cores.
@@ -102,10 +77,21 @@ inline ThreadPool::ThreadPool()
 //! @param nWorkers number of worker threads to create; if `nWorkers = 0`, all
 //!    work pushed to the pool will be done in the main thread.
 inline ThreadPool::ThreadPool(size_t nWorkers)
-  : nWorkers_(nWorkers)
+  : taskManager_{ nWorkers }
 {
-    for (size_t w = 0; w < nWorkers_; w++)
-        this->startWorker();
+    for (size_t id = 0; id < nWorkers; id++) {
+        workers_.emplace_back([this, id] {
+            std::function<void()> task;
+            while (!taskManager_.stopped()) {
+                taskManager_.wait_for_jobs(id);
+                do {
+                    // use inner while to save a few cash misses calling done()
+                    while (taskManager_.try_pop(task, id))
+                        execute(task);
+                } while (!todoList_.empty());
+            }
+        });
+    }
 }
 
 //! destructor joins all threads if possible.
@@ -129,9 +115,10 @@ template<class F, class... Args>
 void
 ThreadPool::push(F&& f, Args&&... args)
 {
-    if (nWorkers_ == 0) {
+    if (workers_.size() == 0) {
         f(args...); // if there are no workers, do the job in the main thread
     } else {
+        todoList_.add();
         taskManager_.push(
           std::bind(std::forward<F>(f), std::forward<Args>(args)...));
     }
@@ -148,11 +135,11 @@ auto
 ThreadPool::pushReturn(F&& f, Args&&... args)
   -> std::future<decltype(f(args...))>
 {
-    using task = std::packaged_task<decltype(f(args...))()>;
-    auto pack = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
-    auto taskPtr = std::make_shared<task>(std::move(pack));
-    taskManager_.push([taskPtr] { (*taskPtr)(); });
-    return taskPtr->get_future();
+    auto task = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
+    using pack_t = std::packaged_task<decltype(f(args...))()>;
+    auto ptr = std::make_shared<pack_t>(std::move(task));
+    this->push([ptr] { (*ptr)(); });
+    return ptr->get_future();
 }
 
 //! maps a function on a list of items, possibly running tasks in parallel.
@@ -202,7 +189,7 @@ ThreadPool::parallelFor(ptrdiff_t begin, size_t size, F&& f, size_t nBatches)
         for (ptrdiff_t i = b.begin; i < b.end; i++)
             f(i);
     };
-    auto batches = createBatches(begin, size, nWorkers_, nBatches);
+    auto batches = createBatches(begin, size, workers_.size(), nBatches);
     auto pushJob = [=] {
         for (const auto& batch : batches)
             this->push(doBatch, batch);
@@ -244,8 +231,12 @@ ThreadPool::parallelForEach(I& items, F&& f, size_t nBatches)
 inline void
 ThreadPool::wait()
 {
-    while (!taskManager_.empty())
-        util::waitAndSync(finishLine_);
+    while (!todoList_.empty()) {
+        todoList_.wait(50);
+        Rcout << "";
+        checkUserInterrupt();
+    }
+    Rcout << "";
 }
 
 //! waits for all jobs to finish and joins all threads.
@@ -264,31 +255,14 @@ ThreadPool::clear()
     taskManager_.clear();
 }
 
-//! spawns a worker thread waiting for jobs to arrive.
-inline void
-ThreadPool::startWorker()
-{
-    workers_.emplace_back([this] {
-        std::function<void()> task;
-        while (!taskManager_.stopped()) {
-            taskManager_.wait_for_jobs();
-
-            finishLine_.start();
-            while (taskManager_.try_pop(task))
-                executeSafely(task);
-            finishLine_.cross();
-        }
-    });
-}
-
-template<class Task>
 inline void
-ThreadPool::executeSafely(Task& task)
+ThreadPool::execute(std::function<void()>& task)
 {
     try {
         task();
+        todoList_.cross();
     } catch (...) {
-        finishLine_.abort(std::current_exception());
+        todoList_.clear(std::current_exception());
     }
 }
 

inst/include/RcppThread/parallelFor.hpp

@@ -15,7 +15,9 @@ namespace RcppThread {
 //! @param begin first index of the loop.
 //! @param end the loop runs in the range `[begin, end)`.
 //! @param f a function (the 'loop body').
-//! @param nThreads deprecated; loop is run on global thread pool.
+//! @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:
@@ -31,8 +33,9 @@ namespace RcppThread {
 //!     x[i] = i;
 //! });
 //! ```
-//! The function dispatches to a global thread pool, so it can safely be nested
-//! or called multiple times with almost no overhead.
+//! 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::parallelFor()`.
 //!
 //! **Caution**: if the iterations are not independent from another,
 //! the tasks need to be synchronized manually (e.g., using mutexes).
@@ -49,24 +52,18 @@ parallelFor(int begin,
     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);
-    util::waitAndSync(finishLine);
+    ThreadPool pool(nThreads);
+    pool.parallelFor(begin, end, std::forward<F>(f), nBatches);
+    pool.join();
 }
 
 //! 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 deprecated; loop is run on global thread pool.
+//! @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:
@@ -82,8 +79,9 @@ parallelFor(int begin,
 //!     xx *= 2;
 //! });
 //! ```
-//! The function dispatches to a global thread pool, so it can safely be nested
-//! or called multiple times with almost no overhead.
+//! 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::parallelForEach()`.
 //!
 //! **Caution**: if the iterations are not independent from another,
 //! the tasks need to be synchronized manually (e.g., using mutexes).
@@ -95,9 +93,9 @@ parallelForEach(I& items,
                 size_t nBatches = 0)
 {
     // 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)); });
+    ThreadPool pool(nThreads);
+    pool.parallelForEach(items, std::forward<F>(f), nBatches);
+    pool.join();
 }
+
 }