Ignore function args when analysing expressions

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: hipercow
 Title: High Performance Computing
-Version: 1.0.39
+Version: 1.0.40
 Authors@R: c(person("Rich", "FitzJohn", role = c("aut", "cre"),
                     email = "rich.fitzjohn@gmail.com"),
              person("Wes", "Hinsley", role = "aut"),

R/parallel.R

@@ -239,11 +239,9 @@ hipercow_parallel_setup <- function(parallel) {
   invisible()
 }
 
+
 hipercow_parallel_setup_future <- function(processes, cores_per_process,
                                            environment) {
-  cli::cli_alert_info(
-    paste0("Creating a future cluster with {processes} process{?es}, ",
-           "each with {cores_per_process} core{?s}"))
 
   # rscript_libs is already set by default to .libPaths() in future::plan
   # but we also want to call set cores and environment.
@@ -276,11 +274,34 @@ hipercow_parallel_setup_parallel <- function(processes, cores_per_process,
 
   # may need some tweaking to find the function.
   # later on we'll also load some packages, source some files
-
   cli::cli_alert_success("Cluster ready to use")
 }
 
 
+hipercow_parallel_teardown <- function(parallel) {
+  if (is.null(parallel$method)) {
+    return()
+  }
+  cli::cli_alert_info("Stopping cluster")
+  switch(parallel$method,
+         future = hipercow_parallel_teardown_future(),
+         parallel = hipercow_parallel_teardown_parallel())
+  cli::cli_alert_success("Cluster stopped")
+}
+
+
+hipercow_parallel_teardown_future <- function() {
+  ## https://github.com/futureverse/future/issues/117
+  future::plan("sequential")
+}
+
+
+hipercow_parallel_teardown_parallel <- function() {
+  cl <- parallel::getDefaultCluster()
+  parallel::stopCluster(cl)
+}
+
+
 ##' @export
 print.hipercow_parallel <- function(x, ...) {
   print_simple_s3(x, "hipercow parallel control (hipercow_parallel)")

R/task-eval.R

@@ -81,6 +81,7 @@ task_eval <- function(id, envir = .GlobalEnv, verbose = FALSE, root = NULL) {
 
     if (!is.null(data$parallel)) {
       hipercow_parallel_setup(data$parallel)
+      withr::defer(hipercow_parallel_teardown(data$parallel))
     }
     check_globals(data$variables$globals, envir, top)
     withr::local_dir(file.path(root$path$root, data$path))

R/util.R

@@ -3,6 +3,11 @@
 }
 
 
+unlist0 <- function(x) {
+  unlist(x, FALSE, FALSE)
+}
+
+
 set_names <- function(x, nms) {
   if (length(nms) == 1 && length(nms) != length(x)) {
     nms <- rep(nms, length(x))
@@ -360,8 +365,16 @@ find_vars <- function(expr, exclude = character()) {
       }
     }
     ret
-  } else {
-    setdiff(all.vars(expr), exclude)
+  } else if (rlang::is_call(expr, "function")) {
+    args <- expr[[2]]
+    body <- expr[[3]]
+    exclude <- c(exclude, names(args))
+    find_vars(body, exclude)
+  } else if (is.recursive(expr)) {
+    found <- unlist0(lapply(expr[-1], find_vars, exclude))
+    setdiff(found %||% character(), exclude)
+  } else if (is.symbol(expr)) {
+    as.character(expr)
   }
 }
 

drivers/windows/DESCRIPTION

@@ -1,6 +1,6 @@
 Package: hipercow.windows
 Title: DIDE HPC Support for Windows
-Version: 1.0.39
+Version: 1.0.40
 Authors@R: c(person("Rich", "FitzJohn", role = c("aut", "cre"),
                     email = "rich.fitzjohn@gmail.com"),
              person("Wes", "Hinsley", role = "aut"),

tests/testthat/test-parallel.R

@@ -85,6 +85,7 @@ test_that("Parallel setup unknown method", {
                "Parallel method 'cactus' unknown.")
 })
 
+
 test_that("Can setup future cluster", {
   mock_plan <- mockery::mock()
   mockery::stub(hipercow_parallel_setup_future,
@@ -146,6 +147,55 @@ test_that("Can set cores and environment variables", {
 })
 
 
+test_that("Can do parallel teardown for future", {
+  mock_parallel_teardown_future <- mockery::mock()
+  mockery::stub(hipercow_parallel_teardown,
+                "hipercow_parallel_teardown_future",
+                mock_parallel_teardown_future)
+
+  suppressMessages(hipercow_parallel_teardown(hipercow_parallel("future")))
+  mockery::expect_called(mock_parallel_teardown_future, 1)
+})
+
+
+test_that("Can do parallel teardown for parallel", {
+  mock_parallel_teardown_parallel <- mockery::mock()
+  mockery::stub(hipercow_parallel_teardown,
+                "hipercow_parallel_teardown_parallel",
+                mock_parallel_teardown_parallel)
+
+  suppressMessages(hipercow_parallel_teardown(hipercow_parallel("parallel")))
+  mockery::expect_called(mock_parallel_teardown_parallel, 1)
+})
+
+
+test_that("future teardown resets plan", {
+  mock_plan <- mockery::mock()
+  mockery::stub(hipercow_parallel_teardown_future, "future::plan",
+                mock_plan)
+  hipercow_parallel_teardown_future()
+  mockery::expect_called(mock_plan, 1)
+  expect_equal(mockery::mock_args(mock_plan)[[1]], list("sequential"))
+})
+
+
+test_that("parallel teardown stops cluster", {
+  e <- new.env()
+  mock_default_cluster <- mockery::mock(e)
+  mock_stop_cluster <- mockery::mock()
+  mockery::stub(hipercow_parallel_teardown_parallel,
+                "parallel::getDefaultCluster",
+                mock_default_cluster)
+  mockery::stub(hipercow_parallel_teardown_parallel,
+                "parallel::stopCluster",
+                mock_stop_cluster)
+  hipercow_parallel_teardown_parallel()
+  mockery::expect_called(mock_default_cluster, 1)
+  mockery::expect_called(mock_stop_cluster, 1)
+  expect_equal(mockery::mock_args(mock_stop_cluster)[[1]], list(e))
+})
+
+
 test_that("can print default parallel control", {
   x <- hipercow_parallel()
   res <- evaluate_promise(print(x))

tests/testthat/test-util.R

@@ -360,6 +360,13 @@ test_that("can find names in multiline expressions with assignments", {
 })
 
 
+test_that("ignore function args when finding variables", {
+  expect_equal(find_vars(quote(f(a, function(x) x + 1))), "a")
+  expect_equal(find_vars(quote(f(a, function(x) x + a / b))), c("a", "b"))
+  expect_equal(find_vars(quote(f(a, function(x, b = 2) x + a / b))), "a")
+})
+
+
 test_that("is_linux false on windows, mac", {
   testthat::skip_on_os(c("linux", "solaris"))
   expect_false(is_linux())

tests/testthat/test-zzz-integration.R

@@ -11,3 +11,43 @@ test_that("can provision a library", {
   suppressMessages(hipercow_provision(root = path_here, show_log = FALSE))
   expect_true(file.exists(file.path(path_there, "hipercow", "lib", "R6")))
 })
+
+
+test_that("Can run task in parallel", {
+  path <- withr::local_tempdir()
+  init_quietly(path)
+  parallel <- hipercow_parallel("parallel")
+  resources <- hipercow_resources(cores = 2)
+
+  id <- withr::with_dir(
+    path,
+    suppressMessages(
+      task_create_expr(
+        parallel::clusterApply(NULL, 1:2, function(x) list(x, Sys.getpid())),
+        parallel = parallel,
+        resources = resources)))
+  dat <- readRDS(path_to_task_file(path, id, "data"))
+  expect_null(dat$variables)
+  expect_equal(dat$parallel$method, "parallel")
+
+  res <- evaluate_promise(
+    withr::with_envvar(
+      c(HIPERCOW_CORES = 2),
+      withr::with_dir(path,  task_eval(id, root = path))))
+  expect_length(res$messages, 4)
+  expect_match(res$messages[[1]],
+               "Creating a parallel cluster with 2 processes")
+  expect_match(res$messages[[2]],
+               "Cluster ready to use")
+  expect_match(res$messages[[3]],
+               "Stopping cluster")
+  expect_match(res$messages[[4]],
+               "Cluster stopped")
+
+  expect_true(res$result)
+
+  res <- task_result(id, root = path)
+  expect_equal(res[[1]][[1]], 1)
+  expect_equal(res[[2]][[1]], 2)
+  expect_true(res[[1]][[2]] != res[[2]][[2]])
+})