Multithreading of growR_run_loop with parallel package.

DESCRIPTION

@@ -31,7 +31,8 @@ Suggests:
     knitr,
     patchwork,
     rmarkdown,
-    testthat (>= 3.0.0)
+    testthat (>= 3.0.0),
+    parallel
 Depends: 
     R (>= 2.10)
 RdMacros: Rdpack

NEWS.md

@@ -4,6 +4,7 @@
 
 * site_name and run_name are printed to console in `growR_run_loop`.
 * PhenologicalAutocut
+* Multithreading for `growR_run_loop` with the `parallel` package.
 
 ## Changed
 

R/run.R

@@ -1,3 +1,7 @@
+pkgs_for_parallelization = c("parallel")
+parallel_error_message = paste0("The following R packages are required in ",
+  "order to run simulations in parallel:", pkgs_for_parallelization)
+
 #' Run growR simulations
 #'
 #' Start the loop over runs specified in the config file.
@@ -15,6 +19,10 @@
 #'   parameters of the modvege_environments. If `FALSE`, initial conditions 
 #'   are taken as the final state values of the simulation of the previous 
 #'   year.
+#' @param n_threads int; *Experimental* The number of parallel threads to use.
+#'   Requires the `parallel` package to be installed and is not tested on 
+#'   Windows. The simulation workload  of the supplied *modvege_environments* is 
+#'   split evenly across *n_threads* different CPUs.
 #' @return A list of the format `[[run]][[year]]` containing clones of 
 #'   the [ModvegeSite] instances that were run. Also write to files, if 
 #'   *output_dir* is nonempty.
@@ -29,66 +37,141 @@
 #' @export
 #' 
 growR_run_loop = function(modvege_environments, output_dir = "", 
-                          independent = TRUE) {
-  # Parse output dir
-  if (output_dir == "") {
-    write_files = FALSE
-  } else {
-    write_files = TRUE
+                          independent = TRUE, n_threads = 1) {
+  # Set up parallelisation
+  if (n_threads > 1) {
+    # Check if suggested packages are present
+    all_installed = rlang::is_installed(pkgs_for_parallelization)
+    if (!all_installed) {
+      stop(parallel_error_message)
+    }
+  }
+
+  # Ensure target directory exists
+  if (output_dir != "") {
+    if (!dir.exists(output_dir)) {
+      logger(sprintf("Creating output directory `%s`.", output_dir), 
+             level = INFO)
+      dir.create(output_dir, recursive = TRUE)
+    }
   }
   n_runs = length(modvege_environments)
   results = list()
-  for (run in 1:n_runs) {
-    results[[run]] = list()
-    logger(sprintf("Starting run %s out of %s.", run, n_runs), level = INFO)
-    run_environment = modvege_environments[[run]]
-    logger(sprintf(" site name: `%s`.", run_environment$site_name), level = INFO)
-    logger(sprintf("  run name: `%s`.", run_environment$run_name), level = INFO)
-    modvege = ModvegeSite$new(run_environment$parameters,
-                              site_name = run_environment$site_name,
-                              run_name = run_environment$run_name
-                              )
-    n_years = length(run_environment$years)
 
-    #-Model-core----------------------------------------------------------------
-    # For each specified year, run ModVege
-    for (i_year in 1:n_years) {
-      this_year = run_environment$years[i_year]
-      logger(sprintf("[Run %s/%s]Simulating year %s (%s/%s)", run, n_runs, 
-                     this_year, i_year, n_years), level = INFO )
-      E = run_environment$get_environment_for_year(this_year)
-      # Run the ModVege Core
-      modvege$run(this_year, E$W, E$M)
+  if (n_threads > 1) {
+      results = parallel::mclapply(modvege_environments, run_loop_parallel, 
+                                   output_dir, independent, 
+                                   mc.cores = n_threads)
+  } else {
+    # Single-thread run
+    for (run in 1:n_runs) {
+      run_environment = modvege_environments[[run]]
+      logger(sprintf("Starting run %s out of %s.", run, n_runs), level = INFO)
+      logger(sprintf(" site name: `%s`.", run_environment$site_name), 
+             level = INFO)
+      logger(sprintf("  run name: `%s`.", run_environment$run_name), 
+             level = INFO)
+      results[[run]] = run_loop_core(run_environment, run, n_runs, output_dir, 
+                                     independent)
+    } # End of loop over runs
+  }
+  logger("All runs completed.", level = INFO)
+  return(results)
+}
+
+#' Run loop in parallel
+#' 
+#' Simple wrapper around [run_loop_core].
+#' 
+#' Prints a message and passes dummy values for the *run* and *n_runs* 
+#' variables, the former of which is not trivially available from within 
+#' mclapply.
+#'
+#' @param run_environment A ModvegeEnvironment instance; The environment with 
+#'   which to call [run_loop_core].
+#' @param ... Further arguments (except *run* and *n_runs*) to be passed to 
+#'   [run_loop_core].
+#' @return A list of [ModvegeSite] instances containing the simulation 
+#'   results for each year specified in *run_environment*.
+#' @seealso [run_loop_core]
+#'
+#' @keywords internal
+#'
+run_loop_parallel = function(run_environment, ...) {
+  pid = Sys.getpid()
+  message = "[PID:%s]Starting run `%s` for site `%s`."
+  logger(sprintf(message, pid, run_environment$run_name, 
+                 run_environment$site_name), 
+         level = INFO)
+  return(run_loop_core(run_environment, "X", "Y", ...))
+}
+
+#' Core of growR_run_loop
+#'
+#' Create the ModvegeSite object from supplied environment and call it's 
+#' `$run()` method for each supplied year.
+#'
+#' @param run_environment A [ModvegeEnvironment] instance containing all the 
+#'   information necessary to simulate a site.
+#' @param run integer Number of the currently carried out run.
+#' @param n_runs integer Total number of runs to carry out.
+#' @param output_dir str Path to directory where output files will be written 
+#'   to.
+#' @param independent boolean Whether or not simulations of subsequent years 
+#'   are treated as independent.
+#' @return A list of [ModvegeSite] instances containing the simulation 
+#'   results for each year specified in *run_environment*.
+#'
+#' @seealso [growR_run_loop]
+#'
+#' @keywords internal
+#'
+run_loop_core = function(run_environment, run, n_runs, output_dir, 
+                         independent) {
+  results_for_env = list()
+  modvege = ModvegeSite$new(run_environment$parameters,
+                            site_name = run_environment$site_name,
+                            run_name = run_environment$run_name
+                            )
+  n_years = length(run_environment$years)
 
-      #-Write-output------------------------------------------------------------
+  #-Model-core----------------------------------------------------------------
+  # For each specified year, run ModVege
+  for (i_year in 1:n_years) {
+    this_year = run_environment$years[i_year]
+    logger(sprintf("[Run %s/%s]Simulating year %s (%s/%s)", run, n_runs, 
+                   this_year, i_year, n_years), level = TRACE )
+    E = run_environment$get_environment_for_year(this_year)
+    # Run the ModVege Core
+    modvege$run(this_year, E$W, E$M)
 
-      if (write_files) {
-        out_file = sprintf("%s%s_%s.dat",
-                           run_environment$site_name,
-                           run_environment$run_name_in_filename,
-                           this_year)
-        out_path = file.path(output_dir, out_file)
-        logger("Entering `ModvegeSite$write_output`", level = TRACE)
-        modvege$write_output(out_path, force = TRUE)
-      }
+    #-Write-output------------------------------------------------------------
 
-      #-Store-output------------------------------------------------------------
-      results[[run]][[i_year]] = modvege$clone(deep = FALSE)
+    if (output_dir != "") {
+      out_file = sprintf("%s%s_%s.dat",
+                         run_environment$site_name,
+                         run_environment$run_name_in_filename,
+                         this_year)
+      out_path = file.path(output_dir, out_file)
+      logger("Entering `ModvegeSite$write_output`", level = TRACE)
+      modvege$write_output(out_path, force = TRUE)
+    }
 
-      #-Adjust-initial-conditions-for-next-year---------------------------------
-      if (!independent) {
-        n_days = modvege$days_per_year
-        for (initial_condition in c("BMGV", "BMGR", "BMDV", "BMDR", "AgeGV", 
-                                    "AgeGR", "AgeDV", "AgeDR")) {
-          previous_value = modvege[[initial_condition]][n_days]
-          for (ic in c(initial_condition, paste0(initial_condition, "0"))) {
-            modvege$parameters[[ic]] = previous_value
-          }
+    #-Store-output------------------------------------------------------------
+    results_for_env[[i_year]] = modvege$clone(deep = FALSE)
+
+    #-Adjust-initial-conditions-for-next-year---------------------------------
+    if (!independent) {
+      n_days = modvege$days_per_year
+      for (initial_condition in c("BMGV", "BMGR", "BMDV", "BMDR", "AgeGV", 
+                                  "AgeGR", "AgeDV", "AgeDR")) {
+        previous_value = modvege[[initial_condition]][n_days]
+        for (ic in c(initial_condition, paste0(initial_condition, "0"))) {
+          modvege$parameters[[ic]] = previous_value
         }
       }
-    } # End of loop over simulation years
-  } # End of loop over runs
-  logger("All runs completed.", level = INFO)
-  return(results)
+    }
+  } # End of loop over simulation years
+  return(results_for_env)
 }