Merge pull request #133 from RcppCore/feature/rcpp-parallel-num-threads

use RCPP_PARALLEL_NUM_THREADS for default num threads

Author: kevinushey

R/options.R

@@ -22,28 +22,20 @@ setThreadOptions <- function(numThreads = "auto", stackSize = "auto") {
    else
       stackSize <- as.integer(stackSize)
 
-   # Call setThreadOptions if using tbb
-   if (!is.null(dllInfo) && isUsingTbb())
-      setTbbThreadOptions(numThreads, stackSize)
-   
+   # set RCPP_PARALLEL_NUM_THREADS
    if (numThreads == -1L)
       Sys.unsetenv("RCPP_PARALLEL_NUM_THREADS")
    else
       Sys.setenv(RCPP_PARALLEL_NUM_THREADS = numThreads)
-}
-
-setTbbThreadOptions <- function(numThreads, stackSize) {
-   .Call(
-      "setThreadOptions",
-      as.integer(numThreads),
-      as.integer(stackSize),
-      PACKAGE = "RcppParallel"
-   )
+   
+   # set RCPP_PARALLEL_STACK_SIZE
+   if (stackSize == 0L)
+      Sys.unsetenv("RCPP_PARALLEL_STACK_SIZE")
+   else
+      Sys.setenv(RCPP_PARALLEL_STACK_SIZE = stackSize)
+   
 }
 
 defaultNumThreads <- function() {
-   .Call(
-      "defaultNumThreads",
-      PACKAGE = "RcppParallel"
-   )
+   .Call("defaultNumThreads", PACKAGE = "RcppParallel")
 }

inst/NEWS

@@ -1,30 +1,38 @@
 RcppParallel 5.0.3 (UNRELEASED)
 ------------------------------------------------------------------------
+
+* setThreadOptions(...) can again be called multiple times per session.
+  The requested number of threads will be used for invocations to parallelFor()
+  and parallelReduce() that don't explicitly request a specific number of threads.
 * The parallelFor() and parallelReduce() functions gain the 'numThreads'
   argument, allowing one to limit the number of threads used for a
   particular computation.
 
 RcppParallel 5.0.2
 ------------------------------------------------------------------------
+
 * setThreadOptions(...) can now only be called once per session, to avoid
   segfaults when compiling RcppParallel / TBB with gcc 10.1. Subsequent
   calls to setThreadOptions(...) are ignored.
 
 RcppParallel 5.0.1
 ------------------------------------------------------------------------
+
 * Fixed compilation issue on OpenSUSE Tumbleweed with -flto=auto
 * Fixed compilation when CPPFLAGS = -I/usr/local/include and a version
   of libtbb is installed there
 
 RcppParallel 5.0.0
 ------------------------------------------------------------------------
+
 * RcppParallel backend can now be customized with RCPP_PARALLEL_BACKEND
   environment variable (supported values are 'tbb' and 'tinythread')
 * Fixed issue when compiling RcppParallel on macOS Catalina
 * Fixed issue when compiling RcppParallel with Rtools40
 
 RcppParallel 4.4.4
 ------------------------------------------------------------------------
+
 * Fixed an issue when compiling RcppParallel with clang-9 on Fedora
 
 RcppParallel 4.4.3

inst/include/RcppParallel.h

@@ -34,6 +34,9 @@ inline void parallelFor(std::size_t begin,
                         std::size_t grainSize = 1,
                         int numThreads = -1)
 {
+   grainSize = resolveValue("RCPP_PARALLEL_GRAIN_SIZE", grainSize, 1);
+   numThreads = resolveValue("RCPP_PARALLEL_NUM_THREADS", numThreads, -1);
+   
 #if RCPP_PARALLEL_USE_TBB
    if (internal::backend() == internal::BACKEND_TBB)
       tbbParallelFor(begin, end, worker, grainSize, numThreads);
@@ -51,6 +54,9 @@ inline void parallelReduce(std::size_t begin,
                            std::size_t grainSize = 1,
                            int numThreads = -1)
 {
+   grainSize = resolveValue("RCPP_PARALLEL_GRAIN_SIZE", grainSize, 1);
+   numThreads = resolveValue("RCPP_PARALLEL_NUM_THREADS", numThreads, -1);
+   
 #if RCPP_PARALLEL_USE_TBB
    if (internal::backend() == internal::BACKEND_TBB)
       tbbParallelReduce(begin, end, reducer, grainSize, numThreads);
@@ -61,6 +67,6 @@ inline void parallelReduce(std::size_t begin,
 #endif
 }
 
-} // namespace RcppParallel
+} // end namespace RcppParallel
 
 #endif // __RCPP_PARALLEL__

inst/include/RcppParallel/Common.h

@@ -1,10 +1,41 @@
 #ifndef __RCPP_PARALLEL_COMMON__
 #define __RCPP_PARALLEL_COMMON__
 
+#include <cerrno>
 #include <cstddef>
+#include <cstdlib>
 
 namespace RcppParallel {
 
+template <typename T, typename U>
+inline int resolveValue(const char* envvar,
+                        T requestedValue,
+                        U defaultValue)
+{
+   // 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   
+   return value;
+}
+
 // 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

inst/include/RcppParallel/TBB.h

@@ -3,7 +3,12 @@
 
 #include "Common.h"
 
+#ifndef TBB_PREVIEW_GLOBAL_CONTROL
+# define TBB_PREVIEW_GLOBAL_CONTROL 1
+#endif
+
 #include <tbb/tbb.h>
+#include <tbb/global_control.h>
 #include <tbb/scalable_allocator.h>
 
 namespace RcppParallel {
@@ -178,6 +183,43 @@ class TBBArenaParallelReduceExecutor
    std::size_t end_;
    std::size_t grainSize_;
 };
+
+class ThreadStackSizeControl
+{
+public:
+   
+   ThreadStackSizeControl()
+      : control_(nullptr)
+   {
+      int stackSize = resolveValue("RCPP_PARALLEL_STACK_SIZE", 0, 0);
+      if (stackSize > 0)
+      {
+         control_ = new tbb::global_control(
+            tbb::global_control::thread_stack_size,
+            stackSize
+         );
+      }
+   }
+   
+   ~ThreadStackSizeControl()
+   {
+      if (control_ != nullptr)
+      {
+         delete control_;
+         control_ = nullptr;
+      }
+   }
+
+private:
+   
+   // COPYING: not copyable
+   ThreadStackSizeControl(const ThreadStackSizeControl&);
+   ThreadStackSizeControl& operator=(const ThreadStackSizeControl&);
+   
+   // private members
+   tbb::global_control* control_;
+   
+};
 
 } // anonymous namespace
 
@@ -186,9 +228,11 @@ inline void tbbParallelFor(std::size_t begin,
                            std::size_t end, 
                            Worker& worker,
                            std::size_t grainSize = 1,
-                           int numThreads = -1)
+                           int numThreads = tbb::task_arena::automatic)
 {
-   tbb::task_arena arena(numThreads == -1 ? tbb::task_arena::automatic : numThreads);
+   ThreadStackSizeControl control;
+   
+   tbb::task_arena arena(numThreads);
    tbb::task_group group;
 
    TBBArenaParallelForExecutor executor(group, worker, begin, end, grainSize);
@@ -200,9 +244,11 @@ inline void tbbParallelReduce(std::size_t begin,
                               std::size_t end, 
                               Reducer& reducer,
                               std::size_t grainSize = 1,
-                              int numThreads = -1)
+                              int numThreads = tbb::task_arena::automatic)
 {
-   tbb::task_arena arena(numThreads == -1 ? tbb::task_arena::automatic : numThreads);
+   ThreadStackSizeControl control;
+   
+   tbb::task_arena arena(numThreads);
    tbb::task_group group;
 
    TBBArenaParallelReduceExecutor<Reducer> executor(group, reducer, begin, end, grainSize);

src/init.cpp

@@ -6,11 +6,9 @@
 
 /* .Call calls */
 extern "C" SEXP defaultNumThreads();
-extern "C" SEXP setThreadOptions(SEXP, SEXP);
 
 static const R_CallMethodDef CallEntries[] = {
    {"defaultNumThreads", (DL_FUNC) &defaultNumThreads, 0},
-   {"setThreadOptions",  (DL_FUNC) &setThreadOptions,  2},
    {NULL, NULL, 0}
 };
 

src/options.cpp

@@ -8,35 +8,6 @@
 #include <string>
 #include <exception>
 
-#include <tbb/task_scheduler_init.h>
-
-extern "C" SEXP setThreadOptions(SEXP numThreadsSEXP, SEXP stackSizeSEXP) {
-
-   static tbb::task_scheduler_init* s_pTaskScheduler = NULL;
-   if (s_pTaskScheduler != NULL)
-      return Rf_ScalarLogical(0);
-
-   int numThreads = Rf_asInteger(numThreadsSEXP);
-   int stackSize = Rf_asInteger(stackSizeSEXP);
-
-   try
-   {
-      s_pTaskScheduler = new tbb::task_scheduler_init(numThreads, stackSize);
-   }
-   catch(const std::exception& e)
-   {
-      const char* fmt = "Error loading TBB: %s\n";
-      Rf_error(fmt, e.what());
-   }
-   catch(...)
-   {
-      const char* fmt = "Error loading TBB: %s\n";
-      Rf_error(fmt, "(Unknown error)");
-   }
-
-   return Rf_ScalarLogical(1);
-}
-
 extern "C" SEXP defaultNumThreads() {
    SEXP threadsSEXP = Rf_allocVector(INTSXP, 1);
    INTEGER(threadsSEXP)[0] = tbb::task_scheduler_init::default_num_threads();
@@ -47,10 +18,6 @@ extern "C" SEXP defaultNumThreads() {
 
 #include <tthread/tinythread.h>
 
-extern "C" SEXP setThreadOptions(SEXP numThreadsSEXP, SEXP stackSizeSEXP) {
-   return R_NilValue;   
-}
-
 extern "C" SEXP defaultNumThreads() {
    SEXP threadsSEXP = Rf_allocVector(INTSXP, 1);
    INTEGER(threadsSEXP)[0] = tthread::thread::hardware_concurrency();