R Plumber结合Future Promise：无法将变量赋值至全局环境的问题

当使用future::plan("multisession")运行Plumber API时，未使用future的test1接口能成功将Test1写入全局环境，但使用future异步执行的test2接口无法将Test2写入全局环境。

复现代码 #

主运行脚本 #

### Set the asynchronous coding
library(promises) ; library(future)
future::plan("multisession")

## Plumber app
library(plumber)

pr <- pr("APIs_TestAsync.R")
pr %>% pr_run()

APIs_TestAsync.R脚本 #

#* Test 1
#* @get test1
#* @serializer unboxedJSON
#* @tag TestFuture
function() {
  
  # Create a dataset
  df<-data.frame(A=c("a", "a", "a", "b", "b", "c"), B=c(1,2,3,4,5,6))
  assign("Test1", df, .GlobalEnv)

  return(
    list(SUM=sum(df$B))
    )
  }


#* Test 2
#* @get test2
#* @serializer unboxedJSON
#* @tag TestFuture
function() {
  
  # Create a future promise
  Prom<-future({
    
      # Wait for 10 seconds (needed to test the asynchronous functions)
      Sys.sleep(10)
  
      # Create a dataset
      df<-data.frame(A=c("a", "a", "a", "b", "b", "c"), B=c(1,2,3,4,5,6))
      assign("Test2", df, .GlobalEnv)

      # Store the list in Prom
      return(
        list(SUM=sum(df$B))
      )

  }, seed=T) # Close future promise
  
  return(Prom) # Return Prom to the API

}

问题原因 #

multisession模式下，future会启动独立的R子进程，每个子进程拥有自己独立的全局环境，和主Plumber进程的全局环境完全隔离。test2中的assign("Test2", df, .GlobalEnv)实际上是把变量写入了future子进程的全局环境，而非主进程的全局环境，因此主进程和其他API接口无法访问该变量。

解决方案 #

1. 使用跨进程共享存储（推荐） #

放弃依赖全局环境，改用支持跨进程访问的存储方案，比如Redis、磁盘文件或共享内存包。这里以Redis为例：

步骤1：安装依赖包

install.packages("redux")

步骤2：修改test2接口存入数据

#* Test 2
#* @get test2
#* @serializer unboxedJSON
#* @tag TestFuture
function() {
  library(redux)
  r <- redux::hiredis()
  
  Prom<-future({
    Sys.sleep(10)
    df<-data.frame(A=c("a", "a", "a", "b", "b", "c"), B=c(1,2,3,4,5,6))
    # 序列化数据后存入Redis
    r$SET("Test2", serialize(df, connection = NULL))
    list(SUM=sum(df$B))
  }, seed=T)
  
  return(Prom)
}

步骤3：新增接口读取数据

#* Read Test2 from Redis
#* @get read_test2
#* @serializer unboxedJSON
function() {
  library(redux)
  r <- redux::hiredis()
  raw_data <- r$GET("Test2")
  
  if (!is.null(raw_data)) {
    df <- unserialize(raw_data)
    return(df)
  } else {
    return(list(message = "Test2 not found"))
  }
}

2. 利用Plumber内置状态管理 #

Plumber提供了pr$state用于存储主进程的全局状态，通过promise的then回调在主进程中更新状态：

修改test2接口

#* Test 2
#* @get test2
#* @serializer unboxedJSON
#* @tag TestFuture
function(req, res) {
  Prom<-future({
    Sys.sleep(10)
    df<-data.frame(A=c("a", "a", "a", "b", "b", "c"), B=c(1,2,3,4,5,6))
    list(SUM=sum(df$B), data=df)
  }, seed=T)
  
  # 异步任务完成后，回到主进程更新Plumber状态
  Prom <- Prom %>% then(function(result) {
    req$pr$state$Test2 <- result$data
    return(list(SUM=result$SUM))
  })
  
  return(Prom)
}

新增接口读取状态数据

#* Read Test2 from Plumber state
#* @get read_test2_state
#* @serializer unboxedJSON
function(req) {
  if (!is.null(req$pr$state$Test2)) {
    return(req$pr$state$Test2)
  } else {
    return(list(message = "Test2 not found in state"))
  }
}

3. 改用非多会话future计划（不推荐） #

如果必须依赖全局环境，可以改用plan("sequential")（同步执行，失去异步优势）或plan("multicore")（仅支持Unix/Linux/macOS，子进程与主进程有部分环境共享，但可靠性不足）。但这种方式会牺牲异步并行的能力，不建议在生产环境使用。

内容的提问来源于stack exchange，提问作者Victor Cz