Defaults: re-score every 10 generations, drop bad hosts, wait 8 h for cores

NAMESPACE

@@ -19,6 +19,7 @@ export(mirrorTerraOptions)
 export(monitorCluster)
 export(numActiveThreads)
 export(plan_psock_min)
+export(plotMachine)
 export(ramUsageGB)
 export(readTransferredObjects)
 export(releaseCapacity)
@@ -34,6 +35,7 @@ export(shippedObjectsDigest)
 export(summaryOutputFolder)
 export(tableFiles)
 export(termsInDEoptim)
+export(testMachine)
 export(transferDirName)
 export(verifyClusterHosts)
 export(visualizeDE)

NEWS.md

@@ -1,5 +1,12 @@
 # clusters (development version)
 
+* Three options now default to the values FireSense production has always set (defined once, in `.optionDefaults`);
+  an explicit setting, including 0, still wins. `clusters.deoptimRescoreEvery` is 10: `DEoptimIterative()` re-scores the
+  surviving population every 10 generations (was never), which converged in about half the generations and reached
+  the best scores without lucky draws. `clusters.onBadHost` is `"drop"`: a host failing verification is dropped with a
+  message (was `"stop"`, which killed the job). `clusters.waitForCores` is 8 hours: a build waits for cores held by
+  other live builds (was 0, which failed at once).
+
 * New `reserveCapacity()`, `releaseCapacity()` and `withCapacity()` book cores and memory for a job that is
   not a cluster -- a FireSense prediction, one SpaDES simulation of about 38 GB and 1-2 cores in a tmux pane of
   `SpaDES.project::experimentTmux()` -- in the same ledger and under the same per-host caps as a cluster

R/DEoptim.R

@@ -197,7 +197,7 @@
 #' Runs [DEoptim::DEoptim()] a few generations at a time (`iterStep` in `...`, default 1),
 #' starting each call from the previous call's population, and caches every chunk with
 #' [reproducible::Cache()], so a fit that is interrupted resumes from its last finished chunk.
-#' Between chunks it can re-score the surviving population (`options(clusters.deoptimRescoreEvery)`),
+#' Between chunks it can re-score the surviving population (`options(clusters.deoptimRescoreEvery)`, default 10 generations; 0 never),
 #' draws progress figures with `SpaDES.core::Plots()` when `figurePath` is a path, and stops early
 #' once the population has stopped improving (see `.deoptimPopulationConverged()`).
 #'
@@ -249,9 +249,8 @@ DEoptimIterative <- function(fn, lower, upper, control, ...,
   objFunArgs$iterStep <- NULL
   ## Can this objective function receive the adaptive prune bound? Asked once, not once per chunk.
   sendPruneAbove <- .fnTakesPruneAbove(fn)
-  ## Re-score the surviving population every `rescoreEvery` generations (0: never); see .rescoreUpdate
-  rescoreEvery <- getOption("clusters.deoptimRescoreEvery")
-  rescoreEvery <- if (is.null(rescoreEvery)) 0L else max(0L, as.integer(rescoreEvery))
+  ## Re-score the surviving population every `rescoreEvery` generations (default 10; 0: never); see .rescoreUpdate
+  rescoreEvery <- max(0L, as.integer(.clustersOption("deoptimRescoreEvery")))
   rescoreArgs <- as.list(getOption("clusters.deoptimRescoreArgs", list()))
   rescoreStats <- NULL
   ## Move the workers to where a new build would put them every `rebalanceEvery` generations (0: never);

R/allocateWhenAvailable.R

@@ -38,7 +38,7 @@
 #' @keywords internal
 .allocateWhenAvailable <- function(probe, total, beta = 0.75,
                                    minFraction = getOption("clusters.minWorkersFraction", 1),
-                                   waitSeconds = getOption("clusters.waitForCores", 0),
+                                   waitSeconds = .clustersOption("waitForCores"),
                                    interval = 60, sleep = Sys.sleep, now = Sys.time, book = NULL,
                                    allocate = NULL, what = "workers") {
   stopifnot(is.function(probe), is.numeric(minFraction), length(minFraction) == 1L,

R/clusters.R

@@ -27,11 +27,11 @@
 #' print(st)
 #' getHostCombination(outs, Npops = 100)
 #'
+#' }
 #' @importFrom data.table as.data.table
 #' @importFrom parallelly makeClusterPSOCK
 #' @importFrom parallel detectCores stopCluster clusterApply
 #' @export
-#' }
 testMachine <- function(NcoresMax = parallel::detectCores(), N = 100, thinning = 5) {
   Ncores <- c(1, seq(floor(NcoresMax/thinning)) * thinning)
   coreTimes <- list(); systemTimes <- list(); estTotTime <- list(); optimisticTotTime <- list()
@@ -73,8 +73,8 @@ testMachine <- function(NcoresMax = parallel::detectCores(), N = 100, thinning =
 #' Map(out = outs, nam = names(outs), function(out, nam)
 #'     do.call(plotMachine, append(out, list(nam = nam))))
 #'
-#' @export
 #' }
+#' @export
 plotMachine <- function(Ncores, coreTimes, estTotTime, optimisticTotTime, systemTimes,
                         N = 100, nam, detectedCores) {
   ncoresUsed <- "NCores used"

R/estimateCluster.R

@@ -290,7 +290,7 @@ plan_psock_min <- function(
     # Same alignment rule as above: one worker per element of `hosts`.
     verifiedHosts <- verifyClusterHosts(cl_verify, hosts = hosts, pkgs = pkgsTopLevel,
                                         libs = master_libs,
-                                        action = getOption("clusters.onBadHost", "stop"))
+                                        action = .clustersOption("onBadHost"))
 
     # parallel::stopCluster(cl_probe)
     # Sys.sleep(2) 
@@ -344,7 +344,7 @@ plan_psock_min <- function(
 
   ## 2) Run the capacity queries (.probeNodes()) everywhere (same code on each worker).
   ##    Repeated until the whole population is free, up to option
-  ##    clusters.waitForCores seconds (default 0: no waiting). A job that has just
+  ##    clusters.waitForCores seconds (default 8 hours; 0: no waiting). A job that has just
   ##    spent hours preparing its inputs should queue for cores, not die
   ##    (2026-09-08: the sixth concurrent build found none free), and it must not
   ##    start with a fraction of its workers either (2026-09-15: fits ran with 5, 2
@@ -366,7 +366,7 @@ plan_psock_min <- function(
     function(alloc) resvId <<- reserveCores(alloc, memPerWorkerGB = memPerWorker())
   alloc_df <- .allocateWhenAvailable(probeCapacity, total = total, beta = beta,
                                      minFraction = getOption("clusters.minWorkersFraction", 1),
-                                     waitSeconds = getOption("clusters.waitForCores", 0),
+                                     waitSeconds = .clustersOption("waitForCores"),
                                      book = book)
   on.exit(if (!is.null(resvId)) try(releaseCores(id = resvId), silent = TRUE), add = TRUE)
 

R/optionDefaults.R

@@ -0,0 +1,9 @@
+## Defaults of the package options, defined once. Each is read as `getOption("clusters.<name>")`, so a
+## user's explicit setting (including 0) wins. These are the values FireSense production has set.
+.optionDefaults <- list(
+  deoptimRescoreEvery = 10L,   # re-score the DEoptim population every k generations; 0 never
+  onBadHost = "drop",          # a host failing verification is dropped; "stop" kills the job
+  waitForCores = 8 * 3600      # seconds to wait for cores held by other live builds; 0 no waiting
+)
+
+.clustersOption <- function(name) getOption(paste0("clusters.", name), .optionDefaults[[name]])

R/reservations.R

@@ -182,7 +182,6 @@ releaseCores <- function(id = NULL, pid = Sys.getpid(),
 #'
 #' @param nodes A data.frame with `host` and `free_est`, as built by
 #'   [plan_psock_min()] from its probe.
-#' @inheritParams reservations
 #' @param path Path to the reservations ledger; defaults to `getOption("clusters.reservationsPath")`.
 #' @param loadWindowMinutes The averaging window, in minutes, of the load average
 #'   `free_est` was measured from (5 for `parallelly::freeCores(memory = "5min")`).

R/reserveCapacity.R

@@ -32,7 +32,7 @@
 #'   other hosts gives a probe that can see them.
 #' @param load_memory,fraction The load-average window and headroom of [parallelly::freeCores()], as in
 #'   [plan_psock_min()].
-#' @param waitSeconds How long to wait for the job to fit (`options(clusters.waitForCores)`, default 0).
+#' @param waitSeconds How long to wait for the job to fit (`options(clusters.waitForCores)`, default 8 hours).
 #' @param interval Seconds between probes.
 #' @param sleep,now Clock functions, replaceable in tests.
 #' @return `reserveCapacity()` an environment, the token: `id` (the reservation's), `host` (where the job
@@ -52,7 +52,7 @@
 #' @export
 reserveCapacity <- function(cores, memGB, hosts = "localhost", probe = NULL,
                             load_memory = "5min", fraction = 0.9,
-                            waitSeconds = getOption("clusters.waitForCores", 0), interval = 60,
+                            waitSeconds = .clustersOption("waitForCores"), interval = 60,
                             sleep = Sys.sleep, now = Sys.time) {
   stopifnot(is.numeric(cores), length(cores) == 1L, cores > 0, is.numeric(memGB), length(memGB) == 1L,
             memGB >= 0, is.character(hosts), length(hosts) >= 1L)

tests/testthat/setup-userState.R

@@ -4,3 +4,6 @@
 ## rows in the real hostSpeed.rds). Tests that set their own path still override this.
 withr::local_options(clusters.reservationsPath = withr::local_tempdir(.local_envir = teardown_env()),
                      .local_envir = teardown_env())
+## The package waits up to 8 hours for cores by default; a test that expects a shortfall to stop at once
+## must not sleep. test-optionDefaults.R unsets this to check the default.
+withr::local_options(clusters.waitForCores = 0, .local_envir = teardown_env())

tests/testthat/test-DEoptimRescore.R

@@ -85,10 +85,15 @@ test_that("a stopped re-scored run resumes from the cache to the same population
   expect_identical(again$DE[[4]]$member$popval, first$DE[[4]]$member$popval)
 })
 
-test_that("re-scoring is off by default", {
-  seen <- newSeen()
-  withr::local_options(clusters.deoptimRescoreEvery = NULL)
-  out <- runRescored(itermax = 3, every = NULL, cachePath = withr::local_tempdir(), seen = seen)
+test_that("with the option unset the population is re-scored every 10 generations", {
+  out <- runRescored(itermax = 10, every = NULL, cachePath = withr::local_tempdir(), seen = newSeen())
+  expect_length(grep("Re-scored", out$messages), 1L)       # after generation 10
+  expect_null(out$DE[[9]]$member$popvalN)
+  expect_true(all(out$DE[[10]]$member$popvalN >= 2L))
+})
+
+test_that("clusters.deoptimRescoreEvery = 0 turns re-scoring off", {
+  out <- runRescored(itermax = 10, every = 0, cachePath = withr::local_tempdir(), seen = newSeen())
   expect_length(grep("Re-scored", out$messages), 0L)
-  expect_null(out$DE[[3]]$member$popvalN)
+  expect_null(out$DE[[10]]$member$popvalN)
 })

tests/testthat/test-librarySync.R

@@ -69,5 +69,5 @@ test_that("probe-cluster teardown cannot mask the outcome, prefix needs one ship
   ## .stopCluster() stops inside try()
   expect_equal(lengths(regmatches(src, gregexpr("\\.stopCluster\\(cl_probe\\)", src))), 2L)
   expect_false(grepl("length\\(existing\\) <= 1L", src))
-  expect_match(src, "clusters.waitForCores")
+  expect_match(src, "clustersOption\\(\"waitForCores\"\\)")
 })

tests/testthat/test-optionDefaults.R

@@ -0,0 +1,24 @@
+## Defaults of the options FireSense production has always set: with the option unset, the package behaves as
+## if it were set to that value. An explicit setting still wins.
+
+test_that("onBadHost drops a failing host unless the user says otherwise", {
+  withr::local_options(clusters.onBadHost = NULL)
+  expect_identical(clusters:::.clustersOption("onBadHost"), "drop")
+  withr::local_options(clusters.onBadHost = "stop")
+  expect_identical(clusters:::.clustersOption("onBadHost"), "stop")
+})
+
+test_that("waitForCores waits up to 8 hours unless the user says otherwise", {
+  withr::local_options(clusters.waitForCores = NULL)
+  expect_identical(clusters:::.clustersOption("waitForCores"), 8 * 3600)
+  expect_identical(eval(formals(clusters:::.allocateWhenAvailable)$waitSeconds), 8 * 3600)
+  expect_identical(eval(formals(reserveCapacity)$waitSeconds), 8 * 3600)
+  withr::local_options(clusters.waitForCores = 0)
+  expect_identical(clusters:::.clustersOption("waitForCores"), 0)
+  expect_identical(eval(formals(clusters:::.allocateWhenAvailable)$waitSeconds), 0)
+})
+
+test_that("deoptimRescoreEvery re-scores every 10 generations unless the user says otherwise", {
+  withr::local_options(clusters.deoptimRescoreEvery = NULL)
+  expect_identical(clusters:::.clustersOption("deoptimRescoreEvery"), 10L)
+})

tests/testthat/test-waitForFullPopulation.R

@@ -90,6 +90,6 @@ test_that("a population larger than every core on the hosts stops at once instea
 test_that("plan_psock_min allocates through the waiting helper and still honours clusters.waitForCores", {
   src <- paste(deparse(clusters:::plan_psock_min), collapse = "\n")
   expect_match(src, ".allocateWhenAvailable(", fixed = TRUE)
-  expect_match(src, "clusters.waitForCores", fixed = TRUE)
+  expect_match(src, "waitForCores", fixed = TRUE)
   expect_match(src, "clusters.minWorkersFraction", fixed = TRUE)
 })