README.md

@@ -2,7 +2,7 @@
 
 This is a Shiny example application, meant to demonstrate how to convert synchronous (traditional) Shiny apps to asynchronous ones.
 
-This is the **synchronous** version. The (naive) asynchronous version is [here](https://github.com/jcheng5/cranwhales/tree/async) ([diff](https://github.com/jcheng5/cranwhales/compare/sync...async?diff=split)). The optimized asynchronous version is [here](https://github.com/rstudio/cranwhales/tree/async2) ([diff](https://github.com/jcheng5/cranwhales/compare/async...async2?diff=split)).
+This is the (naive) **asynchronous** version. The synchronous version is [here](https://github.com/jcheng5/cranwhales) ([diff](https://github.com/jcheng5/cranwhales/compare/sync...async?diff=split)). The optimized asynchronous version is [here](https://github.com/rstudio/cranwhales/tree/async2) ([diff](https://github.com/jcheng5/cranwhales/compare/async...async2?diff=split)).
 
 To learn more about asynchronous programming in R and Shiny, see https://rstudio.github.io/promises/.
 

app.R

@@ -8,6 +8,11 @@ library(glue)
 library(lubridate)
 library(gdata)  # for gdata::humanReadable
 
+library(promises)
+library(future)
+# Leave one core for Shiny itself
+plan(multisession(workers = availableCores() - 1))
+
 source("random-names.R")
 source("modules/detail.R")
 
@@ -59,26 +64,26 @@ server <- function(input, output, session) {
     url <- glue("http://cran-logs.rstudio.com/{year}/{date}.csv.gz")
     path <- file.path("data_cache", paste0(date, ".csv.gz"))
 
-    withProgress(value = NULL, {
-
+    p <- Progress$new()
+    p$set(value = NULL, message = "Downloading data...")
+    future({
       # Download to a temporary file path, then rename to the real
       # path when the download is complete. We do this so other
       # processes/sessions don't use partially downloaded files.
       if (!file.exists(path)) {
         tmppath <- paste0(path, "-", Sys.getpid())
-        setProgress(message = "Downloading data...")
         download.file(url, tmppath)
         if (!file.exists(path)) {
           file.rename(tmppath, path)
         } else {
           file.remove(tmppath)
         }
       }
-      
-      setProgress(message = "Parsing data...")
-      read_csv(path, col_types = "Dti---c-ci", progress = FALSE) %>%
-        filter(!is.na(package))
-    })
+    }) %...>%
+      { p$set(message = "Parsing data...") } %...>%
+      { future(read_csv(path, col_types = "Dti---c-ci", progress = FALSE) %>%
+          filter(!is.na(package))) } %>%
+      finally(~p$close())
   })
 
   # Returns a data frame of just the top `input$count` downloaders of the day,
@@ -93,21 +98,24 @@ server <- function(input, output, session) {
       need(input$count > 0, "Too few downloaders"),
       need(input$count <= 25, "Too many downloaders; 25 or fewer please")
     )
-    data() %>%
-      count(ip_id, country) %>%
-      arrange(desc(n)) %>%
-      head(input$count) %>%
+    data() %...>%
+      count(ip_id, country) %...>%
+      arrange(desc(n)) %...>%
+      head(input$count) %...>%
       mutate(ip_name = factor(ip_id, levels = ip_id,
-        labels = glue("{random_name(length(ip_id), input$date)} [{country}]"))) %>%
+        labels = glue("{random_name(length(ip_id), input$date)} [{country}]"))) %...>%
       select(-country)
   })
 
   # data(), filtered down to the downloads that are by the top `input$count`
   # downloaders
   whale_downloads <- reactive({
-    data() %>%
-      inner_join(whales(), "ip_id") %>%
-      select(-n)
+    promise_all(data = data(), whales = whales()) %...>%
+      with({
+        data %>%
+          inner_join(whales, "ip_id") %>%
+          select(-n)
+      })
   })
 
 
@@ -116,77 +124,82 @@ server <- function(input, output, session) {
   #### "All traffic" tab ----------------------------------------
 
   output$total_size <- renderValueBox({
-    data() %>%
-      pull(size) %>%
-      as.numeric() %>%  # Cast from integer to numeric to avoid overflow warning
-      sum() %>%
-      humanReadable() %>%
+    data() %...>%
+      pull(size) %...>%
+      as.numeric() %...>%  # Cast from integer to numeric to avoid overflow warning
+      sum() %...>%
+      humanReadable() %...>%
       valueBox("bandwidth consumed")
   })
 
   output$total_count <- renderValueBox({
-    data() %>%
-      nrow() %>%
-      format(big.mark = ",") %>%
+    data() %...>%
+      nrow() %...>%
+      format(big.mark = ",") %...>%
       valueBox("files downloaded")
   })
 
   output$total_uniques <- renderValueBox({
-    data() %>%
-      pull(package) %>%
-      unique() %>%
-      length() %>%
-      format(big.mark = ",") %>%
+    data() %...>%
+      pull(package) %...>%
+      unique() %...>%
+      length() %...>%
+      format(big.mark = ",") %...>%
       valueBox("unique packages")
   })
 
   output$total_downloaders <- renderValueBox({
-    data() %>%
-      pull(ip_id) %>%
-      unique() %>%
-      length() %>%
-      format(big.mark = ",") %>%
+    data() %...>%
+      pull(ip_id) %...>%
+      unique() %...>%
+      length() %...>%
+      format(big.mark = ",") %...>%
       valueBox("unique downloaders")
   })
 
   output$all_hour <- renderPlot({
-    whale_ip <- whales()$ip_id
-
-    data() %>%
-      mutate(
-        time = hms::trunc_hms(time, 60*60),
-        is_whale = ip_id %in% whale_ip
-      ) %>%
-      count(time, is_whale) %>%
-      ggplot(aes(time, n, fill = is_whale)) +
-      geom_bar(stat = "identity") +
-      scale_fill_manual(values = c("#666666", "#88FF99"),
-        labels = c("no", "yes")) +
-      ylab("Downloads") +
-      xlab("Hour") +
-      scale_y_continuous(labels = scales::comma)
+    promise_all(data = data(), whales = whales()) %...>%
+      with({
+        whale_ip <- whales$ip_id
+
+        data %>%
+          mutate(
+            time = hms::trunc_hms(time, 60*60),
+            is_whale = ip_id %in% whale_ip
+          ) %>%
+          count(time, is_whale) %>%
+          ggplot(aes(time, n, fill = is_whale)) +
+          geom_bar(stat = "identity") +
+          scale_fill_manual(values = c("#666666", "#88FF99"),
+            labels = c("no", "yes")) +
+          ylab("Downloads") +
+          xlab("Hour") +
+          scale_y_continuous(labels = scales::comma)
+      })
   })
 
   #### "Biggest whales" tab -------------------------------------
 
   output$downloaders <- renderPlot({
-    whales() %>%
-      ggplot(aes(ip_name, n)) +
+    whales() %...>% {
+      ggplot(., aes(ip_name, n)) +
       geom_bar(stat = "identity") +
       ylab("Downloads on this day")
+    }
   })
 
   #### "Whales by hour" tab -------------------------------------
 
   output$downloaders_hour <- renderPlot({
-    whale_downloads() %>%
-      mutate(time = hms::trunc_hms(time, 60*60)) %>%
-      count(time, ip_name) %>%
-      ggplot(aes(time, n)) +
-      geom_bar(stat = "identity") +
-      facet_wrap(~ip_name) +
-      ylab("Downloads") +
-      xlab("Hour")
+    whale_downloads() %...>%
+      mutate(time = hms::trunc_hms(time, 60*60)) %...>%
+      count(time, ip_name) %...>% {
+        ggplot(., aes(time, n)) +
+          geom_bar(stat = "identity") +
+          facet_wrap(~ip_name) +
+          ylab("Downloads") +
+          xlab("Hour")
+      }
   })
 
   #### "Detail view" tab ----------------------------------------

modules/detail.R

@@ -16,61 +16,66 @@ detailViewUI <- function(id) {
 
 detailView <- function(input, output, session, whales, whale_downloads) {
 
-  # When whales() changes, update the selectInput with their names
-  observeEvent(try(silent=TRUE, whales()), {
-    choices <- tryCatch(
-      whales()$ip_name,
-      error = function(err) { character(0) }
-    )
-    updateSelectInput(session, "detail_ip_name",
-      choices = choices,
-      selected = if (input$detail_ip_name %in% choices)
-        input$detail_ip_name
-      else
-        character(0))
-    freezeReactiveValue(input, "detail_ip_name")
+  # When whales() changes, update the selectInput with their names.
+  # The gross try(... %>% catch(~{})) is necessary to swallow both sync and
+  # async errors, because either will cause the session to be immediately and
+  # unceremoniously ended. Something to improve in Shiny v1.2.
+  observeEvent(try(whales() %>% catch(~{})), {
+    whales() %...>%
+      pull(ip_name) %...!%
+      { character(0) } %...>%
+      {
+        updateSelectInput(session, "detail_ip_name",
+          choices = .,
+          selected = if (input$detail_ip_name %in% .)
+            input$detail_ip_name
+          else
+            character(0))
+        freezeReactiveValue(input, "detail_ip_name")
+      }
   })
 
   detail_downloads <- reactive({
     req(input$detail_ip_name, nzchar(input$detail_ip_name))
-    whale_downloads() %>%
-      filter(ip_name == input$detail_ip_name) %>%
-      arrange(time) %>%
+    whale_downloads() %...>%
+      filter(ip_name == input$detail_ip_name) %...>%
+      arrange(time) %...>%
       mutate(package = factor(package, levels = rev(unique(package)), ordered = TRUE))
   })
 
   output$detail_size <- renderValueBox({
-    detail_downloads() %>%
-      pull(size) %>%
-      as.numeric() %>%  # Cast from integer to numeric to avoid overflow warning
-      sum() %>% 
-      humanReadable() %>%
+    detail_downloads() %...>%
+      pull(size) %...>%
+      as.numeric() %...>%  # Cast from integer to numeric to avoid overflow warning
+      sum() %...>% 
+      humanReadable() %...>%
       valueBox("bandwidth consumed")
   })
 
   output$detail_count <- renderValueBox({
-    detail_downloads() %>%
-      nrow() %>%
-      format(big.mark = ",") %>%
+    detail_downloads() %...>%
+      nrow() %...>%
+      format(big.mark = ",") %...>%
       valueBox("files downloaded")
   })
 
   output$detail_uniques <- renderValueBox({
-    detail_downloads() %>%
-      pull(package) %>%
-      unique() %>%
-      length() %>%
-      format(big.mark = ",") %>%
+    detail_downloads() %...>%
+      pull(package) %...>%
+      unique() %...>%
+      length() %...>%
+      format(big.mark = ",") %...>%
       valueBox("unique packages")
   })
 
   # Show every single download from the selected downloader
   output$detail <- renderPlot({
 
-    validate(need(input$detail_ip_name, "Select a downloader from the list above"))    
-    pkg <- levels(detail_downloads()$package)
-
-    detail_downloads() %>% {
+    validate(need(input$detail_ip_name, "Select a downloader from the list above"))
+
+    detail_downloads() %...>% {
+      pkg <- levels(.$package)
+
       ggplot(., aes(time, package)) +
         geom_point() +
         scale_x_time(breaks = seq(hms::hms(0,0,0), by = 60*60*3, length.out = 9),
@@ -83,12 +88,12 @@ detailView <- function(input, output, session, whales, whale_downloads) {
   # Show the downloads that are brushed on output$detail
   output$detail_table <- renderDT({
     req(input$detail_brush)
-    detail_downloads() %>%
-      brushedPoints(input$detail_brush) %>%
+    detail_downloads() %...>%
+      brushedPoints(input$detail_brush) %...>%
       mutate(
         time = as.character(time),
         size = humanReadable(size)
-      ) %>%
+      ) %...>%
       select(-ip_id, -ip_name, -country)
   })
 }