Add support for SnowParam(..., type = "RPSOCK") [Issue #231]

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: BiocParallel
 Type: Package
 Title: Bioconductor facilities for parallel evaluation
-Version: 1.47.0
+Version: 1.47.1
 Authors@R: c(
     person("Jiefei", "Wang", 
         email = "jiefei0804@gmail.com", 
@@ -36,7 +36,7 @@ License: GPL-2 | GPL-3 | BSL-1.0
 SystemRequirements: C++11
 Depends: methods, R (>= 4.1.0)
 Imports: stats, utils, futile.logger, parallel, snow, codetools
-Suggests: BiocGenerics, tools, foreach, BBmisc, doParallel,
+Suggests: BiocGenerics, tools, parallelly (>= 1.48.0), foreach, BBmisc, doParallel,
     GenomicRanges, RNAseqData.HNRNPC.bam.chr14,
     TxDb.Hsapiens.UCSC.hg19.knownGene, VariantAnnotation, Rsamtools,
     GenomicAlignments, ShortRead, RUnit, BiocStyle, knitr, batchtools,

NEWS

@@ -1,3 +1,13 @@
+CHANGES IN VERSION 1.47
+-----------------------
+
+NEW FEATURES
+
+    o (1.47.1) Add support for SnowParam(..., type = "RPSOCK"), which
+    creates a parallel cluster using parallel::makeCluster(..., type
+    = "RPSOCK") via the 'parallelly' package.
+
+
 CHANGES IN VERSION 1.44
 -----------------------
 

R/SnowParam-class.R

@@ -73,7 +73,7 @@
     min(.defaultWorkers(), .snowCoresMax(type))
 }
 
-snowWorkers <- function(type = c("SOCK", "MPI", "FORK")) {
+snowWorkers <- function(type = c("SOCK", "MPI", "FORK", "RPSOCK")) {
     type <- match.arg(type)
     min(.defaultWorkers(), .snowCores(type))
 }
@@ -119,7 +119,7 @@ setOldClass(c("NULLcluster", "cluster"))
 )
 
 SnowParam <- function(workers=snowWorkers(type),
-                      type=c("SOCK", "MPI", "FORK"), tasks=0L,
+                      type=c("SOCK", "MPI", "FORK", "RPSOCK"), tasks=0L,
                       stop.on.error=TRUE,
                       progressbar=FALSE, RNGseed=NULL,
                       timeout=WORKER_TIMEOUT,
@@ -277,13 +277,30 @@ setMethod("bpstart", "SnowParam",
         if (dir.exists(file.path(libPath, "inst")))
             libPath <- file.path(libPath, "inst")
 
-        if (is.null(cargs$snowlib))
-            cargs$snowlib <- libPath
+        if (x$.clusterargs$type %in% c("SOCK", "MPI")) {
+            if (is.null(cargs$snowlib))
+                cargs$snowlib <- libPath
+        }
 
         if (!is.null(cargs$useRscript) && !cargs$useRscript)
             cargs$scriptdir <- libPath
 
-        if (x$.clusterargs$type == "SOCK") {
+        if (x$.clusterargs$type == "RPSOCK") {
+            if (!"rscript_call" %in% names(formals(parallelly::makeNodePSOCK))) {
+                stop("parallelly::makeNodePSOCK() does not support 'rscript_call'")
+            }
+            if (is.null(cargs$rscript_libs))
+                cargs$rscript_libs <- unique(c(dirname(find.package("BiocParallel")), .libPaths()))
+            cargs$rscript_call <- quote(quote({
+              workCommand <- function(master) {
+                BiocParallel::.bpworker_impl(master)
+                FALSE
+              }
+              parallelly:::workRPSOCK(workCommand)
+            }))
+        }
+
+        if (x$.clusterargs$type %in% c("SOCK", "RPSOCK")) {
             cargs$master <- .hostname(x)
             cargs$port <- .port(x)
         }
@@ -363,15 +380,19 @@ setReplaceMethod("bpthreshold", c("SnowParam", "character"),
 ### parallel::SOCKcluster types
 
 setOldClass(c("SOCKcluster", "cluster"))
+setOldClass(c("RichSOCKcluster", "SOCKcluster"))
 
 stopCluster.SOCKcluster <-
     parallel:::stopCluster.default
 
 setAs("SOCKcluster", "SnowParam",
     function(from)
 {
+    type <- sub("cluster$", "", class(from)[1L])
+    if (type == "RichSOCK")
+        type <- "RPSOCK"
     .clusterargs <-
-        list(spec=length(from), type=sub("cluster$", "", class(from)[1L]))
+        list(spec=length(from), type=type)
     prototype <- .prototype_update(
         .SnowParam_prototype,
         .clusterargs = .clusterargs,
@@ -401,6 +422,7 @@ setAs("spawnedMPIcluster", "SnowParam",
 ###
 
 setOldClass(c("SOCK0node", "SOCKnode")) # needed for method dispatch
+setOldClass(c("RichSOCKnode", "SOCK0node"))
 
 .SOCKmanager <- setClass("SOCKmanager", contains = "TaskManager")
 

R/worker.R

@@ -12,7 +12,7 @@
     if (isTRUE(value$dynamic.only))
         return(NULL)
 
-    if (value$static.fun)
+    if (isTRUE(value$static.fun))
         fun <- value$data$fun
     else
         fun <- NULL
@@ -39,12 +39,12 @@
     if (value$type != "EXEC")
         return(value)
 
-    if (value$static.fun)
+    if (isTRUE(value$static.fun))
         value$data$fun <- TRUE
     if (length(value$static.args))
         value$data$args[value$static.args] <- NULL
 
-    if (value$static.fun || length(value$static.args))
+    if (isTRUE(value$static.fun) || length(value$static.args))
         value$dynamic.only <- TRUE
 
     value
@@ -63,7 +63,7 @@
     if (!isTRUE(value$dynamic.only))
         return(value)
 
-    if (value$static.fun)
+    if (isTRUE(value$static.fun))
         value$data$fun <- static_data$fun
     if (length(value$static.args)) {
         value$data$args <- c(value$data$args, static_data$args)

inst/unitTests/test_SnowParam.R

@@ -201,3 +201,45 @@ test_SnowParam_fallback <- function(){
     res <- bplapply(1, function(x) Sys.getpid(), BPPARAM = p)[[1]]
     checkTrue(res != Sys.getpid())
 }
+
+test_SnowParam_RPSOCK <- function()
+{
+    if (!requireNamespace("snow", quietly=TRUE))
+        DEACTIVATED("'snow' package did not load")
+    if (!requireNamespace("parallelly", quietly=TRUE))
+        DEACTIVATED("'parallelly' package did not load")
+
+    param <- SnowParam(2, "RPSOCK", tasks=2)
+    checkIdentical(FALSE, bpisup(param))
+
+    exp <- bplapply(1:2, function(i) Sys.getpid(), BPPARAM=param)
+    checkIdentical(2L, length(unique(unlist(exp))))
+    checkIdentical(FALSE, bpisup(param))
+}
+
+test_SnowParam_coerce_from_RPSOCK <- function()
+{
+    if (!requireNamespace("snow", quietly=TRUE))
+        DEACTIVATED("'snow' package did not load")
+    if (!requireNamespace("parallelly", quietly=TRUE))
+        DEACTIVATED("'parallelly' package did not load")
+
+    cl <- parallel::makeCluster(2L, type = "RPSOCK")
+    p <- as(cl, "SnowParam")
+    checkTrue(validObject(p))
+
+    obs <- tryCatch(bpstart(p), error=conditionMessage)
+    exp <- "'bpstart' not available; instance from outside BiocParallel?"
+    checkIdentical(exp, obs)
+
+    obs <- tryCatch(bpstop(p), warning=conditionMessage)
+    exp <- "'bpstop' not available; instance from outside BiocParallel?"
+    checkIdentical(exp, obs)
+
+    exp <- bplapply(1:2, function(i) Sys.getpid(), BPPARAM=p)
+    checkIdentical(2L, length(unique(unlist(exp))))
+    checkIdentical(TRUE, bpisup(p))
+
+    parallel::stopCluster(cl)
+}
+

man/SnowParam-class.Rd

@@ -43,7 +43,7 @@
 ## constructor
 ## ------------------------------------
 
-SnowParam(workers = snowWorkers(type), type=c("SOCK", "MPI", "FORK"),
+SnowParam(workers = snowWorkers(type), type=c("SOCK", "MPI", "FORK", "RPSOCK"),
     tasks = 0L, stop.on.error = TRUE,
     progressbar = FALSE, RNGseed = NULL,
     timeout = WORKER_TIMEOUT, exportglobals = TRUE, exportvariables = TRUE,
@@ -62,13 +62,13 @@ SnowParam(workers = snowWorkers(type), type=c("SOCK", "MPI", "FORK"),
 ## detect workers
 ## ------------------------------------
 
-snowWorkers(type = c("SOCK", "MPI", "FORK"))
+snowWorkers(type = c("SOCK", "MPI", "FORK", "RPSOCK"))
 }
 
 \details{
 
   \code{SnowParam} is used for distributed memory computing and supports
-  2 cluster types: \sQuote{SOCK} (default) and \sQuote{MPI}. The
+  cluster types: \sQuote{SOCK} (default), \sQuote{MPI}, and \sQuote{RPSOCK}. The
   \code{SnowParam} builds on infrastructure in the \code{snow} and
   \code{parallel} packages and provides the additional features of error
   handling, logging and writing out results.
@@ -149,7 +149,7 @@ snowWorkers(type = c("SOCK", "MPI", "FORK"))
 
   \describe{
     \item{
-      \code{SnowParam(workers = snowWorkers(), type=c("SOCK", "MPI"),
+      \code{SnowParam(workers = snowWorkers(), type=c("SOCK", "MPI", "FORK", "RPSOCK"),
                       tasks = 0L, stop.on.error = FALSE,
                       progressbar = FALSE, RNGseed = NULL,
                       timeout = Inf, exportglobals = TRUE,
@@ -175,13 +175,13 @@ snowWorkers(type = c("SOCK", "MPI", "FORK"))
     the number of cores determined by \code{detectCores} minus 2 unless
     environment variables \code{R_PARALLELLY_AVAILABLECORES_FALLBACK} or
     \code{BIOCPARALLEL_WORKER_NUMBER} are set otherwise. For a
-    \code{SOCK} cluster, \code{workers} can be a \code{character()}
+    \code{SOCK} or \code{RPSOCK} cluster, \code{workers} can be a \code{character()}
     vector of host names.
 
   }
   \item{type}{
     \code{character(1)} Type of cluster to use. Possible values are
-    \code{SOCK} (default) and \code{MPI}. Instead of \code{type=FORK} use
+    \code{SOCK} (default), \code{MPI}, and \code{RPSOCK}. Instead of \code{type=FORK} use
     \code{MulticoreParam}.
   }
   \item{tasks}{