多个节点上的 foreach 迭代器

Question

我正在尝试在 4 个节点的集群上运行模拟，每个节点有 40 个内核/80 个线程，并且在运行并行化 foreach 循环时不断遇到 OUT_OF_MEMORY 错误。我很确定这与我正在通过大 tibble params 的行迭代循环这一事实有关。

我已经尝试将params 更改为一个列表并遍历一个列表（希望这在内存上会更容易）但这没有帮助。另外，我尝试使用一个交互器，据我所知，它应该防止 params 被加载到每个线程中（从而防止 OUT_OF_MEMORY 错误），但这也不起作用。我怀疑发生这种情况是因为我正在处理多个节点（并且我读过类似“它们不共享物理内存”之类的内容），但这只是我的猜测。

这是我最近一次尝试的设置方式：

procs <- 2 * as.numeric(Sys.getenv("SLURM_NTASKS")) 
itx <- iter(params, by = 'row')

registerDoParallel(procs)
tib <- foreach (i = itx,
         .packages=c("tidyverse", "stabledist",  
                       "copula", "VineCopula", "MMDCopula", 'glue')
         ) %dopar% {
    
    # Grab the variables from each row
    alpha <- i[['alpha']]
    d <- i[['delta']]
    s <- i[['seed']]
    a <- i[['mu']] %>% unlist %>% unname
    a_name <- nameTmp(a)
    time <- i[['time']]
    copula <- i[['copula']]
    type <- i[['type']]
    
    set.seed(s)
    a_j <- rev( 1/(1 + (1:(20*TMax))^(d)) )
    X <- Simulate_Lin_Proc(alpha, TMax, a_j, 1)
    tibTmp <- f(...) # Function dependent on the parameters (returns 1 row tibble)
    
    return(tibTmp)
}
stopImplicitCluster()

欢迎提出任何建议。

Answer 1

我推荐四种可能的解决方案。

1. 只要您需要，就将每个对象保留在循环的迭代中。不再使用时将其删除或覆盖。

2. 将循环的向量拆分成块并将每个块的输出写入磁盘。这样可以避免将不断增长的返回值列表保留在内存中。

# Split params vector in chunks of 5000 elements
params_chunks <- split(params, ceiling(seq_along(params)/5000))
    
# Loop over chunks
foreach(P_C = 1:length(params_chunks), .packages=c("tidyverse", "stabledist", "copula", "VineCopula", "MMDCopula", "glue", "foreach")) %dopar% {
   p_c <- params_chunks[[P_C]]
    
   # Loop over elements within chunks
   foreach(p = p_c, .packages=c("tidyverse", "stabledist", "copula", "VineCopula", "MMDCopula", "glue")) %do% {
      ...
      return(tibTmp)
   } %>%
   saveRDS(., paste0(some_directory, P_C, ".RDS"))
   return(NULL)
}

除了将中间列表输出作为 RDS 文件写入磁盘之外，您还可以对它们进行行绑定并将它们作为羽毛文件写入磁盘。这取决于返回的tibTmp 的结构是RDS 还是羽化文件效率更高。

1. 服务器如何在 R 之外处理这个进程？它使用 Openmpi 吗？如果是，我建议使用doMPI 而不是doParallel。这让外部 Openmpi 进程（针对集群服务器量身定制并对其拥有更多控制权）来处理并行化，而不是从 R 内部控制它。

这意味着您无需加载 doParallel 包，而是将以下代码发布到您的 R 脚本中：

if (!is.loaded("mpi_initialize")) {
   library("doMPI")
}

cl <- startMPIcluster(comm=0)
registerDoMPI(cl)
  
.Last <- function(){
   if (is.loaded("mpi_initialize")){
      if (mpi.comm.size(1) > 0){
         mpi.close.Rslaves()
      }
      .Call("mpi_finalize")
   }
}

在你提到的脚本底部：

closeCluster(cl)
mpi.quit()

4. 您还可以限制每个节点的任务数。在 SLURM 作业文件中，您可以使用 --tasks-per-node 选项进行设置。例如。 #SBATCH --tasks-per-node=16 行将每个节点的最大任务数设置为 16。继续减少节点数，直到计算适合节点的物理内存。