handling nested TBB in thread-safe fashion

nCompiler/R/Rcpp_nCompiler_plugin.R

@@ -15,13 +15,11 @@ inlineCxxPlugin <- function(...) {
   uses_nC_inter <- !isFALSE(inlineCxxPlugin_env$uses_nC_inter)
   uses_nList <- !isFALSE(inlineCxxPlugin_env$uses_nList)
   uses_cereal <- !isFALSE(inlineCxxPlugin_env$uses_cereal)
-  uses_TBB <- FALSE # !isFALSE(inlineCxxPlugin_env$uses_TBB) # including here causes error due to #defining FALSE
   include.before <- character()
   if(uses_eigen) include.before <- paste0(include.before, "#define NCOMPILER_USES_EIGEN\n")
   if(uses_nC_inter) include.before <- paste0(include.before, "#define NCOMPILER_USES_NCLASS_INTERFACE\n")
   if(uses_nList) include.before <- paste0(include.before, "#define NCOMPILER_USES_NLIST\n")
   if(uses_cereal) include.before <- paste0(include.before, "#define NCOMPILER_USES_CEREAL\n")
-  if(uses_TBB) include.before <- paste0(include.before, "#define NCOMPILER_USES_TBB\n")
   include.before <- paste0(include.before, "#include <nCompiler/nCompiler_omnibus.h>")
   ans <- Rcpp::Rcpp.plugin.maker(include.before=include.before)()
   ans
@@ -37,9 +35,9 @@ nCompiler_pluginEnv <- new.env()
 make_nCompiler_plugin <- function(nCompiler_pluginEnv) {
   RcppDefaultPlugin <- Rcpp:::Rcpp.plugin.maker()
   force(nCompiler_pluginEnv)
-  ans <- function(...) {
+    ans <- function(...) {
     result <- RcppDefaultPlugin(...)
-    result$env$PKG_CPPFLAGS <- c(result$env$PKG_CPPFLAGS,
+    result$env$PKG_CPPFLAGS <- paste(result$env$PKG_CPPFLAGS,
                                  if(length(nCompiler_pluginEnv$includePaths) > 0)
                                    paste0(
                                      "-I",
@@ -50,6 +48,8 @@ make_nCompiler_plugin <- function(nCompiler_pluginEnv) {
     result$env$PKG_LIBS <- get_nCompLocal_PKG_LIBS_entry()
     ## Makevars doesn't work
     ## result$Makevars <- "CXX_STD=CXX11" does not seem to work
+    if(isTRUE(nCompiler_pluginEnv$uses_TBB))
+      result$env <- setEnvTBB(result$env)
     result
   }
   ans
@@ -74,13 +74,15 @@ make_nCompiler_Eigen_plugin <- function(nCompiler_pluginEnv) {
                                        "")
     # result$env$PKG_CXXFLAGS <- "-std=c++11"
     result$env$PKG_LIBS <- get_nCompLocal_PKG_LIBS_entry()
+    if(!isFALSE(inlineCxxPlugin_env$uses_TBB))
+      result$env <- setEnvTBB(result$env)
     if(isTRUE(get_nOption('compilerOptions')$throwEigenErrors)) {
       # replace include directives to enable Eigen errors
       #preamble = system.file(file.path('include', 'nCompiler',
       #                                 'nCompiler_Eigen_EnableErrors.h'),
       #                       package = 'nCompiler')
       #result$includes = readChar(preamble, file.info(preamble)$size)
-      result$includes = "#define NCOMPILER_HANDLE_EIGEN_ERRORS"
+      result$includes = c("#define NCOMPILER_HANDLE_EIGEN_ERRORS")
     }
     if(isTRUE(get_nOption('compilerOptions')$cppStacktrace)) {
       # add include directives to add stack basic traces
@@ -95,3 +97,13 @@ make_nCompiler_Eigen_plugin <- function(nCompiler_pluginEnv) {
 }
 
 nCompiler_Eigen_plugin <- make_nCompiler_Eigen_plugin(nCompiler_pluginEnv)
+
+setEnvTBB <- function(env) {
+  if(.Platform$OS.type == "windows") {
+    env$PKG_CPPFLAGS <- paste(env$PKG_CPPFLAGS, '-DRCPP_PARALLEL_USE_TBB=1')
+    env$PKG_LIBS <- paste(env$PKG_LIBS,
+                                 '$(shell "${R_HOME}/bin${R_ARCH_BIN}/Rscript.exe"-e "RcppParallel::RcppParallelLibs()")')
+  } else env$PKG_LIBS <- paste(env$PKG_LIBS, 
+                               '$(shell ${R_HOME}/bin/Rscript -e "RcppParallel::RcppParallelLibs()")')
+  return(env)
+}

nCompiler/R/Rexecution.R

@@ -14,6 +14,18 @@ parallel_for <- function(index, range, body, ...) {
 
 #' @export
 parallel_reduce <- function(f, x, init, ...) {
+  if(is.character(f)) {  # Not clear how to convert to char ...
+    operatorDef <- operatorDefEnv[[f]]
+    if(!is.null(operatorDef) && is.null(operatorDef$reduction))
+        stop("`", f, "` is not a valid reduction function/operator")
+  }
+  if(missing(init)) {
+    if(!is.character(f) || is.null(operatorDef) || is.null(operatorDef$reduction))
+      stop("`init` argument is missing and no default value provided for reduction function/operator")
+    init <- operatorDef$reduction
+  }
+  if(identical(f, "pairmin")) f <- "pmin"
+  if(identical(f, "pairmax")) f <- "pmax"
   Reduce(f, x, init)
 }
 

nCompiler/R/compile_aaa_operatorLists.R

@@ -12,6 +12,8 @@ returnTypeCodes <- list(
   promoteToDoubleOrAD = 6L,
   promoteNoLogical = 7L)
 
+liftedBlockOperatorsArg <- c("parallel_for" = 3, "parallel_reduce" = 1) # These are used for flagging when methods in lifted code block will need to have their object reference them explicitly via `obj__.<methodname>`. The values are the argID where the method needs to occur.
+
 returnTypeString2Code <- function(returnTypeString) {
   if(is.character(returnTypeString))
     do.call('switch', c(list("double"), returnTypeCodes))
@@ -140,6 +142,16 @@ assignOperatorDef(
   )
 )
 
+assignOperatorDef(
+  'nClass_method_in_lifted', # This is used for local method calls in the body of lifted code blocks (currently `parallel_{for,reduce}`).
+  list(
+    labelAbstractTypes = list(
+      handler = 'nFunction_or_method_call'),
+    cppOutput = list(
+      handler = 'nClass_method_in_lifted')
+  )
+)
+
 assignOperatorDef(
   'custom_default',
   list(
@@ -228,6 +240,10 @@ updateOperatorDef(
   c('nMatrix', 'nArray'),
   'labelAbstractTypes', 'returnTypeCode', returnTypeCodes$promote
 )
+updateOperatorDef(
+  'nMatrix',
+  'matchDef', val = function(value = 0, nrow = NA, ncol = NA, init = TRUE, fillZeros = TRUE, recycle = TRUE, type = 'double') {}
+)
 
 assignOperatorDef(
   'type_is',
@@ -273,20 +289,26 @@ assignOperatorDef(
 assignOperatorDef(
   c('parallel_for'),
   list(
+    matchDef = function(index, range, body, copyVars, shareVars, nThreads=0) {},
     labelAbstractTypes = list(
       handler = 'ParallelFor'),
     finalTransformations = list(
-      handler = "ParallelFor") ## Creates GeneralFor in the parallel_loop_body class
+      handler = "ParallelFor"), ## Creates GeneralFor in the parallel_loop_body class
+    cppOutput = list(
+      handler = 'ParallelExpr')  
   )
 )
 
 assignOperatorDef(
   c('parallel_reduce'),
-  list(
+  list(  
+    matchDef = function(operator, object, init, nThreads=0) {},
     labelAbstractTypes = list(
       handler = 'ParallelReduce'),
     finalTransformations = list(
-      handler = 'ParallelReduce')
+      handler = 'ParallelReduce'),
+    cppOutput = list(
+      handler = 'ParallelExpr')      
   )
 )
 
@@ -473,9 +495,11 @@ assignOperatorDef(
           )
         ),
         cppOutput = list(
-          handler = 'BinaryOrUnary')
+          handler = 'BinaryOrUnary'),
+        reduction = 0
     )
 )
+updateOperatorDef('-', 'reduction', val = NULL)
 
 assignOperatorDef(
   c('inprod'),
@@ -522,11 +546,13 @@ assignOperatorDef(
     labelAbstractTypes = list(
         handler = 'BinaryCwise',
         returnTypeCode = returnTypeCodes$promoteNoLogical),
-    cppOutput = list()
+    cppOutput = list(),
+    reduction = Inf
   )
 )
 updateOperatorDef('pairmax', 'cppOutput', 'cppString', 'std::max')
 updateOperatorDef('pairmin', 'cppOutput', 'cppString', 'std::min')
+updateOperatorDef('pairmax', 'reduction', val = -Inf)
 
 assignOperatorDef(
   c('pmin', 'pmax'),
@@ -902,7 +928,8 @@ assignOperatorDef(
       )
     ),
     cppOutput = list(
-      handler = 'MidOperator')
+      handler = 'MidOperator'),
+    reduction = 1
   )
 )
 

nCompiler/R/compile_eigenization.R

@@ -5,7 +5,8 @@
 eigenizeUseArgs <- c(
     list(
         setWhich = c(FALSE, TRUE),
-        setRepVectorTimes = c(FALSE, TRUE, TRUE)
+        setRepVectorTimes = c(FALSE, TRUE, TRUE),
+        parallel_reduce = c(FALSE, TRUE, TRUE)
         ))
 
 eigenizeEnv <- new.env()

nCompiler/R/compile_exprClass.R

@@ -218,6 +218,7 @@ wrapExprClassOperator <- function(code, funName, isName = FALSE, isCall = TRUE,
   newExpr
 }
 
+
 insertIndexingBracket <- function(code, argID, index) {
   insertExprClassLayer(code, argID, 'index[')
   setArg(code$args[[argID]], 2, index)