move parallel reducer out of header

Author: kevinushey

RcppParallel.Rproj

@@ -13,7 +13,11 @@ Encoding: UTF-8
 RnwWeave: Sweave
 LaTeX: pdfLaTeX
 
+AutoAppendNewline: Yes
+StripTrailingWhitespace: Yes
+
 BuildType: Package
-PackageInstallArgs: --with-keep.source --clean
+PackageCleanBeforeInstall: No
+PackageInstallArgs: --with-keep.source
 PackageCheckArgs: --as-cran
 PackageRoxygenize: rd,collate,namespace

inst/include/RcppParallel/Common.h

@@ -15,57 +15,57 @@ inline int resolveValue(const char* envvar,
    // if the requested value is non-zero and not the default, we can use it
    if (requestedValue != defaultValue && requestedValue > 0)
       return requestedValue;
-   
+
    // otherwise, try reading the default from associated envvar
    // if the environment variable is unset, use the default
    const char* var = getenv(envvar);
    if (var == NULL)
       return defaultValue;
-   
+
    // try to convert the string to a number
    // if an error occurs during conversion, just use default
    errno = 0;
    char* end;
    long value = strtol(var, &end, 10);
-   
+
    // check for conversion failure
    if (end == var || *end != '\0' || errno == ERANGE)
       return defaultValue;
-   
-   // okay, return the parsed environment variable value   
+
+   // okay, return the parsed environment variable value
    return value;
 }
 
+// Tag type used for disambiguating splitting constructors
+struct Split {};
+
 // Work executed within a background thread. We implement dynamic
 // dispatch using vtables so we can have a stable type to cast
 // to from the void* passed to the worker thread (required because
 // the tinythreads interface allows to pass only a void* to the
 // thread main rather than a generic type / template)
-struct Worker 
-{  
+struct Worker
+{
    // construct and destruct (delete virtually)
    Worker() {}
    virtual ~Worker() {}
-   
+
    // dispatch work over a range of values
-   virtual void operator()(std::size_t begin, std::size_t end) = 0;   
-   
-   // disable copying and assignment
+   virtual void operator()(std::size_t begin, std::size_t end) = 0;
+
 private:
+   // disable copying and assignment
    Worker(const Worker&);
    void operator=(const Worker&);
 };
 
-// Tag type used for disambiguating splitting constructors
-struct Split {};
-
 // Used for controlling the stack size for threads / tasks within a scope.
 class ThreadStackSizeControl
 {
 public:
    ThreadStackSizeControl();
    ~ThreadStackSizeControl();
-   
+
 private:
    // COPYING: not copyable
    ThreadStackSizeControl(const ThreadStackSizeControl&);

inst/include/RcppParallel/TBB.h

@@ -13,110 +13,91 @@
 
 namespace RcppParallel {
 
-struct TBBWorker
-{
-   explicit TBBWorker(Worker& worker) : worker_(worker) {}
-   
-   void operator()(const tbb::blocked_range<size_t>& r) const {
-      worker_(r.begin(), r.end());
-   }
+// This class is primarily used to implement type erasure. The goals here were:
+//
+// 1. Hide the tbb symbols / implementation details from client R packages.
+//    That is, they should get the tools they need only via RcppParallel.
+//
+// 2. Do this in a way that preserves binary compatibility with pre-existing
+//    classes that make use of parallelReduce().
+//
+// 3. Ensure that those packages, when re-compiled without source changes,
+//    can still function as expected.
+//
+// The downside here is that all the indirection through std::function<>
+// and the requirement for RTTI is probably expensive, but I couldn't find
+// a better way forward that could also preserve binary compatibility with
+// existing pre-built pacakges.
+//
+// Hopefully, in a future release, we can do away with this wrapper, once
+// packages have been rebuilt and no longer implicitly depend on TBB internals.
+struct ReducerWrapper {
 
-private:
-   Worker& worker_;
-};
-
-template <typename Reducer>
-struct TBBReducer 
-{  
-   explicit TBBReducer(Reducer& reducer) 
-      : pSplitReducer_(NULL), reducer_(reducer)
-   {
-   }
-   
-   TBBReducer(TBBReducer& tbbReducer, tbb::split)
-      : pSplitReducer_(new Reducer(tbbReducer.reducer_, RcppParallel::Split())),
-        reducer_(*pSplitReducer_)
+   template <typename T>
+   ReducerWrapper(T* reducer)
    {
-   }
-   
-   virtual ~TBBReducer() { delete pSplitReducer_; }
+      self_ = reinterpret_cast<void*>(reducer);
+      owned_ = false;
 
-   void operator()(const tbb::blocked_range<size_t>& r) {
-      reducer_(r.begin(), r.end());
-   }
-   
-   void join(const TBBReducer& tbbReducer) { 
-      reducer_.join(tbbReducer.reducer_); 
+      work_ = [&](void* self, std::size_t begin, std::size_t end)
+      {
+         (*reinterpret_cast<T*>(self))(begin, end);
+      };
+
+      split_ = [&](void* object, Split split)
+      {
+         return new T(*reinterpret_cast<T*>(object), split);
+      };
+
+      join_ = [&](void* self, void* other)
+      {
+         (*reinterpret_cast<T*>(self)).join(*reinterpret_cast<T*>(other));
+      };
+
+      deleter_ = [&](void* object)
+      {
+         delete (T*) object;
+      };
    }
-   
-private:
-   Reducer* pSplitReducer_;
-   Reducer& reducer_;
-};
 
-template <typename Reducer>
-class TBBParallelReduceExecutor
-{
-public:
-   
-   TBBParallelReduceExecutor(Reducer& reducer,
-                             std::size_t begin,
-                             std::size_t end,
-                             std::size_t grainSize)
-      : reducer_(reducer),
-        begin_(begin),
-        end_(end),
-        grainSize_(grainSize)
+   ~ReducerWrapper()
    {
+      if (owned_)
+      {
+         deleter_(self_);
+         self_ = nullptr;
+      }
    }
-   
-   void operator()() const
+
+   void operator()(std::size_t begin, std::size_t end) const
    {
-      TBBReducer<Reducer> tbbReducer(reducer_);
-      tbb::parallel_reduce(
-         tbb::blocked_range<std::size_t>(begin_, end_, grainSize_),
-         tbbReducer
-      );
+      work_(self_, begin, end);
    }
-   
-private:
-   Reducer& reducer_;
-   std::size_t begin_;
-   std::size_t end_;
-   std::size_t grainSize_;
-};
 
-template <typename Reducer>
-class TBBArenaParallelReduceExecutor
-{
-public:
-   
-   TBBArenaParallelReduceExecutor(tbb::task_group& group,
-                                  Reducer& reducer,
-                                  std::size_t begin,
-                                  std::size_t end,
-                                  std::size_t grainSize)
-      : group_(group),
-        reducer_(reducer),
-        begin_(begin),
-        end_(end),
-        grainSize_(grainSize)
+   ReducerWrapper(const ReducerWrapper& rhs, Split split)
    {
+      self_  = rhs.split_(rhs.self_, split);
+      owned_ = true;
+
+      work_    = rhs.work_;
+      split_   = rhs.split_;
+      join_    = rhs.join_;
+      deleter_ = rhs.deleter_;
    }
-   
-   void operator()() const
+
+   void join(const ReducerWrapper& rhs) const
    {
-      TBBParallelReduceExecutor<Reducer> executor(reducer_, begin_, end_, grainSize_);
-      group_.run_and_wait(executor);
+      join_(self_, rhs.self_);
    }
-   
+
 private:
-   
-   tbb::task_group& group_;
-   Reducer& reducer_;
-   std::size_t begin_;
-   std::size_t end_;
-   std::size_t grainSize_;
+   void* self_ = nullptr;
+   bool owned_ = false;
+
+   std::function<void (void*, std::size_t, std::size_t)> work_;
+   std::function<void*(void*, Split)> split_;
+   std::function<void (void*, void*)> join_;
+   std::function<void(void*)> deleter_;
 };
 
 void tbbParallelFor(std::size_t begin,
@@ -125,20 +106,21 @@ void tbbParallelFor(std::size_t begin,
                     std::size_t grainSize = 1,
                     int numThreads = -1);
 
+void tbbParallelReduceImpl(std::size_t begin,
+                           std::size_t end,
+                           ReducerWrapper& wrapper,
+                           std::size_t grainSize = 1,
+                           int numThreads = -1);
+
 template <typename Reducer>
-inline void tbbParallelReduce(std::size_t begin,
-                              std::size_t end, 
-                              Reducer& reducer,
-                              std::size_t grainSize = 1,
-                              int numThreads = -1)
+void tbbParallelReduce(std::size_t begin,
+                       std::size_t end,
+                       Reducer& reducer,
+                       std::size_t grainSize = 1,
+                       int numThreads = -1)
 {
-   ThreadStackSizeControl control;
-   
-   tbb::task_group group;
-   TBBArenaParallelReduceExecutor<Reducer> executor(group, reducer, begin, end, grainSize);
-   
-   tbb::task_arena arena(numThreads);
-   arena.execute(executor);
+   ReducerWrapper wrapper(&reducer);
+   tbbParallelReduceImpl(begin, end, wrapper, grainSize, numThreads);
 }
 
 } // namespace RcppParallel

inst/skeleton/vector-sum.cpp

@@ -19,37 +19,37 @@ using namespace Rcpp;
 using namespace RcppParallel;
 
 struct Sum : public Worker
-{   
+{
    // source vector
    const RVector<double> input;
-   
+
    // accumulated value
    double value;
-   
+
    // constructors
    Sum(const NumericVector input) : input(input), value(0) {}
    Sum(const Sum& sum, Split) : input(sum.input), value(0) {}
-   
+
    // accumulate just the element of the range I've been asked to
    void operator()(std::size_t begin, std::size_t end) {
       value += std::accumulate(input.begin() + begin, input.begin() + end, 0.0);
    }
-   
+
    // join my value with that of another Sum
-   void join(const Sum& rhs) { 
-      value += rhs.value; 
+   void join(const Sum& rhs) {
+      value += rhs.value;
    }
 };
 
 // [[Rcpp::export]]
 double parallelVectorSum(NumericVector x) {
-   
-   // declare the SumBody instance 
+
+   // declare the SumBody instance
    Sum sum(x);
-   
+
    // call parallel_reduce to start the work
    parallelReduce(0, x.length(), sum);
-   
+
    // return the computed sum
    return sum.value;
 }