3.1. Exercice 1, Serveur multi-thread¶
3.1.1. Le Thread principal accueillant les clients.¶
Après la phase d’initialisation qui reste inchangée, le comportement du serveur se décrit comme suit:
- Tant que vrai:
- Accepter un client et récupérer la socket associée \(S\).
- Creer un nouveau Thread \(T\) de type ServiceClient associé à la socket \(S\).
- Démarrer \(T\)
En Java ceci s’exprime comme ci dessous :
la ligne 5 est toujours bloquante et permet de récupérer la socket (le catch try est omis par soucis de clarté)
la ligne 9 crée l’objet exécutable (la socket est passée au cŕeateur de l’objet, on passe aussi un nom sans grande importance), crée le thread et le démarre.
Attention Encore: Si vous utilisez la méthode run() votre programme semblera fonctionner mais les clients seront servis à la queue-leu-leu, en effet en ce cas vous appelez la méthode qui sert un client mais sans créer de thread. Le programme est alors équivalent à celui du serveur séquentiel du TD1. Seule la méthode start() lance le thread et ensuite appelle run().
while(true)
{
System.out.println("[Serveur]: waiting for connexion") ;
ma_connection = gestionnaire_de_connexion.accept();
String c_ip = ma_connection.getInetAddress().toString() ;
int c_port= ma_connection.getPort();
System.out.format("[Serveur] : Arrivée du client IP %s sur le port %d\n", c_ip, c_port);
System.out.format("[Serveur ]: Creation du thread T_%d\n" , c_port);
new Thread( new ServiceClient(ma_connection , "T_" +c_port )).start();
}
A titre de rappel l’initialisation peut se résumer à la création du \(ServerSocket\), avec les sempiternelles lignes try/catch.
public static void main(String[] args)
{
Socket ma_connection=null;
int port = 12000;
ServerSocket gestionnaire_de_connexion;
try{
gestionnaire_de_connexion = new ServerSocket(port);
System.out.println("[Serveur] : Serveur Jouet lancé sur " + (port) );
}
catch (IOException e) {
System.out.format("Cannot create to the server, port %d may be busy\n", port);
System.exit(-1); }
while(true) {
// Boucle infinie acceptant les clients
}
3.1.2. Le Thread servant 1 client¶
L’objet exécutable ServiceClient se résume à sa méthode run qui sert un client. La classe ne possède que deux champs, le nom du thread et la référence à la socket traitées. Le constructeur permet de passer la référence à la socket connectée au client.
public class ServiceClient implements Runnable{
private static String Finish="end";
private Socket ma_connection;
private String id;
public ServiceClient(Socket la_connection, String mid)
{
ma_connection= la_connection;
id=mid;
System.out.format("Thread T__%s créé pour traiter la connection\n",id);
}
}
Pour la méthde \(run\) On peut utiliser n’importe quelle méthode sequentielle servant un client, ie une méthode du TP1. Par exemple on peut utiliser le code ci-dessous où le service est ultra sommaire.
run(){
InputStreamReader isr = new InputStreamReader(ma_connection.getInputStream(), "UTF-8");
flux_entrant = new BufferedReader(isr) ;
ma_sortie = new PrintWriter(ma_connection.getOutputStream() , true);
String message_lu = new String();
int line_num =0 ;
while (true)
{
message_lu = flux_entrant.readLine();}
line_num++;
System.out.format( "%s [line_%d]--> [%s]]\n", id,line_num, message_lu);
}
}
3.1.3. Exemple final de code¶
Au final on peut utiliser les deux classes suivantes, par soucis de concision on a laissé remonter les exceptions. Bien entendu le corps de la méthode \(run()\) de la classe ServiceClient peut être remplacé par votre code.
Pour le répartiteur: Notez que juste que l’on étend la classe SocketServer et qu’en ligne 21 on crée un objet exécutable, on le place dans un thread et on demarre ce thread.
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 | import java.net.*;
import java.io.*;
public class Repartiteur extends ServerSocket {
private final static int port = 12000; /* Port d'écoute */
public Repartiteur() throws IOException {
super(port);
System.out.println("[Serveur] : Serveur Jouet lancé sur " + (port));
}
public void execute() throws IOException {
Socket maConnection;
while (true) {
System.out.println("[Serveur]: waiting for connexion");
maConnection = accept();
String c_ip = maConnection.getInetAddress().toString();
int c_port = maConnection.getPort();
System.out.format("[Serveur] : Arr. Client IP %s sur %d\n", c_ip, c_port);
System.out.format("[Serveur ]: Creation du thread T_%d\n", c_port);
new Thread(new ServiceClient(maConnection, "T_" + c_port)).start();
}
}
public static void main(String[] args) throws IOException {
Repartiteur connectionManager = new Repartiteur();
connectionManager.execute();
}
}
|
Pour le Thread servant un client :
import java.net.*;
import java.util.concurrent.TimeoutException;
import java.io.*;
public class ServiceClient implements Runnable{
//String to finish the communication ici c est ctrl-d
//private static final String Finish="end";
private String Finish=""+(char) 4;
private Socket ma_connection;
private String id;
private void terminer(){
try{
if (ma_connection != null)
{System.out.format("Terminaison pour %s\n", id); ma_connection.close();}
}
catch (IOException e) {
System.out.format("Terminaison pour %s\n", id);
e.printStackTrace();
}
return;
}
public ServiceClient(Socket la_connection, String mid)
{
ma_connection= la_connection;
id=mid;
System.out.format("Thread T__%s créé pour traiter la connection\n",id);
}
public void run(){
// Phase d initialisation
BufferedReader flux_entrant=null;
PrintWriter ma_sortie =null;
try{
InputStreamReader isr = new InputStreamReader(ma_connection.getInputStream(), "UTF-8");
flux_entrant = new BufferedReader(isr) ; // file d'entrée
// flux de sortie en mode autoflush
ma_sortie = new PrintWriter(ma_connection.getOutputStream() , true);
String c_ip = ma_connection.getInetAddress().toString() ;
int c_port= ma_connection.getPort();
System.out.format("[%s] client admis IP %s sur le port %d\n", id,c_ip, c_port);
ma_sortie.format("[%s] : Hello %s sur le port %d, \n" , id, c_ip, c_port );
}
catch (Exception e1) {
System.out.println("Erreur d initialisation") ;e1.printStackTrace();}
String message_lu = new String();
int line_num =0 ;
// Fin de l initialisation
// Boucle principale //
while ( true )
{
try {
message_lu = flux_entrant.readLine();
}
catch (IOException ioe) { ;}
if (message_lu == null){
System.out.println( "Client deconnecté, je termine\n" ) ;
terminer();
return; }
System.out.format( "%s [line_%d]--> [%s]]\n", id,line_num, message_lu);
if (message_lu.contains(Finish) ){
System.out.format ("[%s] : [%s] recu, Transmission finie\n",id,message_lu);
ma_sortie.println("Fermeture de la connexion");
terminer();
return;
}
line_num++;
}
}}
3.1.4. Génération de Threads client¶
Pour le corrigé chaque client va juste compter de 0 à 9 et envoyer les messages 0,1,...,9 puis le message de terminaison \(FINISH\) au serveur. Par ailleurs on a ajouté un petit délai aléatoire entre deux messages afin que les clients soient interlacés (sinon sinon le premier se termine avant même que le second ai commencé).
Classe associée à un client
import java.net.*;
import java.io.*;
public class Thread_Client implements Runnable {
private int serveurPort = 12000;
private String serveurIp = "127.0.0.1";
private final String FINISH = "" + (char) 4;
private String id;
private BufferedReader Tampon_Lecture = null;
private PrintWriter ma_sortie = null;
public Thread_Client( String monId) {id=monId; }
public Thread_Client(String hote, int port, String mon_id) {
this.serveurIp = hote; this.serveurPort = port;
this.id = mon_id;
}
public void run() {
Socket la_connection = null;
try {
la_connection = new Socket(serveurIp, serveurPort);
Tampon_Lecture = new BufferedReader(new InputStreamReader(
la_connection.getInputStream()));
ma_sortie = new PrintWriter(la_connection.getOutputStream(), true);
} catch (IOException e) {
System.out.println(e);System.exit(-1);
}
System.out.format("%s: Contact Reussi avec %s:%d\n", id, serveurIp,
serveurPort);
for (int i = 0; i < 10; i++) {
ma_sortie.format("%s: message %d\n", id, i);
try {
Thread.sleep( (int)( 3000*Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
ma_sortie.format("%s\n",FINISH);
if (la_connection !=null)
try {
la_connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
Le thread principal se contente de créer 10 Threads de la classe Thread_Client et de les démarrer.
Classe principale
public class Gen_Clients {
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
String mon_id = String.format("Client_%d", i);
Thread_Client currrent_client = new Thread_Client(mon_id);
Thread myThread= new Thread(currrent_client);
myThread.start();
}
}
}
3.2. Exercice 2 Arrêt du serveur via une interruption¶
3.2.1. Du coté client :¶
Si on veut arreter un des thread de service il faut empecher que celui-ci bloque lors de l’attente de message du client servi, ce qui arrive lors de l’utilisation de la méthode \(readline()\). On va donc placer un Timeout sur la méthode \(readline()\) ou plus excactement sur la socket que le service de client utilise.
On positionne le timeout sur la socket comme suit :
Socket ma_connection;
ma_connection.setSoTimeout(10);
Dès lors l’appel de la méthode \(readline\) génère une exception quand le delai maximum est dépassé. On la gère avec une paires \(catch/try\)
try {
message_lu = flux_entrant.readLine();
}
catch (SocketTimeoutException to) {continue;}
catch (IOException ioe) { sys.exit(-1);}
Ici en cas d’erreur d’entrée sortie on termine le programme, mais en cas de délai dépassé (SocketTimeoutException) on continue le programme.
Pour rendre le thread ServiceClient interruptible on le modifie très légèrement en remplacant sa boucle principale par la suivante où les modification suivantes ont été apportées:
- ligne 1 on positionne le timeout à 10 millisecondes.
- On a ajouté Le bloc formé par les lignes [4,9] pour terminer le thread en cas d’interruption.
- en ligne 13 on gère l’exception due au timeout, on continue car le timeout est ici attendu
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 | ma_connection.setSoTimeout(10);
while ( true )
{
if(Thread.interrupted())
{
System.out.println( "Service interompu par le serveur, je m'arrete" ) ;
terminer();
return;
}
try {
message_lu = flux_entrant.readLine();
}
catch (SocketTimeoutException to) {continue;}
catch (IOException ioe) { ;}
if (message_lu == null){
System.out.println( "Client deconnecté, je termine\n" ) ;
terminer();
return; }
System.out.format( "%s [line_%d]--> [%s]]\n", id,line_num, message_lu);
if (message_lu.contains(Finish) ){
System.out.format ("[%s] : [%s] recu, Transmission finie\n",id,message_lu);
ma_sortie.println("Fermeture de la connexion");
terminer();
return;
}
line_num++;
}
|
Le reste du code est inchangé et ce trouve ici
Classe ServiceClientI
3.2.2. Du coté serveur :¶
C est ici bien plus simple car il suffit de conserver les références aux Thread créés dans une liste, un tableau ou un contenant et des les interrompre. Le code est presque inchangé. On a juste:
- Ajouté une liste (ligne 15)
- Changé la condition de la boucle (ligne 17)
- Stocké les clients (ligne 26)
Enfin via les lignes 30-31 on demande à tous les threads de service de s’arreter.
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 | import java.net.*;
import java.util.ArrayList;
import java.util.List;
import java.io.*;
public class RepartiteurI extends ServerSocket {
private final static int port = 12000; /* Port d'écoute */
public RepartiteurI() throws IOException {
super(port);
System.out.println("[Serveur] : Serveur Jouet lancé sur " + (port));
}
public void execute() throws IOException {
Socket maConnection;
List<Thread> myThreads = new ArrayList<Thread>();
int numClient = 0;
while (numClient < 5){
System.out.println("[Serveur]: waiting for connexion number "+ numClient);
maConnection = accept();
numClient+=1;
String c_ip = maConnection.getInetAddress().toString();
int c_port = maConnection.getPort();
System.out.format("[Serveur] : Arr. Client IP %s sur %d\n", c_ip, c_port);
System.out.format("[Serveur ]: Creation du thread T_%d\n", c_port);
Thread cThread= new Thread(new ServiceClientI(maConnection, "T_" + c_port));
myThreads.add(cThread);
cThread.start();
}
System.out.format("Closing all Connections\n");
for (Thread item: myThreads)
item.interrupt();
close();
}
public static void main(String[] args) throws IOException {
RepartiteurI connectionManager = new RepartiteurI();
connectionManager.execute();
}
}
|
3.2.3. Exemple d’utilisation d’un champ volatile:¶
On peut par exemple tester une petite classe \(fragileCounter\) qui définit un compteur qui s’arrete quand la methode \(setterminer\) est appelée. Ici la méthode \(main()\) crée un compteur et lui demande de s’arreter après quelque secondes. Dans ce cas jouet l’intérêt semble minime, mais dans une application réelle on trouve potentiellement de nombreux threads travaillant sur des copies du compteur qui ne sont pas nécessairement synchronisée. La déclaration volatil assure que toutes les occurences du champs associé, le booléen \(terminer\), sont immédiatement mises à jours dans tout les threads.
public class fragileCounter implements Runnable {
private volatile Boolean terminer=false;
public void seterminer() { terminer =true;}
public fragileCounter(){};
public void run() {
for (int i =0; !terminer; i++ )
System.out.println(i);
System.out.println("Ordre de terminer recu, terminer=" + terminer);
}
public static void main(String[] args) {
fragileCounter myCounter= new fragileCounter();
Thread mT= new Thread(myCounter);
mT.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
myCounter.seterminer();
}
}
3.3. le schéma producteur consommateur¶
3.3.1. La classe Producteur¶
La classe producteur est un objet exécutable que l’on construit à partir de
- la référence à la file des jobs (qui ici est \(nosJobs\))
- d’un numéro identifiant le producteur (utile pour afficher ce qui se passe)
- de la taille max de la file des jobs (une autre solution serait de l’intégrer à l’objet représentant cette file).
Ce qui donne:
1 2 3 4 5 6 7 8 9 10 11 12 13 | import java.util.List;
class Producteur implements Runnable
{
private final List<String> nosJobs;
private final int maxJobs=4;
private final String iD;
public Producteur(List<String> nosJobs, String iD)
{
this.nosJobs = nosJobs;
this.iD=iD;
}
}
|
Le corps de le l’objet est la méthode \(run()\) qui produit à l’infini des jobs, entre chaque production on ajoute un délai aléatoire. La paire try/catch est due aux lourdeurs de Java.
public void run()
{
int i = 0;
while (true) {
try
{
i++;
prod(i);
Thread.sleep((long) (2000* Math.random()));
}
catch (InterruptedException ex)
{
ex.printStackTrace();
}
}
}
C’est la méthode \(prod()\) qui assure que la file \(nosJobs\) est bien gérée. La conception est la suivante :
- Le bloc est protégé car on commence par s’emparer du drapeau associé à la file d’attente nosJobs (ligne 3). Ceci assure qu’un seul thread modifie la file des jobs en attente.
- Ensuite tant que la file est pleine on attend (ligne 5 et 8), ce qui permet de relacher le drapeau \(nosJobs\) et donc autorise d’autre threads à modifier la file d’attente pendant que la production est mise en attente.
- Quand enfin on peut effectuer une production (on dispose du verrou et la file n’est pas pleine) on effectue une production, on ajoute un job à la file \(nosJobs\), on affiche quelques information sur la sortie standard, dont l’état global de la file.
- Une fois la production terminée on reveille les Thread en attente sur le drapeau \(nosJobs\) en utilisant un \(notify\) (ligne 20)
Remarque:
- Une erreur courante consiste à utiliser un if au lieu d’un while, or il est tout à fait possible d’être reveillé et que la file soit à nouveau pleine. Par exemple ici un autre producteur peut nous reveiller (ie file pleine -> on s’endort, un consommateur consomme et reveille un autre producteur qui à son tour nous reveille après avoir produit).
- En général il est préférable de toujours retester une condition après une attente (ie un wait).
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | private void prod(int i) throws InterruptedException
{
synchronized (nosJobs)
{
while (nosJobs.size() == maxJobs)
{
System.out.format("File pleine %s is waiting\n",iD);
nosJobs.wait();
}
String njob= String.format("T_%d_of_%s",i,iD);
nosJobs.add(njob);
System.out.format("%s Produit %s\n", iD, njob);
System.out.print("File :");
for (String it: nosJobs)
System.out.format(" %s ",it);
System.out.println("");
nosJobs.notify();
}
}
|
3.3.2. La classe Consommateur¶
La classe consommateur est presque identique à la classe Producteur, on pourrait d’ailleurs utiliser une seule classe avec utilisant un paramètre de construction (1 pour le producteur, -1 pour un consommateur). Sa methode run appelle la méthode \(conso\) qui ressemble trait pour trait à la méthode \(prod\).
Hormis les messages d’affichage (qui ont juste un role pédagogique) seules les lignes 4 et 11 diffèrent vraiment :
- le test controle si la file est vide.
- l’action consister à supprimer un job, ici le plus ancien (politique fifo).
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | private void cons(int i) throws InterruptedException{
synchronized (nosJobs)
{
while (nosJobs.size() == 0)
{
System.out.format("File vide, %s is waiting\n", iD);
nosJobs.wait();
System.out.format("%s is awake now \n", iD);
}
String jobDone = nosJobs.remove(0);
System.out.format("%s prend le Job %s\n" ,iD, jobDone );
System.out.print("File :");
for (String it: nosJobs)
System.out.format(" %s ",it);
System.out.println("");
nosJobs.notify();
}
}
|
3.3.3. Le code global¶
Pour simuler le système il suffit :
- de créer la file des job qui est partagée par les Threads, ici nommée \(theJobs\).
- de faire une boucle qui crée un certain nombre (ici 5) de producteurs et consommateurs et les demarre.
Remarque :
- Ici la taille max de la file est définie dans la classe consommateur, ce qui n’est pas très logique, il serait plus logique que ce soit un attribut de la file qui proposerait une methode \(isfull()\).
- Les paramètres (nombre de participants, taille de la file) sont tous codés en dur, vous pouvez vous amuser à les passer en paramètre.
import java.util.ArrayList;
import java.util.List;
public class PC {
public static void main(String[] args) {
List<String> theJobs = new ArrayList<String>();
Thread theTask ;
for (int i=0; i<5; i++){
Consommateur consume = new Consommateur(theJobs, String.format("C_%d",i));
theTask=new Thread(consume);
theTask.start();
Producteur produce = new Producteur(theJobs, String.format("P_%d",i));
theTask =new Thread(produce);
theTask.start();
}
}
}
3.4. Le Tri Bitonique ou Tri par fusion¶
Les 2 classes se trouvent ici, la classe principale Classe MySort et la classe
associée au thread de tri Classe Sorter.java. La méthode main de la classe principale effectue diverses boucles afin de mesurer le temps,
Les méthodes pour trier ou générer les tableaux vont être définies dans la classe \(MySort\), qui est la classe principale.
Le principe est de couper le tableau en 2 parties de même taille (droite et gauche) et de les trier. Si au moins :math”2 CPU sont disponibles on effectue ce tri en parralèle. On crée alors un thread gauche et un thread droit et chacun recoit la moitié des Cpus disponibles.
Ceci donne la méthode \(msort\) dont les paramètres sont une référence au tableau à trier et le nombre de CPUs disponibles.
3.4.1. La méthode msort()¶
On peut comprendre le code comme suit:
- Si le tableau est taille \(1\) il n’y a rien à faire (ligne 2)
- Sinon on coupe le tableau en 2 parties \(left\) et \(right\) (lignes 4-5)
- Si il n’y a qu’un CPU (ligne 8) on effectue 2 appels recursifs (lignes 10,11), i.e on trie l’un après l autre les tableaux left et right.
- Si il y a au moins \(2\) CPUs
- on crée un Thread chargé de trier \(left\) et un Thread chargé de trier \(right\) (lignes 16-17)
- on les demarre (lignes 19-20)
- on attend qu’ils aient terminé leur taches (lignes 22-23).
- Enfin on fusionne \(left\) et \(right\) (ligne 27)
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 | public static void mSort(int[] a, int AvailCPU, String branch) throws InterruptedException {
if (a.length >= 2) {
// Decoupe le tableau en 2 left and right
int[] left = Arrays.copyOfRange(a, 0, a.length / 2);
int[] right = Arrays.copyOfRange(a, a.length / 2, a.length);
// Only one cpu we perform a recursive but sequential sort
if (AvailCPU <= 1)
{
mSort(left, AvailCPU, branch+"0");
mSort(right, AvailCPU, branch+"1");
}
// If 2++ Cpu we create a thread to sort left and right
else
{
Thread lThread = new Thread(new Sorter(left, AvailCPU / 2, branch+"0" ));
Thread rThread = new Thread(new Sorter(right, AvailCPU / 2, branch+"1" ));
lThread.start();
rThread.start();
lThread.join();
rThread.join();
}
// fusionne les tableaux left et right
// La fusion est sequentielle
fusion(left, right, a);
}
}
|
3.4.2. La définition du Thread de tri¶
Ce Thread se contente d’appeler la méthode \(mSort\), ici il effectue aussi quelques affichages uniquement informatifs.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | public class Sorter implements Runnable {
private int[] a;
private int AvailCPU;
private String branch;
public Sorter(int[] a, int AvailCPU, String branch) {
this.a = a;
this.AvailCPU = AvailCPU;
this.branch=branch;
}
public void run() {
try{
//System.out.format("Thread %s sorting an array with size %d\n", branch , a.length);
MySort.mSort(a, AvailCPU, branch);
//System.out.format("Thread %s done\n", branch);
}
catch(Exception e) {;}
}
}
|
Pour trier un tableau \(a\), il suffit alors d’appeler la méthode mSort
MySort.mSort(a, cpuNum, "");
3.4.3. La méthode de fusion de 2 tableau triés¶
Pour obtenir un tableau trié quand \(left\) et \(right\) sont déjà triés il suffit de parcourir les deux tableaux conjoitement et de récupérer le min des deux valeurs courantes. En pratique on va donc déplacer deux indices \(rightI,leftI\) respectivement dans les tableaux \(left,right\).
On peut par exemple effectuer la fusion comme suit.
public static void fusion(int[] left, int[] right, int[] a) {
int leftI = 0;
int rightI = 0;
for (int i = 0; i < a.length; i++) {
if (rightI >= right.length || (leftI < left.length && left[leftI] < right[rightI])) {
a[i] = left[leftI];
leftI++; }
else {
a[i] = right[rightI];
rightI++;
}
}
}
3.4.4. Effectuer des Tests avec 1,2,4 cpus¶
Pour tester la procédure il suffit de créer un tableau aléatoire et d’estimer le temps du tri. On peut créer un tableau aléatoire via la méthode suivante (placée dans la classe principale). Notons ici que nous pourrions aussi bien tester des tableaux ne contenant que des zéros de des uns car trier un tel tableau en utilisant uniquement des comparaisons équivaut à trier un ensemble quelconque de nombres.
On peut par exemple tester en utilisant la méthode main() rudimentaire présentée ci dessous; celle ci trie maxTrial fois un tableau de taille \(10^6\) et affiche le temps approximatif d’un tri, et ce pour 1,2,4,8 cpus. Je vous invite à l’améliorer en ajoutant des arguments ou alors en évaluant le temps du tri pour toutes les tailles de la forme \(2^i, i \in [1,32]\).