R使用foreach并行调用Stan报error writing to connection如何解决

• 多进程写入冲突：6个并行子进程同时调用sink()向同一个log.txt写入内容，触发文件锁冲突，直接导致「error writing to connection」报错，这是重装后N=3数据集也无法运行的核心原因
• 嵌套并行资源耗尽：外层foreach已经启动6个并行进程，每个子进程内又配置options(mc.cores = parallel::detectCores())，stan采样时会再启动和CPU核心数等量的线程，总运行线程数远超CPU承载上限，导致40名患者数据运行时卡死、超时
• 手动终止残留锁：之前手动终止无响应的程序时，残留的僵尸进程仍占用log.txt、output.txt的文件句柄，重装软件未释放这部分文件锁，导致新进程无法写入文件
• 子进程环境缺失：foreach默认不会把主进程的变量、预编译模型传递给子进程，你代码中直接调用compiled_model、it、res_dir等主进程变量，子进程找不到对应对象就会静默崩溃

1. 移除子进程内的并行写log操作，改用进程专属日志或直接输出到foreach指定的output.txt，避免文件锁冲突
2. 关闭嵌套并行，子进程内将stan的运行核心数设为1，外层多进程已经足够利用CPU资源
3. 显式通过foreach的.export参数传递主进程变量，.packages参数预加载依赖包，不要在子进程内重复加载包
4. 每次运行前先关闭所有打开的文件连接，清理残留临时文件

if (run) {
  # 运行前清理旧连接和临时文件
  closeAllConnections()
  unlink("log.txt")
  unlink(res_dir, recursive = TRUE)
  
  duration <- Sys.time()
  # 启动并行集群，预加载依赖包
  cl <- makeCluster(6, outfile="output.txt")
  registerDoParallel(cl)
  
  writeLines(c(""), "log.txt")
  dir.create(res_dir, showWarnings = FALSE)
  
  out <- foreach(i = 1:nrow(it), 
                 # 显式传递主进程变量给子进程
                 .export = c("it", "res_dir", "compiled_model", "param", "n_it", "n_chains", "seed", "model_name", "N", "timearray", "sevdata", "folds"),
                 # 预加载依赖包，避免子进程重复加载
                 .packages = c("tidyverse", "rstan", "scoringRules")) %dopar% {
    
    w <- it$TrainingDay[i]
    f <- it$Fold[i]
    
    # 关闭嵌套并行，每个子进程仅用1核跑stan
    rstan_options(auto_write = TRUE)
    options(mc.cores = 1)
    source("functions.R")
    
    # 调整写入逻辑减少冲突
    cat(paste0("Starting training at week ", w, ", fold ", f, " \n"), file = "log.txt", append = TRUE)
    
    dt_wf <- data.frame(Patient = rep(N,times=1,each=80),
                        Day = timearray,
                        S = sevdata,
                        S_train = sevdata)
    
    idx_pred <- which(!is.na(dt_wf$S) & (dt_wf$Day > w) & (dt_wf$Patient %in% N[folds == f]))
    
    dt_wf$S_train[idx_pred] <- NA
    
    data_stan <- format_data(dt_wf, N, idx_pred)
    
    perf <- data.frame(Patient = dt_wf$Patient[idx_pred],
                       TrainingDay = w,
                       TestingDay = dt_wf$Day[idx_pred],
                       Fold = f,
                       S = dt_wf$S[idx_pred])
    
    ## Fit
    fit <- sampling(compiled_model,
                    data = data_stan,
                    pars = param,
                    iter = n_it,
                    chains = n_chains,
                    seed = seed,
                    control = list(adapt_delta = case_when(model_name %in% c("SSM", "SSMX") ~ 0.99,
                                                           TRUE ~ 0.9)))
    
    ## Prepare ouput
    lpd <- extract(fit, pars = "lpd")[[1]]
    pred <- extract(fit, pars = "S_pred")[[1]]
    smp <- sapply(1:ncol(pred), function(i) {list(pred[, i])})
    
    perf <- perf %>%
      mutate(Mean_pred = apply(pred, 2, mean), # cf. point prediction (mean)
             lpd = apply(lpd, 2, function(x) {log(mean(exp(x)))}), # marginalise lpd
             CRPS = scoringRules::crps_sample(perf[["S"]], t(pred)),
             Samples = smp)
    
    ## Save (intermediate results)
    saveRDS(perf, file = file.path(res_dir, paste0("val_", i, ".rds")))
    
    cat(paste0("Ending training at day ", w, ", fold ", f, " \n"), file = "log.txt", append = TRUE)
    # 每次写入后关闭连接避免残留
    closeAllConnections()
    return(perf)
  }
  stopCluster(cl)
  (duration = Sys.time() - duration)
  
  # Recombine results
  files <- list.files(res_dir, pattern = "val_.*\\.rds")
  if (length(files) < nrow(it)) {
    warning("Number of files (", length(files), ") less than the number of iterations (", nrow(it), "). Some runs may have failed.")
  }
  res <- do.call(rbind,
                 lapply(files,
                        function(f) {
                          readRDS(file.path(res_dir, f))
                        }))
  saveRDS(res, file = res_file)
} else {
  res <- readRDS(res_file)
}

• 若40名患者数据运行仍较慢，可先在主进程预编译stan模型，避免每个子进程重复编译
• 可适当降低stan采样的迭代数n_it，或用optimizing替代sampling做最大后验估计，大幅缩短运行时间

内容的提问来源于stack exchange，提问作者Umar Shehzad