+ 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","algoClust1","algoClust2",
+ "nb_series_per_chunk","nb_items_clust1","ncores_clust",
+ "sep","nbytes","endian","verbose","parll")
+ if (WER=="mix" && ntasks>1)
+ 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 the required
+ # packages, and pass useful variables.
+ if (parll && ntasks>1)
+ require("epclust", quietly=TRUE)
+ indices_medoids = clusteringTask1(
+ inds, getContribs, K1, algoClust1, nb_series_per_chunk, ncores_clust, verbose, parll)
+ if (WER=="mix" && ntasks>1)
+ {
+ if (parll)
+ require("bigmemory", quietly=TRUE)
+ medoids1 = bigmemory::as.big.matrix( getSeries(indices_medoids) )
+ medoids2 = clusteringTask2(medoids1, K2, algoClust2, 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" && ntasks>1)
+ {medoids_file = ".medoids.bin" ; 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, 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 a 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
+
+ if (WER=="mix" && ntasks>1)
+ {
+ indices = seq_len(ntasks*K2)
+ # Now series (synchrones) must be retrieved from medoids_file
+ getSeries = function(inds) getDataInFile(inds, medoids_file, nbytes, endian)
+ # Contributions must be re-computed
+ unlink(contribs_file)
+ index = 1
+ if (verbose)
+ cat("...Serialize contributions computed on synchrones\n")
+ ignored = binarizeTransform(getSeries,
+ function(series) curvesToContribs(series, wav_filt, contrib_type),
+ contribs_file, nb_series_per_chunk, nbytes, endian)
+ }