Upload
lorraine-jug
View
1.193
Download
2
Embed Size (px)
DESCRIPTION
Citation preview
Quel est l’objet
de la programmation
concurrente ?
Tirer parti de la
puissance des
processeurs
1) De quoi cette puissance est-elle faite ?
Architectures muticœurs
2) Conséquences sur la façon de coder ?
3) Quels sont les problèmes des applications « parallèles » ?
Synchronisation
4) Comment s’y prendre pour « synchroniser » les opérations
5) De quels outils dispose-t-on ?
Java 6
Java 7
Java 8
6) Importance de l’algorithmique ?
MULTITHREADING
1995
1995
Si un thread a « asser travaillé »
S’il est en attente d’une ressource lente
S’il est en attente d’un autre thread
1995
Parallèle
SIMD : chaque CPU travaille sur ses propres données
1995
API Java
• Thread, Runnable
• Threads « natifs », « Green threads »
• Moniteur
• synchronized, volatile
1995
API Java
• Thread, Runnable
• Threads « natifs », « Green threads »
• Moniteur
• synchronized, volatile
1995
2004
2004 - 2005
Software Hardware
Processeurs multicœurs
• Fin de la « soupe gratuite »
• La puissance de calcul augmente par le nombre de cœurs, et non plus par l’augmentation de la fréquence de chaque cœur
• Apparition des GPGPU (2007)
Processeurs multicœurs
• Aujourd’hui
– Smartphone : 2 à 4 cœurs
– PC de bureau : 4 à 8 cœurs
– Serveur pro : 128 cœurs
• Demain
– Plateforme exascale : de 100k à 1M nœuds
– Chaque nœud : de 1k à 10k cœurs
Java.util.concurrent
• API de concurrence avancée
– Dispo en Java 4 EDU.oswego
• Nouvelle approche du multithreading
– Nouvelle façon de lancer des threads
– Nouvelle façon de synchroniser les opérations
Problèmes du
multithread ?
1) Il est nécessaire !
2) Data race
Deux threads modifient une variable
« en même temps »
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
Interruption =
catastrophe
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static synchronized Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
public class Singleton {
private static Singleton instance ;
private Singleton() {}
public static synchronized Singleton getInstance() {
if (instance == null) {
instance = new Singleton() ;
}
return instance ; }
}
Lecture ralentie
public class Singleton {
private static Singleton instance ;
public static Singleton getInstance() {
if (instance == null) {
synchronized (Singleton.class) {
if (instance == null) {
instance = new Singleton() ;
}
}
}
return instance ;
}
}
Double check locking
public class Singleton {
private static Singleton instance ;
public static Singleton getInstance() {
if (instance == null) {
synchronized (Singleton.class) {
if (instance == null) {
instance = new Singleton() ;
}
}
}
return instance ;
}
}
Double check locking
BUG !
public enum Singleton {
instance ;
}
Solutions aux « data races »
• Synchronisation de la modification des variables
« garder les variables »
• Et si on ne modifie jamais nos variables ?
Systèmes immutables
SYNCHRONISATION
Java Memory Model
• Objet : définir la valeur d’une variable à un instant donné
• Règle : « toute lecture d’une variable retourne la dernière valeur écrite dans cette variable »
Java Memory Model
• Objet : définir la valeur d’une variable à un instant donné
• Règle : « toute lecture d’une variable retourne la dernière valeur écrite dans cette variable »
• Monothread : contrainte sur le fonctionnement des compilateurs & de la JVM
Java Memory Model
• Objet : définir la valeur d’une variable à un instant donné
• Règle : « toute lecture d’une variable retourne la dernière valeur écrite dans cette variable »
• Multithread multicœur : vrai problème !
« toute lecture … retourne la dernière valeur écrite … »
Java Memory Model
• Multithread multicœur : vrai problème !
« toute lecture … retourne la dernière valeur écrite … »
• Chronologie entre les threads / cœurs
Happens before
• Chronologie = créer des liens entre les lectures et les écritures
• Valeur de r1 ?
– Même thread ⇒ évident
– Threads différents : impossible de prévoir
x = 1
r1 = x
Happens before
• Chronologie = créer des liens entre les lectures et les écritures
• Valeur de r1 ?
– Imposée par la JLS : r1 = 1
x = 1
r1 = x
happens before
Happens before
• Règle :
Un lien « happens before » est établi entre toute écriture synchronisée ou volatile et toute lecture synchronisée ou volatile qui suit
• Existence de ce lien ⇒
Javadoc (java.util.concurrent)
private int index ;
public void incrementonsGaiement() {
index++ ;
}
public void testonsJoyeusement() {
if (index > 10) {
System.out.println("Index est grand !") ;
} }
private int index ;
public void incrementonsGaiement() {
index++ ;
}
public void testonsJoyeusement() {
if (index > 10) {
System.out.println("Index est grand !") ;
} }
Pas de lien « happens before »
private volatile int index ;
public void incrementonsGaiement() {
index++ ;
}
public void testonsJoyeusement() {
if (index > 10) {
System.out.println("Index est grand !") ;
} }
private volatile int index ;
public void incrementonsGaiement() {
index++ ;
}
public void testonsJoyeusement() {
if (index > 10) {
System.out.println("Index est grand !") ;
} }
Lien « happens before »
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
Cas pathologique
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T1 prend le lock
1
2
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T1 prend le lock
1
2
3
4
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T1 prend le lock
1
2
3
4
r2 = 1
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T2 prend le lock
1
4
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T2 prend le lock
1
? ?
4
private int x, y, r1, r2 ;
private Object lock = new Object() ;
void maMethode() {
x = 1 ;
synchronized(lock) {
y = 1 ;
}
}
void maMethodeAussi() {
synchronized(lock) {
r1 = y ;
}
r2 = x ;
}
1) T2 prend le lock
1
? ?
4
r2 = 0 ? r2 = 1 ?
public class Singleton {
private static Singleton instance ;
public static Singleton getInstance() {
if (instance == null) {
synchronized (Singleton.class) {
if (instance == null) {
instance = new Singleton() ;
}
}
}
return instance ;
}
}
Double check locking
Lecture non synchronisée
public class Singleton {
private volatile static Singleton instance ;
public static Singleton getInstance() {
if (instance == null) {
synchronized (Singleton.class) {
if (instance == null) {
instance = new Singleton() ;
}
}
}
return instance ;
}
}
Variable volatile !
PROCESSEUR
private Object o = new Object() ;
private int index = 0 ;
public void uneMethode() {
synchronized (o) {
index++ ;
}
}
private Object o = new Object() ;
private int index = 0 ;
public void uneMethode() {
synchronized (o) {
index++ ;
}
}
private Object o = new Object() ;
private int index = 0 ;
public void uneMethode() {
synchronized (o) {
index++ ;
}
}
« Visibilité »
Barrière de mémoire
• Une instruction assembleur
• Permet de « propager » une modification d’un cœur à l’autre
• Invalide le cache L1, force sa relecture dans L3 ou la mémoire
Optimisation 1
Concevoir des traitements tels que :
• Les données tiennent dans 32kO
• Le code tient dans 32kO
• Et pas de barrière de mémoire
⇒ Exécution à 1GHz
Optimisation 2
• S’arranger pour placer ses données dans des zones contigües de la mémoire
⇒ Divise pas 8 les temps de chargement
Cadeau empoisonné de L3
• Le cache invalide les données ligne par ligne
• Cas de deux threads :
– Le premier exécute une méthode, accède à a
– Le second exécute une autre méthode, accède à b
• Pas de concurrence d’accès sur a et b
Cadeau empoisonné de L3
• Catastrophique pour les perf !
• Solution : recours au « variable padding »
• Utilisé dans le disruptor (LMAX)
public class Sequence {
private static final AtomicLongFieldUpdater<Sequence> updater =
AtomicLongFieldUpdater.newUpdater(Sequence.class, "value");
private volatile long p1 = 7L, p2 = 7L, p3 = 7L, p4 = 7L,
p5 = 7L, p6 = 7L, p7 = 7L ;
private volatile long value = Sequencer.INITIAL_CURSOR_VALUE ;
private volatile long q1 = 7L, q2 = 7L, q3 = 7L, q4 = 7L,
q5 = 7L, q6 = 7L, q7 = 7L;
// ...
public long sumPaddingToPreventOptimisation() {
return p1 + p2 + p3 + p4 + p5 + p6 + p7 + value + q1 +
q2 + q3 + q4 + q5 + q6 + q7;
}
public void setPaddingValue(final long value)
{
p1 = p2 = p3 = p4 = p5 = p6 = p7 = q1 = q2 = q3 =
q4 = q5 = q6 = q7 = value;
}
}
Conclusion
• Prendre en compte le processeur
– Permet de belles optimisations
– Délicat à utiliser !
• Tenir compte de l’ordre des variables en mémoire
Conclusion
• Prendre en compte le processeur
– Permet de belles optimisations
– Délicat à utiliser !
• Tenir compte de l’ordre des variables en mémoire
• Veut-on le faire ?
• Le langage est-il prêt ?
java.util.concurrent
Nouvelle façon de
lancer les threads
La manière du JDK 1.1
Runnable r = new Runnable() {
public void run() {
while (true) {
System.out.println("Vogue la galère !") ;
}
}
} ;
Thread t = new Thread(r) ;
t.start() ;
La manière du JDK 1.1
Runnable r = new Runnable() {
public void run() {
while (true) {
System.out.println("Vogue la galère !") ;
}
}
} ;
Thread t = new Thread(r) ;
t.start() ; 1) Pas de type de retour 2) Pas d’exception 3) Création / destruction
La manière du JDK 5 ExecutorService service =
new ScheduledThreadPoolExecutor(10) ;
Callable<Boolean> task = new Callable<Boolean>() {
public Boolean call() throws Exception {
int i = 0 ;
while (i++ < 1000) {
System.out.println("Vogue la galère !") ;
}
return true ;
}
};
Future<Boolean> f = service.submit(task) ;
Boolean b = f.get(100, TimeUnit.MILLISECONDS) ;
1) Type de retour, exception 2) Cycle de vie géré par le pool
Nouvelle façon de
synchroniser
Interface Lock
private Lock lock = new ReentrantLock() ;
public void maMethode() {
lock.lock() ; // bloque « comme » synchronized
// ...
lock.unlock() ;
} ;
Interface Lock
private Lock lock = new ReentrantLock() ;
public void maMethode() {
lock.tryLock(10, TimeUnit.MILLISECONDS) ; // timeout
// ...
lock.unlock() ;
} ;
Classe Semaphore
private Semaphore s = new Semaphore(5) ;
public void maMethode() {
s.acquire() ; // bloque « comme » synchronized
// ...
s.release() ;
} ;
Classe Semaphore
private Semaphore s = new Semaphore(5) ;
public void maMethode() {
s.tryAcquire(10, TimeUnit.MILLISECONDS) ; // timeout
// ...
s.release() ;
} ;
Classe Semaphore
• Deux méthodes supplémentaires :
availablePermits()
getQueueLength()
Classe CountDownLatch
• Utilisable en initialisation (pas de reset)
private CountDownLatch latch = new CountDownLatch() ;
public void init() {
db.connect() ; // étape « lente »
latch.countDown() ; // ouverture du latch
} ;
public void process() {
latch.await() ; // attente de l’ouverture
} ;
Classe CyclicBarrier
• Synchronisation sur la fin du traitement de plusieurs tâches
public void processOne() {
// traitements divers
barrier.await() ;
} ;
private CyclicBarrier barrier = new CyclicBarrier(4) ;
public void processTwo() {
// traitements divers
barrier.await() ;
} ;
Interface ReadWriteLock
• Gère des paires de Lock
– Un pour la lecture
– Un pour l’écriture
• Lecture : concurrence possible
• Écriture : exclusive
• Visibilité des modifications garantie
Interface ReadWriteLock
private ReadWriteLock rwLock = new ReentrantReadWriteLock() ;
• Pas de contrainte sur les lectures
private ReadWriteLock rwLock = new ReentrantReadWriteLock() ;
public void maMethodeQuiLit() {
Lock readLock = rwLock.readLock() ;
readLock.lock() ;
// traitements de lecture
readLock.unlock() ;
} ;
Interface ReadWriteLock
• Les écritures stoppent les lectures
• Une seule écriture à la fois
private ReadWriteLock rwLock = new ReentrantReadWriteLock() ;
public void maMethodeQuiEcrit() {
Lock writeLock = rwLock.writeLock() ;
writeLock.lock() ;
// traitements de lecture
writeLock.unlock() ;
} ;
Types atomiques
Types atomiques
• AtomicLong, AtomicFloat, etc…
• Compile en une unique instruction assembleur
private AtomicInteger index = new AtomicInteger(0) ;
public void uneMethode() {
long newValue = index.incrementAndGet() ;
}
Types atomiques
• Principe du CAS « compare and swap »
• Prend en entrée :
– Une adresse mémoire
– Deux valeurs « ancienne » & « nouvelle »
• Si la valeur lue à l’adresse correspond à l’ancienne alors la nouvelle est recopiée
• Sinon, retourne false
• Pas de synchronisation !
Types atomiques
// classe AtomicLong
public final long incrementAndGet() {
for (;;) {
long current = get() ;
long next = current + 1 ;
if (compareAndSet(current, next))
return next ;
}
}
Files d’attente
• BlockingQueue
– File d’attente de taille fixe
– add(e), offer(e), put(e) avec timeout
– remove(e), poll(e), take(e)
– element(e), examine(e)
• Thread safe (locks internes)
– Utilisable pour le pattern producteur / consommateur
Tableau immutable
• CopyOnWriteArrayList (et version LinkedList)
– Lecture synchronisée
– Écriture par duplication
• Le tableau lu n’est jamais modifié, donc on peut le partager sans synchronisation
Structures immutables
• Peut-on construire d’autres structures immutables dans l’API Collection ?
– Tables de hachage
• Réponse : oui, et de façon efficace
– Ce qui n’est pas le cas des tableaux
Alternatives à
synchronized
STM Software Transactionnal memory
Philisophie
• Base de données :
• Sur le commit :
– Succès : on continue
– Échec : on annule les modifications
– Règles de visibilité
• Idée de 1986, implémentations 1995
begin
update ... set ... where ...
commit
Même chose dans la JVM ?
• Plusieurs implémentations
• Akka 1.3.1
<dependency>
<groupId>se.scalablesolutions.akka</groupId>
<artifactId>akka-kernel</artifactId>
<version>1.3.1</version>
</dependency>
<repository>
<id>Akka</id>
<name>Akka Maven2 Repository</name>
<url>http://akka.io/repository/</url>
</repository>
Pattern STM
final Ref<Integer> source = new Ref<Integer>(500) ;
final Atomic<Object> atom = new Atomic<Object>() {
@Override
public Object atomically() {
source.swap(source.get() + 1) ;
return null ;
}
} ;
atom.execute() ;
Effectué dans une transaction
Pattern STM
final Ref<Integer> source = new Ref<Integer>(500) ;
final Atomic<Object> atom = new Atomic<Object>() {
@Override
public Object atomically() {
source.swap(source.get() + 1) ;
return null ;
}
} ;
atom.execute() ;
Exécuté dans le thread courant
Pattern STM
• Une transaction qui échoue est rejouée (idem AtomicLong)
– Utilisation du processeur différente de la synchronisation
• Cas applicatifs non accessibles aux Locks
Object o = queue1.remove() ;
queue2.add(o) ;
Pattern STM
• Bons cas applicatifs : faible concurrence d’accès
• Forte concurrence : les cœurs passent leur temps à rejouer le code
Implémentation des STM
• Peut utiliser les Locks
• Peut ne pas les utiliser
Dans ce cas : quid de la visibilité (pas de barrière de mémoire)
Visibilité : comment l’intégrité des données est-elle garantie ?
Implémentation des STM
• Consomme des ressources (mémoire)
– Fonction du nombre de transactions
– Peut être en O(n2) (très mauvais)
• Concurrence avec les Locks, qui n’en consomment pas
Futur des STM
• Architecture Haswell d’Intel (2013)
– Support de TSX (Transactional Synchronisation Extension)
– TSX permet de marquer des portions de code
– La mémoire modifiée n’est visible que du thread qui modifie cette mémoire
– Elle est publiée « instantanément » sur commit
Acteurs
Modèle d’acteurs
1973
• Un acteur réalise un traitement
• Il reçoit un message immutable
• Effectue un traitement, et retourne éventuellement un résultat
Non synchronisé car sans accès concurrent
Modèle d’acteurs
• Utilisé en Erlang
• Plusieurs API Java et sur la JVM (Scala, Clojure)
Akka : écrit en Scala
Akka : fonctionnement
• Akka gère un pool de threads
• On crée des acteurs, on leur envoie des messages
• Un message reçu est traité dans un thread
• Un acteur ne peut pas être exécuté dans deux threads à la fois
• Pas de concurrence d’accès sur son éventuel état
4000! – Akka
public class PrimeFinderActor extends UntypedActor {
public void onReceive(Object o) throws Exception {
List<Integer> bounds = (List<Integer>)o ;
int debut = bounds.get(0) ;
int fin = bounds.get(1) ;
PrimeFactors pfs = new PrimeFactors() ;
for (int i = debut ; i < fin ; i++) {
PrimeFactors pfi = pfs.getPrimeFactors(i) ;
pfs.add(pfi) ;
} // return pfs ;
getContext().reply(pfs) ;
}
}
4000! - ExecutorService public class PrimeFactorCallable
implements Callable<PrimeFactors> {
private int debut, fin ;
public PrimeFactorCallable(int debut, int fin) {
this.debut = debut ; this.fin = fin ;
}
public PrimeFactors call() throws Exception {
PrimeFactors pfs = new PrimeFactors() ;
for (int i = debut ; i < fin ; i++) {
PrimeFactors pfi = pfs.getPrimeFactors(i) ;
pfs.add(pfi) ;
}
return pfs ;
}
}
public static void main(String[] args) {
Future [] futures = new Future [400] ;
for (int i = 0 ; i < futures.length ; i++) {
List<Integer> bound = Collections.unmodifiableList(
Arrays.asList(10*i, 10*(i + 1))) ;
ActorRef primeFactorFinder =
Actors.actorOf(PrimeFinderActor.class).start() ;
futures[i] =
primeFactorFinder.sendRequestReplyFuture(bound) ;
}
PrimeFactors pfs = new PrimeFactors() ;
for (int i = 0 ; i < futures.length ; i++) {
pfs.add((PrimeFactors)futures[i].get()) ;
}
Actors.registry().shutdownAll() ;
}
public static void main(String... args)
throws ExecutionException, InterruptedException {
ExecutorService es = new ScheduledThreadPoolExecutor(10) ;
Future [] futures = new Future [400] ;
for (int i = 0 ; i < futures.length ; i++) {
PrimeFactorCallable callable =
new PrimeFactorCallable(10*i, 10*(i + 1)) ;
futures[i] = es.submit(callable) ;
}
PrimeFactors pfs = new PrimeFactors() ;
for (int i = 0 ; i < futures.length ; i++) {
pfs.add((PrimeFactors)futures[i].get()) ;
}
es.shutdown() ;
}
[2^3989][3^1996][5^996][7^664][11^399][13^331][17^248][19^221][23^180][29^141][31^133][37^110][41^99][43^
95][47^86][53^76][59^68][61^66][67^59][71^56][73^54][79^50][83^48][89^44][97^41][101^39][103^38][107^37][
109^36][113^35][127^31][131^30][137^29][139^28][149^26][151^26][157^25][163^24][167^23][173^23][179^22][1
81^22][191^20][193^20][197^20][199^20][211^18][223^17][227^17][229^17][233^17][239^16][241^16][251^15][25
7^15][263^15][269^14][271^14][277^14][281^14][283^14][293^13][307^13][311^12][313^12][317^12][331^12][337
^11][347^11][349^11][353^11][359^11][367^10][373^10][379^10][383^10][389^10][397^10][401^9][409^9][419^9]
[421^9][431^9][433^9][439^9][443^9][449^8][457^8][461^8][463^8][467^8][479^8][487^8][491^8][499^8][503^7]
[509^7][521^7][523^7][541^7][547^7][557^7][563^7][569^7][571^7][577^6][587^6][593^6][599^6][601^6][607^6]
[613^6][617^6][619^6][631^6][641^6][643^6][647^6][653^6][659^6][661^6][673^5][677^5][683^5][691^5][701^5]
[709^5][719^5][727^5][733^5][739^5][743^5][751^5][757^5][761^5][769^5][773^5][787^5][797^5][809^4][811^4]
[821^4][823^4][827^4][829^4][839^4][853^4][857^4][859^4][863^4][877^4][881^4][883^4][887^4][907^4][911^4]
[919^4][929^4][937^4][941^4][947^4][953^4][967^4][971^4][977^4][983^4][991^4][997^4][1009^3][1013^3][1019
^3][1021^3][1031^3][1033^3][1039^3][1049^3][1051^3][1061^3][1063^3][1069^3][1087^3][1091^3][1093^3][1097^
3][1103^3][1109^3][1117^3][1123^3][1129^3][1151^3][1153^3][1163^3][1171^3][1181^3][1187^3][1193^3][1201^3
][1213^3][1217^3][1223^3][1229^3][1231^3][1237^3][1249^3][1259^3][1277^3][1279^3][1283^3][1289^3][1291^3]
[1297^3][1301^3][1303^3][1307^3][1319^3][1321^3][1327^3][1361^2][1367^2][1373^2][1381^2][1399^2][1409^2][
1423^2][1427^2][1429^2][1433^2][1439^2][1447^2][1451^2][1453^2][1459^2][1471^2][1481^2][1483^2][1487^2][1
489^2][1493^2][1499^2][1511^2][1523^2][1531^2][1543^2][1549^2][1553^2][1559^2][1567^2][1571^2][1579^2][15
83^2][1597^2][1601^2][1607^2][1609^2][1613^2][1619^2][1621^2][1627^2][1637^2][1657^2][1663^2][1667^2][166
9^2][1693^2][1697^2][1699^2][1709^2][1721^2][1723^2][1733^2][1741^2][1747^2][1753^2][1759^2][1777^2][1783
^2][1787^2][1789^2][1801^2][1811^2][1823^2][1831^2][1847^2][1861^2][1867^2][1871^2][1873^2][1877^2][1879^
2][1889^2][1901^2][1907^2][1913^2][1931^2][1933^2][1949^2][1951^2][1973^2][1979^2][1987^2][1993^2][1997^2
][1999^2][2003^1][2011^1][2017^1][2027^1][2029^1][2039^1][2053^1][2063^1][2069^1][2081^1][2083^1][2087^1]
[2089^1][2099^1][2111^1][2113^1][2129^1][2131^1][2137^1][2141^1][2143^1][2153^1][2161^1][2179^1][2203^1][
2207^1][2213^1][2221^1][2237^1][2239^1][2243^1][2251^1][2267^1][2269^1][2273^1][2281^1][2287^1][2293^1][2
297^1][2309^1][2311^1][2333^1][2339^1][2341^1][2347^1][2351^1][2357^1][2371^1][2377^1][2381^1][2383^1][23
89^1][2393^1][2399^1][2411^1][2417^1][2423^1][2437^1][2441^1][2447^1][2459^1][2467^1][2473^1][2477^1][250
3^1][2521^1][2531^1][2539^1][2543^1][2549^1][2551^1][2557^1][2579^1][2591^1][2593^1][2609^1][2617^1][2621
^1][2633^1][2647^1][2657^1][2659^1][2663^1][2671^1][2677^1][2683^1][2687^1][2689^1][2693^1][2699^1][2707^
1][2711^1][2713^1][2719^1][2729^1][2731^1][2741^1][2749^1][2753^1][2767^1][2777^1][2789^1][2791^1][2797^1
][2801^1][2803^1][2819^1][2833^1][2837^1][2843^1][2851^1][2857^1][2861^1][2879^1][2887^1][2897^1][2903^1]
[2909^1][2917^1][2927^1][2939^1][2953^1][2957^1][2963^1][2969^1][2971^1][2999^1][3001^1][3011^1][3019^1][
3023^1][3037^1][3041^1][3049^1][3061^1][3067^1][3079^1][3083^1][3089^1][3109^1][3119^1][3121^1][3137^1][3
163^1][3167^1][3169^1][3181^1][3187^1][3191^1][3203^1][3209^1][3217^1][3221^1][3229^1][3251^1][3253^1][32
57^1][3259^1][3271^1][3299^1][3301^1][3307^1][3313^1][3319^1][3323^1][3329^1][3331^1][3343^1][3347^1][335
9^1][3361^1][3371^1][3373^1][3389^1][3391^1][3407^1][3413^1][3433^1][3449^1][3457^1][3461^1][3463^1][3467
^1][3469^1][3491^1][3499^1][3511^1][3517^1][3527^1][3529^1][3533^1][3539^1][3541^1][3547^1][3557^1][3559^
1][3571^1][3581^1][3583^1][3593^1][3607^1][3613^1][3617^1][3623^1][3631^1][3637^1][3643^1][3659^1][3671^1
][3673^1][3677^1][3691^1][3697^1][3701^1][3709^1][3719^1][3727^1][3733^1][3739^1][3761^1][3767^1][3769^1]
[3779^1][3793^1][3797^1][3803^1][3821^1][3823^1][3833^1][3847^1][3851^1][3853^1][3863^1][3877^1][3881^1][
3889^1][3907^1][3911^1][3917^1][3919^1][3923^1][3929^1][3931^1][3943^1][3947^1][3967^1][3989^1]
Performances
• Les codes se ressemblent
• Utilisation CPU : idem
• Temps de calcul :
– Akka : 2,7s
– ExecutorService : 2,3s
• Différence non significative
Exécution transactionnelle
public class Balance {
public final int montant ;
public Balance(int montant) {
this.montant = montant ;
}
}
public class Retrait {
public final int montant ;
public Retrait(int montant) {
this.montant = montant ;
}
}
Exécution transactionnelle
public class Depot {
public final int montant ;
public Depot(int montant) {
this.montant = montant ;
}
}
public class GetBalance {
}
public class CompteBancaire extends UntypedTransactor {
private final Ref<Integer> balance = new Ref<Integer>(0) ;
public void atomically(final Object message) {
if (message instanceof Depot) {
int montant = ((Depot)message).montant ;
if (montant > 0)
balance.swap(balance.get() + montant) ;
}
if (message instanceof Retrait) {
int montant = ((Retrait)message).montant ;
if (montant > 0) {
if (balance.get() < montant)
throw new IllegalStateException("...") ;
balance.swap(balance.get() - montant) ;
}
}
if (message instanceof GetBalance)
getContext().replySafe((new Balance(balance.get()))) ;
}
}
Exécuté dans une transaction
public class CompteBancaire extends UntypedTransactor {
private final Ref<Integer> balance = new Ref<Integer>(0) ;
public void atomically(final Object message) {
if (message instanceof Depot) {
int montant = ((Depot)message).montant ;
if (montant > 0)
balance.swap(balance.get() + montant) ;
}
if (message instanceof Retrait) {
int montant = ((Retrait)message).montant ;
if (montant > 0) {
if (balance.get() < montant)
throw new IllegalStateException("...") ;
balance.swap(balance.get() - montant) ;
}
}
if (message instanceof GetBalance)
getContext().replySafe((new Balance(balance.get()))) ;
}
}
Fera échouer la transaction
public class Transfert {
public final ActorRef source ;
public final ActorRef destination ;
public final int montant ;
public Transfert(final ActorRef source,
final ActorRef destination,
final int montant) {
this.source = source ;
this.destination = destination ;
this.montant = montant ;
}
}
public class ServiceBancaire extends UntypedTransactor {
public Set<SendTo> coordinate(final Object message) {
if (message instanceof Transfert) {
Set<SendTo> s = new HashSet<SendTo>() ;
Transfert t = (Transfert)message ;
if (t.montant > 0) { // validation
s.add(sendTo(t.destination, new Depot(t.montant))) ;
s.add(sendTo(t.source, new Retrait(t.montant))) ;
return Collections.unmodifiableSet(s) ;
}
}
return nobody() ;
}
}
Ensemble d’appels d’acteurs
public class ServiceBancaire extends UntypedTransactor {
public Set<SendTo> coordinate(final Object message) {
if (message instanceof Transfert) {
Set<SendTo> s = new HashSet<SendTo>() ;
Transfert t = (Transfert)message ;
if (t.montant > 0) { // validation
s.add(sendTo(t.destination, new Depot(t.montant))) ;
s.add(sendTo(t.source, new Retrait(t.montant))) ;
return Collections.unmodifiableSet(s) ;
}
}
return nobody() ;
}
}
Ensemble d’appels d’acteurs
public class ServiceBancaire extends UntypedTransactor {
public Set<SendTo> coordinate(final Object message) {
if (message instanceof Transfert) {
Set<SendTo> s = new HashSet<SendTo>() ;
Transfert t = (Transfert)message ;
if (t.montant > 0) { // validation
s.add(sendTo(t.destination, new Depot(t.montant))) ;
s.add(sendTo(t.source, new Retrait(t.montant))) ;
return Collections.unmodifiableSet(s) ;
}
}
return nobody() ;
}
}
Appel dans une transaction
Exécution transactionnelle
• Chaque acteur gère un compte
• Il possède un état, mais qui n’est jamais partagé
• Les transferts sont gérés par un acteur sans état
• Toutes les opérations sont transactionnelles
Bilan STM & acteurs
• STM & acteurs : deux approches alternatives à synchronized
• Offrent des fonctionnalités supplémentaires
• Vont bénéficier du support de l’assembleur
• Pas de miracle au niveau performance
Parallélisation des
calculs
Nouveautés Java 7 & 8
• Pattern Fork / Join
• API non incluse en Java 7 : parallel arrays
• Méthode en Java 8 : parallel()
Objet :
• Automatiser la parallélisation sur les tableaux et les collections
• Ne plus avoir à écrire de code de parallélisation
Java 7 : fork / join
Fork / Join
• Traite des calculs sur de grandes quantités de données, numériques ou non
• Gère des tâches
– Si elle est trop grosse elle se divise : Fork
– Sinon elle s’exécute et retourne son résultat
• Une tâche qui s’est divisée récupère les résultats de ses sous-tâches : Join
Fork / Join
• Gère un pool de threads
• Chaque thread possède une file d’attente qui stocke les tâches
• Quand sa file d’attente est vide, il va voler du travail à son voisin
Fork / Join
• Chaque tâche travaille sur ses propres données
• Pas de bloc synchronisé
• Pas de variables partagées
• On doit donc « dupliquer » des informations
• Cas du traitement des tableaux de grande taille ?
Parallel Arrays
Java 6 &7
Parallel Arrays
• API non incluse dans le JDK
• Peut être utilisée en Java 6
• Fork / Join dans un package jsr166y.*
• Avantage : API auto-porteuse
Pourquoi les parallel arrays
ne sont-ils pas dans
l’API Java 7 ?
Java 8
Sept. 2013
Syntaxe
• Du light !
int maxAge =
persons.map(p -> p.getAge()).reduce(0, Integer::max) ;
Collection<Person> oldies =
persons.filter(p -> p.age > 40).into(new ArrayList()) ;
Syntaxe
• Du light !
int maxAge =
persons.map(p -> p.getAge()).reduce(0, Integer::max) ;
Collection<Person> oldies =
persons.filter(p -> p.age > 40).into(new ArrayList()) ;
Type Collection
Syntaxe
• Du light !
int maxAge =
persons.map(p -> p.getAge()).reduce(0, Integer::max) ;
Collection<Person> oldies =
persons.filter(p -> p.age > 40).into(new ArrayList()) ;
D’où viennent les méthodes map() et filter() ??
Syntaxe
• Du light !
• Évolution de l’API Collection, méthodes vituelles d’extension
int maxAge =
persons.map(p -> p.getAge()).reduce(0, Integer::max) ;
Collection<Person> oldies =
persons.filter(p -> p.age > 40).into(new ArrayList()) ;
Méthodes virtuelles d’ext.
• On ajoute des méthodes dans les interfaces, avec leurs implémentation par défaut
public interface Collection<E> {
public boolean add(E e) ;
public boolean addAll(Collection<? extends E> c) ;
// la litanie des méthodes de Collection
// méthode virtuelle d'extension
public void sort(Comparator<? super E> comp)
default Collections.sort(comp) ;
}
Syntaxe
• Du light !
• Support du parallélisme
int maxAge =
persons.map(p -> p.getAge()).reduce(0, Integer::max) ;
Collection<Person> oldies =
persons.filter(p -> p.age > 40).into(new ArrayList()) ;
Collection<Person> oldies =
persons.parallel().filter(p -> p.age > 40)
.into(new ArrayList()) ;
Interface spliterable()
• Extension d’iterable()
• Type de retour de parallel()
• Le fork / join devient un framework système
• Les fonctionnalités des parallel arrays sont intégrées via les lambdas
• On gagne java.util.parallel
Algorithmique
Une anecdote
Le 24/2/2012 Heinz Kabutz lance un défi :
• Déterminer les 10 premiers et 10 derniers octets du nombre Fibonacci (1 milliard)
• Taille du nombre écrit dans un fichier = ~180Mo
Une anecdote
• Record à battre : 5600s sur un CPU 8 cœurs
Une anecdote
• Record à battre : 5600s sur un CPU 8 cœurs
• 1ère amélioration : 3300s sur un CPU 4 cœurs
Une anecdote
• Record à battre : 5600s sur un CPU 8 cœurs
• 1ère amélioration : 3300s sur un CPU 4 cœurs
• 2ème amélioration : 51s sur un cœur
Une anecdote
Par quel miracle ?
• Utilisation de GNU Multiple Precision Arithmetic Lib (GMP)
• Changement d’implémentation de BigInteger
• Changement d’algorithme de calcul de Fibonacci
• Suppression de l’utilisation de la récursivité
Par quel miracle ?
• Utilisation de GNU Multiple Precision Arithmetic Lib (GMP)
• Changement d’implémentation de BigInteger
• Changement d’algorithme de calcul de Fibonacci
• Suppression de l’utilisation de la récursivité
Quel rapport avec
Fork / Join ou la
parallélisation ?
Algorithmique
L’optimisation des fonctions linéaires a gagné un facteur 43M sur les temps de traitement en 15 ans :
• 1k est dû à la vitesse des processeurs
• 43k est dû à l’amélioration des algorithmes
Moralité
• Utiliser les bons algorithmes de traitement avant de penser à la parallélisation
Moralité
• Utiliser les bons algorithmes de traitement avant de penser à la parallélisation
• S’assurer que les algorithmes sont bien parallélisables !!
Moralité
• Utiliser les bons algorithmes de traitement avant de penser à la parallélisation
• S’assurer que les algorithmes sont bien parallélisables !!
Quick sort : se parallélise mal
• Merge sort consomme de la mémoire
Moralité
• Utiliser les bons algorithmes de traitement avant de penser à la parallélisation
• S’assurer que les algorithmes sont bien parallélisables !!
Recuit simulé : ne se parallélise pas, perd ses propriétés de convergence
Conclusion
Conclusion
Une nouvelle période charnière se déroule en ce moment
En 1995 : la mémoire est devenue une ressource gérée par la JVM
En 2012 : le CPU est en train de devenir une ressource gérée par la JVM (et le serveur aussi !)
Les langages et les processeurs s’adaptent les uns aux autres
Conclusion
Les applications vont utiliser des traitements en parallèle de façon transparente
Mais les pièges se trouvent au niveau des algorithmes, ils seront plus durs à éviter
Nouveaux challenges pour les développeurs, et de nouvelles opportunités !