Fix #2

DESCRIPTION

@@ -36,9 +36,7 @@ Depends:
   R (>= 4.0.0)
 Imports:
   crew (>= 0.1.0),
-  processx,
   R6,
-  rlang,
   utils
 Suggests:
   knitr (>= 1.30),

NAMESPACE

@@ -8,6 +8,4 @@ importFrom(crew,crew_assert)
 importFrom(crew,crew_class_launcher)
 importFrom(crew,crew_launcher)
 importFrom(crew,crew_random_name)
-importFrom(processx,run)
-importFrom(rlang,abort)
 importFrom(utils,globalVariables)

NEWS.md

@@ -1,6 +1,8 @@
 # crew.cluster 0.0.2.9000 (development)
 
-
+* Submit workers asynchronously (#2).
+* Use `system2()` instead of `processx` to submit workers (#2).
+* Add a `verbose` argument to the SGE launcher to optionally print `system2()` stdout and stderr.
 
 # crew.cluster 0.0.2
 

R/crew_controller_sge.R

@@ -43,6 +43,7 @@ crew_controller_sge <- function(
   sge_cores = NULL,
   sge_gpu = NULL,
   sge_lines = NULL,
+  verbose = FALSE,
   auto_scale = "demand"
 ) {
   router <- crew::crew_router(
@@ -77,7 +78,8 @@ crew_controller_sge <- function(
     sge_memory_gigabytes_limit = sge_memory_gigabytes_limit,
     sge_cores = sge_cores,
     sge_gpu = sge_gpu,
-    sge_lines = sge_lines
+    sge_lines = sge_lines,
+    verbose = verbose
   )
   controller <- crew::crew_controller(
     router = router,

R/crew_launcher_sge.R

@@ -57,6 +57,8 @@
 #'   added to the SGE job script just after the more common flags.
 #'   An example would be `sge_lines = "module load R"` if your SGE cluster
 #'   supports R through an environment module.
+#' @param verbose Logical, whether to see console output and error messages
+#'   when submitting worker.
 crew_launcher_sge <- function(
   name = NULL,
   seconds_launch = 60,
@@ -81,7 +83,8 @@ crew_launcher_sge <- function(
   sge_memory_gigabytes_limit = NULL,
   sge_cores = NULL,
   sge_gpu = NULL,
-  sge_lines = NULL
+  sge_lines = NULL,
+  verbose = FALSE
 ) {
   name <- as.character(name %|||% crew::crew_random_name())
   launcher <- crew_class_launcher_sge$new(
@@ -108,7 +111,8 @@ crew_launcher_sge <- function(
     sge_memory_gigabytes_limit = sge_memory_gigabytes_limit,
     sge_cores = sge_cores,
     sge_gpu = sge_gpu,
-    sge_lines = sge_lines
+    sge_lines = sge_lines,
+    verbose = verbose
   )
   launcher$validate()
   launcher
@@ -146,6 +150,8 @@ crew_class_launcher_sge <- R6::R6Class(
     sge_gpu = NULL,
     #' @field sge_lines See [crew_launcher_sge()].
     sge_lines = NULL,
+    #' @field verbose See [crew_launcher_sge()].
+    verbose = NULL,
     #' @description SGE launcher constructor.
     #' @return an SGE launcher object.
     #' @param name See [crew_launcher_sge()].
@@ -172,6 +178,7 @@ crew_class_launcher_sge <- R6::R6Class(
     #' @param sge_cores See [crew_launcher_sge()].
     #' @param sge_gpu See [crew_launcher_sge()].
     #' @param sge_lines See [crew_launcher_sge()].
+    #' @param verbose See [crew_launcher_sge()].
     initialize = function(
       name = NULL,
       seconds_launch = NULL,
@@ -196,7 +203,8 @@ crew_class_launcher_sge <- R6::R6Class(
       sge_memory_gigabytes_limit = NULL,
       sge_cores = NULL,
       sge_gpu = NULL,
-      sge_lines = NULL
+      sge_lines = NULL,
+      verbose = NULL
     ) {
       super$initialize(
         name = name,
@@ -224,51 +232,7 @@ crew_class_launcher_sge <- R6::R6Class(
       self$sge_cores <- sge_cores
       self$sge_gpu <- sge_gpu
       self$sge_lines <- sge_lines
-    },
-    #' @description Launch a local process worker which will
-    #'   dial into a socket.
-    #' @details The `call` argument is R code that will run to
-    #'   initiate the worker. Together, the `launcher`, `worker`,
-    #'   and `instance` arguments are useful for
-    #'   constructing informative job names.
-    #' @return A handle object to allow the termination of the worker
-    #'   later on.
-    #' @param call Text string with a namespaced call to [crew_worker()]
-    #'   which will run in the worker and accept tasks.
-    #' @param launcher Character of length 1, name of the launcher.
-    #' @param worker Positive integer of length 1, index of the worker.
-    #'   This worker index remains the same even when the current instance
-    #'   of the worker exits and a new instance launches.
-    #'   It is always between 1 and the maximum number of concurrent workers.
-    #' @param instance Character of length 1 to uniquely identify
-    #'   the current instance of the worker.
-    launch_worker = function(call, launcher, worker, instance) {
-      name <- name_job(
-        launcher = launcher,
-        worker = worker,
-        instance = instance
-      )
-      lines <- c(
-        paste("#$ -N", name),
-        self$script(),
-        paste("R -e", shQuote(call))
-      )
-      script <- tempfile()
-      writeLines(text = lines, con = script)
-      on.exit(unlink(script))
-      processx::run(command = self$sge_qsub, args = script)
-      name
-    },
-    #' @description Terminate a local process worker.
-    #' @return `NULL` (invisibly).
-    #' @param handle A process handle object previously
-    #'   returned by `launch_worker()`.
-    terminate_worker = function(handle) {
-      processx::run(
-        command = self$sge_qdel,
-        args = handle,
-        error_on_status = FALSE
-      )
+      self$verbose <- verbose
     },
     #' @description Validate the launcher.
     #' @return `NULL` (invisibly). Throws an error if a field is invalid.
@@ -303,7 +267,8 @@ crew_class_launcher_sge <- R6::R6Class(
       fields <- c(
         "sge_cwd",
         "sge_envvars",
-        "sge_log_join"
+        "sge_log_join",
+        "verbose"
       )
       for (field in fields) {
         crew::crew_assert(
@@ -332,6 +297,59 @@ crew_class_launcher_sge <- R6::R6Class(
       }
       invisible()
     },
+    #' @description Launch a local process worker which will
+    #'   dial into a socket.
+    #' @details The `call` argument is R code that will run to
+    #'   initiate the worker. Together, the `launcher`, `worker`,
+    #'   and `instance` arguments are useful for
+    #'   constructing informative job names.
+    #' @return A handle object to allow the termination of the worker
+    #'   later on.
+    #' @param call Text string with a namespaced call to [crew_worker()]
+    #'   which will run in the worker and accept tasks.
+    #' @param launcher Character of length 1, name of the launcher.
+    #' @param worker Positive integer of length 1, index of the worker.
+    #'   This worker index remains the same even when the current instance
+    #'   of the worker exits and a new instance launches.
+    #'   It is always between 1 and the maximum number of concurrent workers.
+    #' @param instance Character of length 1 to uniquely identify
+    #'   the current instance of the worker.
+    launch_worker = function(call, launcher, worker, instance) {
+      name <- name_job(
+        launcher = launcher,
+        worker = worker,
+        instance = instance
+      )
+      lines <- c(
+        paste("#$ -N", name),
+        self$script(),
+        paste("R -e", shQuote(call))
+      )
+      script <- name_script(name = name)
+      writeLines(text = lines, con = script)
+      system2(
+        command = self$sge_qsub,
+        args = shQuote(script),
+        stdout = if_any(self$verbose, "", FALSE),
+        stderr = if_any(self$verbose, "", FALSE),
+        wait = FALSE
+      )
+      name
+    },
+    #' @description Terminate a local process worker.
+    #' @return `NULL` (invisibly).
+    #' @param handle A process handle object previously
+    #'   returned by `launch_worker()`.
+    terminate_worker = function(handle) {
+      unlink(name_script(name = handle), force = TRUE)
+      system2(
+        command = self$sge_qdel,
+        args = shQuote(handle),
+        stdout = if_any(self$verbose, "", FALSE),
+        stderr = if_any(self$verbose, "", FALSE),
+        wait = FALSE
+      )
+    },
     #' @description Generate the job script.
     #' @details Includes everything except the worker-instance-specific
     #'   job name and the worker-instance-specific

R/crew_package.R

@@ -20,9 +20,7 @@
 #' @family help
 #' @importFrom crew crew_assert crew_class_launcher crew_launcher
 #'   crew_random_name
-#' @importFrom processx run
 #' @importFrom R6 R6Class
-#' @importFrom rlang abort
 #' @importFrom utils globalVariables
 NULL
 

R/utils_names.R

@@ -2,7 +2,11 @@ name_job <- function(launcher, worker, instance) {
   out <- paste(launcher, worker, instance, sep = "-")
   alpha <- all(grepl(pattern = "^[[:alpha:]]", x = out))
   if (!alpha) {
-    out <- paste0("sge-", out)
+    out <- paste0("worker-", out)
   }
   out
 }
+
+name_script <- function(name) {
+  file.path(tempdir(), paste0(name, ".sh"))
+}

inst/WORDLIST

@@ -95,7 +95,7 @@ CloudWatch
 MERCHANTABILITY
 NONINFRINGEMENT
 Tidyselect
-Bischel
+Bischl
 Surmann
 Kubernetes
 namespaced

tests/sge/minimal.R

@@ -4,7 +4,8 @@ controller <- crew_controller_sge(
   name = "my_workflow",
   workers = 1L,
   seconds_idle = 300,
-  sge_lines = paste0("module load R/", getRversion())
+  sge_lines = paste0("module load R/", getRversion()),
+  verbose = TRUE
 )
 controller$start()
 controller$push(

tests/testthat/test-utils_names.R

@@ -1,4 +1,10 @@
 test_that("name_job()", {
   expect_equal(name_job("a", "b", "c"), "a-b-c")
-  expect_equal(name_job("1", "b", "c"), "sge-1-b-c")
+  expect_equal(name_job("1", "b", "c"), "worker-1-b-c")
+})
+
+test_that("name_script()", {
+  name <- name_job("a", "b", "c")
+  script <- name_script(name)
+  expect_equal(script, file.path(tempdir(), "a-b-c.sh"))
 })