Использование 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)

Довольно просто ... мы запустили несколько случайныхлес параллельно.Кажется, эффективная память.Мы могли бы объединить их позже или сделать что-то еще.Handy.Ницца.Хорошо себя ведет.

Теперь рассмотрим следующий сценарий, сохраненный как 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)
}

Во втором сценарии мы получили большой список в нашей исходной среде.Мы не вызываем список и не вносим его в нашу параллельную функцию.Я установил размер списка, чтобы он, вероятно, был проблемой, по крайней мере, на 32-гигабайтной машине.

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

source(scrip.loc)

rf.parallel_2(n=10)

Когда мы запускаем второй скрипт, мы получаем ~ 3 ГБ (размер нашего большогоlist) * количество рабочих потоков, установленных для кластера, дополнительный материал вокруг.Если мы запускаем содержимое второго скрипта в среде без источников, это не поведение;скорее, мы получаем один ~ 3 ГБ список, распараллеленная функция выполняется без проблем, и на этом все.

Итак ... как / почему рабочие среды отбирают элементы ненужных переменных из родительской среды?Почему это происходит только в сценариях с источником?Как я могу смягчить это, когда у меня есть большой и сложный сценарий с источником, который имеет параллельные части (но может иметь 3-10 ГБ промежуточных данных, переносимых вокруг)?

Соответствующие или похожие потоки:

Использование parLapply и clusterExport внутри функции

clusterExport, среда и область видимости переменных

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")

Поэтому для оценки FUN на работнике работник должен знать как определение FUN, так и содержаниесреды FUN был определен в. R позволяет работнику узнать о FUN с помощью serialize().Следствие легко увидеть

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