[RÉSOLU] mclapply qui finit avec un seul thread actif

[Dominique Soudant]

Bonjour (ça fait un moment que je n'avais pas posté),

Contexte
43000+ séries temporelles à traiter, je fais ça en 12 lots de 3600+ séries, 2 modèles par série (i.e. 2 max de vraisemblance) le sujet n'étant pas ce que je fais mais le volume que ça représente.
Parallélisation à coup de mclapply
24 proc (double xeon) 64 gig de RAM, debian 10.

Problème
Je commence bien avec mes 24 proc qui marnent comme des fous. Et puis quelques heures après y'en a plus qu'un qui bosse.

Pistes
1) je me suis dit : équirépartition des séries sur les proc mais toutes les séries ne prennent pas le même temps => mc.preschedule=FALSE permet de ne pas faire cette équirépartition a priori (j'ai peut être mal compris) et de continuer à répartir les tâches sur tous les fils. Mise en oeuvre : c'est pas ça, au bout de quelques heures toujours qu'un seul fil qui tourne.
2) je regarde l'utilisation de la mémoire. J'ai l'impression que beaucoup de RAM est utilisée genre plus de 50GO sur les 64GO, je me dis que c'est peut être ça le pb et un peu de recherche sur internet me laisse supposer que cela pourrait être le cas => je modifie la fonction appelée par mclapply en la finissant par un rm(list=ls()) et gc(). Mise en oeuvre : c'est pas ça, au bout de quelques heures j'ai 3500 séries de traitées mais plus qu'un fil actif et il va mettre des heures à traiter les 100+ qui restent ... À toutes fins utiles (top) :

Code : Tout sélectionner

%Cpu(s):  4,2 us,  0,0 sy,  0,0 ni, 95,7 id,  0,1 wa,  0,0 hi,  0,0 si,  0,0 st
MiB Mem :  64350,6 total,  55698,8 free,   2036,5 used,   6615,3 buff/cache
MiB Swap:      0,0 total,      0,0 free,      0,0 used.  61763,4 avail Mem

  PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND                                                                                                                                   
 7953           20   0 1460956   1,1g   8964 R 100,0   1,8 342:28.79 R

Donc avant j'avais ma used qui était à 50GO, là je suis à 2GO, c'est mieux.

J'aime pas trop VIRT=1460956, 1.5TO ? Ça me semble pas cool.

Toute aide est la bienvenue.
Merci d'avance
@+
DS

[Mickael Canouil]

Bonjour,

un exemple reproductible simple peut-être ?

Sinon, je ne peux que recommender l'usage du futureverse (https://www.futureverse.org/) pour la parallélisation.
Dans votre cas, se serait une combinaison des extensions {future} et {futre.apply}.

Cordialement,

[Dominique Soudant]

Hello,
Merci de l'intérêt.
Malheureusement non, 43000 fichiers de l'ordre d1MO chacun et un code de plusieurs millier de lignes réparties en fonctions et en fichiers. C'est pour ça que je pars du diagnostique. Mais je vais y reflechir.
Merci
@+
DS

[Mickael Canouil]

Ce n'est pas tant le contenu, que le code qui gère la parallelisation qui importe ici.

Voir par exemple viewtopic.php?f=3&t=10522

[Logez Maxime]

Bonjour,

Tu peux avoir des problèmes de mémoires, mais surtout des problèmes de répartition des travaux entre tes processus fils.
Si tu as regroupé ensemble les séries chronologiques qui prennent le plus de temps alors, tu peux très bien avoir 11 processus qui ont finit et tu attends le résultat du dernier. Quand il y a de fortes différences de temps de calcul il te faut utiliser un cluster avec du load balancing comme tu peux le faire avec les fonctions parLapplyLB ou encore avec foreach en specifiant .inorder = FALSE. Tu auras toujours le problème d'avoir un temps de calcul qui dépende du calcul le plus lent. Il te faut peut-être revoir ta façon lancer ton calcul, en séparant tes séries chronologiques différemment, par exemple en paquets plus petits ( << 3600), pour que les séries chronologiques les plus lentes se répartissent entre plusieurs éléments.Dans un cas comme ça, le temps de calcul lier au fonctionnement du cluster va forcément augmenter, mais il sera compenser par une meilleur utilisation des processus fils. Il y a une sorte de compromis à trouver entre le nombre d'opérations que tu réalises à chaque fois et le nombre de processus a lancer par le cluster. Celui-ci dépend notamment du temps de calcul de chaque élément simple. Et après il y a bien sur la question de la mémoire ...

Après pour être sur que les calculs sont terminés pour les autres coeurs, tu peux très bien générer des .Rdata à la fin de chaque opération et les ouvrir par ailleurs.

Cordialement,
Maxime

[Dominique Soudant]

Hello,

OK :

Code : Tout sélectionner

setwd(.DERIVED.DATA.SETS)
load("SATCHLOROA_Filtrage.rda")
Data00 <- SATCHLOROA_Filtrage

setwd(.OUT)

options(mc.cores = parallel:::detectCores())
getOption("mc.cores")

cat("Nb séries : ", length(unique(Data00$Série)))
Sys.time()     
if("package:parallel" %in% search()){
  mclapply(unique(Data00$Série)
          ,function(x) TraitementSérie(Data00[Data00$Série == x,]
                                      ,Log=TRUE
                                      ,PeriodicFactorForm=NULL
                                      ,NbMaxModele=2
                                      ))

} else {
  by(Data00
    ,Data00$Série
    ,function(x) TraitementSérie(x
                                ,Log=TRUE
                                ,PeriodicFactorForm=NULL
                                ,NbMaxModele=2
#                                ,debug=TRUE
                                )
    )
}
Sys.time()

TraitementSérie est une fonction qui construit une fonction-modèle, en fait l'optimisation par max de vraisemblance et écrit le résultat dans un fichier rda.

@+
DS

[Dominique Soudant]

Salut Maxime, content de te relire.

Je pensais que l'option mc.preschedule=FALSE

if set to TRUE then the computation is first divided to (at most) as many jobs are there are cores and then the jobs are started, each job possibly covering more than one value. If set to FALSE then one job is forked for each value of X. The former is better for short computations or large number of values in X, the latter is better for jobs that have high variance of completion time and not too many values of X compared to mc.cores.

résolvait ce pb de répartition entre les cœurs et donc (cf. ci-dessus) à l'usage ça change rien au symptôme "plus qu'un cœur qui tourne". En seconde analyse, toutes les séries font la même longueur 20ans*24 données par an, ça veut pas dire qu'elles prennent le même temps à être traitées, mais intuitivement et par expérience ça devrait pas être très différent. Du coup j'ai repassé mc.preschedule=TRUE (valeur par défaut).

Je vais essayer autre chose, parLapplyLB ça doit être facile à mettre en oeuvre.

Sinon je génère des rda pour chaque série et j'ai aussi une trace d'exécution qui s'alimente au cours de l'exécution.

Merci
@+
DS

[Logez Maxime]

re,

Content de te lire aussi, ça faisait un moment ...

dans ton code tel qu'il est écrit, tu ne permets pas le load balançing, en tout cas ce n'était pas explicite du fait que l'option n'était pas présentée.
Si je ne dis pas de bêtises, chaque coeur va traiter d'un modèle à la fois.
Les processus sont envoyés par vague, ici 24 processus lancés en même temps.
Une fois les 24 processus terminés, une nouvelle vague de 24 processus sera envoyée.
Si tu as un processus lent au milieu des 24 alors, une fois les 23 processus rapides terminés ils sont en standby jusqu'à ce que le processus le plus lent soit fini et qu'une nouvelle vague de processus puisse être lancée.
Ca pourrait expliquer pourquoi des coeurs s'arrêtent de marner pendant qu'un seul tourne.
Je ne connais pas les fonctionnalités de 'future', mais je peux te dire comment je ferais avec le package parrallel.

Code : Tout sélectionner

# en parallel
cl <- makeCluster(parallel:::detectCores(), type = "FORK")
parLapplyLB(cl, unique(Data00$Série), function(x) TraitementSérie(...))
stopCluster(cl)
 

Après je suis tombé la deussus et ça semble assez intéressant :
https://www.r-bloggers.com/2020/12/goin ... cing-in-r/

Cordialement,
Maxime

[Dominique Soudant]

Impec merci, je vais tester demain (j'ai lancé un run plus tôt dans la journée, fin à H+13 ...)
@+
DS

[Mickael Canouil]

Si j'avais à l'écrire, voici à quoi ressemblerait le bout de code fourni (en conservant le côté test de la présence des extensions, d'ailleurs parallel fait partie de R base, il est donc impossible qu'il ne soit pas installé sauf corruption de l'installation de R).
A noter que le choix de la méthode de parallélisation (fork comme multicore) va avoir un impact important selon les cycles de lecture/écriture en mémoire ou sur disques.

Au vu du code, il serait mieux que la fonction TraitementSérie dispose d'un argument "path" ou "directory" plutôt qu'employer "setwd".

Code : Tout sélectionner

source("TraitementSerie.R") # source the function "TraitementSerie"

Data00 <- get(load(file.path(.DERIVED.DATA.SETS, "SATCHLOROA_Filtrage.rda")))

if (require(future)) {
  plan("multisession", workers = min(c(availableCores(), 4)))  # e.g., use max 4 if 4 are available.
  # plan("multicore", workers = min(c(availableCores(), 4)))
}

message("Nb séries : ", length(unique(Data00$Série)))
timestamp()
if (nzchar(system.file(package = "future.apply")) && require(future)) {
  results <- future.apply::future_by(
    data = Data00,
    INDICES = Data00$Série,
    FUN = TraitementSerie,
    Log = TRUE,
    output_directory = .OUT,
    PeriodicFactorForm = NULL,
    NbMaxModele = 2,
    future.globals = TRUE, # default, to use global environnement objects/variables
    future.packages = NULL, # default, specify any packages required/imported inside "TraitementSerie"
    future.scheduling = ceiling(length(unique(Data00$Série)) / nbrOfWorkers()) # number of tasks per "worker" / "core", default = 1.
  )
} else {
  results <- by(
    data = Data00,
    INDICES = Data00$Série,
    FUN = TraitementSerie,
    Log = TRUE,
    output_directory = .OUT,
    PeriodicFactorForm = NULL,
    NbMaxModele = 2
  )
}
timestamp() 

EDIT : Si "SATCHLOROA_Filtrage.rda", est juste un data.frame ou un objet R, en lieu et place d'un environnement, il serait plus judicieux d'utiliser writeRDS et readRDS pour lire/écrire des fichiers ".rds" pour éviter de charger des choses dans l'environnement.

Code : Tout sélectionner

Data00 <- get(load(file.path(.DERIVED.DATA.SETS, "SATCHLOROA_Filtrage.rda")))

Deviendrait

Code : Tout sélectionner

Data00 <- readRDS(file.path(.DERIVED.DATA.SETS, "SATCHLOROA_Filtrage.rds"))

[Dominique Soudant]

Hello,

Merci beaucoup Mickaël, je vais garder tout ça en tête, en particulier j'avais jamais entendu parler des RDS. Faut dire que le bout de code que j'ai donné existe par ailleurs pour ~60 variables*10 laboratoires, donc quand je fais une modif, faut que je prenne de l'élan.

Merci encore
@+
DS

[Dominique Soudant]

Hello,

feedback : 5H13 avec parLapplyLB au lieu de 12H, la révolution est en marche.

Mickaël> je garde l'existence de "future" en tête. Cela dit, les programmes que je fais sont supposés être utilisés sur le long terme (on utilise encore des programmes que j'ai fait en 2007) et pour cela je me limite dans les packages utilisés (e.g. pas de ggplot, dplyr etc) et privilégie ce qui vient avec "base".

Merci à tous les deux
@+
DS