Adds parallel implementation using foreach

DESCRIPTION

@@ -37,13 +37,16 @@ Imports:
     mgcv,
     xtable,
     Rdpack,
-    data.table
+    data.table,
+    foreach,
+    parallel
 Suggests:
     gridExtra,
     grid,
     png,
     Hmisc,
-    rbenchmark
+    rbenchmark,
+    doParallel
 License: GPL (>= 3) | file LICENSE
 URL: https://github.com/vwmaus/dtwSat/
 BugReports: https://github.com/vwmaus/dtwSat/issues

NAMESPACE

@@ -87,6 +87,8 @@ importFrom(dtw,asymmetric)
 importFrom(dtw,rabinerJuangStepPattern)
 importFrom(dtw,symmetric1)
 importFrom(dtw,symmetric2)
+importFrom(foreach,"%dopar%")
+importFrom(foreach,foreach)
 importFrom(grDevices,gray.colors)
 importFrom(grDevices,terrain.colors)
 importFrom(lubridate,"day<-")

R/twdtwApply.R

@@ -344,8 +344,8 @@ twdtwApply.twdtwRaster = function(x, y, weight.fun, dist.method, step.matrix, n,
   fun
 }
 
-fasttwdtwApply = function(x, y, dist.method="Euclidean", step.matrix = symmetric1, n=NULL, progress = "text", ncores = 1,
-                          span=NULL, min.length=0, breaks=NULL, from=NULL, to=NULL, by=NULL, overlap=0.5, fill = 255, filepath="", ...){
+fasttwdtwApply = function(x, y, dist.method="Euclidean", step.matrix = symmetric1, n=NULL, progress = "text", ncores = 1, paralle = FALSE,
+                          span=NULL, min.length=0, breaks=NULL, from=NULL, to=NULL, by=NULL, overlap=0.5, fill = 255, filepath="", chunksize, minrows=1, ...){
   # x = rts
   # y = temporal_patterns
   # dist.method="Euclidean"
@@ -431,7 +431,7 @@ fasttwdtwApply = function(x, y, dist.method="Euclidean", step.matrix = symmetric
     names(vv) <- names(out)
   }
 
-  bs <- blockSize(x@timeseries[[1]])
+  bs <- blockSize(x@timeseries[[1]], chunksize = chunksize, minrows = minrows)
   bs$array_rows <- cumsum(c(1, bs$nrows*out[[1]]@ncols))
   pb <- pbCreate(bs$n, progress)
 
@@ -452,9 +452,14 @@ fasttwdtwApply = function(x, y, dist.method="Euclidean", step.matrix = symmetric
     })
 
     # Apply TWDTW analysis
-    twdtw_results <- parallel::mclapply(ts, mc.cores = ncores, FUN = twdtwReduceTime, y = y, breaks = breaks, ...)
+    twdtw_results <- foreach(
+      i = ts, 
+      .combine = 'rbind'
+    ) %dopar% {
+      twdtwReduceTime(x = i, y = y, breaks = breaks, ...)
+    }
 
-    twdtw_results <- data.table::rbindlist(twdtw_results)[,c("label","distance")]
+    # twdtw_results <- data.table::rbindlist(twdtw_results)[,c("label","distance")]
     twdtw_label <- matrix(twdtw_results$label, ncol = length(breaks)-1, byrow = TRUE)
     twdtw_distance <- matrix(twdtw_results$distance, ncol = length(breaks)-1, byrow = TRUE)
 

R/zzz.R

@@ -34,6 +34,7 @@
 #' @import ggplot2
 #' @import methods
 #' @import rgdal 
+#' @importFrom foreach foreach %dopar%
 #' @importFrom proxy dist pr_DB
 #' @importFrom reshape2 melt
 #' @importFrom scales pretty_breaks date_format percent

examples/fast_twdtw.R

@@ -38,16 +38,23 @@
   # Create temporal patterns 
   temporal_patterns <- createPatterns(training_ts, freq = 8, formula = y ~ s(x))
 
-  # Set TWDTW weight function 
-  # log_fun <- logisticWeight(-0.1, 50)
-
-  # Run fast-TWDTW analysis
+  # Run sequential fast-TWDTW analysis
+  foreach::registerDoSEQ()
   system.time(
-    # The logistic time weigh is codeded in Fortran: TODO: add logit parameters to function call
-    # parallel uses parallel::mclapply - not so much implementation
-    fast_lucc <- dtwSat:::fasttwdtwApply(x = rts, y = temporal_patterns, ncores = 1, progress = 'text')
+    # The logistic time weigh is in the Fortran code: TODO: add logit parameters to function call
+    fast_lucc <- dtwSat:::fasttwdtwApply(x = rts, y = temporal_patterns, progress = 'text', minrows = 27)
   )
 
+  # Run parallel fast-TWDTW 
+  cl <- parallel::makeCluster(parallel::detectCores(), type = "FORK")
+  doParallel::registerDoParallel(cl)
+  foreach::getDoParRegistered()
+  system.time(
+    fast_lucc <- dtwSat:::fasttwdtwApply(x = rts, y = temporal_patterns, progress = 'text', minrows = 27)
+  )
+  foreach::registerDoSEQ()
+  parallel::stopCluster(cl)
+
   # Plot TWDTW distances for the first year 
   plot(fast_lucc, type = "distance", time.levels = 1)