Asynchronous Programming in R

To kick-Start Developing a Scalable shiny App

Abstract

The notebook is a practical walk-through of implementing asynchronous programming in R using package future and promises, mainly for the purpose of kick-starting developing a scalable shiny application. Indeed, all 3 packages come with very well-structured official tutorials already. This notebook serves more as a minimally sufficient one-stop reference for developers to quickly hands-on on the topic and get ready for the actual application development.

1 Using future

library(future)
print(installed.packages()["future", "Version"])

[1] "1.14.0"

The package future (Bengtsson (2019)) allows a single-threaded R process to launch a task in another thread to achieve asynchronous programming. It is the main entry point for a task to be asynchronous.

To make an expression a Future, we simply wrap it with a future function call:

f <- future({
  # Supposedly anything expensive to compute here.
  1 + 1
})

Then we can collect the result of a Future:

print(value(f))

[1] 2

In order for a Future to be useful, we need to configure the execution plan. That is, how we’d like to run a given task in a separate process.

1.1 Execution Plan

1.1.1 Sequential Plan

The default execution plan for future is sequential:

plan()  # Check the current execution plan.

sequential:
- args: function (expr, envir = parent.frame(), substitute = TRUE, lazy = FALSE, seed = NULL, globals = TRUE, local = TRUE, earlySignal = FALSE, label = NULL, ...)
- tweaked: FALSE
- call: NULL

It means that all function calls are executed sequentially in the current R session. So the following future expression will block the session for 1 second:¹

plan(sequential)  # Set explicitly to a sequential plan.

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})

str(f)  # A SequentialFuture object.

Classes 'SequentialFuture', 'UniprocessFuture', 'Future', 'environment' <environment: 0x3dbb7f8> 

=== Execution Time: 1.01056122779846 Sec ===

The return value from a future is refered to as a promise in the asynchronous programming terminology. As we discussed already, to access the bounded value from a future expression (i.e., a promise) we use value function:

print(value(f))

[1] "I'm from the future!"

Or a syntactic sugar can be used:

v %<-% {
  Sys.sleep(1)
  "I'm from the future!"
}

print(v)

[1] "I'm from the future!"

=== Execution Time: 1.01539325714111 Sec ===

1.1.2 Multi-Processing Plan

A SequentialFuture is of course not very useful at all. The advantage of asynchgronous programming is to have multi-processing execution plan such that our function calls become non-blocking (asynchronous). To enable such setting:

plan(multiprocess)
plan()

multiprocess:
- args: function (expr, envir = parent.frame(), substitute = TRUE, lazy = FALSE, seed = NULL, globals = TRUE, workers = availableCores(), gc = FALSE, earlySignal = FALSE, label = NULL, ...)
- tweaked: FALSE
- call: plan(multiprocess)

Depending on the running platform, the actual execution plan will be either multicore using fork (for Linux and macOS) or multisession using multiple R sessions (for Windows). So multiprocess is just a platform-independent convenience plan for multi-threading.

It is worth noting that the behavior is technically NOT the same between multicore and multisession, especially when it comes to variable scoping, which we will discussed latter.

Under a multiprocess plan now our future expression becomes non-blocking:

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})  # The call will return immediately.

print(class(f))  # A MultiprocessFuture object.

[1] "MulticoreFuture"    "MultiprocessFuture" "Future"            
[4] "environment"       

=== Execution Time: 0.0269737243652344 Sec ===

Again we can use value to access the bounded value. But keep in mind that value is a blocking call. If the promised value is not yet resolved, it will wait until resolved:

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})  # The call will return immediately.

print(value(f))  # This is blocking since `value` is waiting for `f` to be resolved.

[1] "I'm from the future!"

=== Execution Time: 1.03893327713013 Sec ===

One interesting fact about the pipe sugar %<-% in a multiprocess plan is that it is still non-blocking even if it seems to contain the value call implicitly:

v %<-% {
  Sys.sleep(1)
  "I'm from the future!"
}

=== Execution Time: 0.0218565464019775 Sec ===

This is because %<-% is lazy in its implicit value call. Only when the value is actually used will the call to value be executed.²

To see this in action:

v %<-% {
  Sys.sleep(1)
  "I'm from the future!"
}  # Non-blocking.

print(v)  # Blocking.

[1] "I'm from the future!"

=== Execution Time: 1.05203032493591 Sec ===

When using the future pipe operator, one can still access the promise object without value. This is done by futureOf:

v %<-% {
  Sys.sleep(1)
  "I'm from the future!"
}  # Non-blocking.

print(class(futureOf(v)))  # A Future object.

[1] "MulticoreFuture"    "MultiprocessFuture" "Future"            
[4] "environment"       

1.1.3 Nested Futures

future expression can be nested and the corresponding plan can be configured accordingly. By default the second-layer future will be sequential no matter what plan the first layer uses.

To specify the so-called future topology, for example a sequential for the first layer and a multiprocess for the second layer, we can give a list to the plan call:

plan(list(sequential, multiprocess))

1.2 Scope of a future

The scope of a future expression can be tricky. In this section we will spend some time experimenting a bit on its scoping nature.

1.2.1 Global Variables

Since future relies on a separate R process to handle asynchronous calls, how does it deal with global variable in the master process? By default future will automatically scan the given expression and determine which variable to copy for the separate thread to access. This is controlled by the following call:

future(expr, envir=parent.frame(), globals=TRUE)

Setting globals=TRUE enables the auto-search feature, which in theory can result in error by overlooking variables required by expr. The search includes the envir (by default it is the calling environment) AND its parents until found (a.k.a. lexical scoping):

x <- 42
y <- 64  # Search up to here for y.
k <- function() {
  x <- 0  # Search up to here for x.
  g <- function() {
    f <- future({
      list(x=x, y=y)
    }, globals=TRUE)
    value(f)
  }
  g()
}
k()

$x
[1] 0

$y
[1] 64

The search behavior can be controlled explicitly by setting globals to a variable name vector and specify the enviroment to search for:

x <- 42  # Exposed to future explicitly.
y <- 64  # Exposed to future explicitly.
k <- function() {
  x <- 0  # This is ignored since it is not in the specified search environment.
  g <- function() {
    f <- future({
      list(x=x, y=y)
    }, globals=c("x", "y"), envir=.GlobalEnv)
    value(f)
  }
  g()
}
k()

$x
[1] 42

$y
[1] 64

1.2.2 Inconsistency between Plans

When it comes to scoping, future is not very consistent across different execution plan. In this section we discuss several notable examples that can be confusing.

Auto-Search Scope Inconsistency

To see how the global is NOT copied correctly by the auto-search feature, consider the following example:

plan(multisession)
z <- 10
f <- future({
  get("z", envir=environment(), inherits=TRUE)  # Error.
})
tryCatch(value(f), error=function(e) print(e))

<simpleError in get("z", envir = environment(), inherits = TRUE): object 'z' not found>

A contrasting example is to make the expression easy for the future to understand that we need the variable:

plan(multisession)
z <- 10
f <- future({
  get("z", envir=environment(), inherits=TRUE)  # This now is working even BEFORE the eval to z.
  z  # This makes the auto-search work and export the variable from global.
})
tryCatch(value(f), error=function(e) print(e))

[1] 10

But the same function call works perfectly under either multicore or sequential:

plan(sequential)
z <- 10
f <- future({
  get("z", envir=environment(), inherits=TRUE)
})
tryCatch(value(f), error=function(e) print(e))

[1] 10

plan(multicore)  # This is not supported in Windows and will fallback to a sequential plan.
z <- 10
f <- future({
  get("z", envir=environment(), inherits=TRUE)
})
tryCatch(value(f), error=function(e) print(e))

[1] 10

Global Copy Inconsistency

Remember that all globals are copied to the future block, which means in-place change will not reflect on the original object even if the object is mutable:

plan(multisession)
e <- new.env()  # A global mutable.
e$x <- 0

f <- future({
  e$x <- 42  # `e` is automatically searched and accessable, but changed in a copy.
})
invisible(value(f))

ls.str(e)  # The original copy is intact.

x :  num 0

plan(multicore)  # This is not supported in Windows and will fallback to a sequential plan.
e <- new.env()  # A global mutable.
e$x <- 0

f <- future({
  e$x <- 42
})
invisible(value(f))

ls.str(e)

x :  num 0

However, the above fact only holds for a multiprocess plan. Under sequential plan the global original is indeed changed in-place:

plan(sequential)
e <- new.env()  # A global mutable.
e$x <- 0

f <- future({
  e$x <- 42
})
invisible(value(f))

ls.str(e)  # The original copy has been modified!

x :  num 42

Super Assignment Inconsistency

Super assignment (<<-) behaves differently by execution plan. For a sequential plan:

plan(sequential)
x <- 0
f <- future({
  x <<- 42
  y <- 0
})
invisible(value(f))
x

[1] 42

But for a multisession plan:

plan(multisession)
x <- 0
f <- future({
  x <<- 42
  y <- 0
})
invisible(value(f))
x

[1] 0

Or for a multicore plan:

plan(multicore)  # This is not supported in Windows and will fallback to a sequential plan.
x <- 0
f <- future({
  x <<- 42
  y <- 0
})
invisible(value(f))
x

[1] 0

1.3 Non-Blocking Resolution Check

Since now a future expression may be or may not be resolved already, and a value call to the promised value will block if the value is not yet resolved, it becomes important to have the ability that allows us to check whether a promise is already resolved, without being blocked.

This is exactly what the function resolved is doing:

plan(multiprocess)

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})

resolved(f)  # Return immediately and `f` is not yet resolved.

[1] FALSE

Sys.sleep(1)

resolved(f)  # Return immediately and `f` should be already resolved.

[1] TRUE

Now it is theoretically possible to create one non-blocking future thread to wait for another non-blocking future thread:

# Educational purpose only.
# This pattern may not be very useful in practice.
f1 <- future({
  Sys.sleep(3)
  "I'm from the future!"
})

f2 <- future({
  while ( TRUE ) {
    if ( resolved(f) ) {
      value(f)
      break
    }
  }
})

Whether such pattern is useful at all depends on the actual use case. But in the next section we will learn how to control even more on a promise to arrive at a full-fletched asynchronous programming framework in R.

1.4 Error Propogation

Error from a future expression will propogate to the actual bounded value but not the Future object itself.

fe <- future(stop("Error from the future!"))
print(class(fe))  # No exception before access the value.

[1] "MulticoreFuture"    "MultiprocessFuture" "Future"            
[4] "environment"       

tryCatch(value(fe), error=function(e) print(e))

<simpleError in eval(quote(stop("Error from the future!")), new.env()): Error from the future!>

For explicit error handling it is better to resort to using promises.

2 Using promises

Using promises (Cheng (2018)) is one big step ahead of future, enabling even more flexibility on asynchronous programming. But it also drastically changes how we should write our code–specifically, in a promise-style.

library(promises)
print(installed.packages()["promises", "Version"])

[1] "1.0.1"

# Fix the plan to multiprocess for promises.
plan(multiprocess)

2.1 From future to promises

A future expression return a promise. Such promise can be converted explicitly to a promise object using then:

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})

p <- then(f, onFulfilled=function(v) v)
print(class(p))  # A promise object.

[1] "promise"

str(p)

List of 3
 $ then   :function (onFulfilled = NULL, onRejected = NULL)  
 $ catch  :function (onRejected)  
 $ finally:function (onFinally)  
 - attr(*, "class")= chr "promise"
 - attr(*, "promise_impl")=Classes 'Promise', 'R6' <Promise [pending]> 

Unlike the original return value from a future, a promise, even under multiprocess, is always resolved:

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})

p <- then(f, onFulfilled=function(v) v)
resolved(p)  # This is always, immediately, TRUE.

[1] TRUE

=== Execution Time: 0.0291604995727539 Sec ===

Put it differently, resolved only works (or is only meaningful) for a Future object but not for a promise object.

We loosely call the return value of a future as a promise as well since this is the common wording used in asynchronous programming. But here we explicitly refer to the object class promise implemented by the R package promises. When we refer to the general concept of promise, we will avoid using the syntax highlight for code for readers’ ease of distinguishing the difference.

A Future is not, but can be a promise. Indeed, when a Future object is fed to a then function, it will be immediately converted to a promise-like object by attribute assignment. This can be easily seen in the following code:

f <- future({
  Sys.sleep(1)
  "I'm from the future!"
})
str(f)  # A Future, before called with a then function.

Classes 'MulticoreFuture', 'MultiprocessFuture', 'Future', 'environment' <environment: 0x4c28a60> 

p <- then(f, onFulfilled=function(v) v)
str(f)  # The same Future, after called with then and has been attached promise attributes.

Classes 'MulticoreFuture', 'MultiprocessFuture', 'Future', 'environment' <environment: 0x4c28a60> 
 - attr(*, "converted_promise")=List of 3
  ..$ then   :function (onFulfilled = NULL, onRejected = NULL)  
  ..$ catch  :function (onRejected)  
  ..$ finally:function (onFinally)  
  ..- attr(*, "class")= chr "promise"
  ..- attr(*, "promise_impl")=Classes 'Promise', 'R6' <Promise [pending]> 

A Future is either resolved or not resolved. While a promise is either pending, fulfilled, or rejected (due to error).

f <- future({
  Sys.sleep(.1)
  "I'm from the future!"
})

p <- then(f, onFulfilled=function(v) v)
print(p)  # Still pending.

<Promise [pending]>

Due to the specialty of the notebook rendering environment, a promise is always shown as pending no matter how long we wait for. For an actual R session the result will be something like:

<Promise [fulfilled: character]>

2.2 No Way Back Once promised

How do we extract the bounded value from a promise, like what we do with a Future by using the value function?

It turns out that, we cannot.

Indeed, a promise is ALWAYS a promise. There is simply no way back once we pipe our task into a promise. The design philosophy is that we never know when the promised value will be available, and hence the return value from a promise must always be a promise.

To process the return value from a promise, we simply chain it with another promise:

f <- future(1 + 1)
p1 <- then(f, onFulfilled=function(v) v + 1)
p2 <- then(p1, onFulfilled=function(v) v + 2)

2.3 Error Handling

When a then job failed at either its onFulfilled or onRejected task, it returns a rejected promise with the corresponding error type. Rejection does NOT propagate, though. When a then job processes a rejected promise, it will go to the onRejected branch (by default simply propagate the error if not specified) and if that task is done without error, the resulting value is a fulfilled promise.

The following example illustrates the above idea:

plan(sequential)

f <- future(1 + 1)

# The first promise will fail at its onFuilfilled task (on purpose).
p1 <- then(
  f,  # This is a fulfilled promise.
  onFulfilled=function(v) stop(v),  # Hence we follow this branch. (And then will fail.)
  onRejected=function(e) e
)

# The second promise will go to the onRejected branch since the first promise failed.
p2 <- then(
  p1,  # This is failed on its own onFulfilled task.
  onFulfilled=function(v) print(sprintf("From onFulfilled: %s", v)),
  onRejected=function(e) print(sprintf("From onRejected: %s", e))  # Hence we go here.
)



p1

<Promise [pending]>

p2

<Promise [pending]>

The first promise object p1 will have a value of

<Promise [rejected: simpleError]>

after running its own onFulfilled task. But the second promise object p2 instead will have a value of

<Promise [fulfilled: character]>

after running its own onRejected task and also print the message

[1] "From onRejected: Error in onFulfilled(value): 2"

Then same logic applies to Future object (i.e., converted promise):

# Error at the future expression.
f <- future(stop("Something just went wrong at the very beginning."))

# The first promise will go for the onRejected task, and return successfully.
p1 <- then(
  f,
  onFulfilled=function(v) {
    print("From onFulfilled.")
    v
  },
  onRejected=function(e) {
    print("From onRejected")
    e
  }
)

# Since the onRejected task is successfully excecuted, p1 becomes a fulfilled promise.
# p2 then will go for the onFulfilled task.
p2 <- then(
  p1,
  onFulfilled=function(v) {
    print("From onFulfilled.")
    v
  },
  onRejected=function(e) {
    print("From onRejected")
    e
  }
)

Sys.sleep(3)  # Wait longer to allow for potential backend overhead.

p1
p2

This time p1 will print

[1] "From onRejected"

with a value of

<Promise [fulfilled: simpleError]>

And p2 will print

[1] "From onFulfilled."

with a value of

<Promise [fulfilled: simpleError]>

Both promises are fulfilled (i.e., no further error at the branch) on their corresponding task, whether it is a branch task of onFulfilled or onRejected.

2.4 Syntactic Sugars

There are several syntactic sugars available when using the then API.

Lambda Expression

f <- future({
  Sys.sleep(.1)
  "I'm from the future!"
})

p <- then(f, ~{
  .  # Return the Future value as-is.
})

Promise-Aware Pipe

f <- future({
  Sys.sleep(.1)
  "I'm from the future!"
}) %...>% {
  cat(.)  # This won't output to the notebook code chunk.
}

The pipe %...>% only supports onFulfilled function. In such case the onRejected task is simply an error propagation.

For onRejected function one can use %...!% instead. And the onFulfilled task is a simple identity.

For a complete custom branch handling one should always use the then API explicitly.

3 Async with Reactive Programming

Reactive programming in R is introduced by the well-known shiny package (Chang et al. (2018)) which facilitates the ease of web application development purely using R, usually for data-driven dashboard building purpose.

A huge limitation about such web app is that R is a single-threaded process. Reactive programming itself does not provide asynchrony. In order for a web app to be scalable for a multi-session use case, reactive programming must combine with asynchronous programming.

The good news is that shiny has come fully support for future and promises under its reactive programming framework after its major release of v1.1.

3.1 A Quick Recap: Reactive Programming

library(shiny)
print(installed.packages()["shiny", "Version"])

[1] "1.3.2"

Broadly speaking there are two types of handler in reactive programmning for shiny: value handler and event handler.

3.1.1 Value Handler

To register an object whose value changes reactively (according to, say, user input from a web app portal), we can use reactiveVal or reactiveValues. reactiveVal is designed for a single value (single object) while reactiveValues is designed for a list of values (multiple objects).

To initialize a reactive value without a default (NULL as default):

r <- reactiveVal(value=NULL)

To update the value we simply call:

r(x)  # `x` is the new value assigned reactively to `r`.

And to retrieve the value we can call without argument:

r()

In general reactiveVal is only callable under a reactive context (all the render* function such as renderText, renderPrint). To test it interactively, we can also use isolate to directly retrieve the value non-reactively:

r <- reactiveVal(value=NULL)
isolate(r())

NULL

r(42)
isolate(r())

[1] 42

To initialize a list of reactive values:

rl <- reactiveValues()

reactiveValues can be directly update by using the list assignment syntax:

rl$a <- 1
rl$b <- 1

To test it:

rl <- reactiveValues()
rl$a <- 1
rl$b <- 1
isolate(rl$a + rl$b)

[1] 2

We can create reactive functions which depend on reactive values. The function will re-execute everytime any of the dependent reactive values change.

rl <- reactiveValues(a=1, b=2)

rf <- reactive({
  rl$a + rl$b
})

isolate(rf())

[1] 3

Or we can create reactive observer functions similar to reactive but only for its side-effects (i.e., no return value):

rl <- reactiveValues(a=1, b=2)

# Nothing will be printed since the value doesn't change without a reactive context.
observe({
  print(rl$a + rl$b)
})

3.1.2 Event Handler

We can also create handlers that explicitly respond to UI component or other reactive events defined by a reactive function.

For example, we can react to a user button click by a eventReactive function:

r <- eventReactive(input$run, {expr})  # Assume `run` is the id for a button in the UI code.

Not just UI component, eventReactive can react to anything reactive:

r <- reactiveVal()
r(123)
er <- eventReactive(r(), 456)
isolate(er())

[1] 456

We can do the same but only for side-effects by using observeEvent:

observeEvent(input$run, {expr})  # Assume `run` is the id for a button in the UI code.

3.1.3 A Minimum App Template

For completeness, the following code chunk provides a minimum single-file working shiny app with a user input button and a output text window.

# Define frontend code.
ui <- fluidPage(

  titlePanel("Test Reactive Programming in R"),
  actionButton("do", "Do something."),
  uiOutput("out")

)

# Define backend code.
server <- function(input, output, session) {

  observeEvent(input$do, {
    output$out <- renderText({
      "Something."
    })
  })

}

# Launch the app.
shinyApp(ui=ui, server=server, options=list(port=8787))

3.2 Promise-Aware Reactive Expression

All reactive functions provided by shiny are both future and promise aware. It means that reactive expressions can be a Future:

rl <- reactiveValues(a=1, b=2)

rf <- reactive({
  future(rl$a + rl$b)
})

isolate(value(rf()))

[1] 3

or a promise:

rl <- reactiveValues(a=1, b=2)

rf <- reactive({
  future(rl$a + rl$b) %...>% {
    .
  }
})

isolate(str(rf()))

List of 3
 $ then   :function (onFulfilled = NULL, onRejected = NULL)  
 $ catch  :function (onRejected)  
 $ finally:function (onFinally)  
 - attr(*, "class")= chr "promise"
 - attr(*, "promise_impl")=Classes 'Promise', 'R6' <Promise [pending]> 

3.3 Experimental App on Async

To see how one session is blocking another in a single-threaded app, here is a minimum app for pure illustration purpose:

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


exec_plan <- commandArgs(trailingOnly=TRUE)[1]
if ( is.na(exec_plan) ) exec_plan <- "sequential"

plan(exec_plan)


# Define frontend code.
ui <- fluidPage(

  titlePanel("Async Shiny App"),
  textOutput("time"),
  actionButton("do", "Do some heavy works."),
  verbatimTextOutput("out")

)


do_heavy_work <- function() {
  st <- Sys.time()

  Sys.sleep(5) # Or anything expensive here.

  et <- Sys.time()
  list(st=st, et=et)
}


# Define backend code.
server <- function(input, output, session) {

  output$time <- renderText({
    invalidateLater(1000, session)
    paste("The current time is", Sys.time())
  })

  observeEvent(input$do, {
    st <- Sys.time()  # This only record when the app starts process the input but NOT when the user hit the button.
    output$out <- renderText({
      future(do_heavy_work()) %...>% {
        paste(
          "Heavy work done!",
          sprintf("Started at %s", st),
          sprintf("Ended at %s", .$et),
          sprintf("Time used: %s", .$et - st),
          sep="\n"
        )
      }
    })
  })

}

# Launch the app.
shinyApp(ui=ui, server=server, options=list(port=8787))

To play around with the actual app, run:

Rscript app.R

for synchronous mode.

The app will be listening on 127.0.0.1:8787. Open more than 1 tab and click the do button for all the opening tabs. For a sequential app we shall observe the timer stop jumping when there is any other session (browser tab) still working and hence blocking. The more other sessions are still working, the more times you will see the timer jump and stop. The start time recorded in the output text is always roughly 5 seconds before it ended. However the start time is NOT the time we hit the do button.

Now close the app. For asynchronous mode run:

Rscript app.R multiprocess

and investigate again.

The timer will still stop jumping when we hit the do button. But this time it didn’t stop more than once. It only stop for its own session.

And the timer is stopping because we didn’t implement the timer asynchronously. In the experimental app only the do_heavy_work operation is implemented in asynchronous mode.

In general we will only make those heavy tasks asynchronous instead of trying to make everything asynchronous.

4 Closing Remarks

Up to now we’ve equiped with the basic knowledge to start developing asynchronous web application using R. There are far more features then what we just discussed here for all these packages: future, promises, and shiny. But we will leave it for exploration and retain the scope of this notebook at a entry-level.

5 References

Bengtsson, Henrik. 2019. Future: Unified Parallel and Distributed Processing in R for Everyone. https://CRAN.R-project.org/package=future.

Chang, Winston, Joe Cheng, JJ Allaire, Yihui Xie, and Jonathan McPherson. 2018. Shiny: Web Application Framework for R. https://CRAN.R-project.org/package=shiny.

Cheng, Joe. 2018. Promises: Abstractions for Promise-Based Asynchronous Programming. https://CRAN.R-project.org/package=promises.

---

1. In this notebook we use a custom knit_hooks to time the relevant code chunk to showcase the blocking/non-blocking execution time.↩

2. Readers should not confuse the laziness here with the lazy function argument in a future call. By setting lazy=FALSE (which is the default) in a future call it enables the expression to start execution immediately, otherwise not.↩