R doParallel 进度条来监控完成的工作

Question

我正在尝试编写一个具有多线程功能的 CRAN 包。我使用 实现了一个完美的解决方案doSNOW，但该软件包已被 CRAN 团队标记为“已取代”，他们要求我切换到一个doParallel解决方案。这很好，但是我找不到一种方法来监视已完成的作业数量，doParallel就像我使用doSNOW. 这是我的doSNOW解决方案：

# Set up parameters
nthreads<-2
nreps<-100
funrep<-function(i){
    Sys.sleep(0.1)
    res<-c(log2(i),log10(i))
    return(res)
}
# doSNOW solution
library(doSNOW)
cl<-makeCluster(nthreads)
registerDoSNOW(cl)
pb<-txtProgressBar(0,nreps,style=3)
progress<-function(n){
    setTxtProgressBar(pb,n)
}
opts<-list(progress=progress)
i<-0
output<-foreach(i=icount(nreps),.combine=c,.options.snow=opts) %dopar% {
    s<-funrep(i)
    return(s)
}
close(pb)
stopCluster(cl)

这是之前 Stack Overflow 帖子中建议的 doParallel 解决方案。但是，如您所见，它不会在作业完成时打印进度，它只会在最后合并结果时打印。

# doParallel solution
library(doParallel)
progcombine<-function(){
  count<-0
  function(...) {
    count<<-count+length(list(...))
    setTxtProgressBar(pb,count)
    utils::flush.console()
    c(...)
  }
}
cl <- makeCluster(nthreads)
registerDoParallel(cl)
output<-foreach(i = icount(nreps),.combine=progcombine()) %dopar% {
    funrep(i)
}
stopCluster(cl)

您能否建议我使用doParallel或至少不使用被取代的来监视作业状态完成的解决方案doSNOW？可能带有进度条，也可能带有多操作系统功能。非常感谢！

Answer 1

（免责声明：我是 Progressr 包和未来框架的作者）

当使用doFuture作为foreach的并行后端时， progressr包可以实现这一点：

library(progressr) ## use progressr for procession updates
library(doFuture)  ## attaches also foreach and future
registerDoFuture() ## tell foreach to use futures
plan(multisession) ## parallelize over a local PSOCK cluster

xs <- 1:5

with_progress({
  p <- progressor(along = xs) ## create a 5-step progressor
  y <- foreach(x = xs) %dopar% {
    p()                       ## signal a progression update
    Sys.sleep(6.0-x)
    sqrt(x)
  }
})

默认设置用于utils::txtProgressBar()进展报告，但您可以更改此设置。progress::progress_bar()例如，以下内容将通过和报告进度更新beepr::beep()：

library(progressr) ## use progressr for procession updates
library(doFuture)  ## attaches also foreach and future
registerDoFuture() ## tell foreach to use futures
plan(multisession) ## parallelize over a local PSOCK cluster

xs <- 1:5

with_progress({
  p <- progressor(along = xs) ## create a 5-step progressor
  y <- foreach(x = xs) %dopar% {
    p()                       ## signal a progression update
    Sys.sleep(6.0-x)
    sqrt(x)
  }
})

您还可以为每个进度更新添加消息，例如

progressr::handlers("progress", "beepr")

FYI，是基本上whereplan(multisession, workers = 2)的缩写。plan(cluster, workers = cl)clcl <- parallel::makeCluster(2L)

附言。Progressr 包的目标是为进度更新提供最小的、可持续的、可扩展的和统一的 API。这对于使用的迭代器框架是不变的。

聚苯硫醚。Progressr API 正在开发中；它可能需要一段时间才能识别出真实的自我。

Answer 2

我找不到 doParallel 的解决方案（我认为它不支持作业完成的进度条），但也许您可以尝试新的包pbapply：

# pbapply solution
library(pbapply)
cl<-parallel::makeCluster(nthreads)
invisible(parallel::clusterExport(cl=cl,varlist=c("nreps")))
invisible(parallel::clusterEvalQ(cl=cl,library(utils)))
result<-pblapply(cl=cl,
                 X=1:nreps,
                 FUN=funrep)
parallel::stopCluster(cl)