CLEANUP: Un-export all pico_p2p_nnn() functions

Author: HenrikBengtsson

.github/workflows/R-CMD-check.yaml

@@ -38,7 +38,7 @@ jobs:
       _R_CHECK_SUGGESTS_ONLY_: true
       _R_CHECK_THINGS_IN_TEMP_DIR_: true
       _R_CHECK_TESTS_NLINES_: 300
-      RCMDCHECK_ERROR_ON: error
+      RCMDCHECK_ERROR_ON: warning
       ## Specific to futures
       R_FUTURE_CONNECTIONS_ONMISUSE: "error[details=TRUE]"
       R_FUTURE_GLOBALENV_ONMISUSE: "error"

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.1.0-9010
+Version: 0.1.0-9011
 Description: Implementation of the 'Future' API <doi:10.32614/RJ-2021-048> that resolves futures on a peer-to-peer ('P2P') compute environment. By using this future backend, you and your friends can share your spare compute resources with each other.
 Depends:
   future (>= 1.67.0)

NAMESPACE

@@ -11,24 +11,8 @@ S3method(resolved,PicoP2PFuture)
 S3method(result,PicoP2PFuture)
 export(PicoP2PFutureBackend)
 export(cluster)
-export(future_id)
 export(host_cluster)
 export(p2p_cluster_name)
-export(pico_hosted_channels)
-export(pico_p2p_have_future)
-export(pico_p2p_hello)
-export(pico_p2p_hosted_clusters)
-export(pico_p2p_receive_future)
-export(pico_p2p_receive_result)
-export(pico_p2p_send_future)
-export(pico_p2p_send_result)
-export(pico_p2p_take_on_future)
-export(pico_p2p_wait_for)
-export(pico_pipe)
-export(pico_receive_message)
-export(pico_receive_message_dataframe)
-export(pico_send_message)
-export(pico_terminate)
 export(pico_username)
 export(worker)
 importFrom(callr,r_bg)

R/PicoP2PFutureBackend-class.R

@@ -138,7 +138,7 @@ launchFuture.PicoP2PFutureBackend <- function(backend, future, ...) {
   FutureRegistry(reg, action = "add", future = future, earlySignal = FALSE)
 
   ## 3. Launch future, i.e. submit it to the Pico P2P cluster dispatcher
-  future <- dispatch_future(future)
+  future <- pico_p2p_dispatch_future(future)
 
   invisible(future)
 } ## launchFuture()
@@ -189,94 +189,6 @@ waitForWorker <- function(...) {
 }
 
 
-#' @importFrom callr r_bg
-#' @importFrom utils file_test
-dispatch_future <- function(future) {
-  send_future <- function(topic, name, host = host, ssh_args = ssh_args, future_id, file, to, via, duration) {
-    pico <- future.p2p::pico_pipe(topic, user = name, host = host, ssh_args = ssh_args)
-    m <- future.p2p::pico_p2p_hello(pico, type = "client")
-
-    ## 2. Announce future
-    repeat {
-      m1 <- future.p2p::pico_p2p_have_future(pico, future = file, duration = duration)
-      m2 <- future.p2p::pico_p2p_wait_for(pico, type = "offer", futures = m1[["future"]], expires = m1[["expires"]])
-      if (m2[["type"]] != "expired") break
-    }
-
-    ## 3. Send future to workers
-    worker <- m2[["from"]]
-    stopifnot(is.character(worker), nzchar(worker))
-    m3 <- future.p2p::pico_p2p_send_future(pico, future = file, to = worker, via = via)
-
-    ## 4. Remove temporary file
-    file.remove(file)
-
-    ## 5. Wait for and receive FutureResult file
-    path <- file.path(dirname(dirname(file)), "results")
-    tryCatch({
-      file <- future.p2p::pico_p2p_receive_result(pico, via = via, path = path)
-    }, interrupt = function(int) {
-      cat(file = "foo.log", "interrupted\n")
-    })
-
-    invisible(file)
-  }
-
-  debug <- isTRUE(getOption("future.p2p.debug"))
-  if (debug) {
-    mdebug_push("dispatch_future()...")
-    on.exit(mdebugf_pop())
-  }
-
-  ## Get backend
-  backend <- future[["backend"]]
-  stopifnot(inherits(backend, "FutureBackend"))
-
-  cluster <- backend[["cluster"]]
-  name <- backend[["name"]]
-  host <- backend[["host"]]
-  ssh_args <- backend[["ssh_args"]]
-  via <- via_transfer_uri()
-  
-  ## 1. Put future on the dispatcher queue
-  void <- p2p_dir("results")
-  future[["file"]] <- saveFuture(future, path = p2p_dir("queued"))
-  
-  if (debug) mdebugf("File: %s", sQuote(future[["file"]]))
-
-  ## 1. Connect to pico and say hello
-  cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
-    topic <- sprintf("%s/future.p2p", basename(cluster))
-  } else {
-    topic <- sprintf("%s/future.p2p", cluster)
-  }
-
-  args <- list(
-    topic = topic,
-    name = name,
-    host = host,
-    ssh_args = ssh_args,
-    future_id = future_id(future),
-    file = future[["file"]],
-    via = via,
-    duration = getOption("future.p2p.duration.request", 10.0)
-  )
-  if (debug) {
-    mstr(args)
-  }
-
-  rx <- r_bg(send_future, args = args, supervise = TRUE)
-  future[["rx"]] <- rx
-
-  future[["pico_via"]] <- via
-
-  ## Update future state
-  future[["state"]] <- "running"
-  
-  invisible(future)
-}
-
 
 #' @importFrom utils file_test
 p2p_dir <- function(dir = c("queued", "running", "results")) {

R/pico.R

@@ -31,7 +31,6 @@
 #' This function relies on the <https://pico.sh> services.
 #'
 #' @importFrom processx process
-#' @export
 pico_pipe <- function(topic = NULL, command = c("pipe", "pub", "sub", "ls", "help"), args = c(), host = "pipe.pico.sh", ssh_args = NULL, ...) {
   command <- match.arg(command)
   if (command %in% c("pipe", "pub", "sub")) {
@@ -91,14 +90,12 @@ print.pico <- function(x, ...) {
 }
 
 #' @rdname pico_pipe
-#' @export
 pico_terminate <- function(p, ...) {
   stopifnot(inherits(p, "pico_pipe"))
   p$process$kill()
 }
 
 #' @rdname pico_pipe
-#' @export
 pico_send_message <- function(p, message, newline = TRUE, ...) {
   stopifnot(inherits(p, "pico_pipe"))
   stopifnot(length(message) == 1L, is.character(message), !is.na(message))
@@ -107,6 +104,9 @@ pico_send_message <- function(p, message, newline = TRUE, ...) {
 }
 
 
+#' @param df (data.frame) Data frame to send as a message.
+#'
+#' @rdname pico_pipe
 pico_send_message_dataframe <- function(p, df) {
   msg <- unlist(df, use.names = TRUE)
   msg <- sprintf("%s=%s", names(msg), msg)
@@ -118,15 +118,16 @@ pico_send_message_dataframe <- function(p, df) {
 
 
 #' @rdname pico_pipe
-#' @export
 pico_receive_message <- function(p, n = 1L, ...) {
   stopifnot(inherits(p, "pico_pipe"))
   stopifnot(length(n) == 1L, is.numeric(n), !is.na(n), n > 0L)
   p$process$read_output_lines(n)
 }
 
 
-#' @export
+#' @param pattern (character string; optional) A regular expression so scan for.
+#'
+#' @rdname pico_pipe
 pico_receive_message_dataframe <- function(p, ..., pattern = NULL) {
   msg <- pico_receive_message(p, ...)
 
@@ -150,8 +151,9 @@ pico_receive_message_dataframe <- function(p, ..., pattern = NULL) {
 }
 
 
-
-#' @export
+#' @param timeout (numeric scalar) Timeout (in seconds).
+#'
+#' @rdname pico_pipe
 pico_hosted_channels <- function(host = "pipe.pico.sh", ssh_args = NULL, timeout = 10.0) {
   username <- pico_username()
   t_max <- proc.time()[3] + timeout

R/pico_p2p.R

@@ -12,7 +12,6 @@ now_str <- function(when = pico_p2p_time()) {
   format(when, format = "%FT%T")
 }
 
-#' @export
 future_id <- function(future, ...) {
   if (inherits(future, "Future")) {
     id <- paste(future[["uuid"]], collapse = "-")
@@ -23,7 +22,7 @@ future_id <- function(future, ...) {
   id
 }
 
-#' @export
+
 pico_p2p_hello <- function(p, from = p$user, type = c("worker", "client", "cluster"), expires = NULL, duration = 60*60, ...) {
   type <- match.arg(type)
 
@@ -58,7 +57,7 @@ pico_p2p_expired <- function() {
   )
 }
 
-#' @export
+
 pico_p2p_wait_for <- function(p, type, futures = NULL, expires = NULL, duration = 60, delay = 0.1, ...) {
   if (is.null(expires)) {
     expires <- Sys.time() + duration
@@ -95,7 +94,6 @@ pico_p2p_wait_for <- function(p, type, futures = NULL, expires = NULL, duration
 
 
 #' @importFrom utils file_test
-#' @export
 pico_p2p_have_future <- function(p, future, duration = getOption("future.p2p.duration.request", 60), from = p$user, ...) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -136,7 +134,6 @@ pico_p2p_have_future <- function(p, future, duration = getOption("future.p2p.dur
   pico_send_message_dataframe(p, m)
 }
 
-#' @export
 pico_p2p_take_on_future <- function(p, to, future, duration = 60, from = p$user, ...) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -165,7 +162,6 @@ pico_p2p_take_on_future <- function(p, to, future, duration = 60, from = p$user,
   pico_send_message_dataframe(p, m)
 }
 
-#' @export
 pico_p2p_send_future <- function(p, future, to, via = via_transfer_uri(), duration = 60, from = p$user, ...) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -215,7 +211,6 @@ pico_p2p_send_future <- function(p, future, to, via = via_transfer_uri(), durati
 }
 
 
-#' @export
 pico_p2p_receive_future <- function(p, via, duration = 60) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -244,7 +239,6 @@ pico_p2p_receive_future <- function(p, via, duration = 60) {
 
 
 #' @importFrom future result
-#' @export
 pico_p2p_send_result <- function(p, future, via, duration = 60) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -269,7 +263,6 @@ pico_p2p_send_result <- function(p, future, via, duration = 60) {
 }
 
 
-#' @export
 pico_p2p_receive_result <- function(p, via, duration = 60, path = tempdir()) {
   debug <- isTRUE(getOption("future.p2p.debug"))
   if (debug) {
@@ -291,11 +284,100 @@ pico_p2p_receive_result <- function(p, via, duration = 60, path = tempdir()) {
 }
 
 
-#' @export
 pico_p2p_hosted_clusters <- function(host = "pipe.pico.sh", ssh_args = NULL, timeout = 10.0) {
   clusters <- pico_hosted_channels(host, ssh_args = ssh_args, timeout = timeout)
   keep <- grep("/future.p2p$", clusters$name)
   clusters <- clusters[keep, ]
   clusters$name <- sub("/future.p2p$", "", clusters$name)
   clusters
 }
+
+
+
+#' @importFrom callr r_bg
+#' @importFrom utils file_test
+pico_p2p_dispatch_future <- function(future) {
+  send_future <- function(topic, name, host = host, ssh_args = ssh_args, future_id, file, to, via, duration) {
+    pico <- future.p2p:::pico_pipe(topic, user = name, host = host, ssh_args = ssh_args)
+    m <- future.p2p:::pico_p2p_hello(pico, type = "client")
+
+    ## 2. Announce future
+    repeat {
+      m1 <- future.p2p:::pico_p2p_have_future(pico, future = file, duration = duration)
+      m2 <- future.p2p:::pico_p2p_wait_for(pico, type = "offer", futures = m1[["future"]], expires = m1[["expires"]])
+      if (m2[["type"]] != "expired") break
+    }
+
+    ## 3. Send future to workers
+    worker <- m2[["from"]]
+    stopifnot(is.character(worker), nzchar(worker))
+    m3 <- future.p2p:::pico_p2p_send_future(pico, future = file, to = worker, via = via)
+
+    ## 4. Remove temporary file
+    file.remove(file)
+
+    ## 5. Wait for and receive FutureResult file
+    path <- file.path(dirname(dirname(file)), "results")
+    tryCatch({
+      file <- future.p2p:::pico_p2p_receive_result(pico, via = via, path = path)
+    }, interrupt = function(int) {
+      cat(file = "foo.log", "interrupted\n")
+    })
+
+    invisible(file)
+  }
+
+  debug <- isTRUE(getOption("future.p2p.debug"))
+  if (debug) {
+    mdebug_push("dispatch_future()...")
+    on.exit(mdebugf_pop())
+  }
+
+  ## Get backend
+  backend <- future[["backend"]]
+  stopifnot(inherits(backend, "FutureBackend"))
+
+  cluster <- backend[["cluster"]]
+  name <- backend[["name"]]
+  host <- backend[["host"]]
+  ssh_args <- backend[["ssh_args"]]
+  via <- via_transfer_uri()
+  
+  ## 1. Put future on the dispatcher queue
+  void <- p2p_dir("results")
+  future[["file"]] <- saveFuture(future, path = p2p_dir("queued"))
+  
+  if (debug) mdebugf("File: %s", sQuote(future[["file"]]))
+
+  ## 1. Connect to pico and say hello
+  cluster_owner <- dirname(cluster)
+  if (cluster_owner == pico_username()) {
+    topic <- sprintf("%s/future.p2p", basename(cluster))
+  } else {
+    topic <- sprintf("%s/future.p2p", cluster)
+  }
+
+  args <- list(
+    topic = topic,
+    name = name,
+    host = host,
+    ssh_args = ssh_args,
+    future_id = future_id(future),
+    file = future[["file"]],
+    via = via,
+    duration = getOption("future.p2p.duration.request", 10.0)
+  )
+  if (debug) {
+    mstr(args)
+  }
+
+  rx <- r_bg(send_future, args = args, supervise = TRUE)
+  future[["rx"]] <- rx
+
+  future[["pico_via"]] <- via
+
+  ## Update future state
+  future[["state"]] <- "running"
+  
+  invisible(future)
+}