在并行 foreach 计算中使用标准 R 闪亮进度条

Question

我正在尝试使用 doParallel 后端在并行 foreach 循环中使用标准 R 闪亮进度条。但是，这会导致以下错误消息：

警告：{中的错误：任务 1 失败 -“'session' 不是 ShinySession 对象。”

代码（最小工作示例）

library(shiny)
library(doParallel)

ui <- fluidPage(
  actionButton(inputId = "go", label = "Launch calculation")
)

server <- function(input, output, session) {

  workers=makeCluster(2)
  registerDoParallel(workers)

  observeEvent(input$go, {
    Runs=c(1:4)
    Test_out=foreach(i=Runs, .combine=cbind, .inorder=TRUE, .packages=c("shiny"),.export=c("session")) %dopar% { 
      pbShiny = shiny::Progress$new()
      pbShiny <- Progress$new(session,min = 0, max = 10)
      on.exit(pbShiny$close())
      test_vec=rep(0,100)

      for(i in 1:10){
        test_vec=test_vec+rnorm(100)
        pbShiny$set(message="Simulating",detail=paste(i),
                  value=i)
        Sys.sleep(0.2)
      }

    }
  })
}

shinyApp(ui = ui, server = server)

如果我按顺序运行 foreach 循环（使用 registerDoSEQ()），代码就会运行。 有谁知道如何解决这个问题？

---

总体目标

• 使用 doParallel 在并行 foreach 循环中向用户显示进度 闪亮的后端
• 用户应该知道工人的数量和 每个工人的进度和/或整体进度

---

在以下链接下有一个类似的问题，但由于没有提供工作示例，因此没有得到解决：

Utilizing parallel foreach for progress bar in R Shiny

Answer 1

doParallel 包是并行包的扩展，如此处的文档所示。

https://cran.r-project.org/web/packages/doParallel/doParallel.pdf

阅读并行包的文档，我们看到它实现了 3 种不同的方法来实现并行性。请记住，R 是一种单线程语言。

1. 父进程与工作进程或子进程通信的新 R 会话。
2. 通过分叉
3. 使用操作系统级别的工具

您可以在此处找到此信息，

https://stat.ethz.ch/R-manual/R-devel/library/parallel/doc/parallel.pdf

这样做的结果是子进程无法与父进程通信，直到它完成计算并返回一个值。这是据我所知。

因此，无法在工作进程中勾选进度条。

完全披露，我没有使用过 doParallel 包，并且关于闪亮的文档是有限的。

---

替代解决方案

有一个类似的包，但是有关于闪亮的大量文档。这些是 futures 和 promises 和 ipc 包。 futures 和promises 启用异步编程，而ipc 启用进程间通信。为了进一步帮助我们，它还具有AsyncProgress() 功能。

这是一个我们同步勾选两个计数器的示例。

示例

library(shiny)
library(future)
library(promises)
library(ipc)

plan(multisession)


ui <- fluidPage(
  actionButton(inputId = "go", label = "Launch calculation")
)

server <- function(input, output, session) {

  observeEvent(input$go, {

    progress = AsyncProgress$new(message="Complex analysis")

    future({
      for (i in 1:15) {
        progress$inc(1/15)
        Sys.sleep(0.5)
      }

      progress$close()
      return(i)
    })%...>%
      cat(.,"\n")

    Sys.sleep(1)

    progress2 = AsyncProgress$new(message="Complex analysis")

    future({
      for (i in 1:5) {
        progress2$inc(1/5)
        Sys.sleep(0.5)
      }

      progress2$close()

      return(i)
    })%...>%
      cat(.,"\n")

    NULL
  })
}

shinyApp(ui = ui, server = server)

您的代码已修改

这是您编写的代码，稍作修改以分离出许多异步进程。任何工作都可以在工作器中执行，例如您创建的向量并添加rnorm。 （此处未显示）

library(shiny)
library(future)
library(promises)
library(ipc)

plan(multisession)

ui <- fluidPage(
  actionButton(inputId = "go", label = "Launch calculation")
)

server <- function(input, output, session) {

  observeEvent(input$go, {
    Runs=c(1:4) #define the number of runs
    progress = list() #A list to maintain progress for each run

    for(j in Runs){
      progress[[j]] = AsyncProgress$new(message="Complex analysis")
      future({
        for (i in 1:10) {
          progress[[j]]$inc(1/10)
          Sys.sleep(0.2)
        }
        progress[[j]]$close()
        return(i)
    })%...>%
        cat(.,'\n')
    }

    NULL
  })
}

shinyApp(ui = ui, server = server)

以上代码是此处ipc文档中代码的修改版本：

http://htmlpreview.github.io/?https://github.com/fellstat/ipc/blob/master/inst/doc/shinymp.html

其他资源：

https://rstudio.github.io/promises/articles/overview.html

Answer 2

我想我找到了解决运行次数超过核心数的情况的解决方案。

我搜索了嵌套的未来进程，发现如下页面：

https://cran.r-project.org/web/packages/future/vignettes/future-3-topologies.html

我将代码更改如下。这会按核心顺序运行作业，并相应地更新相应的进度条。

library(shiny)
library(future)
library(promises)
library(ipc)
library(listenv)

plan(list(multiprocess, sequential))

ui <- fluidPage(
  actionButton(inputId = "go", label = "Launch calculation")
)

server <- function(input, output, session) {

  observeEvent(input$go, {
    x <- listenv()
    Runs=12 #define the number of runs
    N=availableCores()
    Tasks=rep(0,N) #Number of sequential tasks per core
    Tasks[1:(Runs-(ceiling(Runs/N)-1)*N)]=ceiling(Runs/N)
    if((Runs-(ceiling(Runs/N)-1)*N)<N){
      Tasks[(Runs-(ceiling(Runs/N)-1)*N+1):N]=ceiling(Runs/N)-1
    }

    progress = list() #A list to maintain progress for each run

    for(j in 1:N){

      for(l in 1:Tasks[j]){
        progress[[(l-1)*N+j]] = AsyncProgress$new(message=paste("Complex analysis, core ",j," , task ",l))
      }

    x[[j]]%<-%{
      for(l in 1:Tasks[j]){
        for (i in 1:10) {
          progress[[(l-1)*N+j]]$inc(1/10)
          Sys.sleep(0.5)
        }
        progress[[(l-1)*N+j]]$close()
      }
    }
    }

    NULL
  })
}

shinyApp(ui = ui, server = server)