Experimental work on forked computation in shiny app using the package "parallel".

README.md

Shiny asynchronous jobs

Experiemental work on forked computation in shiny app using the package "parallel".

Does not work on windows.

Example

Rscript run.R

Usage

library(shiny)
source("helper.R")

# Define an expensive function

expensiveFunction <- function(){
  for(i in 1:3){
    Sys.sleep(1) 
  }
  return(1:10)
}

# define a feedback function. res$data is the result, res$message is a process message.

feedbackFunction <- function(res){
  output$plotExpensive <- renderPlot( barplot(res$data) )
}

# Do it in a forked process. Resulting 'pid' object is the process id. Can be used with tools::pskill to stop process.

pid <- do( 
  expr = expensiveFunction() , 
  feedback = feedbackFunction()
  )

helper.R

#' Exacute expression in a forked process (unix only)
#' @param expr {expression} Expression to evaluate
#' @param feedback {function} Function to handle worker feedback. Take one argument : res. Res is a list with two member : data and message. 
#' @param loading {expression} Expression to evaluate before the start of the process
#' @param refreshRateSeconds {numeric} Number of second between read refresh
#' @param maxTimeSeconds {numeric} Maximum time for the process before ending. 
#' @return process pid
#' @export
do <- function(expr, feedback, loading=NULL, refreshRateSeconds= 1, maxTimeSeconds = Inf ){
  library(parallel)
  
  if(!is.null(loading)){ 
    eval(loading,envir=parent.frame())
  }
  
  # Create a new external job
  job <- mcparallel(expr)
  # Get current time
  start <- Sys.time()
  # List in output
  res <- list(
    data = NULL,
    message = NULL
    )

  # Internal nested observer. 
  observe({
    # get diff time
    elapsed <- as.numeric(
      Sys.time() - start,
      units="secs"
      )

    # test for timeout and ollect result
    if( isTRUE( elapsed > maxTimeSeconds ) ){
     res$message <- sprintf(
       "Timeout. Process took more than %s [s]"
       , maxTimeSeconds
       )
    }else{
     res$data <- mccollect(
       jobs = job, 
       wait=F
       )[[1]]
    }

    # evaluate result
    if(all(sapply(res,is.null))){
      invalidateLater( millis=refreshRateSeconds )
    }else{
      tools::pskill( job$pid )
      if( !is.null( res$message )) stop( res$message )
      feedback( res )
    }
  })
  return(job$pid)
}

run.R

library(shiny)


runApp(".",port=2323,launch.browser=F)

server.R

#
# Author Fred Moser
# Date 26/09/2016
# Session info
#    R version 3.3.1 (2016-06-21)
#    Platform: x86_64-apple-darwin14.5.0 (64-bit)
#    Running under: OS X 10.10.5 (Yosemite)


library(shiny)
library(parallel)
source("helper.R")


#' Server function
#' @param input {reactivevalues}
#' @param output {shinyoutput}
#' @param session {ShinySession}
server <- function(input, output, session){
  # on click handler
  observeEvent(input$numTestNonBlocking,{ 
    do(
      refreshRateSecond=0.5, 
      maxTimeSecond=4,
      loading={
      session$sendCustomMessage("innerHTML",
        list(
          id = "plotExpensive",
          value = "Loading, please wait"
          )
        )
      }
      ,
      expr={
        for(i in 1:2){
        Sys.sleep(1) 
        }
        rnorm(1:input$numTestNonBlocking)
      },function( res ){
        output$plotExpensive <- renderPlot( barplot(res$data) )
      })
})
  # plot
  output$plotTest <-  renderPlot({
    barplot( rnorm( 1:abs(input$numTest) ))
  })
}

ui.R

#' Shiny UI
ui <- bootstrapPage(
  tags$head(
    tags$script('
      Shiny.addCustomMessageHandler("innerHTML", 
        function(e) {
          el = document.getElementById(e.id);
          if(el){
            el.innerHTML=e.value
          }
        }) 
      ')),
      column(12,
        tagList(
          tags$h1("Shiny Asynchronous Job"),
          tags$p("Asynchronous processes in shiny app with feedback function."),
          fluidRow(
            tags$div(class="col-md-6",
              tags$h2("Expensive job"),
              tags$p("Launch a function asynchronously: other processes will not be blocked"),
              sliderInput(
                inputId="numTestNonBlocking",
                label="Enter a number",
                value=10,
                min=0,
                max=100
                ),
              plotOutput("plotExpensive")
              ),
            tags$div(class="col-md-6",
              tags$h2("Simple job"),
              tags$p("Launch a function on the main thread : other processes will be blocked"),
              sliderInput(
                inputId="numTest",
                label="Enter a number",
                value=10,
                min=0,
                max=100
                ),
              plotOutput("plotTest")
              )
            )
          )
        )
      )