Synchronisation de groupe de threads [Sources]

[millie]

Lors de calcul en parallèle, il peut être utile de mettre des points de synchronisation pour attendre que tous les threads soient arrivés au même point avant de continuer les calculs.

Par exemple, si vous effectuez un même traitement dans une boucle

Code :

1
2
3
 
for(int i = 0; i < 50; i++)
  traitement(i);

Il est parfois possible de paralléliser sous la forme :

Code :

1
2
3
4
5
6
7
8
9
 
int nbThread = 3; //nombre total de threads
 
 
//chaque thread effectue une partie des traitement
for(int i = 0; i<50; i+= nbThread) {
  traitement(i);
  attendreAutresThreads();
}

Il peut être utile d'attendre les autres threads pour éviter les incohérences dans les calculs (typiquement, on peut faire des traitements sur une image sans avoir à mettre un synchronized sur l'image car chaque thread va modifier des pixels différents).

Code :

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
/**
 * 
 */
package millie.thread;
 
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
/**
 * @author millie
 * 
 */
public class PointSyncThreadGroup {
	private Set<Thread> threads = new HashSet<Thread>();
 
	private Object mutex = new Object();
 
	private ReentrantLock waitLock = new ReentrantLock();
	Condition notFull  = waitLock.newCondition(); 
 
	private boolean isStarted = false;
	private boolean isFinished = false;
 
	private int threadWaiting = 0;
 
 
 
	/**
	 * Permet d'ajouter un thread au groupe
	 *  
	 * @param t
	 */
	public void add(Thread t) {
		synchronized (mutex) {
			threads.add(t);
		}
	}
 
	/**
	 *  Démarre l'ensemble des threads
	 *  
	 *  @throws IllegalThreadStateException si le groupe a déjà démarré
	 *
	 */
	public void startAll() {
		synchronized (mutex) {
			if(isStarted)
				throw new IllegalThreadStateException("Started before");
 
			for (Thread t : threads)
				t.start();
			isStarted = true;
		}
	}
 
	/**
	 *  Permet d'attendre tous les autres threads
	 *   
	 * @param t
	 * @throws InterruptedException
	 */
	public void waitAll(Thread t) throws InterruptedException {
		synchronized (mutex) {
			if (!isStarted)
				throw new IllegalThreadStateException("Not started");
			if(isFinished)
				throw new IllegalThreadStateException("Is finished");
		}
 
		waitLock.lock();
		try {
			threadWaiting++;
 
			if (threadWaiting != threads.size()) //pas de while, le cas ne peut pas se produire
				notFull.await();
			else {
				threadWaiting = 0;
				notFull.signalAll();
			}
 
		} finally {
			waitLock.unlock();
		}
 
	}
 
	/**
	 *  Attend la fin de tous les threads
	 *   
	 * @throws InterruptedException
	 */
	public void joinAll() throws InterruptedException {
		for (Thread t : threads)
			t.join();
		synchronized (mutex) {
			isFinished = true;
		}
	}
 
}

Code :

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
/**
 * 
 */
package millie.thread;
 
class MonThread extends Thread {
	private PointSyncThreadGroup g;
 
	private int nbThread;
 
	public MonThread(PointSyncThreadGroup g, int nbThread) {
		this.g = g;
		this.nbThread = nbThread;
	}
 
	public void run() {
		for (int i = 0; i < 5; i++) {
 
			//traitement métier
			for (int j = 0; j < 5; j++)
				System.err.println(nbThread + " - " + j);
 
			try {
				g.waitAll(this);
				System.err.println("FIN ATTENTE");
			} catch (InterruptedException e) {
				System.err.println(e.getMessage());
			}
		}
	}
}
 
/**
 * @author millie
 * 
 */
public class Test {
	public static void main(String[] args) throws InterruptedException {
		try {
			PointSyncThreadGroup group = new PointSyncThreadGroup();
 
			MonThread t1 = new MonThread(group, 1);
			MonThread t2 = new MonThread(group, 2);
			MonThread t3 = new MonThread(group, 3);
			group.add(t1);
			group.add(t2);
			group.add(t3);
 
			group.startAll();
			group.joinAll();
		} catch (Exception e) {
			System.err.println(e.getMessage());
			e.printStackTrace();
		}
	}
}

Evidemment, si un thread n'attend pas les autres, les autres vont l'attendre et le processus risque de bloquer.