Multithread producteurs consommateurs

**barney2008** · 25/04/2008, 18h15

Bonjour,

vous devez tous connaitre le problème du multithread consommateur/producteur.

Je suis actuellement en train de le coder et j'ai un problème lors de l'execution du programme.

Lorsque ma file d'attente est pleine, le programme se bloque et m'affiche Bus Error (core dumped).

J'ai beau cherché je n'arrive pas a trouver mon erreur.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
 
/*    Bus Error (core dumped)  */
 
 
 
/***************************************************************************************************************************/
 
/*gcc -Wall -lpthread -l posix4 -o  proconsmulti proconsmulti.c*/
 
/***************************************************************************************************************************/
 
/* définition des bibliothèques */
#define _REENTRANT
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <signal.h>
#include <string.h>
#include <sys/types.h>
#include <time.h>
 
#define TAILLE_FILE_ATTENTE 5 /* Taille de la file d'attente */
 
/* Déclaration Etat production / consommation */
int etatprod=1; /*1 si en production, 0 sinon */
int etatcons=1; /*1 si en consommation, 0 sinon */
int arretprod=0; /* 1 si on souhaite arreter la production, 0 sinon */
int arretcons=0; /* 1 si on souhaite arreter la consommation, 0 sinon */
 
/* Déclaration du terminal */
FILE * terminal;
 
 
/* Variables globales */
int nbprod;
int nbcons;
 
/* déclaration des variables conditionnelles */
pthread_cond_t cond_ecriture = PTHREAD_COND_INITIALIZER;;
pthread_cond_t cond_lecture = PTHREAD_COND_INITIALIZER;;
 
/* Déclaration File d'attente */
int nbplalibres=TAILLE_FILE_ATTENTE;
char * file_attente[TAILLE_FILE_ATTENTE];
int positionlecture=0;
int positionecriture=0;
 
/* Tableau des identifiants des threads */
pthread_t* t_conso;
pthread_t* t_prod;
pthread_t t_supervision;
 
/* Déclaration des mutex lecture, ecriture , producteurs et consommateurs */
pthread_mutex_t mutex_ecriture = PTHREAD_MUTEX_INITIALIZER;/* mutex permettant d'ecrire dans la file */
pthread_mutex_t mutex_lecture = PTHREAD_MUTEX_INITIALIZER;/* mutex permettant de lire la file */
pthread_mutex_t mutex_case_libre = PTHREAD_MUTEX_INITIALIZER;
 
 
/* Fonction gérant les erreurs */
void ctrlArg (int argc, char **argv, int nbArg) 
/* controle du nombre d'argument d'un programme */
{
	int i;
	if (argc!=nbArg+1) /* si le nombre d'argument est incorrect */
	{
		fprintf(stderr,"Nombre d'arguments incorrect\nUsage : %s",argv[0]);
		for(i=1;i<=nbArg;i++)
		fprintf(stderr," argument%d",i);
		fprintf(stderr,"\n");
		exit(1);
	}
}
 
void exitOnErrSyst (char *nomFonction, char *texte)
/* Gestion d'une erreur système */
{
	fprintf(stderr,"ERREUR fonction %s - %s",nomFonction,texte);
	perror("");
	exit(1);
}
 
 
/***************************************************************************************************************************/
 
/* Thread du producteur */
void *producteur(void *arg)
{
       int r=0;
 
       char tampon[5][10];
       strcpy(tampon[0],"Banane");
       strcpy(tampon[1],"Ananas");
       strcpy(tampon[2],"Pomme");
       strcpy(tampon[3],"Cidre");
       strcpy(tampon[4],"raisin");
 
       time_t t;
       char * date;
       int idpro=(int)arg;
 
       /* On attend les ordres du superviseur */
       while(1)
       {
               /* Tant qu'il n'y a pas de production ou si on souhaite arreter la production */
               while( (etatprod==0) || (nbplalibres==0) || (arretprod ==1) )
               {
 
		       if(arretprod==1) /* si on souhaite arreter la production */
                       {
                               pthread_exit(NULL);
                       }
                       if(etatprod==0) /* si il n'y a pas de production */
                       {
                               fprintf(stdout,"\n\nSupervision : producteur %d en sommeil\n",idpro);
                       	       if( (pthread_cond_wait(&cond_ecriture,&mutex_ecriture)) != 0)
                       	       {
                               		fprintf(stdout,"\n\nAttente producteur");
                               }
                       } 		       
               }
               sleep(1); /* Attende de production : 1sc entre 2 productions */
	       file_attente[positionecriture]=(char*)malloc(sizeof(char)*(strlen(tampon[r])+1));
 
 
               if( (pthread_mutex_lock(&mutex_case_libre)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_lock","blocage case libre");		   	    
               }
	       nbplalibres--;
	       if( (pthread_mutex_unlock(&mutex_case_libre)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_unlock","libération case libre");		   	    
               }
 
               if( (pthread_mutex_lock(&mutex_ecriture)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_lock","blocage Mutex ecriture");
               }
 
               /* si il y a de la place, on ecrit un message dans la file et on récupère le temps*/
               r=(r+1)%5;
               file_attente[positionecriture]=tampon[r];
               t=time(NULL);
               date=ctime(&t);
               /* Ecriture dans un fichier du message producteur */
               fprintf(stdout,"\n\n%s Producteur numéro %d écriture case %d produit %s\n",date,idpro,positionecriture,file_attente[positionecriture]);
               positionecriture = (positionecriture+1)%TAILLE_FILE_ATTENTE;
               if( (pthread_mutex_unlock(&mutex_ecriture)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_unlock","libération Ecriture file");
               }
               /* On signale qu'il y a quelque chose à lire dans cette case car la case a été écrite */
               if( (pthread_cond_signal(&cond_lecture)) != 0)
               {
                       fprintf(stdout,"\n\nErreur Variable conditionnelle lecture");
                       exit(1);
               }
       }
}
 
 
/***************************************************************************************************************************/
 
 
/* Thread du consommateur */
void *consommateur(void *arg)
{
       time_t t;
       char * date;
 
       int idcons=(int)arg;
       /* On attend les ordres du superviseur */
       while(1)
       {
               /* Tant qu'il n'y a pas de consommation ou si on souhaite arreter la consommation */
               while( (etatcons==0) || (arretcons ==1) || (nbplalibres==TAILLE_FILE_ATTENTE) )
               {
                       if(arretcons==1) /* si on souhaite arreter la consommation */
                       {
                               pthread_exit(NULL);
                       }
                       if(etatcons==0) /* si il n'y a pas de consommation */
                       {
                               fprintf(stdout,"\n\nSupervision : consommateur %d en sommeil\n",idcons);
                       	       if( (pthread_cond_wait(&cond_lecture,&mutex_lecture)) != 0)
                       	       {
                               		fprintf(stdout,"\n\nAttente consommateur");
                               }
                       }		       
               }
               sleep(2); /* Attende de consommation : 2sc entre 2 consommations */
 
 
               if( (pthread_mutex_lock(&mutex_case_libre)) != 0)
 
               {
                       exitOnErrSyst("pthread_mutex_lock","blocage case libre");		   	    
               }
	       nbplalibres++;
	       if( (pthread_mutex_unlock(&mutex_case_libre)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_unlock","libération case libre");		   	    
               }
 
               if( ( pthread_mutex_lock(&mutex_lecture)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_lock","blocage Mutex ecriture");
               }
 
               t=time(NULL);
               date=ctime(&t);
               /* Ecriture dans un fichier du message consommateur */
               fprintf(stdout,"\n\n%s Consommateur numéro %d lecture case %d j'aime %s\n",date,idcons,positionlecture,file_attente[positionlecture]);
               free(file_attente[positionlecture]);
	       positionlecture=(positionlecture + 1)%TAILLE_FILE_ATTENTE;
               if( (pthread_mutex_unlock(&mutex_lecture)) != 0)
               {
                       exitOnErrSyst("pthread_mutex_unlock","libération lecture file");
               }
               /* On signale qu'il y a quelque chose à lire dans cette case car la case a été lue */
               if( (pthread_cond_signal(&cond_ecriture)) != 0)
               {
                       fprintf(stdout,"Erreur Variable conditionnelle ecriture");
                       exit(1);
               }
       }
}
 
 
/***************************************************************************************************************************/
 
 
/* Thread Superviseur */
 
void* superviseur(void *arg)
{
       int caract,retour_chariot;
       int i,j;
 
       while(1)
       {
               retour_chariot=getchar();
	       fprintf(terminal,"\n\t*******MENU SUPERVISEUR*******\n");
      	       fprintf(terminal,"\n\t1 : Suspension Consommation\n");
       	       fprintf(terminal,"\n\t2 : Reprise Consommation\n");
       	       fprintf(terminal,"\n\t3 : Suspension Production\n");
               fprintf(terminal,"\n\t4 : Reprise Production\n");
      	       fprintf(terminal,"\n\t5 : Suspension Consommation et Production\n");
               fprintf(terminal,"\n\t6 : Arret de l'application\n");
               fprintf(terminal,"\n\t*******Tapez votre choix*******\n");
	       caract=getchar();
	       switch(caract)
               {
                       case '1' : /*Suspension Consommation */
                       { 
			       etatcons=0;
			       fprintf(terminal,"\n\t*******Consommation suspendue*******\n");
                       break;
		       }
                       case '2' : /*Reprise Consommation */
                       {
		               etatcons=1;
                               if( (pthread_cond_broadcast(&cond_lecture)) != 0)
                               {
                                      	fprintf(stdout,"Erreur reprise consommateurs");
 
                               }
                               fprintf(terminal,"\n\t*******Reprise Consommation*******\n");
                       break;
		       }
                       case '3' : /*Suspension Production */
                       {
		               etatprod=0;
			       fprintf(terminal,"\n\t*******Production suspendue*******\n");
                       break;
		       }
                       case '4' : /*Reprise Production */
                       {
		               etatprod=1;
                               if( (pthread_cond_broadcast(&cond_ecriture)) != 0)
                               {
                                      	fprintf(stdout,"Erreur reprise producteur");
 
                               }
                               fprintf(terminal,"\n\t*******Reprise production*******\n");
                       break;
		       }
                       case '5' : /*Suspension Consommation & Production */
                       {
		               etatprod=0;
                               etatcons=0;
			       fprintf(terminal,"\n\t*******Production & Consommation suspendues*******\n");
                       break;
		       }
                       case '6' : /*Arret de l'application */
                       {
		               /* premiere etape : envoi de reprise de tous les producteurs */
                               etatprod=1;
                               if( (pthread_cond_broadcast(&cond_ecriture)) != 0)
                               {
                                      	fprintf(stdout,"\n\nErreur reprise producteur");
 
                               }
                               /* envoi de reprise de tous les consommateurs */
                               etatcons=1;
                               if( (pthread_cond_broadcast(&cond_lecture)) != 0)
                               {
                                      	fprintf(stdout,"\n\nErreur reprise consommateurs");
 
                               }
                               /* envoi du signal de fin des consommateurs et producteurs */
                               arretprod=1;
                               arretcons=1;
 
			       fprintf(terminal,"\n\t*******Production & Consommation suspendues*******\n");
 
                               /* Attente terminaison producteurs avant de quitter la supervision */
                               for(i=0;i<nbprod;i++)
                               {
                                       if( pthread_join(t_prod[i],NULL) != 0)
                                       {
                                       		exitOnErrSyst("pthread_join","Attente terminaison thread producteurs");
                                       }
                                       fprintf(terminal,"\n\nSupervision : producteur %d termine\n",i+1);
                               }
                               /* Attente terminaison consommateur avant de quitter la supervision */
                               for(j=0;j<nbcons;j++)
                               {
                                       if( pthread_join(t_conso[j],NULL) != 0)
                                       {
                                              exitOnErrSyst("pthread_join","Attente terminaison thread consommateurs");
                                       }
                                       fprintf(terminal,"\n\nSupervision : consommateur %d termine\n",j+1);
                               }
                               /* libération des threads qui attentent quelque chose a lire */
                               if((pthread_cond_broadcast(&cond_lecture)) != 0)
                               {
                                       fprintf(stdout,"Erreur Attente lecture possible");
                                       exit(1);
                               }
                               pthread_exit(NULL);
                       break;
		       }
                       default : 
		       {
		       		fprintf(terminal,"\n\t*******ERREUR DE SAISIE*******\n"); 
		       break;
		       }
               }
       }
}
 
 
 
/***************************************************************************************************************************/
 
 
/* Main */
 
int main(int argc, char**argv)
{
 
       int fd,i,j,nb[2];
       ctrlArg (argc,argv,2);
       nbprod=atoi(argv[1]);
       nbcons=atoi(argv[2]);
       /* nb permet de savoir le nombre de consommateurs et de producteurs pour le thread de supervision */
       nb[1]=nbprod;
       nb[2]=nbcons;
       if( (fd=(open("/dev/tty",O_WRONLY|O_APPEND))) == -1)
       {
               exitOnErrSyst("open","ouverture terminal");
       }
       if( (terminal=fdopen(fd,"ab+")) == NULL)
       {
               exitOnErrSyst("fdopen","ouverture du terminal en écriture");
       }
       fprintf(terminal,"\n\nVous avez opté pour %d producteurs et %d consommateurs",nb[1],nb[2]);
       /* création des identifiants des threads */
       t_conso = (pthread_t*)malloc(nbcons*sizeof(pthread_t));
       t_prod = (pthread_t*)malloc(nbprod*sizeof(pthread_t));
 
       /* Creation thread consommateur */
       for(i=0;i<nbcons;i++)
       {
               if( pthread_create(&t_conso[i],NULL,consommateur,(void*)i+1) != 0)
               {
                       exitOnErrSyst("pthread_create","thread consommateur");
               }
       }
       /* Creation thread producteur */
       for(j=0;j<nbprod;j++)
       {
               if( pthread_create(&t_prod[j],NULL,producteur,(void*)j+1) != 0)
               {
                        exitOnErrSyst("pthread_create","thread producteur");
               }
       }
       /* Création thread supervision */
       if( pthread_create(&t_supervision,NULL,superviseur,NULL) != 0)
               {
                        exitOnErrSyst("pthread_create","thread supervision");
               }
       /* Terminaison thread supervision */
       if( pthread_join(t_supervision,NULL) != 0)
       {
               exitOnErrSyst("pthread_join","Attente terminaison thread de supervision");
       }
       fprintf(terminal,"\n\t*******ARRET DE L'APPLICATION*******\n");
       return 0;
}

En attente de votre réponse, bonne soirée.

**Celelibi** · 27/04/2008, 00h15

Salut,

Je vais avoir quelques trucs à dire à propos de ton code, je vais essayer de le faire dans l'ordre.
Dans la commande de compilation : c'est quoi cette librairie posix4 ? J'ai pas de telle lib chez moi, et ça compile très bien sans.
Le #define _RENTRANT il sert à quoi ?

D'un point de vue général, le problème des producteurs/consommateurs se résout très simplement avec juste deux sémaphores et un mutex.

Dans la fonction main, pourquoi ouvrir un terminal et pas afficher simplement afficher les messages sur la sortie standard/sortie d'erreur ?

Dans le superviseur, d'un point de vue purement syntaxique, les accolades dans les case du switch sont inutiles. Je dirais même plus, ça nuit à la lisibilité du code.

Dans tes producteurs et consommateurs, tu as des races conditions.

Tu sais, le truc qui fait que le programme bug quand t'as pas eut de chance avec l'ordonnanceur.
Si tu code une solution à ce problème avec des variables condition, je suppose que tu as vu en cours ce que sont les moniteurs et comment ils marchent.
Donc tu devrais savoir qu'il n'y a qu'un seul thread à la fois qui se trouve dans le moniteur. Donc il y a un mutex qui protège l'ensemble des méthodes du moniteur. Et les variables condition (qui n'ont que peu de sens hors d'un moniteur), lorsqu'on fait un wait dessus, doivent relâcher le mutex du moniteur histoire qu'un autre thread puisse entrer dans le moniteur et faire un signal sur la variable condition et réveiller un de ceux qui attendent sur le wait.
Et donc la race condition se trouve sur le fait que le mutex que tu donne à pthread_cond_wait n'est pas verrouillé au début, donc il peut arriver tout et n'importe quoi (en particulier un pthread_cond_signal) entre le moment où tu test et le moment où tu wait.

De plus tu fais de l'attente active quand le tampon est plein ou vide, ça c'est vraiment très mal ! L'attente active on en fait pas dans la programmation normale, ça bouffe du cpu pour rien.

Je peut-être te conseiller ceci à propos du problème producteur/consommateur. http://www.geinac.net/articles/artic...r_consommateur
Tu peux aussi lire le reste de la page pour t'informer sur les sémaphores et moniteurs afin de mieux les utiliser.

**barney2008** · 27/04/2008, 12h26

Salut,

je ne peux pas utiliser de sémaphores, car je demande par l'intermédiaire du superviseur de stopper la production ou consommation, ou de reprendre ou d'arreter le programme.

Le #define RENTRANT sert à définir les threads sous Sun. Je n'ai pas eu plus d'explications.

Ensuite, je dois ouvrir un terminal car je dois rediriger les annonces du superviseur sur le terminal, et le message des producteurs/ consommateurs dans un fichier (pas d'affichage à l'écran).

Les accolades, j'ai rien à dire, ca sert à rien ...

Je suppose que mon soucis vient donc dans la suite de vos propos. Le problème avec le cond_wait qui n'est pas verrouillé, et le gros problème de mon programme, la file d'attente pleine qui fait arreter mon programme.

Je vais essayer de corriger cela.

Comment fait-on pour passer en attente "passive" ?

Merci, a bientôt.

**Celelibi** · 27/04/2008, 16h31

Pour faire de l'attente passive, il faut que les "pauses" soient dues à un sémaphore "vidé", un mutex verrouillé ou un wait sur une variable condition.

Si tu tiens à garder les conditions, commence par écrire ton code sous forme de moniteur. cf le lien donné précédemment.

Mais sache qu'il y a une équivalence stricte entre sémaphores et moniteurs. Il n'y a pas plus de problèmes qu'on peut résoudre avec les moniteurs qu'avec les sémaphores. De plus il est possible de passer de l'un à l'autre par des méthodes formelles.
Donc l'argument du superviseur qui doit arrêter la production ou consommation est un faux argument.

Par exemple, tu peux gérer l'arrête avec une variable booléenne et un sémaphore pour l'arrêt de la production.
Pour cela, avant chaque production tu test si tu dois arrêter la production alors tu décrémente le sémaphore (qui était initialisé à 0), et pour relancer la production, le superviseur re-change le booléen indiquant l'arrêt, puis incrémente le sémaphore ce qui va libérer le producteur. Ça c'est pour le cas d'un unique producteur, mais si tu as compris le principe, tu pourra certainement généraliser à n consommateurs.
Regarde aussi du côté des barrières de synchronisations, notamment sur wikipedia, ça pourrait te servir.

Sinon, c'est vrai que j'ai pas vraiment cherché pour ton problème de Bus Error. De toutes façons un programme présentant des races conditions de ce type finira forcément par bugger.
Donc commence par gérer correctement la concurrence (au moins sur papier), après on regarde les bugs plus en détail.

PS : le vouvoiement me dérange un peu. Developpez est une grande famille non ?

Multithread producteurs consommateurs

Linux

Discussions similaires

Partager

Partager