#' @param nbytes Number of bytes to serialize a floating-point number; 4 or 8
#' @param endian Endianness to use for (de)serialization. Use "little" or "big" for portability
#' @param verbose Level of verbosity (0/FALSE for nothing or 1/TRUE for all; devel stage)
+#' @param parll TRUE to fully parallelize; otherwise run sequentially (debug, comparison)
#'
#' @return A matrix of the final medoids curves (K2) in rows
#'
nb_series_per_chunk=50*K1, min_series_per_chunk=5*K1, #chunk size
sep=",", #ASCII input separator
nbytes=4, endian=.Platform$endian, #serialization (write,read)
- verbose=FALSE)
+ verbose=FALSE, parll=TRUE)
{
# Check/transform arguments
- if (!is.matrix(getSeries) && !is.function(getSeries) &&
- !methods::is(getSeries, "connection" && !is.character(getSeries)))
+ if (!is.matrix(getSeries) && !bigmemory::is.big.matrix(getSeries)
+ && !is.function(getSeries)
+ && !methods::is(getSeries,"connection") && !is.character(getSeries))
{
- stop("'getSeries': matrix, function, file or valid connection (no NA)")
+ stop("'getSeries': [big]matrix, function, file or valid connection (no NA)")
}
K1 = .toInteger(K1, function(x) x>=2)
K2 = .toInteger(K2, function(x) x>=2)
nbytes = .toInteger(nbytes, function(x) x==4 || x==8)
# Serialize series if required, to always use a function
- bin_dir = ".epclust.bin/"
+ bin_dir = ".epclust_bin/"
dir.create(bin_dir, showWarnings=FALSE, mode="0755")
if (!is.function(getSeries))
{
getSeries = function(inds) getDataInFile(inds, series_file, nbytes, endian)
}
- # Serialize all computed wavelets contributions onto a file
+ # Serialize all computed wavelets contributions into a file
contribs_file = paste(bin_dir,"contribs",sep="") ; unlink(contribs_file)
index = 1
nb_curves = 0
if (verbose)
cat("...Compute contributions and serialize them\n")
- repeat
- {
- series = getSeries((index-1)+seq_len(nb_series_per_chunk))
- if (is.null(series))
- break
- contribs_chunk = curvesToContribs(series, wf, ctype)
- binarize(contribs_chunk, contribs_file, nb_series_per_chunk, sep, nbytes, endian)
- index = index + nb_series_per_chunk
- nb_curves = nb_curves + nrow(contribs_chunk)
- }
+ nb_curves = binarizeTransform(getSeries,
+ function(series) curvesToContribs(series, wf, ctype),
+ contribs_file, nb_series_per_chunk, nbytes, endian)
getContribs = function(indices) getDataInFile(indices, contribs_file, nbytes, endian)
if (nb_curves < min_series_per_chunk)
if (nb_series_per_task < min_series_per_chunk)
stop("Too many tasks: less series in one task than min_series_per_chunk!")
+ runTwoStepClustering = function(inds)
+ {
+ if (parll)
+ require("epclust", quietly=TRUE)
+ indices_medoids = clusteringTask1(
+ inds, getContribs, K1, nb_series_per_chunk, ncores_clust, verbose, parll)
+ if (WER=="mix")
+ {
+
+
+
+
+#TODO: getSeries(indices_medoids) BAD ; il faudrait une big.matrix de medoids en entree
+ #OK en RAM il y en aura 1000 (donc 1000*K1*17519... OK)
+ #...mais du coup chaque process ne re-dupliquera pas medoids
+
+
+ medoids2 = computeClusters2(getSeries(indices_medoids),
+ K2, getSeries, nb_curves, nb_series_per_chunk, ncores_clust, verbose, parll)
+ binarize(medoids2, synchrones_file, nb_series_per_chunk, sep, nbytes, endian)
+ return (vector("integer",0))
+ }
+ indices_medoids
+ }
+
# Cluster contributions in parallel (by nb_series_per_chunk)
indices_all = if (random) sample(nb_curves) else seq_len(nb_curves)
indices_tasks = lapply(seq_len(ntasks), function(i) {
})
if (verbose)
cat(paste("...Run ",ntasks," x stage 1 in parallel\n",sep=""))
-# cl = parallel::makeCluster(ncores_tasks)
-# parallel::clusterExport(cl, varlist=c("getSeries","getContribs","K1","K2",
-# "nb_series_per_chunk","ncores_clust","synchrones_file","sep","nbytes","endian"),
-# envir = environment())
- # 1000*K1 indices [if WER=="end"], or empty vector [if WER=="mix"] --> series on file
-# indices = unlist( parallel::parLapply(cl, indices_tasks, function(inds) {
- indices = unlist( lapply(indices_tasks, function(inds) {
-# require("epclust", quietly=TRUE)
-
- browser() #TODO: CONTINUE DEBUG HERE
-
- indices_medoids = clusteringTask(inds,getContribs,K1,nb_series_per_chunk,ncores_clust)
+ if (WER=="mix")
+ {synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)}
+ if (parll)
+ {
+ cl = parallel::makeCluster(ncores_tasks)
+ varlist = c("getSeries","getContribs","K1","K2","verbose","parll",
+ "nb_series_per_chunk","ncores_clust","sep","nbytes","endian")
if (WER=="mix")
- {
- medoids2 = computeClusters2(
- getSeries(indices_medoids), K2, getSeries, nb_series_per_chunk)
- binarize(medoids2, synchrones_file, nb_series_per_chunk, sep, nbytes, endian)
- return (vector("integer",0))
- }
- indices_medoids
- }) )
-# parallel::stopCluster(cl)
+ varlist = c(varlist, "synchrones_file")
+ parallel::clusterExport(cl, varlist=varlist, envir = environment())
+ }
+
+ # 1000*K1 indices [if WER=="end"], or empty vector [if WER=="mix"] --> series on file
+ if (parll)
+ indices = unlist( parallel::parLapply(cl, indices_tasks, runTwoStepClustering) )
+ else
+ indices = unlist( lapply(indices_tasks, runTwoStepClustering) )
+ if (parll)
+ parallel::stopCluster(cl)
getRefSeries = getSeries
- synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)
if (WER=="mix")
{
indices = seq_len(ntasks*K2)
index = 1
if (verbose)
cat("...Serialize contributions computed on synchrones\n")
- repeat
- {
- series = getSeries((index-1)+seq_len(nb_series_per_chunk))
- if (is.null(series))
- break
- contribs_chunk = curvesToContribs(series, wf, ctype)
- binarize(contribs_chunk, contribs_file, nb_series_per_chunk, sep, nbytes, endian)
- index = index + nb_series_per_chunk
- }
+ ignored = binarizeTransform(getSeries,
+ function(series) curvesToContribs(series, wf, ctype),
+ contribs_file, nb_series_per_chunk, nbytes, endian)
}
+
+
+#TODO: if ntasks==1, c'est deja terminé
+
# Run step2 on resulting indices or series (from file)
if (verbose)
cat("...Run final // stage 1 + stage 2\n")
- indices_medoids = clusteringTask(
- indices, getContribs, K1, nb_series_per_chunk, ncores_tasks*ncores_clust)
- medoids = computeClusters2(getSeries(indices_medoids),K2,getRefSeries,nb_series_per_chunk)
+ indices_medoids = clusteringTask1(
+ indices, getContribs, K1, nb_series_per_chunk, ncores_tasks*ncores_clust, verbose, parll)
+ medoids = computeClusters2(getSeries(indices_medoids), K2,
+ getRefSeries, nb_curves, nb_series_per_chunk, ncores_tasks*ncores_clust, verbose, parll)
# Cleanup
unlink(bin_dir, recursive=TRUE)
}) )
}
-# Helper for main function: check integer arguments with functiional conditions
+# Check integer arguments with functional conditions
.toInteger <- function(x, condition)
{
if (!is.integer(x))