r - R parLapply不平行

我目前正在开发一个R包，该包将通过“并行”包使用并行计算来解决某些任务。

当使用我的包函数内部定义的集群时，我会遇到一些非常尴尬的行为，其中parLapply函数将作业分配给工作人员，然后等待其完成以将工作分配给下一个工作人员。
或者至少通过观察日志文件“ cluster.log”和unix shell中正在运行的进程的列表，这似乎正在发生。

下面是我的包中声明的原始函数的模型版本：

.parSolver <- function( varMatrix, var1 ) {

    no_cores <- detectCores()

    #Rows in varMatrix
    rows <- 1:nrow(varMatrix[,])

    # Split rows in n parts
    n <- no_cores
    parts <- split(rows, cut(rows, n))

    # Initiate cluster
    cl <- makePSOCKcluster(no_cores, methods = FALSE, outfile = "/home/cluster.log")
    clusterEvalQ(cl, library(raster))
    clusterExport(cl, "varMatrix", envir=environment())
    clusterExport(cl, "var1", envir=environment())


    rParts <- parLapply(cl = cl, X = 1:n, fun = function(x){
        part <- rasterize(varMatrix[parts[[x]],], raster(var1), .....)
        print(x)
        return(part)
        })

    do.call(merge, rParts)
}

笔记：


我使用makePSOCKcluster是因为我希望代码可以在Windows和Unix系统上运行，尽管此特定问题仅在Unix系统中表现出来。
栅格化和栅格化功能在library（raster）中定义，并导出到集群。


对我来说，奇怪的是，如果我在全局环境中执行parSolver函数的完全相同的代码，那么每件事都会顺利进行，所有工作人员同时执行一项工作，并且任务立即完成。
但是，如果我做类似的事情：

library(myPackage)

varMatrix <- (...)
var1 <- (...)
result <- parSolver(varMatrix, var1)

出现所描述的问题。

但是，这似乎是一个负载平衡问题，无法说明为什么它在一种情况下可以正常工作而在另一种情况下不能正常工作。

我在这里想念什么吗？
提前致谢。

最佳答案

我不认为parLapply按顺序运行。更有可能的是，它运行效率低下，使其似乎顺序运行。

我有一些改进建议：


不要在parSolver中定义worker函数
不要将所有varMatrix导出到每个工人
在parSolver之外创建集群


第一点很重要，因为如您的示例所示，parSolver中定义的所有变量将与匿名worker函数一起序列化，并由parLapply发送给worker。通过在任何函数之外定义worker函数，序列化不会捕获任何不需要的变量。

第二点避免了不必要的套接字I / O，并使用了更少的内存，从而使代码更具可伸缩性。

这是一个伪造的但自成体系的示例，类似于您的示例，它展示了我的建议：

# Define worker function outside of any function to avoid
# serialization problems (such as unexpected variable capture)
workerfn <- function(mat, var1) {
    library(raster)
    mat * var1
}

parSolver <- function(cl, varMatrix, var1) {
    parts <- splitIndices(nrow(varMatrix), length(cl))
    varMatrixParts <- lapply(parts, function(i) varMatrix[i,,drop=FALSE])
    rParts <- clusterApply(cl, varMatrixParts, workerfn, var1)
    do.call(rbind, rParts)
}

library(parallel)
cl <- makePSOCKcluster(3)
r <- parSolver(cl, matrix(1:20, 10, 2), 2)
print(r)

请注意，这利用了clusterApply函数来迭代varMatrix的行块列表，因此不需要将整个矩阵发送给每个人。它还避免了对clusterEvalQ和clusterExport的调用，从而简化了代码，并使代码效率更高。