R语言future包二次调用异步函数陷入无限等待问题求助

你遇到的问题核心是persistent集群future的资源未正确清理，加上函数内重复创建集群并设置执行计划导致的状态冲突。让我一步步帮你分析和解决：

问题根源分析 #

• persistent = TRUE的副作用：当你使用persistent = TRUE创建cluster future时，worker会保持持久连接，即使调用stopCluster(cl)，底层的socket连接可能没有完全关闭或清理干净。第二次运行函数时，新创建的集群尝试抢占这些残留资源，导致通信阻塞。
• 执行计划（plan）未重置：第一次运行函数时你设置了plan(cluster, workers = cl)，但函数结束后没有重置回默认计划。第二次运行时，future可能还在尝试使用已被销毁的旧集群，导致任务提交后无法得到响应，陷入无限等待。
• 重复创建集群的冗余操作：你的代码每次调用函数都创建新的PSOCK集群，搭配persistent = TRUE完全没有意义——persistent是为了在同一个worker上多次提交任务，而你每次都销毁集群，反而会导致资源泄漏。

修复后的示例代码 #

下面是调整后的版本，解决了上述所有问题：

library(future)
library(parallel)

asynchronousfunction <- function(){
  # 获取可用核心数
  core_num <- availableCores()
  # 创建PSOCK集群
  cl <- parallel::makePSOCKcluster(core_num)
  # 设置集群执行计划，显式指定persistent = FALSE（默认值，此处标注更清晰）
  plan(cluster, workers = cl, persistent = FALSE)
  
  # 第一次提交future任务
  threads <- lapply(1:core_num, function(index){
    future::future({Sys.getpid()}, workers = cl[[index]])
  })
  # 等待所有任务完成
  while(!all(resolved(threads))){
    Sys.sleep(0.1)
  }
  
  # 第二次提交future任务
  threads <- lapply(1:core_num, function(index){
    future::future({Sys.getpid()}, workers = cl[[index]])
  })
  # 等待所有任务完成
  while(!all(resolved(threads))){
    Sys.sleep(0.1)
  }
  
  # 关键清理步骤：先重置执行计划，再停止集群
  plan(sequential)  # 重置回默认的串行执行计划
  stopCluster(cl = cl)
}

asynchronousfunction() # 首次调用正常运行
asynchronousfunction() # 二次调用也能正常执行

关键调整点 #

• 移除persistent = TRUE：因为你每次函数调用都会销毁集群，持久化worker完全没有必要，反而会导致资源残留。
• 函数结束时重置执行计划：用plan(sequential)将future的执行计划重置回默认状态，避免下次运行时尝试使用已销毁的集群。
• 统一使用future::future()：不需要直接调用future::cluster()，通过plan(cluster)后，future()会自动使用集群执行，代码更简洁规范。

额外优化建议 #

如果你确实需要复用worker（比如多次调用函数不想重复创建集群），可以把集群创建和计划设置移到函数外部，这样只初始化一次集群，多次调用函数都复用它：

# 全局初始化集群和执行计划
core_num <- availableCores()
cl <- parallel::makePSOCKcluster(core_num)
plan(cluster, workers = cl, persistent = TRUE)

asynchronousfunction <- function(){
  # 第一次提交任务
  threads <- lapply(1:core_num, function(index){
    future::future({Sys.getpid()}, workers = cl[[index]])
  })
  while(!all(resolved(threads))){
    Sys.sleep(0.1)
  }
  
  # 第二次提交任务
  threads <- lapply(1:core_num, function(index){
    future::future({Sys.getpid()}, workers = cl[[index]])
  })
  while(!all(resolved(threads))){
    Sys.sleep(0.1)
  }
}

asynchronousfunction()
asynchronousfunction()

# 所有任务完成后统一清理
plan(sequential)
stopCluster(cl)

这样既避免了重复创建集群的开销，也能保证多次调用函数不会出现资源冲突的问题。

内容的提问来源于stack exchange，提问作者Harvard Alexander