From 4092acea7804a55681d00b6f12c404479019bc73 Mon Sep 17 00:00:00 2001 From: Henrik Bengtsson Date: Fri, 3 Jul 2026 19:57:19 +0200 Subject: [PATCH] Add support for SnowParam(..., type = "RPSOCK") [Issue #231] --- DESCRIPTION | 4 ++-- NEWS | 10 ++++++++ R/SnowParam-class.R | 34 +++++++++++++++++++++----- R/worker.R | 8 +++---- inst/unitTests/test_SnowParam.R | 42 +++++++++++++++++++++++++++++++++ man/SnowParam-class.Rd | 12 +++++----- 6 files changed, 92 insertions(+), 18 deletions(-) diff --git a/DESCRIPTION b/DESCRIPTION index 1e16fea..beb6c02 100644 --- a/DESCRIPTION +++ b/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, diff --git a/NEWS b/NEWS index a638da6..dcca58d 100644 --- a/NEWS +++ b/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 ----------------------- diff --git a/R/SnowParam-class.R b/R/SnowParam-class.R index f42ffe7..406056d 100644 --- a/R/SnowParam-class.R +++ b/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,6 +380,7 @@ setReplaceMethod("bpthreshold", c("SnowParam", "character"), ### parallel::SOCKcluster types setOldClass(c("SOCKcluster", "cluster")) +setOldClass(c("RichSOCKcluster", "SOCKcluster")) stopCluster.SOCKcluster <- parallel:::stopCluster.default @@ -370,8 +388,11 @@ stopCluster.SOCKcluster <- 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") diff --git a/R/worker.R b/R/worker.R index 000cac9..2fbe8ac 100644 --- a/R/worker.R +++ b/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) diff --git a/inst/unitTests/test_SnowParam.R b/inst/unitTests/test_SnowParam.R index 6f3795d..c1cc441 100644 --- a/inst/unitTests/test_SnowParam.R +++ b/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) +} + diff --git a/man/SnowParam-class.Rd b/man/SnowParam-class.Rd index 4dfa245..bbc79bd 100644 --- a/man/SnowParam-class.Rd +++ b/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}{