diff --git a/NAMESPACE b/NAMESPACE index fee1ebc..eca1187 100644 --- a/NAMESPACE +++ b/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) diff --git a/NEWS.md b/NEWS.md index 98171f0..29e3439 100644 --- a/NEWS.md +++ b/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 diff --git a/R/DEoptim.R b/R/DEoptim.R index ec0c73c..b60d91a 100644 --- a/R/DEoptim.R +++ b/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); diff --git a/R/allocateWhenAvailable.R b/R/allocateWhenAvailable.R index 966d605..7128a83 100644 --- a/R/allocateWhenAvailable.R +++ b/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, diff --git a/R/clusters.R b/R/clusters.R index 598c508..25241b1 100644 --- a/R/clusters.R +++ b/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" diff --git a/R/estimateCluster.R b/R/estimateCluster.R index 3981ced..24d2ab8 100644 --- a/R/estimateCluster.R +++ b/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) diff --git a/R/optionDefaults.R b/R/optionDefaults.R new file mode 100644 index 0000000..35f100c --- /dev/null +++ b/R/optionDefaults.R @@ -0,0 +1,9 @@ +## Defaults of the package options, defined once. Each is read as `getOption("clusters.")`, 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]]) diff --git a/R/reservations.R b/R/reservations.R index eaf055f..5197b7b 100644 --- a/R/reservations.R +++ b/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")`). diff --git a/R/reserveCapacity.R b/R/reserveCapacity.R index fb39747..3a3a6d8 100644 --- a/R/reserveCapacity.R +++ b/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) diff --git a/man/DEoptimIterative.Rd b/man/DEoptimIterative.Rd index 3b7b762..0910bc2 100644 --- a/man/DEoptimIterative.Rd +++ b/man/DEoptimIterative.Rd @@ -61,7 +61,7 @@ A list with one \code{\link[DEoptim:DEoptim]{DEoptim::DEoptim()}} result per chu Runs \code{\link[DEoptim:DEoptim]{DEoptim::DEoptim()}} a few generations at a time (\code{iterStep} in \code{...}, default 1), starting each call from the previous call's population, and caches every chunk with \code{\link[reproducible:Cache]{reproducible::Cache()}}, so a fit that is interrupted resumes from its last finished chunk. -Between chunks it can re-score the surviving population (\code{options(clusters.deoptimRescoreEvery)}), +Between chunks it can re-score the surviving population (\code{options(clusters.deoptimRescoreEvery)}, default 10 generations; 0 never), draws progress figures with \code{SpaDES.core::Plots()} when \code{figurePath} is a path, and stops early once the population has stopped improving (see \code{.deoptimPopulationConverged()}). } diff --git a/man/dot-allocateWhenAvailable.Rd b/man/dot-allocateWhenAvailable.Rd index 67a11df..82d0d6f 100644 --- a/man/dot-allocateWhenAvailable.Rd +++ b/man/dot-allocateWhenAvailable.Rd @@ -9,7 +9,7 @@ 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, diff --git a/man/plotMachine.Rd b/man/plotMachine.Rd index 9e8739c..9209152 100644 --- a/man/plotMachine.Rd +++ b/man/plotMachine.Rd @@ -35,3 +35,13 @@ plotMachine( \description{ Plot the outputs from \code{testMachine} } +\examples{ +\dontrun{ +## Needs the `outs` timings from runTests() on real hosts. + +par(mfrow = c(2,length(hosts))) +Map(out = outs, nam = names(outs), function(out, nam) + do.call(plotMachine, append(out, list(nam = nam)))) + +} +} diff --git a/man/reserveCapacity.Rd b/man/reserveCapacity.Rd index 79c211a..67c06e4 100644 --- a/man/reserveCapacity.Rd +++ b/man/reserveCapacity.Rd @@ -13,7 +13,7 @@ reserveCapacity( probe = NULL, load_memory = "5min", fraction = 0.9, - waitSeconds = getOption("clusters.waitForCores", 0), + waitSeconds = .clustersOption("waitForCores"), interval = 60, sleep = Sys.sleep, now = Sys.time @@ -42,7 +42,7 @@ other hosts gives a probe that can see them.} \item{load_memory, fraction}{The load-average window and headroom of \code{\link[parallelly:freeCores]{parallelly::freeCores()}}, as in \code{\link[=plan_psock_min]{plan_psock_min()}}.} -\item{waitSeconds}{How long to wait for the job to fit (\code{options(clusters.waitForCores)}, default 0).} +\item{waitSeconds}{How long to wait for the job to fit (\code{options(clusters.waitForCores)}, default 8 hours).} \item{interval}{Seconds between probes.} diff --git a/man/testMachine.Rd b/man/testMachine.Rd index a8c9329..3c67359 100644 --- a/man/testMachine.Rd +++ b/man/testMachine.Rd @@ -16,3 +16,29 @@ testMachine(NcoresMax = parallel::detectCores(), N = 100, thinning = 5) \description{ Test machines } +\examples{ +\dontrun{ +## Builds an SSH cluster and times it; needs hosts, so not run by checks. +# example code +hosts <- c("97", "106", "184", "189", "213", "217", "220")#, "102") +hosts <- makeHosts(ipbase = "spades", hosts) +hosts <- c(hosts, "132.156.148.105", "localhost") +cl <- parallelly::makeClusterPSOCK(hosts, + rscript = c("nice", "/usr/local/bin/Rscript")) +parallel::clusterExport(cl, c("testMachine", "pkgs")) +parallel::clusterEvalQ(cl, { + if (!require("Require")) install.packages("Require", + repos = c("https://predictiveecology.r-universe.dev", + getOption("repos"))) + Require::Require(pkgs) +}) + +st <- system.time(outs <- parallel::clusterApply(cl, seq_along(cl), function(x) + testMachine(N = 1, NcoresMax = 10, thinning = 3))) + +names(outs) <- hosts +print(st) +getHostCombination(outs, Npops = 100) + +} +} diff --git a/tests/testthat/setup-userState.R b/tests/testthat/setup-userState.R index a40d5ae..21bbc3e 100644 --- a/tests/testthat/setup-userState.R +++ b/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()) diff --git a/tests/testthat/test-DEoptimRescore.R b/tests/testthat/test-DEoptimRescore.R index 7c3ae4f..9f7c1e6 100644 --- a/tests/testthat/test-DEoptimRescore.R +++ b/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) }) diff --git a/tests/testthat/test-librarySync.R b/tests/testthat/test-librarySync.R index 345eea1..6b0c135 100644 --- a/tests/testthat/test-librarySync.R +++ b/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\"\\)") }) diff --git a/tests/testthat/test-optionDefaults.R b/tests/testthat/test-optionDefaults.R new file mode 100644 index 0000000..7ac9e33 --- /dev/null +++ b/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) +}) diff --git a/tests/testthat/test-waitForFullPopulation.R b/tests/testthat/test-waitForFullPopulation.R index e3e3869..f80dd52 100644 --- a/tests/testthat/test-waitForFullPopulation.R +++ b/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) })