- cl = parallel::makeCluster(ncores_tasks)
- #1000*K1 (or K2) indices (or NOTHING--> series on file)
- indices = unlist( parallel::parLapply(cl, indices_tasks, function(inds) {
- clusteringTask(inds, getSeries, getSeries, getCoefs, K1, K2*(WER=="mix"),
- nb_series_per_chunk,ncores_clust,to_file=TRUE)
- }) )
- parallel::stopCluster(cl)
-
- getSeriesForSynchrones = getSeries
- synchrones_file = paste(bin_dir,"synchrones",sep="")
+
+ if (parll && ntasks>1)
+ {
+ # Initialize parallel runs: outfile="" allow to output verbose traces in the console
+ # under Linux. All necessary variables are passed to the workers.
+ cl = parallel::makeCluster(ncores_tasks, outfile="")
+ varlist = c("getSeries","getContribs","K1","K2","algo_clust1","algo_clust2",
+ "nb_series_per_chunk","nb_items_clust","ncores_clust","sep",
+ "nbytes","endian","verbose","parll")
+ if (WER=="mix")
+ varlist = c(varlist, "medoids_file")
+ parallel::clusterExport(cl, varlist, envir = environment())
+ }
+
+ # This function achieves one complete clustering task, divided in stage 1 + stage 2.
+ # stage 1: n indices --> clusteringTask1(...) --> K1 medoids
+ # stage 2: K1 medoids --> clusteringTask2(...) --> K2 medoids,
+ # where n = N / ntasks, N being the total number of curves.
+ runTwoStepClustering = function(inds)
+ {
+ # When running in parallel, the environment is blank: we need to load required
+ # packages, and pass useful variables.
+ if (parll && ntasks>1)
+ require("epclust", quietly=TRUE)
+ indices_medoids = clusteringTask1(
+ inds, getContribs, K1, nb_series_per_chunk, ncores_clust, verbose, parll)
+ if (WER=="mix")
+ {
+ if (parll && ntasks>1)
+ require("bigmemory", quietly=TRUE)
+ medoids1 = bigmemory::as.big.matrix( getSeries(indices_medoids) )
+ medoids2 = clusteringTask2(medoids1, K2, getSeries, nb_curves, nb_series_per_chunk,
+ nbytes, endian, ncores_clust, verbose, parll)
+ binarize(medoids2, medoids_file, nb_series_per_chunk, sep, nbytes, endian)
+ return (vector("integer",0))
+ }
+ indices_medoids
+ }
+
+ # Synchrones (medoids) need to be stored only if WER=="mix"; indeed in this case, every
+ # task output is a set of new (medoids) curves. If WER=="end" however, output is just a
+ # set of indices, representing some initial series.
+ if (WER=="mix")
+ {medoids_file = paste(bin_dir,"medoids",sep="") ; unlink(medoids_file)}
+
+ if (verbose)
+ {
+ message = paste("...Run ",ntasks," x stage 1", sep="")
+ if (WER=="mix")
+ message = paste(message," + stage 2", sep="")
+ cat(paste(message,"\n", sep=""))
+ }
+
+ # As explained above, indices will be assigned to ntasks*K1 medoids indices [if WER=="end"],
+ # or nothing (empty vector) if WER=="mix"; in this case, medoids (synchrones) are stored
+ # in a file.
+ indices <-
+ if (parll && ntasks>1)
+ unlist( parallel::parLapply(cl, indices_tasks, runTwoStepClustering) )
+ else
+ unlist( lapply(indices_tasks, runTwoStepClustering) )
+ if (parll && ntasks>1)
+ parallel::stopCluster(cl)
+
+ # Right before the final stage, two situations are possible:
+ # a. data to be processed now sit in binary format in medoids_file (if WER=="mix")
+ # b. data still is the initial set of curves, referenced by the ntasks*K1 indices
+ # So, the function getSeries() will potentially change. However, computeSynchrones()
+ # requires a function retrieving the initial series. Thus, the next line saves future
+ # conditional instructions.
+ getRefSeries = getSeries
+