switched parallelization to doParallel

R/Rphree_Kinetics.R

 ##  Functions for dealing with simulations with kinetics
 
 ### Marco De Lucia, delucia@gfz-potsdam.de, 2009-2018
-### Time-stamp: "Last modified 2018-05-04 20:47:46 delucia"
+### Time-stamp: "Last modified 2018-05-16 16:45:40 delucia"
 
 ##' This function just runs the generated input buffer - or a list
 ##' thereof - through \code{phreeqc}. Obviously it requires the
@@ -11,7 +11,7 @@
 ##' RPhreeFile("phreeqc.dat", is.db=TRUE ))}
 ##'
 ##' Currently parallelization is achieved using
-##' \code{parallel::mclapply}, this can change in the future. At the
+##' \code{parallel::parLapply}, this can change in the future. At the
 ##' moment the function doesn't check that the SELCTED_OUTPUT block is
 ##' well-formed, so this is responsibility of the user.
 ##' @title Run the generated input buffer using phreeqc
@@ -47,8 +47,10 @@ RunPQC <- function(input, procs=1) {
         }
     } else {
         ## go parallel!
-        ## call to parallel::mclapply
-        res <- parallel::mclapply(input, .runPQC, mc.silent=TRUE, mc.cores=procs)
+
+        ## old one
+        ## res <- parallel::mclapply(input, .runPQC, mc.silent=TRUE, mc.cores=procs)
+        res <- parallel::parLapply(input, .runPQC)
 
         ## a is the string containing the rbind of each element of the list
         a <- paste("rbind(",paste("res[[",1:procs,"]]",sep="", collapse = ", "),")")

R/Rphree_ReactTrans.R

 ##  Functions for dealing with surrogate simulations
 
 ### Marco De Lucia, delucia@gfz-potsdam.de, 2009-2018
-### Time-stamp: "Last modified 2018-05-04 20:35:35 delucia"
+### Time-stamp: "Last modified 2018-05-16 16:51:41 delucia"
 
 
 ##' Todo
@@ -223,7 +223,7 @@ ReactTranspBalanceEq <- function(setup, init, maxtime, step=c("time","iter","fix
                                  surrogate.FUN, model, baleq, tol=1E-9, call_pqc=TRUE)
     
 {
-
+    require(doParallel)
     if (surrogate) {
         if (missing(surrogate.FUN))
             stopmsg("Need a valid function to apply surrogate instead of chemistry")
@@ -243,6 +243,12 @@ ReactTranspBalanceEq <- function(setup, init, maxtime, step=c("time","iter","fix
     immobile <- setup$immobile
     initsim  <- setup$initsim
 
+    if (procs > 1) {
+        registerDoParallel(procs)
+        msg("Registered default doParallel cluster with ", procs, "nodes")
+    }
+    cl <- getDefaultCluster()
+
     n  <- setup$n
     L  <- setup$len  # m
     U  <- setup$U    # m3/s
@@ -366,6 +372,7 @@ ReactTranspBalanceEq <- function(setup, init, maxtime, step=c("time","iter","fix
     on.exit({
         msg("MAIN LOOP interrupted during iteration",iter,"!!")
         msg("returning last calculated chemistry.")
+        parallel::stopCluster()
         print(traceback())
         msg(" Bye.")
         attr(state_C,"timing") <- timing
@@ -460,6 +467,9 @@ ReactTranspBalanceEq <- function(setup, init, maxtime, step=c("time","iter","fix
         
     }
     on.exit()
+    if (procs > 1)
+        parallel::stopCluster()
+    
     msg("total time of chemistry: ",round(sum(timing[,3]),3),"seconds")
     msg("total number of simulations: ",sum(timing[,1]))
     msg(" Bye.")
@@ -524,7 +534,12 @@ ReactTranspBalanceKin <- function(setup, init, maxtime, step=c("time","iter","fi
     kin      <- setup$kin
     initsim  <- setup$initsim
     ann      <- setup$ann
-    
+
+    if (procs > 1) {
+        registerDoParallel(procs)
+        msg("Registered default doParallel cluster with ", procs, "nodes")
+    }
+
     n  <- setup$n
     L  <- setup$len  # m
     U  <- setup$U    # m3/s

R/Surr_Train.R

@@ -2,7 +2,7 @@
 
 ### Marco De Lucia, delucia@gfz-potsdam.de, 2009-2018
 ### Janis Jatnieks, janisj@gfz-potsdam.de, jatnieks@janis.es
-### Time-stamp: "Last modified 2018-05-16 16:30:07 delucia"
+### Time-stamp: "Last modified 2018-05-16 16:38:01 delucia"
 ### Licence: LGPL version 2.1
 
 ## resolve dependencies automatically
@@ -29,10 +29,8 @@ start_up <- function(session_path=FALSE, ## to disable set to a non-string
 {
     gc()
     
-    if (Sys.info()[['sysname']]=="Linux") {
-        cat("\nYou are not on Windows, loading doMC!\n")
-        list.of.packages = c(list.of.packages,"doMC")
-    }
+    msg("loading doParallel")
+    list.of.packages = c(list.of.packages,"doParallel")
     
     ## load session data first
     ## if (class(session_path)[1]=="character" & file.exists(session_path) ) { 
@@ -57,11 +55,8 @@ start_up <- function(session_path=FALSE, ## to disable set to a non-string
     
     loadsuccess <- lapply(list.of.packages, require, character.only=TRUE)
     
-    if (!Sys.info()[["sysname"]]=="Windows") {
-        cat(":: Registering parallelization with ", use_cores, "cores\n")
-        registerDoMC( use_cores ) ## set up parallelization
-    }
-
+    msg("Registering parallelization with ", use_cores, "cores")
+    registerDoParallel( use_cores ) ## set up parallelization
 
     ### MDL: some global definitions
     ## we use this to swich calls between Dice and caret or direct calls
@@ -493,12 +488,9 @@ FastClust <- function(dfrm, clustnum=7, g=TRUE, link_type="complete") {
 ##' @export
 safe_get_cores <- function() {
     ## register parallelization backends on anyhing but win, due to missing fork()
-    if ( !.Platform$OS.type=='windows') {
-        require(doMC)
-        require(parallel)
-        ## detect using 
-        report_cores = detectCores() }
-    else report_cores = 1
+    require(doParallel)
+    ## detect using 
+    report_cores <- detectCores() 
     return( report_cores )
 }
 
@@ -1992,7 +1984,7 @@ Train <- function(surrogate_types = c(),
     ##         cat("\nManually forcing",use_cores,"cores for parallelization!\n")  
     ##         registerDoMC( use_cores ) 
     ##     }
-    ## } else { cat("\nTried to set number of processing cores manually, but doMC does not appear to be loaded, ignoring!\n\n") } 
+    ## } else { cat("\nTried to set number of processing cores manually, but doParallel does not appear to be loaded, ignoring!\n\n") } 
     
     ## cleanup, so no old data strs can affect new batch by mistake
     mrm( c("Tin","Oin","Oout","Cout","Fin","Fout","Tout","SRes","Sperf","valOUT","perftop","Residuals") )