R中的并行处理 - 使用mclapply()和pbmclapply()设置种子

Question

我正在使用R语言中的parallel包中的mclapply()并行运行模拟程序，希望能够跟踪每个函数调用的进度。因此，我决定改用pbmcapply包中的pbmclapply()，以便每次运行模拟时都有一个进度条(pbmclapply()是为mclapply()特别创建的包装器，因此它们应该具有相同的功能，除了进度条)。

我可以使用mclapply()设置种子并得到可重复的结果，没有任何问题，但奇怪的是，每次运行pbmclapply()都会得到不同的结果。下面是一个相当简单的 reprex 示例。

例如，这是使用mcapply()：

## GIVES THE SAME RESULT EACH TIME IT IS RUN
library(parallel)
RNGkind("L'Ecuyer-CMRG")
set.seed(1)
x <- mclapply(1:100, function(i) {rnorm(1)}, mc.cores = 2)
y <- do.call(rbind, x)
z <- mean(y)
print(mean(z))

使用 pbmclapply() 的代码与上述代码相同：

## GIVES DIFFERENT RESULTS EACH TIME IT IS RUN
library(pbmcapply)
RNGkind("L'Ecuyer-CMRG")
set.seed(1)
x <- pbmclapply(1:100, function(i) {rnorm(1)}, mc.cores = 2)
y <- do.call(rbind, x)
z <- mean(y)
print(mean(z))

上面两个代码块唯一的区别是第二个使用了pbmclapply()，而第一个使用了mclapply()。然而，每次运行第一个块时都会得到一致的结果，而每次运行第二个块时都会得到不同的结果（尽管种子设置方式相同）。这两个函数之间播种程序有什么区别？我希望能得到任何关于发生这种情况的反馈。谢谢！

- bob

Answer 1

问题在于pbmcapply包中的utils.R文件运行以下代码行：

if (isTRUE(mc.set.seed))
      mc.set.stream()

如果我们将这与在parallel包中运行mclapply()函数时调用的内容进行比较，我们可以看到它的运行方式：

if (mc.set.seed) 
        mc.reset.stream()

这会影响结果，因为重置流将允许代码从全局设置的种子运行，而运行设置流将使用初始种子设置为新的随机起始值。 我们可以在下面附加的函数中看到这一点：

mc.reset.stream <- function () 
{
    if (RNGkind()[1L] == "L'Ecuyer-CMRG") {
        if (!exists(".Random.seed", envir = .GlobalEnv, inherits = FALSE)) 
            sample.int(1L)
        # HERE! sets the seed to the global seed variable we set
        assign("LEcuyer.seed", get(".Random.seed", envir = .GlobalEnv, 
            inherits = FALSE), envir = RNGenv)
    }
}

mc.set.stream <- function () 
{
    if (RNGkind()[1L] == "L'Ecuyer-CMRG") {
        assign(".Random.seed", get("LEcuyer.seed", envir = RNGenv), 
            envir = .GlobalEnv)
    }
    else {
        if (exists(".Random.seed", envir = .GlobalEnv, inherits = FALSE)) 
            rm(".Random.seed", envir = .GlobalEnv, inherits = FALSE)
    }
}

我认为这种变化可能是由于使用mclapply函数时存在问题，当您想在设置种子后多次调用mclapply函数时，它会使用相同的随机数。（即通过重新设置r会话，你应该能以相同的顺序获得相同的结果，在pbmclapply中第一次我得到了0.143，然后是0.064，最后是-0.015）。通常这是首选的行为，因此您可以多次调用该函数。更多信息请参见R doesn't reset the seed when "L'Ecuyer-CMRG" RNG is used?。

您可以通过将.customized_mcparallel函数定义中的行从mc.set.stream()更改为mc.reset.stream()来测试这两种实现之间的差异。在此处，我已简化了软件包中函数调用，删除了进度条，并仅保留计算（还删除了错误检查）和设置随机种子。 （另外请注意，这些功能将不再在Windows机器上运行，只能在Linux或MacOS上运行）。

library(pbmcapply)
RNGkind("L'Ecuyer-CMRG")
set.seed(1)
pbmclapply <- function()  {

  pkg <- asNamespace('pbmcapply')
  .cleanup <- get('.cleanup', pkg)


  progressMonitor <- .customized_mcparallel({

    mclapply(1:100, function(i) {
            rnorm(1)
        }, mc.cores = 2, mc.preschedule = TRUE, mc.set.seed = TRUE,
                       mc.cleanup = TRUE, mc.allow.recursive = TRUE)
  })

  # clean up processes on exit
  on.exit(.cleanup(progressMonitor$pid), add = T)

  # Retrieve the result
  results <- suppressWarnings(mccollect(progressMonitor$pid)[[as.character(progressMonitor$pid)]])

  return(results)
}

.customized_mcparallel <- function (expr, name, detached = FALSE){
  # loading hidden functions
  pkg <- asNamespace('parallel')
  mcfork <- get('mcfork', pkg)
  mc.advance.stream <- get('mc.advance.stream', pkg)
  mcexit <- get('mcexit', pkg)
  mcinteractive <- get('mcinteractive', pkg)
  sendMaster <- get('sendMaster', pkg)
  mc.set.stream <- get('mc.set.stream', pkg)
  mc.reset.stream <- get('mc.reset.stream', pkg)

  f <- mcfork(F)
  env <- parent.frame()
  mc.advance.stream()
  if (inherits(f, "masterProcess")) {

    mc.set.stream()
    # reset the group process id of the forked process
    mcinteractive(FALSE)

    sendMaster(try(eval(expr, env), silent = TRUE))
    mcexit(0L)
  }

  f
}

x <- pbmclapply()
y <- do.call(rbind, x)
z <- mean(y)
print(z)

为了完全解决问题，我最好的建议是在您自己的代码中重新实现这些函数（我将函数从pbmcapply粘贴复制了一下并进行了一些微小修改），或者通过分叉该包并在utils.R文件中用mc.reset.seed替换mc.set.seed。目前我想不出更简单的解决方案，但希望这能解释清楚问题。

- Joel Kandiah

---

非常好的回答。顺便说一下：我打算授予您赏金，但故意将其保持开放状态，以吸引更多的流量到问题和答案上，两者都是值得社区额外点赞的典范。感谢您分享您的见解！ - merv

Answer 2

Joel Kandiah提出了一个很好的问题并给出了优秀的答案！

另一个解决方案是将您的代码放入R-Markdown文件中。编织该文件将始终得到相同的结果。但是显示进度更加复杂。您还可以通过Rscript从命令行运行代码：

Rscript yourfile.R

这样做每次都会得到相同的结果，因为您总是从头开始。它将显示进度，并且您还可以将输出重定向到文件中。对于长时间运行的模拟，调用Rscript比使用GUI更加稳健。

不确定这是否足够满足您的需求，但仍然想分享一下，因为它对我非常有效，而且不需要更改pbmclapply。

- Johannes Titz