Convergence_Plot: Refactor to use Set_parallel function for cluster management

Author: elgabbas

NAMESPACE

@@ -129,7 +129,6 @@ export(SaveSessionInfo)
 export(Scale_0_1)
 export(ScrapLinks)
 export(ScriptLocation)
-export(SetChainName)
 export(Set_parallel)
 export(Solve2)
 export(Solve2vect)

R/General_Set_parallel.R

@@ -5,9 +5,7 @@
 #' Set up or stop parallel processing plan
 #'
 #' Configures parallel processing with `future::plan()` or stops an existing
-#' plan. On Windows, it uses `snow::makeSOCKcluster()` with `future::cluster`;
-#' on Linux/macOS, it uses `future::multicore`. When stopping, it resets to
-#' sequential mode.
+#' plan. When stopping, it resets to sequential mode.
 #'
 #' @param NCores Integer. Number of cores to use. If `NULL`, defaults to
 #'   sequential mode.
@@ -43,19 +41,29 @@ Set_parallel <- function(
     NCores = 1L, Stop = FALSE, Cat = TRUE, Level = 0L, Future_maxSize = 8) {
 
   # Validate NCores input
-  NCores <- ifelse(is.null(NCores) || NCores < 1, 1L, as.integer(NCores))
+  NCores <- ifelse(
+    (is.null(NCores) || NCores < 1),
+    1L, as.integer(NCores))
+
+  # NCores can not be more than the available cores
+  NCores <- ifelse(
+    NCores > parallelly::availableCores(),
+    {
+      warning(
+        paste0(
+          "`NCores` > number of available cores. ",
+          "It was reset to the number of available cores",
+          parallelly::availableCores()),
+        call. = FALSE)
+      parallelly::availableCores()
+    },
+    NCores)
 
   if (Stop) {
     if (Cat) {
       IASDT.R::CatTime("Stopping parallel processing", Level = Level)
     }
 
-    if (.Platform$OS.type == "windows" &&
-        exists("c1", envir = parent.frame())) {
-      try(snow::stopCluster(get("c1", envir = parent.frame())), silent = TRUE)
-      rm("c1", envir = parent.frame())
-    }
-
     # Stop any running future plan and reset to sequential
     future::plan("future::sequential", gc = TRUE)
 
@@ -75,17 +83,7 @@ Set_parallel <- function(
       .local_envir = parent.frame())
 
     if (NCores > 1) {
-
-      if (.Platform$OS.type == "windows") {
-        # Windows: Use cluster (SOCK) since multicore is unavailable
-        c1 <- snow::makeSOCKcluster(NCores)
-        # assigns to the calling function's environment
-        assign("c1", c1, envir = parent.frame())
-        future::plan("future::cluster", workers = c1)
-      } else {
-        # Linux/Mac: Use multicore for better performance
-        future::plan("future::multicore", workers = NCores)
-      }
+      future::plan("future::multisession", workers = NCores)
     } else {
       future::plan("future::sequential", gc = TRUE)
     }

R/Mod_Convergence_Plot.R

@@ -468,12 +468,14 @@ Convergence_Plot <- function(
 
   FileConv_Beta <- IASDT.R::Path(Path_Convergence, "Convergence_Beta.RData")
 
-
   if (file.exists(FileConv_Beta)) {
 
     IASDT.R::CatTime("Loading plotting data", Level = 1)
     PlotObj_Beta <- IASDT.R::LoadAs(FileConv_Beta)
 
+    rm(Obj_Beta, envir = environment())
+    invisible(gc())
+
   } else {
 
     HTML1 <- "<span style='color:blue;'><b>"
@@ -497,6 +499,7 @@ Convergence_Plot <- function(
     HTML1 <- "<span style='color:blue;'><b>"
     HTML2 <- "</b></span><span style='color:grey;'>"
     HTML3 <- "</span>"
+    HTML4 <- "&nbsp;&nbsp;&nbsp;&mdash;&nbsp;&nbsp;&nbsp;"
     VarsDesc <- tibble::tribble(
       ~Variable, ~VarDesc,
       "bio1", "annual mean temperature",
@@ -525,9 +528,9 @@ Convergence_Plot <- function(
           "{HTML1}{stringr::str_to_sentence(Variable)}\\
         {HTML2} [{VarDesc}]{HTML3} - {Term}"),
         VarDesc = stringr::str_replace(
-          VarDesc, " - L$", "&nbsp&nbsp&mdash&nbsp&nbspLinear"),
+          VarDesc, " - L$", paste0(HTML4, "Linear")),
         VarDesc = stringr::str_replace(
-          VarDesc, " - Q$", "&nbsp&nbsp&mdash&nbsp&nbspQuadratic"),
+          VarDesc, " - Q$", paste0(HTML4, "Quadratic")),
         Variable = paste0(Variable, "_", Term),
         Term = NULL) %>%
       dplyr::bind_rows(LinearTerms)
@@ -629,9 +632,6 @@ Convergence_Plot <- function(
 
     # Prepare working on parallel
     IASDT.R::Set_parallel(NCores = min(NCores, nrow(Beta_DF)), Level = 2)
-    if (.Platform$OS.type == "windows") {
-      on.exit(try(snow::stopCluster("c1"), silent = TRUE), add = TRUE)
-    }
     on.exit(future::plan("future::sequential", gc = TRUE), add = TRUE)
 
     # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
@@ -773,7 +773,7 @@ Convergence_Plot <- function(
     # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
 
     # Stopping cluster
-    IASDT.R::Set_parallel(Stop = TRUE, Cat = TRUE, Level = 2)
+    IASDT.R::Set_parallel(Stop = TRUE, Level = 2)
 
     rm(Beta_DF, BetaNames, envir = environment())
     invisible(gc())
@@ -803,17 +803,15 @@ Convergence_Plot <- function(
   # Prepare working on parallel
   IASDT.R::Set_parallel(
     NCores = min(NCores, nrow(BetaTracePlots_ByVar)), Level = 2)
-
-  if (.Platform$OS.type == "windows") {
-    on.exit(try(snow::stopCluster("c1"), silent = TRUE), add = TRUE)
-  }
   on.exit(future::plan("future::sequential", gc = TRUE), add = TRUE)
 
   # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
 
   IASDT.R::CatTime("Save plots", Level = 2)
+  VarNames <- BetaTracePlots_ByVar$Variable
+
   BetaTracePlots_ByVar0 <- future.apply::future_lapply(
-    X = BetaTracePlots_ByVar$Variable,
+    X = VarNames,
     FUN = function(x) {
 
       VarDesc <- BetaTracePlots_ByVar %>%
@@ -903,16 +901,16 @@ Convergence_Plot <- function(
     future.seed = TRUE,
     future.globals = c("BetaTracePlots_ByVar", "NRC", "Path_Convergence"),
     future.packages = c(
-      "dplyr", "ggplot2", "magrittr", "purrr", "IASDT.R", "ggtext",
-      "tibble", "tidyr", "cowplot", "grid"))
+      "tidyr", "dplyr", "ggplot2", "purrr", "ggtext",
+      "tibble", "cowplot", "grDevices", "IASDT.R"))
 
   rm(BetaTracePlots_ByVar0, BetaTracePlots_ByVar, envir = environment())
   invisible(gc())
 
   # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
 
   # Stopping cluster
-  IASDT.R::Set_parallel(Stop = TRUE, Cat = TRUE, Level = 2)
+  IASDT.R::Set_parallel(Stop = TRUE, Level = 2)
 
   # # ..................................................................... ###
 
@@ -933,10 +931,6 @@ Convergence_Plot <- function(
   # Prepare working on parallel
   IASDT.R::Set_parallel(
     NCores = min(NCores, nrow(BetaTracePlots_BySp)), Level = 2)
-
-  if (.Platform$OS.type == "windows") {
-    on.exit(try(snow::stopCluster("c1"), silent = TRUE), add = TRUE)
-  }
   on.exit(future::plan("future::sequential", gc = TRUE), add = TRUE)
 
   # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
@@ -1018,17 +1012,17 @@ Convergence_Plot <- function(
       rm(PlotTitle, envir = environment())
       return(invisible(NULL))
     },
-    future.scheduling = 1, future.seed = TRUE,
+    future.seed = TRUE,
     future.globals = c(
       "MarginType", "BetaTracePlots_BySp", "Path_Convergence_BySp", "Beta_NRC"),
     future.packages = c(
       "dplyr", "coda", "ggplot2", "ggExtra", "ggtext", "IASDT.R",
-      "stringr", "gtools", "cowplot", "purrr"))
+      "stringr", "gtools", "cowplot", "purrr", "grDevices"))
 
   # # |||||||||||||||||||||||||||||||||||||||||||||||||||||||| ##
 
   # Stopping cluster
-  IASDT.R::Set_parallel(Stop = TRUE, Cat = TRUE, Level = 2)
+  IASDT.R::Set_parallel(Stop = TRUE, Level = 2)
 
   rm(BetaTracePlots_BySp0, envir = environment())
 

R/Mod_Prep4HPC.R

@@ -1300,18 +1300,28 @@ Mod_Prep4HPC <- function(
 
   IASDT.R::CatTime("Save data to disk")
 
+  SetChainName <- function(Obj, Chain) {
+    if (is.null(Obj) || is.null(Chain)) {
+      stop("Obj and Chain cannot be empty", call. = FALSE)
+    }
+    Obj %>%
+      unlist() %>%
+      as.vector() %>%
+      stats::setNames(paste0("Chain", unlist(Chain)))
+  }
+  
   Model_Info <- Model_Info %>%
     tidyr::nest(
       Post_Path = Post_Path, Path_ModProg = Path_ModProg,
       Chain = Chain, Command_HPC = Command_HPC, Command_WS = Command_WS,
       Post_Missing = Post_Missing) %>%
     dplyr::mutate(
-      Post_Path = purrr::map2(Post_Path, Chain, IASDT.R::SetChainName),
-      Chain = purrr::map2(Chain, Chain, IASDT.R::SetChainName),
-      Command_HPC = purrr::map2(Command_HPC, Chain, IASDT.R::SetChainName),
-      Command_WS = purrr::map2(Command_WS, Chain, IASDT.R::SetChainName),
-      Path_ModProg = purrr::map2(Path_ModProg, Chain, IASDT.R::SetChainName),
-      Post_Missing = purrr::map2(Post_Missing, Chain, IASDT.R::SetChainName),
+      Post_Path = purrr::map2(Post_Path, Chain, SetChainName),
+      Chain = purrr::map2(Chain, Chain, SetChainName),
+      Command_HPC = purrr::map2(Command_HPC, Chain, SetChainName),
+      Command_WS = purrr::map2(Command_WS, Chain, SetChainName),
+      Path_ModProg = purrr::map2(Path_ModProg, Chain, SetChainName),
+      Post_Missing = purrr::map2(Post_Missing, Chain, SetChainName),
       Post_Aligned = NA)
 
   save(Model_Info, file = Path_ModelDT)

R/Mod_SetChainName.R

@@ -193,7 +193,7 @@ reference:
   - AddMissingCols
 - subtitle: Other functions
   contents:
-  - SetChainName
+  - Set_parallel
   - CurrOS
   - System
   - CheckURL