Forward all 'host' and 'ssh_args' arguments to pico_username(), etc.

Author: HenrikBengtsson

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: future.p2p
 Title: A Peer-to-Peer Compute Cluster via Futureverse
-Version: 0.3.0-9017
+Version: 0.3.0-9018
 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.
 Imports:
     future (>= 1.67.0),

R/PicoP2PFutureBackend-class.R

@@ -100,7 +100,7 @@ PicoP2PFutureBackend <- function(cluster = p2p_cluster_name(), host = "pipe.pico
   core <- structure(core, class = c("PicoP2PFutureBackend", "MultiprocessFutureBackend", "FutureBackend", class(core)))
 
   if (!p2p_can_connect(cluster, name = name, host = host, ssh_args = ssh_args)) {
-    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   core
@@ -245,19 +245,21 @@ print.PicoP2PFutureBackend <- function(x, ...) {
   cat(sprintf("P2P cluster: %s\n", sQuote(backend[["cluster"]])))
   cat(sprintf("P2P client ID: %s\n", sQuote(backend[["name"]])))
 
-  clusters <- pico_p2p_hosted_clusters(backend[["host"]], backend[["ssh_args"]])
+  host <- backend[["host"]]
+  ssh_args <- backend[["ssh_args"]]
+  clusters <- pico_p2p_hosted_clusters(host = host, ssh_args = ssh_args)
   cat(sprintf("P2P clusters you are hosting: [n=%d]\n", nrow(clusters)))
   for (kk in seq_len(nrow(clusters))) {
     cluster <- clusters[kk, ]
     users <- strsplit(cluster$users, split = ",", fixed = TRUE)[[1]]
-    users <- unique(c(users, pico_username()))
+    users <- unique(c(users, pico_username(host = host, ssh_args = ssh_args)))
     users <- paste(users, collapse = ", ")
     cat(sprintf(" %2d. %s (users: %s)\n", kk, sQuote(cluster$name), users))
   }
 
   cat("Message board:\n")
   cat(sprintf(" - Server: %s\n", backend[["host"]]))
-  username <- pico_username(backend[["host"]], backend[["ssh_args"]])
+  username <- pico_username(host = host, ssh_args = ssh_args)
   cat(sprintf(" - Username: %s\n", sQuote(username)))
 
   cat("Data transfer tools:\n")
@@ -273,7 +275,7 @@ print.PicoP2PFutureBackend <- function(x, ...) {
 
 p2p_can_connect <- function(cluster, name, host = "pipe.pico.sh", ssh_args = NULL, timeout = 10.0) {
   cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
+  if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
     topic <- sprintf("%s/future.p2p", basename(cluster))
   } else {
     topic <- sprintf("%s/future.p2p", cluster)

R/host_cluster.R

@@ -16,25 +16,25 @@
 #'
 #' @importFrom future resolve plan sequential
 #' @export
-host_cluster <- function(cluster = p2p_cluster_name(users), users = character(0L), host = "pipe.pico.sh", ssh_args = NULL, duration = 14*24*60*60) {
+host_cluster <- function(cluster = p2p_cluster_name(users, host = host, ssh_args = ssh_args), users = character(0L), host = "pipe.pico.sh", ssh_args = NULL, duration = 14*24*60*60) {
   stopifnot(length(cluster) == 1L, is.character(cluster), !is.na(cluster), nzchar(cluster))
 
   parts <- strsplit(cluster, split = "/", fixed = TRUE)[[1]]
   okay <- FALSE
   if (length(parts) == 1L) {
     okay <- TRUE
     cluster_name <- cluster
-    cluster_owner <- pico_username()
+    cluster_owner <- pico_username(host = host, ssh_args = ssh_args)
     cluster <- sprintf("%s/%s", cluster_owner, cluster_name)
   } else if (length(parts) == 2L) {
     cluster_owner <- parts[1]
-    if (cluster_owner == pico_username()) {
+    if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
       okay <- TRUE
       cluster_name <- parts[2]
     }
   }
   if (!okay) {
-    stop(sprintf("Argument 'cluster' must be of format '{owner}/{name}' or '{name}' where '{owner}' is your Pico username (%s): %s", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Argument 'cluster' must be of format '{owner}/{name}' or '{name}' where '{owner}' is your Pico username (%s): %s", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   stopifnot(

R/p2p_client_id.R

@@ -9,13 +9,13 @@
 #' otherwise `{pico_name}/friends`.
 #' 
 #' @export
-p2p_cluster_name <- function(users = character(0)) {
+p2p_cluster_name <- function(users = character(0), ...) {
   users <- unique(users)
   name <- getOption("future.p2p.cluster")
   if (is.null(name)) {
-    users <- setdiff(users, pico_username())
+    users <- setdiff(users, pico_username(...))
     name <- if (length(users) == 0) "personal" else "friends"
-    name <- sprintf("%s/%s", pico_username(), name)
+    name <- sprintf("%s/%s", pico_username(...), name)
   }
   name
 }
@@ -30,9 +30,9 @@ p2p_cluster_name <- function(users = character(0)) {
 #' @keywords internal
 p2p_client_id <- local({
   name <- NULL
-  function() {
+  function(...) {
     if (is.null(name)) {
-      user <- pico_username()
+      user <- pico_username(...)
       hostname <- Sys.info()[["nodename"]]
       pid <- Sys.getpid()
       name <<- sprintf("%s@%s:%d", user, hostname, pid)

R/pico.R

@@ -156,7 +156,7 @@ pico_receive_message_dataframe <- function(p, ..., pattern = NULL) {
 #'
 #' @rdname pico_pipe
 pico_hosted_channels <- function(host = "pipe.pico.sh", ssh_args = NULL, timeout = 10.0) {
-  username <- pico_username()
+  username <- pico_username(host = host, ssh_args = ssh_args)
   t_max <- proc.time()[3] + timeout
   pattern_1 <- sprintf(".*[[:blank:]]%s/([^:]+):", username)
   pattern_2 <- sprintf(".*[[:blank:]]%s/([^:]+):[[:blank:]]+[(]Access List:[[:blank:]]+(.*)[)]", username)

R/pico_p2p.R

@@ -465,7 +465,7 @@ pico_p2p_dispatch_future <- function(future) {
 
   ## 1. Connect to pico and say hello
   cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
+  if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
     topic <- sprintf("%s/future.p2p", basename(cluster))
   } else {
     topic <- sprintf("%s/future.p2p", cluster)

R/pico_username.R

@@ -25,6 +25,11 @@ pico_username <- local({
         })
       }
 
+      ## Special case
+      if (grepl("^pipe[.]", host)) {
+        host <- sub("^pipe[.]", "", host)
+      }
+      
       ssh_config <- list(options = ssh_args, host = host)
       args <- c(ssh_config[["options"]], ssh_config[["host"]], "user")
       if (debug) {
@@ -41,7 +46,9 @@ pico_username <- local({
       if (!is.null(status)) {
         stop(sprintf("Failed to infer pico.sh username. Exit code %s", status))
       }
-      stopifnot(length(out) >= 3)
+      if (length(out) < 3) {
+        stop(sprintf("pico_username(): Received unexpected results: [n=%s]\n%s", length(out), paste(out, collapse = "\n")))
+      }
       username <<- structure(out[1], id = out[2], created_on = as.POSIXct(sub("T", " ", out[3])), class = "pico_username")
     }
 

R/worker.R

@@ -16,7 +16,7 @@
 #'
 #' @importFrom processx poll
 #' @export
-worker <- function(cluster = p2p_cluster_name(), host = "pipe.pico.sh", ssh_args = NULL, duration = 60*60) {
+worker <- function(cluster = p2p_cluster_name(host = host, ssh_args = ssh_args), host = "pipe.pico.sh", ssh_args = NULL, duration = 60*60) {
   parts <- strsplit(cluster, split = "/", fixed = TRUE)[[1]]
   if (length(parts) != 2L) {
     stop(sprintf("Argument 'cluster' must be of format '{owner}/{name}': %s", sQuote(cluster)))
@@ -47,7 +47,7 @@ worker <- function(cluster = p2p_cluster_name(), host = "pipe.pico.sh", ssh_args
   info("assert connection to p2p cluster %s", sQuote(cluster))
   worker_id <- p2p_worker_id()
   if (!p2p_can_connect(cluster, name = worker_id, host = host, ssh_args = ssh_args)) {
-    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   channel_prefix <- sprintf("%s_%s", .packageName, session_uuid())
@@ -103,7 +103,7 @@ worker <- function(cluster = p2p_cluster_name(), host = "pipe.pico.sh", ssh_args
 
   info("connect worker %s to p2p cluster %s", sQuote(worker_id), sQuote(cluster))
   cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
+  if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
     topic <- sprintf("%s/future.p2p", basename(cluster))
   } else {
     topic <- sprintf("%s/future.p2p", cluster)
@@ -379,12 +379,12 @@ run_worker <- function(cluster, worker_id, host, ssh_args, duration, channels) {
 
   info("assert connection to p2p cluster %s", sQuote(cluster))
   if (!p2p_can_connect(cluster, name = worker_id, host = host, ssh_args = ssh_args)) {
-    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username())))
+    stop(sprintf("Cannot connect to P2P cluster %s - make sure they have given you (%s) access", sQuote(cluster), sQuote(pico_username(host = host, ssh_args = ssh_args))))
   }
 
   info("connect background worker process %s to p2p cluster %s for %s until %s", sQuote(worker_id), sQuote(cluster), format(duration), expires)
   cluster_owner <- dirname(cluster)
-  if (cluster_owner == pico_username()) {
+  if (cluster_owner == pico_username(host = host, ssh_args = ssh_args)) {
     topic <- sprintf("%s/future.p2p", basename(cluster))
   } else {
     topic <- sprintf("%s/future.p2p", cluster)