在源脚本中使用 parLapply 会导致内存泄漏

Question

这可能是哲学家的问题...（或@Steve Weston 或@Martin Morgan）

我在使用 parLapply 时遇到了一些内存泄漏问题，在研究了足够多的线程之后，我认为这个问题是有根据的。我花了一些时间来尝试解决这个问题，虽然我对观察到的行为发生的原因有所了解，但我不知道如何解决它。

将以下内容视为源脚本，另存为：parallel_question.R

rf.parallel<-function(n=10){
  library(parallel)
  library(randomForest)

  rf.form<- as.formula(paste("Final", paste(c('x','y','z'), collapse = "+"), sep = " ~ "))

  rf.df<-data.frame(Final=runif(10000),y=runif(10000),x=runif(10000),z=runif(10000))

  rf.df.list<-split(rf.df,rep(1:n,nrow(rf.df))[1:nrow(rf.df)])

  cl<-makeCluster(n)
  rf.list<-parLapply(cl,rf.df.list,function(x,rf.form,n){
    randomForest::randomForest(rf.form,x,ntree=100,nodesize=10, norm.votes=FALSE)},rf.form,n)
  stopCluster(cl)

  return(rf.list)
  }

我们使用以下方式获取和运行脚本：

scrip.loc<-"G:\\Scripts_Library\\R\\Stack_Answers\\parallel_question.R"

source(scrip.loc)

rf.parallel(n=10)

相当直接......我们并行运行了几个随机森林。似乎是内存效率。我们可以稍后将它们组合起来，或者做其他事情。便利。好的。举止得体。

现在考虑以下脚本，保存为parallel_question_2.R

rf.parallel_2<-function(n=10){
  library(parallel)
  library(magrittr)
  library(randomForest)

  rf.form<- as.formula(paste("Final", paste(c('x','y','z'), collapse = "+"), sep = " ~ "))

  rf.df<-data.frame(Final=runif(10000),y=runif(10000),x=runif(10000),z=runif(10000))

  large.list<-rep(rf.df,10000)

  rf.df.list<-split(rf.df,rep(1:n,nrow(rf.df))[1:nrow(rf.df)])

  cl<-makeCluster(n)
  rf.list<-parLapply(cl,rf.df.list,function(x,rf.form,n){

    randomForest::randomForest(rf.form,x,ntree=100,nodesize=10, norm.votes=FALSE)},rf.form,n)

  stopCluster(cl)

  return(rf.list)
}

在第二个脚本中，我们的源环境中有一个很大的列表。我们没有调用列表或将其带入我们的并行函数。我已将列表的大小设置为在至少 32gb 的机器上可能是个问题。

scrip.loc<-"G:\\Scripts_Library\\R\\Stack_Answers\\parallel_question_2.R"

source(scrip.loc)

rf.parallel_2(n=10)

当我们运行第二个脚本时，我们最终会携带大约 3gb（我们的大列表的大小）* 设置到集群的工作线程数，以及周围的其他材料。如果我们在非源环境中运行第二个脚本的内容，这不是行为；相反，我们得到一个约 3gb 的列表，并行化的函数运行没有问题，到此结束。

那么.. 工作环境如何/为什么从父环境中获取不必要的变量元素？为什么它只发生在源脚本中？当我有一个源代码、大型且复杂的脚本时，我该如何缓解这种情况，该脚本具有并行化的子部分（但可能携带 3-10gb 的中间数据）？

相关或类似主题：

Using parLapply and clusterExport inside a function

clusterExport, environment and variable scoping

Answer 1

parLapply(cl, X, FUN, ...) 的签名将FUN 应用于X 的每个元素。工作人员需要知道FUN，因此将其序列化并发送给工作人员。什么是 R 函数？它是定义函数的代码，和定义函数的环境。为什么是环境？因为在 R 中引用在 FUN 之外定义的变量是合法的，例如，

f = function(y) x + y
x = 1; f(1)
## [1] 2

作为第二种复杂性，R 允许函数更新函数外部的变量

f = function(y) { x <<- x + 1; x + y }
x = 1; f(1)
## [1] 3

在上面，我们可以想象我们可以弄清楚f()的环境的哪些部分需要被看到（只有变量x），但一般这种分析是不可能不实际评估的函数，例如f = function(y, name) get(name) + y; x = 1; f(1, "x")

因此，要在 worker 上评估 FUN，worker 需要知道 FUN 的定义和定义 FUN 的环境的内容。R 让工人通过使用serialize() 了解FUN。结果很容易看出

f = function(n) { x = sample(n); length(serialize(function() {}, NULL)) }
f(1)
## [1] 754
f(10)
## [1] 1064
f(100)
## [1] 1424

环境中较大的对象会导致更多信息发送给工作人员/由工作人员使用。

如果你仔细想想，到目前为止的描述意味着整个 R 会话 应该被序列化到工作线程（或者到磁盘，如果 serialize() 被用于保存对象） - -f()中隐式函数的环境包括f()的主体，也包括f()的环境，即全局环境，以及全局环境的环境，即搜索路径…… （查看environment(f) 和parent.env(.GlobalEnv)）。 R 有一个任意规则，它在全局环境中停止。所以不要使用隐式的function() {}，而是在 .GlobalEnv 中定义它

g = function() {}
f = function(n) { x = sample(n); length(serialize(g, NULL)) }
f(1)
## [1] 592
f(1000)
## [1] 592

还要注意，这会对可以序列化的函数产生影响。例如，如果g() 在下面的代码中被序列化，它将“知道”x

f = function(y) { x = 1; g = function(y) x + y; g() }
f(1)
## [1] 2

但在这里它不知道——它知道定义它的环境中的符号，但不知道调用它的环境中的符号。

rm(x)
g = function(y) x + y
f = function(y) { x = 1; g() }
f()
## Error in g() : object 'x' not found

在你的脚本中，你可以比较

cl = makeCluster(2)
f = function(n) {
    x = sample(n)
    parLapply(
        cl, 1,
        function(...)
            length(serialize(environment(), NULL))
    )
}
f(1)[[1]]
## [1] 256
f(1000)[[1]]
## [1] 4252

与

g = function(...) length(serialize(environment(), NULL))
f = function(n) {
    x = sample(n)
    parLapply(cl, 1, g)
}
f(1)[[1]]
## [1] 150
f(1000)[[1]]
## [1] 150

Answer 2

在处理接近尾声时，我将近 50 GB 的数据传回 parLapply，这并不理想。

我最终创建了一个名为parLapply 的新函数。我将它放在我的嵌套循环中，在那里创建了一个新环境，将父环境设置为.GlobalEnv，只将所需的变量传递给新环境，然后将该环境传递给clusterExport。

有关环境的详细信息，我推荐this 博客文章。另外，我发现 Ethan McCallum 和 Stephen Weston 的 Parallel R 书很有帮助。在第 15-17 页，“雪”包中对此问题进行了讨论。