Diviser pour calculer plus vite Fork - Join - i3s.unice.frriveill/programmation-concurrente/cours07-fork... · Head Tail a P: Dequeue() => a Q: Enqueue(d) Simple Code, easy to prove

[email protected]://www.i3s.unice.fr/~riveill

Diviser pour calculer plus vite Fork - Join

Commentmesurerl’accélera=ondueauparallélisme

2

Accéléra=onmaximumàloid’Amdhal

•  p pourcentageducodeexécutableenparallèle•  (1–p):pourcentageducodeexécutéenséquen=el•  p/n:exécu=onducodeparallèlesurncœurs

•  Letempsd’exécu=onthéoriqueévoluedoncen:•  T=(1-p)+p/n

•  Accélera=on A=1/T•  A=1/((1–p)+p/n)•  Sin→∞alorsA→1/(1–p)

3

Quelquesexemplesissuesduprojet

•  Nombredethreads entre2^0e2^9•  Pourcentageducodeexécutéenparallèle

•  Chezcertains P=0,05•  Chezd’autres P=0,95

•  Tempsd’exécu=onthéorique•  Tempspourfairesor=runepersonneduterrainconstantenmoyenne:t•  Tempspourfairesor=rXpersonnesenséquen=el:X*t•  Tempspourfairesor=rXpersonnesenparallèle:T(X)=((1-p)+p/n)*X*t

•  Lesmesuresdoiventêtreentreles2bornes(etleplusprochepossiblede100%deparallélisme)•  avec4coeurs,etp=100%àT(X)=X*t/4•  avec4coeursetp=0%àT(X)=X*t•  Pouréliminertquel’onneconnaitpas,onpeutprésenterlesrésultatssousla

formed’unra=o•  Intérêtdesmesures:est-cequel’observa=onestconformeàcequel’on

a.end•  Onprésentelesrésultatssurdescourbes•  Onessaiedetrouverdesexplica=onsraisonnable

4

Applica=onnumérique(surd’autresdonnées)

5

Application numérique : §  p = 0,25

§  a = 3,7 à 32 cœurs §  a → 4 quand n → ∞

§  p = 0,95 §  a = 12,55 à 32 cœurs §  a = 17,42 à 128 cœurs §  a = 20 quand n → ∞

§  p = 0,96 §  a = 14,28 à 32 cœurs : +12.1%, §  a = 21,05 à 128 cœurs : +17.2% §  a = 25 quand n → ∞

⇒ Ça vaut le coup de se battre pour paralléliser quelques pourcents

Commentaugmenterleparallélisme

Op=on1:réduirelatailledessec=onscri=ques•  Passerduverrouillageàgrosgrain(coarse-grainedlocking)

èàunverrouillageàgrainfin(fine-grainedlocking)•  Effet:augmentelepdanslaloid’Amdahl•  Comment:diviserleslonguessec=onscri=quespardenombreusespe=tes

sec=oncri=ques•  Augmentegénéralementlacomplexitéduprogramme

•  Travailnécessaire,complexe•  Maispasvraimentdetechniquegénéralepourlefaire…•  Parconséquentpasdequoifaireuneséancedecourssaufàvousdonnerdes

sec=onscri=quesàreécrire.•  Op=on2:améliorerlesalgorithmesdeverrouillage

•  CfcoursTMPCenSI5•  Op=on3:supprimerlesverrousenadoptantunealgorithmiquenon

bloquante•  Cfcoursdelasemaineprochaine(introduc=on)+coursTMPCenSI5

•  Op=on4:changercomplètementlemodèlemémoireaveclesmémoirestransac=onnelles•  CfcoursTMPCenSI5 6

Solu=onsnaturelles

1.  Prendre les verrous les plus fins possibles...

7

75%Unshared

25%Shared

c cc cc cc c

CoarseGrained

c

cc

c

c

c

c c

c cc cc cc c

FineGrained c c

cc

cc

cc

Thereasonwegetonly2.9speedup

75%Unshared

25%Shared

Finegrainedparallelismhashugeperformancebenefit

A Concurrent FIFO Queue

Object lock

b c d

Tail Head

a

P: Dequeue() => a Q: Enqueue(d)

Simple Code, easy to prove correct

Contention and sequential bottleneck

Fine Grain Locks

b c d

Tail Head

a

P: Dequeue() => a Q: Enqueue(d)

Finer Granularity, More Complex Code

Verification nightmare: worry about deadlock, livelock… Complex boundary cases: empty queue, last item Worry how to acquire multiple locks

Avousd’imaginerunesolu=onpourlapremière/secondeversionduprojet

10

Dis=nguerlestâchesàfaireetlesthreadsquivontlesexécuter

11

Unpeud’algorithmiquepourcommencer

•  Jusqu’àprésentonal’associa=on•  1tacheestdirectementassociéà1thread•  Etonaautantdethreadquedetacheàexécuter

•  Nouspouvonsmaintenantposerdesques=onsdenatureplusalgorithmique:•  Commentobtenirungaindeperformanceenadaptantledécoupagetache/

threadetenop=misantlenombredethread

•  Supposonsquel’onsouhaiteaddi=onnerpointparpointdeuxtableauxdemêmetaille:

// Semantics: C = A + B static void add (int[] C, int[] A, int[] B) { for (int i = 0; i < C.length; i++) C[i] = A[i] + B[i] ; }

•  LacomplexitéestO(n).

•  Siondisposedeplusieursprocesseurs,onpeutespérerfairemieux,carlesinstruc=onsC[i]=A[i]+B[i]sontindépendanteslesunesdesautres.•  LesarchitecturesGPUsontparfaitementadaptéespourcetypedecalcul

•  Présentéepeutêtredanslecadredececours12

Unparallélismeforcené?

•  Uneapprocheradicaleconsisteraitàlancernthreadsenparallèle:static void add (int[] C, int[] A, int[] B) {

for (int i = 0; i < C.length; i++) { fork { C[i] = A[i] + B[i]; } }

joinAll ; }

•  Est-ceunebonneidée?•  D’unpointdevuepra=que,non,cen’estpasunebonneidée.

•  Créerunthreadcoutecher•  Ilfautaugrandminimumplusieursmilliersd’instruc=onspourlancer

etarrêterunprocessus«léger».•  Lancerunthreadpoureffectuerunetachetrèscourten’aaucunsens!

13

Uneversionmoinsagressive

•  Onpeutlancerunnombredethreadsmoinsimportantendécoupantletableauenuneséried’intervalles:static void addChunk (int[] C, int[] A, int[] B) { final int CHUNK = ... ; for (int c = 0; c < C.length; c += CHUNK) {

fork { for (int i = c; i < c + CHUNK; i++) C[i] = A[i] + B[i] ; }

}

joinAll; }

•  CommentchoisirlavaleurdeCHUNK?

14

Pourbienfaire,ilfauts’adapteràlacharge?

•  Sionapprocesseurs:•  ChoisirCHUNK=n/p•  Donclancerpthreads,peutsemblernaturel.

•  Maissicertainsprocesseurssontdéjàoccupespard’autresprocessus(lourdsoulégers):•  Ilyaurapartagedutemps,d’oùinefficacité.

•  Orlenombredeprocesseursdéjàoccupespeutvarieraucoursdutemps!

•  Pasfacile...

15

Versuneabstrac=ondeplus:lestaches

•  Pourfaciliterlarépar==ondynamiquedutravail,onpeut:•  introduireuneno=ondetâche,•  poserquefork/joincréent/a.endentunetâche,•  u=liserunnombrefixedeprocessuslégers(workerthreads)pourexécuter

cestâches,•  cequisupposeunalgorithmeefficacederépar==ondestâchesentre

travailleurs(«scheduling»)

•  Plusieurslibrairiesimplémententce.eidée:•  IntelThreadBuildingBlocks•  Microso�TaskParallelLibrary•  Java7ForkJoin–c’estcequenousallonsétudier.

16

Unexempled’u=lisa=ondumodèleFork/Join

•  Lorsqueletraitementestdécomposableensous-tâchesindépendantes•  Existenced'algorithmes

parallèlesspécifiques

•  Ladécomposi=onensous-tâchespeutêtre:•  sta=que:découperuntableau

enzonesfixes•  dynamique:découvrirune

arborescencedefichiers•  A.en=onàmaîtriserlevolume

detâchescréées

•  récursive

Etape1miseenœuvredujoinAll

18

JoinAll=barrière

•  Idée:touslesprocessusfranchissentaumêmemoment“labarrière”

•  ImplémentaIon:•  Facileàimplémenterparunmoniteurou

avecdessémaphores•  ExisteenJava7,enposix

•  PrincipeenJava7:•  Unprocessuscréeunobjetbarrière•  ildonnelenombreNdeprocessusa.endu

•  Chaqueprocessusexécute«await»surl’objetbarrière

•  QuandNesta.eint,labarrières’ouvre

19

LabarrièreJava

public class MonRunnable implements Runnable { private CyclicBarrier barrier; string name = Thread.currentThread().getName(); public MonRunnable(CyclicBarrier barrier) { this.barrier = barrier; } public void run() { System.out.println(threadName + " is started"); Thread.sleep(400*new Random().nextInt(10)); System.out.println(threadName + " is waiting on barrier"); barrier.await(); System.out.println(threadName + " has crossed the barrier"); } } 20

LabarrièreJava

main(String args[]) { //creating CyclicBarrier with 3 Threads needs to call await() CyclicBarrier cb = new CyclicBarrier(3, new Runnable(){ public void run(){ //This task will be executed once all thread reaches barrier System.out.println("All parties are arrived at barrier"); } }); //starting each of thread Thread t1 = new Thread(new MonRunnable(cb), "Thread 1"); Thread t2 = new Thread(new MonRunnable(cb), "Thread 2"); Thread t3 = new Thread(new MonRunnable(cb), "Thread 3"); System.out.println("Début démarrage des threads"); t1.start(); t2.start(); t3.start(); System.out.println("Fin démarrage des threads"); }

21

LabarrièreJava

•  Unexempled’exécu=onDébut démarrage des threads Fin démarrage des threads

Thread 3 is started

Thread 1 is started

Thread 2 is started

Thread 1 is waiting on barrier



All parties are arrived at barrier, lets play Thread 3 has crossed the barrier

Thread 2 has crossed the barrier

Thread 1 has crossed the barrier

•  Evidemmentdanslacasdetâchescycliques,ilfautréini=aliserlabarrière•  reset()

22

MiseenœuvrenaïvedelabarrièreenJava

Class Barrier { private final int N_THREADS;

int arrived;

public Barrier (int n) {

N_THREADS = n;

}

public synchronized void aWait () {

arrived++;

if (arrived < N_THREADS) {

wait();

} else {

arrived = 0;

notifyAll();

}

}

} 23

SansunphénomènebiencurieuxappeléàSpuriouswakeup

Spuriouswakeuph6p://docs.oracle.com/javase/7/docs/api/java/lang/Object.html

•  Athreadcanalsowakeupwithoutbeingno=fied,interrupted,or=mingout,aso-calledspuriouswakeup.

•  Whilethiswillrarelyoccurinprac=ce,applica=onsmustguardagainstitbytes=ngforthecondi=onthatshouldhavecausedthethreadtobeawakened,andcon=nuingtowaitifthecondi=onisnotsa=sfied.Inotherwords,waitsshouldalwaysoccurinloops,likethisone:

synchronized (obj) {

while (<condition does not hold>)

obj.wait(timeout);

... // Perform action}

•  Formoreinforma=ononthistopic,seeSec=on3.2.3inDougLea's"ConcurrentProgramminginJava(SecondEdi=on)"(Addison-Wesley,2000),orItem50inJoshuaBloch's"Effec=veJavaProgrammingLanguageGuide"(Addison-Wesley,2001).

24

Arriveenpar=culiersurLinuxcarlamiseenœuvredelaJVMreposesurlespthreadquiengendrelephénomènede«Spuriuswakeup»

MiseenœuvredelabarrièreJava(avantlejava7)

Class Barrier { private final int N_THREADS;

int[] counts = new int[] {0, 0};

int current = 0;

public Barrier (int n) {

N_THREADS = n;

}

public synchronized void aWait () {

int my = current;

counts[my]++;

if (counts[my] < N_THREADS)

while (counts[my] < N_THREADS) wait();

else {

current ^= 1;

counts[current] = 0;

notifyAll();

}}} 25

PriseencompteduSpuriouswakeup

Associerdestâchesàdesthreads

26

Contrôleduparallélisme

•  Leparallélismec’estbienmieuxsionpeuteffecIvementcalculerenparallèle•  Exploita=ond’unel’architecturemul=-cœur•  Java

•  Nomduthreadcourant:Thread.currentThread().getName()•  Nombredecœursurleprocesseur:Run=me.getRunCme().availableProcessors();•  Quelestl’intérêtdelancer10threadsionaque4cœurs?

•  Jusqu’àprésentoncréaitdesthreadsparallèlessansjamaisseposerlaques=ondunombredethreadsquiétaiteffec=vementexécutéen//•  Unmodèle:les«pools»dethread

•  Créerunethreadcoûtecher•  AlorsonrecycleJ

27

Pooldethread

Principe•  OncréeunensembledeNthreads

•  Généralementmoinsquedeprocesseurs/coeursexistants•  Ouaucontrairelégèrementpluspourprendreencomptelesopéra=ons

bloquantes(entrée/sor=e,réseau)

•  Oncréeunensembledetâche•  Lenombredépendgénéralementduproblèmeàrésoudre

•  Lestâchessontsuccessivementexécutéesparlesthread

MiseenœuvreenJavaàl’aided’unenouvellehiérarchie:Executor

28

ClasseExecutorsuneautremanièred’ac=verunethread

public class MonRunnable implements Runnable { public void run(){

System.out.println("Debut execution dans le thread "

+Thread.currentThread().getName());

//On simule un traitement long

Thread.sleep(4000);

System.out.println("Fin execution dans le thread "

+Thread.currentThread().getName());

} }

•  Solu=ondéjàconnueThread thread = new Thread(new MonRunnable());

thread.start();

•  Enu=lisantl’interfaceExecutorExecutor executor = Executors.newSingleThreadExecutor();

executor.execute(new MonRunnable()); 29

ClasseExecutors:exécu=onséquen=elledestâches

List<Runnable> runnables = new ArrayList<Runnable>(); //création de 4 tâches

runnables.add(new MonRunnable());




//création ‘executor’ mono-thread

ExecutorService executor = Executors.newSingleThreadExecutor();

//exécution des tâches selon le modèle choisi

for(Runnable r : runnables){

executor.execute(r);

}

//attente de la terminaison de toutes les tâches

executor.shutdown();

30


31

Serveur

CPU A

B

C

A

BC

Avantage•Facileàu=liser

LeCPUestinac=fpendantlesE/S

Filecontenantlesrequêtesena.ente


•  Evidemmentlestracesd’exécu=onsont:Debut execution dans le thread pool-1-thread-1 Fin execution dans le thread pool-1-thread-1

Debut execution dans le thread pool-1-thread-1

Fin execution dans le thread pool-1-thread-1





32

ClasseExecutors:exécu=onparallèledestâches

List<Runnable> runnables = new ArrayList<Runnable>(); //création de 4 tâches





//création ‘executor’ pool de 2 threads

ExecutorService executor = Executors.newFixedThreadPool(2); //Executors.newSingleThreadExecutor();

//exécution des taches selon le modèle choisi

for(Runnable r : runnables){

executor.execute(r);

}

//attente de la terminaison de toutes les taches

executor.shutdown(); 33

Seulchangement


34

Avantage•Moinsderequêtedanslafiled’a.ente•Minimizelacréa=on/destruc=ondethreadDésavantage•Nécessitéde«tunning»pour=rerlemeilleurpar=dumatériel•Pasfacileàme.reenœuvreavantJava7.0

Serveur

CPU A

B

C

A

C

CPU B

Unnombreop=mumdethreadpeutêtrecrééetrecyclé


•  Unexempled’exécu=ondesthreadsDebut execution dans le thread pool-1-thread-1 Debut execution dans le thread pool-1-thread-2







35

Appeldeméthodeavecfutur(oucommentrécupérerunrésultatplustard)

36

Unmodèle:l’appelasynchroneavecavecfuture

•  Appelsynchrone

•  Appeldeméthode•  Appeldeprocédureàdistance

(RPC)•  Client-serveur

37

•  Appelasynchroneavecfutur

•  Acteurs/objetsac=fs•  Appelaynchronedeprocédureà

distance(A-RPC)

appelantclient

appeléserveur

execute()

returns()

appelantclient

appeléserveur

submit()

returns()

futur

new()

put()

get()

RunnableversusCallable<T>

Runnable Callable<T>

IntroducedinJava1.0 IntroducedinJava1.5aspartofjava.u=l.concurrentlibrary

Runnablecannotbeparametrized Callableisaparametrizedtypewhosetypeparameterindicatesthereturntypeofitsrunmethod

Classesimplemen=ngRunnableneedstoimplementrun()method

Classesimplemen=ngCallableneedstoimplementcall()method

Runnable.run()returnsnoValue Callable.call()returnsavalueofTypeT

CannotthrowCheckedExcep=ons CanthrowCheckedExcep=ons

publicinterfaceRunnable { void run(); }

publicinterfaceCallable<V>{ V call() throwsException; }

38

ClasseCallable

public class MonCallable

implements Callable<Integer> { public Integer call() {

System.out.println("Debut execution”);

//Simulation traitement long

Thread.sleep(4000);

System.out.println("Fin execution); return new Random().nextInt(10);

}

}

39

LaclasseFuture<T>

// création d’un “Executor”

ExecutorService executor = Executors.newSingleThreadExecutor();

//Appel de la tache et récupération d’un Future<V>

Future<Integer> future =

executor.submit(new MonCallable());

System.out.println("Apres submit"); //On a besoin du résultat on appel : future.get()

// appel bloquant

System.out.println("Resultat=" + future.get());

System.out.println("Avant shutdown");

execute.shutdown();

40

Appelasynchroneavecfutur

•  ExempledetraceApres submit Debut execution

Fin execution

Resultat=6

Avant shutdown

41

Miseenoeuvredumodèlefork-joinenJava

42

Miseenoeuvredupa.ern‘forkjoin’enJava

•  Exemple=réde:h.p://fr.openclassrooms.com/informa=que/cours/apprenez-a-programmer-en-java/depuis-java-7-le-pa.ern-fork-join

•  Leproblème•  Parcourirtouslesrépertoiresdemon$HOMEetafficherlenombredefichier

contenantlesuffixe‘psd’

•  Sedécomposeassezaisémentensous-problèmesindépen-dants•  L’analysed’unrépertoirepeut

êtreréalisédemanièrerécur-siveparanalysedessousré-pertoire

43

Algorithme

•  Danschaquerépertoire:•  Onlanceunenouvelle

analysepourchaquesous-répertoires

•  Oncomptelenombredefichiercorrespondantaufiltre

•  Ona.endlaterminaisondesanalysesdessousrépertoiresetonaddi=onnelenombredefichiertrouvé

44

•  Approcheséquen=elle•  Ilya92fichier(s)portant

l'extension*.psd•  Tempsdetraitement:67723

45

•  U=lisa=onde4coeurs•  Ilya92fichier(s)portant

l'extension*.psd•  Tempsdetraitement:78679•  Maistempsd’a.entepresque3

foispluscours

Lepa.ernforkjoin

•  ForkJoinPool:représenteunpooldethreadquireçoitlalistedestâchesàréaliser

•  invoque:lancelestâchespasséesenparamètredemanièresynchrones

•  execute:lancelestâchespasséesenparamètredemanièreasynchrones

•  submit:lancelatâchepasséeenparamètredemanièreasynchroneetrenvoieunobjetfuturimmédiatement

•  ForkJoinTask:représenteunetâcheunique

•  invoke():exécu=ondelatâcheparlathreadcourante

•  fork():lanceuneautretâchedanslemêmepoolquelathreadcourante

•  join():retournelerésultatdel’exécu=ondece.etâche

•  UnForkJoinPoolexécutedesForkJoinTasks

46

<interface> Executor

<interface> ExecutorService

ForkJoinPool

T invoke(task) void execute(task) Future<T> submit(task)

<interface> Future<V>

<abstract> ForkJoinTask<V>

RecursiveTask<V>

RecursiveAction

compute(), invoke(), fork(), join(), isDone(), get() ...

ForkJoinTask

•  RecursiveAction•  ImplémenteFuture<Void> •  Modéliseuntraitementquine

renvoiepasdevaleur•  Peutnéanmoinsmodifierles

donnéespasséesenparamètre•  Ex:trieruntableau"in-place"

•  RecursiveTask<V>•  ImplémenteFuture<V>•  Modéliseuntraitementrécursif

quirenvoieunevaleur•  Ex:calculerlatailletotale

d'unearborescencedefichiers

compute() estlaméthodeexcécutéeparlatâche•  équivalentàrun()pourlesthreadLestâchespeuventêtreappeléesdemanièresynchroneouasynchrone•  Synchrone:invoke() •  Asynchrone:fork()

Main

//Créa=ondel'objetFolderScannerFolderScannerfs=newFolderScanner(chemin,filtre);//Analyseresultat=fs.sequen=alScan();//Untransparentquisuitdécrit://sequen=alScan

48

//Nombredeprocesseursdisponiblesintprocesseurs=Run=me.getRun=me().availableProcessors();//Créa=ondupooldethreadForkJoinPoolpool=newForkJoinPool(processeurs);//Créa=ondel'objetFolderScannerRecursiveTask<Long>fs=newFolderScanner(chemin,filtre);//Analyseresultat=pool.invoke(fs);//exécutecompute//Untransparentquisuitdécrit://compute

SynchroneouAsynchrone?(ForkJoinPool)

Resultat = pool.invoke(fs); estéquivalentà:

pool.invoke(fs);

Bla bla bla resultat = fs.join();

49

Resultat = pool.execute(fs); estéquivalentà:

pool.execute(fs); Bla bla bla resultat = fs.join();

FolderScannerpublicclassFolderScanner{publiclongsequen=alScan(){//Récupéra=ondelalistedesfichiersdurépertoired’analyseDirectoryStream<Path>lis=ng=Files.newDirectoryStream(path));for(Pathnom:lis=ng){if(Files.isDirectory(nom.toAbsolutePath())){//S'ils'agitd'undossierilestanalyséresult+=newFolderScanner(nom.toAbsolutePath(),this.filter).sequen=alScan();}

//Récupéra=ondelalistedesfichiersrépondantaufiltreDirectoryStream<Path>lis=ng=Files.newDirectoryStream(path,this.filter));for(Pathnom:lis=ng){//Pourchaquefichier,onincrémentelecompteurresult++;}returnresult; 50

publicclassFolderScannerextendsRecursiveTask<Long>publicLongcompute(){//Récupéra=ondelalistedesfichiersdurépertoired’analyseDirectoryStream<Path>lis=ng=Files.newDirectoryStream(path));for(Pathnom:lis=ng){if(Files.isDirectory(nom.toAbsolutePath())){//S'ils'agitd'undossierilestanalysésubTasks.add(newFolderScanner(nom.toAbsolutePath(),this.filter).fork());}} //Récupéra=ondelalistedesfichiersrépondantaufiltreDirectoryStream<Path>lis=ng=Files.newDirectoryStream(path,this.filter));for(Pathnom:lis=ng){//Pourchaquefichier,onincrémentelecompteurresult++;}

//Récupéra=ondurésultatdetouteslessous-tâchesfor(ForkJoinTask<Long>subTask:subTasks)result+=subTask.join();

returnresult;

Synchroneouasynchrone?(ForkJoinTask/RecursiveTask/RecursiveAc=on)

if(Files.isDirectory(nom.toAbsolutePath())){//S'ils'agitd'undossierilestanalyséForkJoinTask<Long>fs=newFolderScanner(nom.toAbsolutePath(),this.filter)

.invoke();}

}fs.invoke();Blablablafs.join(); 51

if(Files.isDirectory(nom.toAbsolutePath())){//S'ils'agitd'undossierilestanalyséForkJoinTask<Long>fs=newFolderScanner(nom.toAbsolutePath(),this.filter)

.fork();}}

fs.fork();Blablablafs.join();

Asynchronismeettâchesrécursives(RecursiveTask)

// correct protected Long compute() { List<ForkJoinTask<Long>> subTasks = new ArrayList<>(); long size = 0; for(File f : root.listFiles()) { if (f.isDirectory()) { subTasks.add(new Task(f).fork()); } else { size += f.length(); } } for (ForkJoinTask<Long> subTask : subTasks) { size += subTask.join(); } return size; }

// Incorrect protected Long compute() { ForkJoinTask<Long> subTask; long size = 0; for (File f : root.listFiles()) { if (f.isDirectory()) { subtask = new Task(f).fork(); size += subtask.join(); } else { size += f.length(); } } return size; }

/*l’asynchronismesouhaité(fork)n’estpasexploité:ona.endtoutdesuite*/

Bonnespra=ques

•  A.en=onaucoûtdeges=on•  Lebénéficeobtenuenparallélisantuntraitement

doitêtresupérieuraucoûtdeges=onparleframework•  Trouverlabonnegranularité

•  A.en=onàlaconsomma=onmémoire•  Grossesstructures:préférerlamodifica=on"in-place"•  Découvertedynamiquedessous-tâches

•  A.en=onàlacomplexité•  Op=misa=onprématurée?•  Maintenance

Op=misa=ons

•  FrameworkFork/Join•  Threads>=Degrédeparallélisme

•  Lestâchesena.entedejoin()sontmisesdecôtépourperme.reàd'autrestâchesd'êtretraitées

•  Tâches•  Eviterlasynchronisa=onmanuelle

•  U=liserfork()etjoin()uniquement

•  Op=miserlagranularitéàl'aidedessta=s=quesdupool•  getQueuedSubmissionCount(),getStealCount()...•  Possibilitéd'op=misa=ondynamique

Quelquesréférencesquim’ontaidéespourpréparercecours

•  APIJava7,packagejava.u=l.concurrenth.p://download.java.net/jdk7/docs/api/

•  EtudedeDougLeah.p://gee.cs.oswego.edu/dl/papers/�.pdf

•  Desblogs/tutoriaux

•  h.p://www.oracle.com/technetwork/ar=cles/java/fork-join-422606.html•  h.p://blog.paumard.org/2011/07/05/java-7-fork-join/•  h.p://sdz.tdct.org/sdz/le-framework-executor.html•  h.p://fr.openclassrooms.com/informa=que/cours/apprenez-a-programmer-en-java/depuis-java-7-le-pa.ern-fork-join

Etsionformalisaitunpeu?

56

Del’algorithmeitéra=fàl’algorithmeparallèle

•  Nousallonstenterderépondreàcedeuxques=ons=•  Commentu=lisercesmécanismespourobtenirungaindeperformance?•  Commentprédirelegaindeperformancequel’onpeutespérer?

•  Exemple:addi=onpointparpointdeuxtableauxdemêmetaille// Semantics: C = A + B static void add (int[] C, int[] A, int[] B) { for (int i = 0; i < C.length; i++) C[i] = A[i] + B[i] ; }

•  LacomplexitéestO(n).•  Siondisposedeplusieursprocesseurs,onpeutespérerfairemieux,carles

instruc=onsC[i]=A[i]+B[i]sontindépendanteslesunesdesautres.•  Parallélisa=onàoutrance:1threadparcelluledutableau

static void add (int[] C, int[] A, int[] B) { for (int i = 0; i < C.length; i++) { fork { C[i] = A[i] + B[i]; } } joinAll ;

} 57

Unparallélismecontrôlé

•  Est-ceunebonneidée?•  D’unpointdevuepra=que,non,cen’estpasunebonneidée.

•  Créerunthreadcoutecher•  Ilfautaugrandminimumplusieursmilliersd’instruc=onspourlanceretarrêter

unprocessus«léger».•  Lancerunthreadpoureffectuerunetachetrèscourten’aaucunsens!

•  Onpeutlancerunnombredethreadsmoinsimportantendécoupantletableauenuneséried’intervalles:static void addChunk (int[] C, int[] A, int[] B) { final int CHUNK = ... ; for (int c = 0; c < C.length; c += CHUNK) {

fork { for (int i = c; i < c + CHUNK; i++) C[i] = A[i] + B[i] ; } } joinAll;

} •  CommentchoisirlavaleurdeCHUNK?

58

Uneversion«diviserpourrégner»

•  Onpeutsouventexprimerrécursivementladécomposi=onentâches:static void addRec(int[] C, Int[] A, int[] B, int lo, int hi) {

if (hi - lo <= CHUNK) for (int i=lo; i<hi; i++) C[i] = A[i] + B[i] ; else { int limit = lo+(hi-lo)/2; fork{ addRec(C, A, B, lo, limit); }

fork{ addRec(C, A, B, limit, hi); }

joinAll ; }

}

•  Lesalgorithmes«diviserpourrégner»sontfondamentalementparallèles.

59

Diagrammesdetâches:addChunk

•  Pourcomprendrelecomportementdecesalgorithmes,ilestu=ledereprésenterlestâchesetleursdépendancessousformedegraphe.

•  PouraddChunk,legrapheressembleàceci:

•  Chaquesommetreprésenteunesuite(plusoumoinscoûteuse)d’instruc=onsexécutéesséquen=ellement.

60

Diagrammesdetâches:addRec

61

Diagrammesdetâches

•  Danslecasgénéral,onpeutavoirungrapheorientéacyclique,symétrique

•  Lastructuredecegraphepermetdeprédirequelgaindeperformanceonpeutespérer,aumaximum,lorsquelenombredeprocesseurscroît.

•  Deuxquan=tésjouentunrôleprimordial:•  letempsd’exécu=onséquen=el•  letempsd’exécu=onparallèle

62

Travail

•  Letempsd’exécu=onséquen=elestlecoutcumulédetouteslestaches

•  C’estletempsnécessairepourexécutercestachesavecunseulprocesseur

•  OnlenoteT1.

63

Profondeur

•  Letempsd’exécu=onparallèleestlecoûtcumulédestâchessurlechemincri=que(chemin“lepluslent”)carletempsd’exécu=onestlesupdestempsdechaqueprocessusmisenparallèle

•  C’estletempsnécessairepourexécutercestâchesavecuneinfinitédeprocesseurs

•  OnlenoteT∞.

Parallélisme

•  Mêmesil’ondisposed’uneinfinitédeprocesseurs,legaindeperformanceestdoncaumaximum•  T1/T∞=tempsséquen=el/tempsparallèle

•  Ce.equan=téestappelée«maximumspeedup»ouparallélisme

•  Onpeutévaluer(asympto=quement,danslepirecas)letravailetlaprofondeuràl’aidedestechniquesd’analysequevousconnaissez.

•  Onappliquelesrègles:•  T1(I1||I2)=T1(I1)+T1(I2)•  T∞(I1||I2)=max(T∞(I1),T∞(I2))•  où«I1||I2»estuneabrévia=onpour«forkI1;forkI2;joinAll».

65

Analyseasympto=que:exemple1

static void addRec(int[] C, int[] A, int[] B, int lo, int hi) { if (hi - lo <= CHUNK) for (int i = lo;i < hi;i++) C[i] = A[i] + B[i] ; else { int limit = lo + (hi - lo) / 2; fork { addRec(C, A, B, lo, limit); } fork { addRec(C, A, B, limit, hi); }

joinAll;

}

}

•  Nousavonsleséqua=onsderécurrencesuivantes,pournassezgrand:•  T1(n)=2T1(n/2)+O(1)•  T∞(n)=T∞(n/2)+O(1)

•  D’ou...?

66


•  Nousavonsleséqua=onsderécurrencesuivantes,pournassezgrand:•  T1(n)=2T1(n/2)+O(1) d’oùT1(n)=Θ(n)•  T∞(n)=T∞(n/2)+O(1) d’oùT∞(n)=Θ(logn)

•  Leparallélismeestéleve:T1/T∞=Θ(n/log(n)).•  C’estbonsigne.

67


//SemanCcs:B[lo..hi]=sort(A[lo..hi])//T[lo..hi)canbeusedastemporarystoragestaIcvoidmergeSortRecstaIcvoidmergeSortRec(int[]B,int[]A,intlo,inthi,int[]T){

if(hi-lo==0)return;if(hi-lo==1)B[lo]=A[lo];else{ intlimit=lo+(hi-lo)/2; spawn{mergeSortRec(T,A,lo,limit,B);} spawn{mergeSortRec(T,A,limit,hi,B);} sync; merge(B,T,lo,limit,hi);//auxiliaryfuncCon}

}Ànouveau:

•  T1(n)=2T1(n/2)+f(n)•  T∞(n)=T∞(n/2)+f(n)•  f(n)estlecoûtdemerge...

68


•  Lafonc=onauxiliairemergeeffectuelafusiondedeuxtableauxtriés://SemanCcs:B[lo..hi)=sort(T[lo..hi))//Requires:T[lo..limit)andT[limit..hi)aresortedstaIcvoidmerge(int[]B,int[]T,intlo,intlimit,inthi){

intdst=lo;intsrc1=lo;intsrc2=limit;while(src1<limit&&src2<hi) B[dst++]=T[src1]<T[src2]?T[src1++]:T[src2++];while(src1<limit) B[dst++]=T[src1++];while(src2<hi) B[dst++]=T[src2++];

}•  SacomplexiteestO(n).

69


•  Nousavonsdoncleséqua=onsderécurrence:•  T1(n)=2T1(n/2)+O(n) d’oùwork(n)=Θ(nlogn)•  T∞(n)=T∞(n/2)+O(n) d’oùT∞(n)=Θ(n)

•  Leparallélismeestfaible:T1/T∞=Θ(nlogn/n)=Θ(logn).

70


•  Enfait,lecoutparallèleestmauvaiscarunepor=onnon-négligeableducalculsefaitenséquen=el,c’estlaloid’Amdahl.•  Dansnotrecaspar=culier:lespeedupnepeutdépasser1/B(Bpor=on

séquen=elle)

•  Pourentrecomplet,laloideGustafsonmontrequ’onaaussiintérêtàaugmenterlenombrededonnéessurlequelfairefonc=onneruncodeparallèle

•  Pouraugmentéleparallélismeetaméliorerletempsd’exécu=on,onva

probablementdevoirparalléliseraussilafonc=onauxiliairemerge.

71

Queretenir?

•  Nousavons:•  présentélesthreadsetlesopéra=onsélémentaires«fork»et«join»•  suggèréuneno=onplusabstraitedetachedotéed’opéra=onsanalogues•  regardélamiseenœuvredecepa.erndeprogramma=onenJava

•  suggèréque,pourespérerobtenirungaindeperformance,ilfautcontrôlerlagranularitédestachesetlesdépendancesentretaches.

72

UnmodèleprochemaiscomplémentaireMapReduce

73

Seraprésentéendétaildanscertainsparcoursde5èmeannée

MapReduce

•  Unparadigme/modèledeprogramma=onpourledéveloppementlogicielquipermet•  Uneparallélisa=onautoma=que•  Del’équilibragedecharge•  Desop=misa=onssurlestransfertsdisquesetréseaux•  Delatoléranceauxpannes

•  Trouvesesoriginesdanslesméthodesmapetreduce/folddeslangagesfonc=onnels.

•  Basésurtroisopéra=onsprincipales•  Map:unefonc=onappliquéeàchaquepar=edesdonnéesd’entréepour

générerunesériedecouple<clé,valeur>•  Shuffle:regroupementdescouples<clé;valeur>parclédis=ncte•  Reduce:uneautrefonc=onappliquéeàchacundesgroupesainsigénéré

(doncparclédis=ncte).

74

MapReduce

Lecoded’unprogrammeestdonc1.  Lecturedesdonnées2.  Map

•  AchaqueélémentdufluxrepéréparunecléK1,Mapassocieunlistedecouple<clé,valeur>

•  Map:<K1,V1>àlist(K2,V2)

3.  Shuffle/Sort:regrouperlessor=esetlestrierselonlesclésK24.  Reduce

•  ExécutéunefoislaphaseMapterminée•  Agréga=onenlistedetouteslesvaleursintermédiairesassociéesàunecléK2

•  Agréga=onestuneopéra=onassocia=ve•  Reduce:<K2,list(V2)>àlist(K3,V3)–biensouvent:K2=K3

5.  Ecrirelesdonnées

75

PourquoiMap/Reduce

•  Traiterdesvolumesdedonnéessanscessegrandissantnécessiteuneplusfortepuissancedecalcul

•  Lacadencedesprocesseursstagnedepuisledébutdesannées2000•  L’évolu=onsetourneverslamul=plica=ondescœursetl’exploita=ondu

parallélisme

•  Ledéveloppementdesprogrammesparallèledevientunbesoincroissantetrépandu •  C’étaitauparavantexcep=onnel

•  Formaliseruneméthodologiecommune:•  Pourledéveloppementdeprogramme•  Pourleurmiseenœuvredemanièreaiséesurdiverstypesd’architecture•  Rendrelesexécu=onsadaptablesenfonc=onduvolumededonnéesàtraiter

76

Map/reduce

•  MéthodologiequiaétéformaliséepourlapremièrefoisparGoogleen2004•  MapReduce:SimplifiedDataProcessingonLargeClusters•  h.ps://sta=v/googleusercontent.com/media/reserach.google.com/en//

archive/mapreduce-osdi04/pdf•  Intérêtdel’approche

•  Perme.redeparalléliseruneexécu=onsurautantdemachinequenécéssaire

•  Traiterdesvolumesdedonnéesaussilargequenécessaire•  Miseenœuvre

•  Endéveloppantdeuxfonc=onssimples:mapetreduce•  Quelquesoitlenombredemachineetlevolumededonnées

•  Evolu=onportéepardegrandsgroupesprivées(Google,Yahoo)maisaussipardesfonda=onsliéesaulogiciellibreouOpenSource(Apache)•  Hadoop:projetapache,principaleimplémenta=onduframework

77

Map/Reduce

•  Au=lisersi•  Onsouhaitetravaillersurdegrandsvolumesdedonnées(del’ordreduGB/

TB)•  Onsouhaiterésoudreunproblèmecomplexe,trèsconsommateurde

ressourcesprocesseurs•  Etplusgénéralement,lorsqu’unproblèmedoitêtreparallélisécarson

exécu=onsurunemachineuniquen’estpasréaliste

•  Anepasu=lisersi•  Levolumededonnéesestlimitéetl’exécu=onsurunemachineuniqueest

viable•  Lesdonnéesd’entréesnesontpassegmentable/découpablesefficacement

•  Ilfautquelestâchesàréalisersoientindépendanteslesunesdesautres

78

Map/Reduce–unexemple

•  Compterlesoccurrencesd’unmotdansunensembledefichiers•  Le«hello,World»deMapReduce

•  Donnéesini=ales:•  Lecontenudel’intégralitédelabibliothèquena=onale

•  Objec=fs•  Connaîtrelesmotslespluspopulairesetlesmotslesmoinspopulaires/

•  Leproblèmeportesuruntrèsgrandsvolumededonnées•  Ilestparallélisable:onpeutcompterlesmotsdechaquelivre,dechaque

pagepuisfairelasynthèse

79


•  Donnéesini=ales

•  L.Aragon,LaroseetleRéséda,1943•  Oncommencepardécouperlesdonnéesd’entrée,detellesortequela

fradmenta=onrésulteendesblocsdedonnéessuscep=blesd’êtretraitésindépendammentsansinfluencesurlerésultatfinal•  Exemple:chaquelignedutexted’entréedevientun«bloc»dedonnéesqu’il

faut«ne.oyer»(suppressiondesmajuscules,desaccents,delaponctua=on,etc.)

80

CeluiquicroyaitaucielCeluiquin’ycroyaitpas[…]FouquifaitledélicatFouquisongeàsesquerelles


•  Données

•  onconsidèreconceptuellementces4blocsdedonnéesd'entréecomme4couples<clef,valeur>pourlesquelslaclefestnulleetlavaleurestcons=tuéedelalignedetexte

•  L’opéra=onmapvaproduireunesériedecouples<clef,valeur>–etcescouplesserontlorsdel'opéra=onsuivante(shuffle)regroupésparclefdis=ncte.

•  L’opéra=onreduceauraquandàellepourbutderéduirecesgroupestriésparclefdis=ncte;lechoixdelaclefàu=liserestdonccrucial.

81

CeluiquicroyaitaucielCeluiquin’ycroyaitpas

Fouquifaitledélicat

Fouquisongeàsesquerelles


•  Ici,ongénèrecommeclefauseindel'opéra=onmaplemotlui-même;etcommevaleurcorrespondante,laconstante«1»(quelquesoitlemotrencontré).

•  Lerôledelafonc=onmapseradoncdeparcourirlaligned'entrée,etpourchaquemotdis=nctdegénéreruncouple•  <clef,valeur>=<mot,1>.

82

Celuiquicroyaitauciel

Celuiquin’ycroyaitpas

Fouquifaitledélicat

Fouquisongeàsesquerelles

<celui,1><qui,1><croyait,1><au,1>…

<celui,1><qui,1><ny,1><croyait,1>…

<fou,1><qui,1><fait,1><le,1>…

<fou,1><qui,1><songe,1><a,1>…


•  Aprésexécu=ondemapetdeshuffle,lescouples<clef,valeur>ainsigénérésserontregroupés(parleframeworkd'exécu=onduprogrammesurlecluster,parexempleHadoop)parclefdis=ncte:

•  Chacundecesgroupesdis=nctsserapasséenentréedelafonc=onreduce.

83

<celui,1><qui,1><croyait,1><au,1>…

<celui,1><qui,1><ny,1><croyait,1>…

<fou,1><qui,1><fait,1><le,1>…

<fou,1><qui,1><songe,1><a,1>…

<celui,1><celui,1>

<qui,1><qui,1><qui,1><qui,1>

<croyait,1><croyait,1>

<au,1>

<ny,1>

<ciel,1>


•  Quandàl’opéra=onreduce,ellevarecevoirungroupedecouples•  <clef,valeur>enentrée:ceuxquicorrespondentàunedesclefs

dis=nctes.•  Sonrôlevasimplementêtredeconserverlaclefunique,d'addi=onnerles

valeursdetouslescouples<clef,valeur>reçusenentrée,etgénérerununiquecouple<clef,valeur>ensor=e,composédelaclefuniqueetdutotalobtenu.

84

<celui,1><celui,1>

<qui,1><qui,1><qui,1><qui,1>

<croyait,1><croyait,1>

<celui,2>

<qui,4>

<croyait,2>


•  Autermedel'exécu=onduprogrammemapreducedanssonensemble,onob=endraunesériedecouples<clef,valeur>:unpourchacundesmotsuniquesprésentsdansletexte.Pourchacunedecesclefs,onauralenombred'occurencesdumotdansletexte.

•  Commeleprogrammeaétéimplémentéàpar=rdelaméthodologiemap/reduce,ilestparallélisable;etpourraitsiondisposed'assezdemachiness'exécuterenquelquessecondesmêmesuruntexted'entréetrèsvolumineux.

85

qui:4celui:2croyait:2fou:2au:1ciel:1ny:1pas:1fait:1…

Schémagénéral

86

(source:documentaConHadoop)

Unexemplemoinstrivial

87

Legraphesocial

•  Ondisposedelabasededonnéesd'unréseausocialcontenantplusieursmillionsd'u=lisateurs.Pourchacund'entreeux,onaunelisted'autresu=lisateurs:lesamisdel'u=lisateurcourantsurleréseau.

•  Onchercheàgénérer,pourchaquecoupled'u=lisateursdis=ncts,lalistedesamisqu'ilsontencommun.

•  Onnepeutpaseffectuerce.eopéra=onparlebiaisd'unerequêtesurlabasededonnéesrela=onnellesansunimpactimmensesurleserveurduréseausocial,poten=ellementbloquantpourlabase,etdoncpourlesitelui-même.

•  Parconséquent,onvacréerunetâchemap/reducepourréglerleproblème,etl'exécuteràintervallesrégulierssurunexportdelabasededonnéesrela=onnelle;onimagineensuitequelesrésultatsserontré-importésdansunetabledédiéeàcetusage.

88

Graphesocial

•  Ici,lesdonnéesd’entréesontsouslaformeU=lisateurèamis•  AèB,C,D•  BèA,C,D,E•  CèA,B,D,E•  DèA,B,C,E•  EèB,C,D•  Lesdonnéessontfragmentables(par«ligne»i.e.paru=lisateur)

•  Choixdelaclé•  Onvachoisirunechaînedecaractère,souslaforme:«A-B»•  Ce.eclef,parexemple,désignera«lesamisencommundesu=lisateursAetB».

•  Lera=onnelderrièrecechoixvientdufaitquelaclefreprésentel'élémentautourduquelsontregroupéslesdonnéeslorsdel'opéra=onintermédiaire(shuffle);ellecorrespondaupointcommunentrelescouples(clef;valeur)quivaperme.rederéduirelesdonnéeslorsdel'opéra=onreduce.

•  Parailleurs,onferaégalementensortequelaclefsoittoujourstriéeparordrealphabé=que(clef«B-A»seraexpriméesouslaforme«A-B»).

89

Graphesocial

•  L'opéra=onmapestaucoeurdelarésolu=onduproblème.Ellevaprendrelalistedesamisfournieenentrée,etvagénérertouteslesclefsdis=nctespossiblesàpar=rdece.eliste.Pourchacunedecesclef,ellevagénéreruncouple(clef;valeur)dontlavaleurseralalisted'amis,tellequelle.

•  Cetraitementpeutparaîtrecontre-intui=f,maisilvaàtermenousperme.red'obtenir,pourchaqueclefdis=ncte,deuxcouples(clef;valeur):lesdeuxlistesd'amisdechacundesu=lisateursquicomposentlaclef.

•  Parexemple,•  pourlapremièreligne:AèB,C,D•  Onob=endralescouples(clef;valeur):

•  (A-B;BCD)•  (A-C;BCD)•  (A-D;BCD)

90

Graphesocial

•  Lepseudo-codedel'opéra=onmap

UTILISATEUR=[PREMIEREPARTIEDELALIGNE]POURAMIdans[RESTEDELALIGNE],FAIRE:SIUTILISATEUR<AMI:CLEF=UTILISATEUR+"-"+AMISINON:CLEF=AMI+"-"+UTILISATEURGENERERCOUPLE(CLEF;[RESTEDELALIGNE])

91

GrapheSocial

•  Pourlasecondeligne:BèA,C,D,E•  Onob=endraainsi

•  (A-B,ACDE),(B-C,A,C,D,E),(B-D,A,C,D,E),(B-E,A,C,D,E)•  Etcsurles5lignesd’entrée•  Unefoisl'opéra=onMAPeffectuée,l'étapedeshufflevarécupérerles

couples(clef;valeur)detouslesfragmentsetlesgrouperparclefdis=ncte.Lerésultatsurlabasedenosdonnéesd'entrée•  Pourlaclef"A-B":valeurs"ACDE"et"BCD"•  Pourlaclef"A-C":valeurs"ABDE"et"BCD"•  Pourlaclef"A-D":valeurs"ABCE"et"BCD"•  Pourlaclef"B-C":valeurs"ABDE"et"ACDE"•  Pourlaclef"B-D":valeurs"ABCE"et"ACDE"•  Pourlaclef"B-E":valeurs"ACDE"et"BCD"•  Pourlaclef"C-D":valeurs"ABCE"et"ABDE"•  Pourlaclef"C-E":valeurs"ABDE"et"BCD"•  Pourlaclef"D-E":valeurs"ABCE"et"BCD»

•  Onob=entbien,pourchaqueclef«USER1-USER2»,deuxlistesd'amis:lesamisdeUSER1etceuxdeUSER2

92

GrapheSocial

•  L’opéra=onreduceserévèleparconséquentévidente:ellevarecevoirchacundecesgroupesdedeuxcouples(clef;valeur),parcourirlesdeuxlistesd'amispourgénérerunelisted'amiscommuns,etrenvoyerunseulcouple(clef;valeur),dontlaclefseraiden=queàlaclefdis=nctecorrespondantaugrouped'entrée,etlavaleurserace.elisted'amiscommuns.

•  PseudocodeLISTE_AMIS_COMMUNS=[]//Listevideaudépart.SILONGUEUR(VALEURS)!=2,ALORS://Nedevraitpasseproduire.RENVOYERERREURSINON:POURAMIDANSVALEURS[0],FAIRE:SIAMIEGALEMENTPRESENTDANSVALEURS[1],ALORS://Présentdanslesdeuxlistesd'amis,onl'ajoute.LISTE_AMIS_COMMUNS+=AMIRENVOYERLISTE_AMIS_COMMUNS

93

Graphesocial

•  Aprèsexécu=ondel'opéra=onREDUCEpourlesvaleursdechaqueclefunique,onob=endradonc,pouruneclef«A-B»,lesu=lisateursquiapparaissentdanslalistedesamisdeAetdanslalistedesamisdeB.Autrementdit,onob=endralalistedesamisencommundesu=lisateursAetB.

•  Lerésultat:•  "A-B":"C,D"•  "A-C":"B,D"•  "A-D":"B,C"•  "B-C":"A,D,E"•  "B-D":"A,C,E"•  "B-E":"C,D"•  "C-D":"A,B,E"•  "C-E":"B,D"•  "D-E":"B,C"

•  OnsaitainsiqueAetBontpouramiscommunslesu=lisateursCetD,ouencorequeBetContpouramiscommunslesu=lisateursA,DetE.

94

Graphesocial

•  Lesdeuxopéra=onsmapetreducesontsommestoutesrela=vementsimples,maiscombinéesàunframeworkd'exécu=onmapreduce,elleseffectuentuntravailcomplexe.

•  Parailleurs,latâcheestparallélisable:onpeutl'exécuteraussivitequenécessairedumomentqu'onaassezdemachinesauseinduclusterd'exécu=on.

•  Danslecadredel'exemple,onimaginequecetypedetâcheseraitexécutétouteslesheuressurunclusterHadoopdédiéàceteffet;perme.anttouteslesheuresdeme.reàjourlatablerela=onnelle«classique»contenantleslistesd'amisencommunsavecuncoûtinexistantpourlabasededonnéesrela=onnelle.

95

Autresexemples

•  Uneentreprisedisposed'uncomptetwi.erpoursonserviceaprésvente,recevantplusieursdizainesdemilliersdetweetsparjour.Ellechercheàdéterminerletauxdesa=sfac=ondesesclientsàpar=rducomptetwi.er.Chaqueheure,lestweetsreçussontexportésauseind'unfichiertexte.•  clef:undescripteurdesen=mentclient(«sa=sfait»,«insa=sfait»ou«a.ente»,

«inconcluant»).•  map:génèreuncouple(clef;valeur)parsen=mentclientdétecté(mot

correspondantàunelisteprédéfinie).•  Sideuxsen=mentscontradictoiresdétectés:renvoyerinconcluant.•  Onrenvoieuncouple(clef;valeur)pourchaquefragmentdesdonnéesd'entrée:chaque

tweet•  reduce:addi=onnelesvaleursassociéesàlaclefunique;renvoieletotalpour

valeur(iden=queaureducerducompteurd'occurencesdemots).•  Exemple

•  Conclusion:lorsdeladernièreheure,33%detweetsexprimantdelasa=sfac=on.

•  Enimportant,heureparheure,lesdonnéesauseind'unebasededonnéesrela=onnelle,onpourraobtenirenpermanencedessta=s=quessurlasa=sfac=onàpar=rdetwi.er. 96

"@acmeVotreserviceclientestnul""@acme30mind'a.ente...trèsinsa=sfait""Trèssa=sfaitparunproduitsuper!!@acme""mercid'avoirRT@acme""@acmeproduitdéjàcassé,superinsa=sfait!»

("insa=sfait";1)("insa=sfait";1)("sa=sfait";1)("inconcluant";1)("inconcluant";1)

("insa=sfait";2)("sa=sfait";1)("inconcluant";2)

Conclusion

•  ApplicaIonscourantesdemap/reduceetHadoop:•  Analysedelogsetdonnéesengénéral,validaIondedonnées,

recoupements,filtrage,etc.=>traitementssurdesvolumesdedonnéesmassifs.

•  ExécuIondetâchesintensivesenCPUdemanièredistribuée(simulaIonsscienIfiques,encodagevidéo,etc.).

•  …etbiensouventlesdeux:tâchesintensivesenCPUsurdesvolumesdedonnéesmassifs(entraînementdemodèlesenmachinelearning,etc.)

•  Pluscomplexe,voirecontre-producIfàappliquersurtoutproblèmeoùlafragmentaIondesdonnéesd'entréeposeproblème(enfaitsurtoutproblèmeoùlastratégiedudiviserpourrégnern'estpasviable).

97

98

Exemple(Miseenoeuvre)

•  L’u=lisateurprogrammelesfonc=onsMap/Reduce•  EnpseudoJava

Map(String input_key, String input_value) {

foreach word w in input_value do {

EmitIntermediate (w, 1);

}

Reduce (String key, Iterator<Integer> intermediate_value) { int result =0;

foreach v in intermediate_value do {

result += v;

}

•  Leframeworkfournilesfonc=onsSPLIT/SORTmaisaussilages=ondufluxdedonnées

99

ForkJoin versus MapReduce

ForkJoin MapReduce

Singlemachine Clustermachine

Takesadvantageofmul=plecores Takeadvantagesofclustermachine.Massivelyscalable.

ForkJoinrecursivelypar==onsataskintoseveralsubtasks,onasinglemachine

Doesonlyonebigsplit,withnocommunica=onbetweenthepartsun=lthereducestep.

Startsquicklyandscaleswellforsmallinputs(<5MB),butitcannotprocesslargerinputsduetothesizerestric=onsofshared-memory,singlenodearchitectures.

MapReducetakestensofsecondstostartup,butscaleswellformuchlargerinputs(>100MB)onacomputecluster.

NofaulttoleranceNodistribu=on

Generally,faulttolerantimplementa=on,transparenttaskdistribu=on

ExistinJavasinceJava7 Existsince2004PopularizedbygoogleOpensourceimplementa=on:hadoop(h.p://hadoop.apache.org/) 100

h.p://www.i3s.unice.fr/~riveill

Q&A

101

Livreu=lisépourcechapitre

•  Doug Lea •  Professor•  StateUniversityofNewYorkat

Oswegoh.p://g.oswego.edu

102

Documents

Diviser pour calculer plus vite Fork - Join - i3s.unice.frriveill/programmation-concurrente/cours07-fork... · Head Tail a P: Dequeue() => a Q: Enqueue(d) Simple Code, easy to prove