在 shiny 中使用并行包

Question

我正在为我创建的模拟器创建一个闪亮的应用程序。为了加快模拟速度，我使用了 parallel 包。

我的应用程序在不并行化我的代码时运行良好，尽管它很慢。但是，当我并行化时，出现以下错误：

Error in checkForRemoteErrors(val) : 
  3 nodes produced errors; first error: Operation not allowed without an active reactive context. (You tried to do something that can only be done from inside a reactive expression or observer.)

这是我的 ui.R 和 server.R 的删节版：

ui.R

library(shiny)

shinyUI(fluidPage(
  titlePanel("Simulator"),

  fluidRow(
    column(6,
           fluidRow(
             column(5,
                    helpText("Choose 9 bitcoins for firm 1"),
                    selectizeInput("firm1bit1", label = "Bitcoin 1:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit2", label = "Bitcoin 2:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit3", label = "Bitcoin 3:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit4", label = "Bitcoin 4:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit5", label = "Bitcoin 5:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit6", label = "Bitcoin 6:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit7", label = "Bitcoin 7:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit8", label = "Bitcoin 8:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm1bit9", label = "Bitcoin 9:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    helpText("Choose the maximum number of transactions for firm 1"),
                    selectizeInput("firm1transacts", label = "Firm 1 maximum number of transactions:", 
                                   choices = data$max_transactions, options =
                                     list(maxOptions = 7))
             ),
             column(5,
                    helpText("Choose 9 bitcoins for firm 2"),
                    selectizeInput("firm2bit1", label = "Bitcoin 1:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit2", label = "Bitcoin 2:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit3", label = "Bitcoin 3:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit4", label = "Bitcoin 4:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit5", label = "Bitcoin 5:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit6", label = "Bitcoin 6:",
                                   choices = data$bitcoin, options = 
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit7", label = "Bitcoin 7:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit8", label = "Bitcoin 8:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    selectizeInput("firm2bit9", label = "Bitcoin 9:",
                                   choices = data$bitcoin, options =
                                     list(maxOptions = 7)),
                    helpText("Choose the maximum number of transactions for firm 2"),
                    selectizeInput("firm2transacts", label = "Firm 2 maximum number of transactions:", 
                                   choices = data$max_transactions, options =
                                     list(maxOptions = 7))
             ),
             submitButton("Simulate")
           ))
  )
))

server.R

cl <- makeCluster(detectCores()-1, 'PSOCK')

shinyServer(function(input, output, session){

  firm1bits <- reactive({c(input$firm1bit1, input$firm1bit2, input$firm1bit3,
                            input$firm1bit4, input$firm1bit5, input$firm1bit6,
                            input$firm1bit7, input$firm1bit8, input$firm1bit9)})
  firm2bits <- reactive({c(input$firm2bit1, input$firm2bit2, input$firm2bit3,
                            input$firm2bit4, input$firm2bit5, input$firm2bit6,
                            input$firm2bit7, input$firm2bit8, input$firm2bit9)})
  firm1max <- reactive({input$firm1transacts})
  firm2max <- reactive({input$firm2transacts})

  reactive({clusterExport(cl, varlist=c("firm1bits", "firm2bits", "firm1max",
                                        "firm2max"))})
  gameResults <- reactive({parSapply(cl, 1:1000, function(i){
    simulate_bitcoin_Twoway(firm1bits(), firm2bits(), firm1max(), firm2max())
  })})
})

我想重申，当我不使用 parSapply() 而是使用 replicate() 时，代码仍然有效。问题不在其他函数，例如simulate_bitcoin_Twoway().

Answer 1

由于您没有提供 MCVE ，所以这是一个比其他任何事情都更疯狂的猜测。

当您调用 clusterExport 时，您会在集群中分配反应变量。 parSapply 在集群上执行 simulate_bitcoin_Twoway，每个 worker 都有单独的环境，而不包含 reactive 块。由于反应值需要反应上下文，因此整个操作失败。

为了解决这个问题，我会尝试在本地评估反应式表达式并分发返回值：

gameResults <- reactive({
    firm1bits_v <- firm1bits()
    firm2bits_v <- firm2bits()
    firm1max_v <- firm1max()
    firm2max_v <- firm2max()

    clusterExport(cl, varlist=c(
        "firm1bits_v", "firm2bits_v", "firm1max_v", "firm2max_v"))

    parSapply(cl, 1:1000, function(i ){
        simulate_bitcoin_Twoway(firm1bits_v, firm2bits_v, firm1max_v, firm2max_v)
    })
})

如果上述方法不起作用，您可以尝试依赖反应值，但在 isolate 块内的集群上进行评估。

编辑:

这是一个完整的工作示例：

library(shiny)
library(parallel)
library(ggplot2)

cl <- makeCluster(detectCores()-1, 'PSOCK')
sim <- function(x, y, z) {
    c(rnorm(1, mean=x), rnorm(1, mean=y), rnorm(1, mean=z))
}

shinyApp(
    ui=shinyUI(bootstrapPage(
        numericInput("x", "x", 10, min = 1, max = 100),
        numericInput("y", "y", 10, min = 1, max = 100),
        numericInput("z", "z", 10, min = 1, max = 100),
        plotOutput("plot")
    )),

    server=shinyServer(function(input, output, session){
        output$plot <- renderPlot({
            x <- input$x
            y <- input$y
            z <- input$z
            clusterExport(
               cl, varlist=c("x", "y", "z", "sim"),
               envir=environment())

            mat <- t(parSapply(cl, 1:1000, function(i) {
                sim(x, y, z)
            }))
            ggplot(
                as.data.frame(mat),
                aes(x=V1, y=V2, col=cut(V3, breaks=10))) + geom_point()
        })
    })
)

请注意 clusterExport 的 envir 参数。默认情况下 clusterExport 在全局环境中搜索，其中定义在闭包中的变量不可见。

Answer 2

我遇到了完全相同的问题，尽管我使用的是 doParallel 和 foreach 软件包。我没有在我的应用程序中明确定义任何反应值，但我在 foreach 块中引用了 input，这当然是默认情况下的反应值。

在尝试了许多不同的事情之后，我发现最简单的解决方案是我可以简单地在 foreach 中包含一个 isolate 语句。但是，由于 isolate 使得这些变量不依赖于 foreach 循环之外的任何内容，因此我们需要导出 input 向量以及 isolate 函数本身。在您的情况下，您还需要导出所有反应值。

给我错误的代码

Operation not allowed without an active reactive context

看起来像这样：

  optiResults<-foreach(i=seq(1,3),
  .combine = rbind,
  ) %dopar% {
    print("hello")
    rv = input$power
    thingy = data.frame(matrix(0,1,2))
  }

简单的解决方案是这样做：

  optiResults<-foreach(i=seq(1,3),
  .combine = rbind,
  .export = c("isolate","input")
  ) %dopar% {
    print("hello")
    isolate({
    rv = input$power
    thingy = data.frame(matrix(0,1,2))
    })
  }

在 shiny 中使用并行包

Using parallel package in shiny

parallel-processing

r

shiny

ui.R

server.R