通过 Snow 在并行代码中使用 Rcpp 来创建集群

Question

我在 Rcpp 中编写了一个函数，并使用 inline 对其进行了编译。现在，我想在不同的内核上并行运行它，但我遇到了一个奇怪的错误。这是一个最小的示例，其中函数 funCPP1 可以自行编译并运行良好，但无法被 snow 的 clusterCall 函数调用。该函数作为单个进程运行良好，但在并行运行时会出现以下错误：

Error in checkForRemoteErrors(lapply(cl, recvResult)) : 
  2 nodes produced errors; first error: NULL value passed as symbol address

这是一些代码：

## Load and compile
library(inline)
library(Rcpp)
library(snow)
src1 <- '
     Rcpp::NumericMatrix xbem(xbe);
     int nrows = xbem.nrow();
     Rcpp::NumericVector gv(g);
     for (int i = 1; i < nrows; i++) {
      xbem(i,_) = xbem(i-1,_) * gv[0] + xbem(i,_);
     }
     return xbem;
'
funCPP1 <- cxxfunction(signature(xbe = "numeric", g="numeric"),body = src1, plugin="Rcpp")

## Single process
A <- matrix(rnorm(400), 20,20)
funCPP1(A, 0.5)

## Parallel
cl <- makeCluster(2, type = "SOCK") 
clusterExport(cl, 'funCPP1') 
clusterCall(cl, funCPP1, A, 0.5)

Answer 1

想一想——内联有什么作用？它为您创建一个 C/C++ 函数，然后将其编译并链接到可动态加载的共享库中。那个人坐在哪里？在 R 的临时目录中。

因此，您尝试了正确的做法，将调用该共享库的 R 前端传送到另一个进程（该进程有另一个临时目录！！），但这并没有在那里获取 dll / so 文件。

因此，建议是创建一个本地包，安装它并让两个 Snow 进程加载并调用它。

（和往常一样：在 rcpp-devel 列表中可能会有更好质量的答案，该列表的 Rcpp 贡献者比 SO 的阅读者更多。）

Answer 2

老问题，但我在查看顶部 Rcpp 标签时偶然发现了它，所以也许这个答案仍然有用。

我认为当您编写的代码经过完全调试并执行您想要的操作时，德克的答案是正确的，但是为示例中的一小段代码编写新包可能会很麻烦。相反，您可以做的是导出代码块，导出编译源代码并运行助手的“助手”函数。这将使 CXX 函数可用，然后使用另一个辅助函数来调用它。例如：

# Snow must still be installed, but this functionality is now in "parallel" which ships with base r.
library(parallel)

# Keep your source as an object
src1 <- '
     Rcpp::NumericMatrix xbem(xbe);
     int nrows = xbem.nrow();
     Rcpp::NumericVector gv(g);
     for (int i = 1; i < nrows; i++) {
      xbem(i,_) = xbem(i-1,_) * gv[0] + xbem(i,_);
     }
     return xbem;
'
# Save the signature
sig <- signature(xbe = "numeric", g="numeric")

# make a function that compiles the source, then assigns the compiled function 
# to the global environment
c.inline <- function(name, sig, src){
    library(Rcpp)
    funCXX <- inline::cxxfunction(sig = sig, body = src, plugin="Rcpp")
    assign(name, funCXX, envir=.GlobalEnv)
}
# and the function which retrieves and calls this newly-compiled function 
c.namecall <- function(name,...){
    funCXX <- get(name)
    funCXX(...)
}

# Keep your example matrix
A <- matrix(rnorm(400), 20,20)

# What are we calling the compiled funciton?
fxname <- "TestCXX"

## Parallel
cl <- makeCluster(2, type = "PSOCK") 

# Export all the pieces
clusterExport(cl, c("src1","c.inline","A","fxname")) 

# Call the compiler function
clusterCall(cl, c.inline, name=fxname, sig=sig, src=src1)

# Notice how the function now named "TestCXX" is available in the environment
# of every node?
clusterCall(cl, ls, envir=.GlobalEnv)

# Call the function through our wrapper
clusterCall(cl, c.namecall, name=fxname, A, 0.5)
# Works with my testing

我写了一个包 ctools （无耻的自我推销）包含了用于集群计算的并行包和 Rhpc 包中的许多功能，均带有 PSOCK 和 MPI。我已经有一个名为“c.sourceCpp”的函数，它以与上面大致相同的方式在每个节点上调用“Rcpp::sourceCpp”。现在我看到了它的用处，我将添加一个“c.inlineCpp”来执行上述操作。

编辑：

根据 Coatless 的评论，Rcpp::cppFunction() 实际上不需要 c.inline 帮助器，尽管 c.仍然需要 namecall。

src2 <- '
 NumericMatrix TestCpp(NumericMatrix xbe, int g){
        NumericMatrix xbem(xbe);
        int nrows = xbem.nrow();
        NumericVector gv(g);
        for (int i = 1; i < nrows; i++) {
            xbem(i,_) = xbem(i-1,_) * gv[0] + xbem(i,_);
        }
        return xbem;
 }
'

clusterCall(cl, Rcpp::cppFunction, code=src2, env=.GlobalEnv)

# Call the function through our wrapper
clusterCall(cl, c.namecall, name="TestCpp", A, 0.5)

Answer 3

我通过在每个集群节点上获取一个带有所需 C 内联函数的 R 文件来解决这个问题：

clusterEvalQ(cl, 
    {
     library(inline)
     invisible(source("your_C_func.R"))
    })

并且您的文件 your_C_func.R 应包含 C 函数定义：

c_func <- cfunction(...)