diff --git a/NEWS.md b/NEWS.md index d2106035..4b3fcd6e 100644 --- a/NEWS.md +++ b/NEWS.md @@ -23,6 +23,12 @@ * Futures failed on R (< 4.0.0). This bug was introduced in **future** 1.68.0 (2025-11-17). + + * When a cluster worker that `plan()` had started was relaunched after + it was canceled, interrupted or had died, `plan()` never shut the + relaunched worker down, because the cluster registry kept the old + node. The worker process and its connection were left running until + garbage collection. # Version 1.76.0 [2026-09-24] diff --git a/R/backend_api-11.ClusterFutureBackend-class.R b/R/backend_api-11.ClusterFutureBackend-class.R index 7d289ff7..95fffa52 100644 --- a/R/backend_api-11.ClusterFutureBackend-class.R +++ b/R/backend_api-11.ClusterFutureBackend-class.R @@ -1231,6 +1231,7 @@ requestNode <- local({ workers[[node_idx]] <- node2 backend[["workers"]] <- workers + clusterRegistry$replaceNode(workers, node_idx, node2) node <- node2 if (debug) { @@ -1609,6 +1610,7 @@ handleInterruptedFuture <- local({ ## Update backend backend[["workers"]] <- workers + clusterRegistry$replaceNode(workers, node_idx, node2) node <- NULL @@ -1743,6 +1745,18 @@ clusterRegistry <- local({ cluster } ## startCluster() + ## When a worker is relaunched, the registry must track the new node, + ## otherwise stopCluster() closes the old, already closed, node and leaves + ## the relaunched worker and its connection behind + replaceNode <- function(workers, idx, node) { + if (is.null(cluster)) return(invisible(FALSE)) + if (!identical(attr(workers, "name", exact = TRUE), attr(cluster, "name", exact = TRUE))) { + return(invisible(FALSE)) + } + cluster[[idx]] <<- node + invisible(TRUE) + } ## replaceNode() + stopCluster <- function(debug = FALSE) { if (debug) { mdebug_push("Stopping existing cluster ...") @@ -1800,6 +1814,7 @@ clusterRegistry <- local({ list( getCluster = getCluster, startCluster = startCluster, + replaceNode = replaceNode, stopCluster = stopCluster ) }) ## clusterRegistry() diff --git a/inst/testme/test-cancel-relaunch-cluster.R b/inst/testme/test-cancel-relaunch-cluster.R new file mode 100644 index 00000000..b62e0100 --- /dev/null +++ b/inst/testme/test-cancel-relaunch-cluster.R @@ -0,0 +1,40 @@ +#' @tags cancel +#' @tags detritus-files +#' @tags detritus-connections +#' @tags multisession + +library(future) + +message("Relaunched cluster workers are shut down with the plan ...") + +covr_testing <- ("covr" %in% loadedNamespaces()) + +## Terminated workers can leave truncated coverage traces. +if (!covr_testing) { + n0 <- nrow(showConnections()) + + plan(multisession, workers = I(1)) + + ## Cancel a running future, which terminates its worker + f <- future({ Sys.sleep(30); 42 }) + Sys.sleep(1.0) + f <- cancel(f) + + ## The next future relaunches the terminated worker + f2 <- future(42) + stopifnot(value(f2) == 42) + + ## Keep a reference, so that garbage collection cannot hide the leak + backend <- plan("backend") + + plan(sequential) + + ## The relaunched worker must be closed by plan(), not by garbage collection + n <- nrow(showConnections()) + message("Connections: before = ", n0, ", after = ", n) + stopifnot(n <= n0) + rm(list = "backend") + gc() +} + +message("Relaunched cluster workers are shut down with the plan ... done") diff --git a/tests/test-cancel-relaunch-cluster.R b/tests/test-cancel-relaunch-cluster.R new file mode 100644 index 00000000..2d530ad3 --- /dev/null +++ b/tests/test-cancel-relaunch-cluster.R @@ -0,0 +1,4 @@ +#! /usr/bin/env Rscript +## This runs testme test script inst/testme/test-cancel-relaunch-cluster.R +## Don't edit - it was autogenerated by inst/testme/deploy.R +future:::testme("cancel-relaunch-cluster")