update developer interface

- export worker loop implementation, .bpworker_impl - rename args to try to more clearly convey backend / worker as main arg - .bpstop_impl stops workers (sends .DONE() semaphore)
Bioconductor · Jan 22, 2019 · 1eb8b90 · 1eb8b90
1 parent 60f4c68
commit 1eb8b90
Show file tree

Hide file tree

Showing 11 changed files with 134 additions and 109 deletions.
diff --git a/NAMESPACE b/NAMESPACE
@@ -105,9 +105,6 @@ exportMethods(
 S3method(print, remote_error)
 S3method(print, bplist_error)
 
-S3method(bploop, MPInode)
-S3method(bploop, SOCKnode)
-S3method(bploop, SOCK0node)
 S3method(bploop, lapply)
 S3method(bploop, iterate)
 
@@ -118,6 +115,7 @@ S3method(bploop, iterate)
 export(
     .BiocParallelParam_prototype, .prototype_update, .prettyPath,
     .send_to, .recv_any, .send, .recv, .close, .send_all, .recv_all,
-    .bpstart_impl, .bpstop_impl, .bplapply_impl, .bpiterate_impl,
+    .bpstart_impl, .bpstop_impl, .bpworker_impl,
+    .bplapply_impl, .bpiterate_impl,
     .error_worker_comm
 )
diff --git a/R/DeveloperInterface.R b/R/DeveloperInterface.R
@@ -1,83 +1,84 @@
-## .BiocParallelParam_prototype, .prototype_update, .prettyPath,
-## .bpstart_impl, .bpstop_impl, .bplapply_impl, .bpiterate_impl,
-## .error_worker_comm
+##
+## see NAMESPACE section for definitive exports
+##
 
 ## server
+
 setGeneric(
     ".send_to",
-    function(cluster, node, value) standardGeneric(".send_to"),
-    signature = "cluster"
+    function(backend, node, value) standardGeneric(".send_to"),
+    signature = "backend"
 )
 
 setGeneric(
     ".recv_any",
-    function(cluster) standardGeneric(".recv_any"),
-    signature = "cluster"
+    function(backend) standardGeneric(".recv_any"),
+    signature = "backend"
 )
 
 setGeneric(
     ".send_all",
-    function(cluster, value) standardGeneric(".send_all"),
-    signature = "cluster"
+    function(backend, value) standardGeneric(".send_all"),
+    signature = "backend"
 )
 
 setGeneric(
     ".recv_all",
-    function(cluster) standardGeneric(".recv_all"),
-    signature = "cluster"
+    function(backend) standardGeneric(".recv_all"),
+    signature = "backend"
 )
 
 ## client
 
 setGeneric(
     ".send",
-    function(cluster, value) standardGeneric(".send"),
-    signature = "cluster"
+    function(worker, value) standardGeneric(".send"),
+    signature = "worker"
 )
 
 setGeneric(
     ".recv",
-    function(cluster) standardGeneric(".recv"),
-    signature = "cluster"
+    function(worker) standardGeneric(".recv"),
+    signature = "worker"
 )
 
 setGeneric(
     ".close",
-    function(cluster) standardGeneric(".close"),
-    signature = "cluster"
+    function(worker) standardGeneric(".close"),
+    signature = "worker"
 )
 
-## default implementation -- SNOW cluster
+## default implementation -- SNOW backend
 
 setMethod(
     ".send_all", "ANY",
-    function(cluster, value)
+    function(backend, value)
 {
-    for (node in seq_along(cluster))
-        .send_to(cluster, node, value)
+    for (node in seq_along(backend))
+        .send_to(backend, node, value)
 })
 
 setMethod(
     ".recv_all", "ANY",
-    function(cluster)
+    function(backend)
 {
-    replicate(length(cluster), .recv_any(cluster), simplify=FALSE)
+    replicate(length(backend), .recv_any(backend), simplify=FALSE)
 })
 
 setMethod(
     ".send_to", "ANY",
-    function(cluster, node, value)
+    function(backend, node, value)
 {
-    parallel:::sendData(cluster[[node]], value)
+    parallel:::sendData(backend[[node]], value)
     TRUE
 })
 
 setMethod(
     ".recv_any", "ANY",
-    function(cluster)
+    function(backend)
 {
     tryCatch({
-        parallel:::recvOneData(cluster)
+        parallel:::recvOneData(backend)
     }, error  = function(e) {
         ## indicate error, but do not stop
         .error_worker_comm(e, "'.recv_any()' data failed")
@@ -86,17 +87,17 @@ setMethod(
 
 setMethod(
     ".send", "ANY",
-    function(cluster, value)
+    function(worker, value)
 {
-    parallel:::sendData(cluster, value)
+    parallel:::sendData(worker, value)
 })
 
 setMethod(
     ".recv", "ANY",
-    function(cluster)
+    function(worker)
 {
     tryCatch({
-        parallel:::recvData(cluster)
+        parallel:::recvData(worker)
     }, error = function(e) {
         ## indicate error, but do not stop
         .error_worker_comm(e, "'.recv()' data failed")
@@ -105,7 +106,7 @@ setMethod(
 
 setMethod(
     ".close", "ANY",
-    function(cluster)
+    function(worker)
 {
-    parallel:::closeNode(cluster)
+    parallel:::closeNode(worker)
 })
diff --git a/R/SnowParam-class.R b/R/SnowParam-class.R
@@ -261,17 +261,13 @@ setMethod("bpstop", "SnowParam",
     if (!bpisup(x))
         return(invisible(x))
 
-    tryCatch({
-        res <- capture.output(parallel::stopCluster(bpbackend(x)))
-    }, error=function(err) {
-        txt <- sprintf("failed to stop %s cluster: %s",
-                       sQuote(class(bpbackend(x))[[1]]),
-                       conditionMessage(err))
-        stop(paste(strwrap(txt, exdent=2), collapse="\n"), call.=FALSE)
-    })
+    x <- .bpstop_impl(x)
+    cluster <- bpbackend(x)
+    for (i in seq_along(cluster))
+        .close(cluster[[i]])
     bpbackend(x) <- .NULLcluster()
 
-    .bpstop_impl(x)
+    x
 })
 
 setMethod("bpisup", "SnowParam",

diff --git a/R/SnowParam-utils.R b/R/SnowParam-utils.R
@@ -12,7 +12,7 @@ bprunMPIslave <- function() {
     Rmpi::mpi.comm.set.errhandler(comm)
     Rmpi::mpi.comm.disconnect(intercomm)
 
-    bploop(snow::makeMPImaster(comm))
+    .bpworker_impl(snow::makeMPImaster(comm))
 
     Rmpi::mpi.comm.disconnect(comm)
     Rmpi::mpi.quit()
@@ -57,7 +57,7 @@ bprunMPIslave <- function() {
             }
         })
         node <- structure(list(con = con), class = "SOCK0node")
-        bploop(node)
+        .bpworker_impl(node)
     }, detached=TRUE)
 }
 

diff --git a/R/bploop.R b/R/bploop.R
@@ -19,24 +19,27 @@
          time = time, log = log, sout = sout)
 }
 
+.DONE <-
+    function()
+{
+    list(type = "DONE")
+}
+
 ### - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 ### Worker loop used by SOCK, MPI and FORK.  Error handling is done in
 ### .composeTry.
 
-bploop <- function(manager, ...)
-    UseMethod("bploop")
-
-.bploop.worker <- function(manager, ...)
+.bpworker_impl <- function(worker)
 {
     repeat {
         tryCatch({
-            msg <- .recv(manager)
+            msg <- .recv(worker)
             if (inherits(msg, "error"))
                 ## FIXME: try to return error to manager
                 break                   # lost socket connection?
 
             if (msg$type == "DONE") {
-                .close(manager)
+                .close(worker)
                 break
             } else if (msg$type == "EXEC") {
                 ## need local handler for worker read/send errors
@@ -63,18 +66,14 @@ bploop <- function(manager, ...)
                 value <- .VALUE(
                     msg$data$tag, value, success, t2 - t1, log, sout
                 )
-                .send(manager, value)
+                .send(worker, value)
             }
         }, interrupt = function(e) {
             NULL
         })
     }
 }
 
-bploop.MPInode <- .bploop.worker
-bploop.SOCKnode <- .bploop.worker
-bploop.SOCK0node <- .bploop.worker
-
 ### - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 ### Manager loop used by SOCK, MPI and FORK
 
@@ -183,6 +182,9 @@ bploop.SOCK0node <- .bploop.worker
 ## bploop.lapply(): derived from snow::dynamicClusterApply.
 ##
 
+bploop <- function(manager, ...)
+    UseMethod("bploop")
+
 bploop.lapply <-
     function(manager, X, FUN, ARGFUN, BPPARAM, ...)
 {

diff --git a/R/bpstart-methods.R b/R/bpstart-methods.R
@@ -122,9 +122,9 @@ setMethod("bpstart", "missing",
 }
 
 .bpstart_impl <-
-    function(x, ...)
+    function(x)
 {
-    ## common actions once bpisup(x)
+    ## common actions once bpisup(backend)
 
     ## logging
     if (bplog(x))

diff --git a/R/bpstop-methods.R b/R/bpstop-methods.R
@@ -11,9 +11,20 @@ setMethod("bpstop", "missing",
 ## .bpstop_impl: common functionality after bpisup() is no longer TRUE
 ##
 
+.bpstop_nodes <-
+    function(x)
+{
+    cluster <- bpbackend(x)
+    for (i in seq_along(cluster))
+        .send_to(cluster, i, .DONE())
+
+    TRUE
+}
+
 .bpstop_impl <-
     function(x)
 {
+    bpisup(x) && .bpstop_nodes(x)
     .ClusterManager$drop(x$.uid)
     invisible(x)
 }
diff --git a/inst/RSOCKnode.sh b/inst/RSOCKnode.sh
@@ -7,5 +7,5 @@ ${RPROG:-R} --vanilla <<EOF > ${OUT:-/dev/null} 2>&1 &
 
 loadNamespace("snow")
 options(timeout=getClusterOption("timeout"))
-BiocParallel::bploop(snow::makeSOCKmaster())
+BiocParallel::.bpworker_impl(snow::makeSOCKmaster())
 EOF
diff --git a/inst/snow/RSOCKnode.R b/inst/snow/RSOCKnode.R
@@ -25,5 +25,5 @@ local({
 
     if (port == "") port <- getClusterOption("port")
 
-    BiocParallel::bploop(snow::makeSOCKmaster(master, port))
+    BiocParallel::.bpworker_impl(snow::makeSOCKmaster(master, port))
 })