-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy path05_producer_consumer.c
136 lines (115 loc) · 2.6 KB
/
05_producer_consumer.c
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
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
pthread_t *producers;
pthread_t *consumers;
sem_t mutex,empty,full;
int *buf, buf_pos=-1, prod_count, con_count, buf_len;
int produce(pthread_t self){
int i = 0;
int p = 1 + rand()%40;
while(!pthread_equal(*(producers+i),self) && i < prod_count)
i++;
printf("\nProducer %d produced %d \n\n",i+1,p);
return p;
}
void consume(int p,pthread_t self){
int i = 0;
while(!pthread_equal(*(consumers+i),self) && i < con_count)
i++;
printf("Buffer:");
for(i=0;i<=buf_pos;++i)
printf("%d ",*(buf+i));
printf("\nConsumer %d consumed %d \nCurrent buffer len: %d\n\n",i+1,p,buf_pos);
}
void print_buf_full()
{
printf("Cannot Insert to Buffer, Buffer is full\n");
}
void print_buf_empty()
{
printf("Buffer is Empty, Cannot Consume.\n");
}
void* producer(void *args){
while(1){
int p = produce(pthread_self());
++buf_pos;
if(buf_pos>=buf_len)
{
print_buf_full();
sleep(3 + rand()%3);
--buf_pos;
continue;
}
sem_wait(&empty);
sem_wait(&mutex);
*(buf + buf_pos) = p;
sem_post(&mutex);
sem_post(&full);
sleep(1 + rand()%3);
}
return NULL;
}
void* consumer(void *args){
int c;
while(1){
if(buf_pos<=-1)
{
print_buf_empty();
sleep(3 + rand()%4);
continue;
}
sem_wait(&full);
sem_wait(&mutex);
c = *(buf+buf_pos);
consume(c,pthread_self());
--buf_pos;
sem_post(&mutex);
sem_post(&empty);
sleep(1+rand()%5);
}
return NULL;
}
int main(void){
int i,err;
srand(time(NULL));
sem_init(&mutex,0,1);
sem_init(&full,0,0);
printf("Enter the number of Producers: ");
scanf("%d",&prod_count);
producers = (pthread_t*) malloc(prod_count*sizeof(pthread_t));
printf("Enter the number of Consumers: ");
scanf("%d",&con_count);
consumers = (pthread_t*) malloc(con_count*sizeof(pthread_t));
printf("Enter buffer capacity: ");
scanf("%d",&buf_len);
buf = (int*) malloc(buf_len*sizeof(int));
sem_init(&empty,0,buf_len);
for(i=0;i<prod_count;i++){
err = pthread_create(producers+i,NULL,&producer,NULL);
if(err != 0){
printf("Error creating producer %d: %s\n",i+1,strerror(err));
}else{
printf("Successfully created producer %d\n",i+1);
}
}
for(i=0;i<con_count;i++){
err = pthread_create(consumers+i,NULL,&consumer,NULL);
if(err != 0){
printf("Error creating consumer %d: %s\n",i+1,strerror(err));
}else{
printf("Successfully created consumer %d\n",i+1);
}
}
for(i=0;i<prod_count;i++){
pthread_join(*(producers+i),NULL);
}
for(i=0;i<con_count;i++){
pthread_join(*(consumers+i),NULL);
}
return 0;
}