使用POSIX線程解決“生產者/消費者”問題

使用POSIX線程解決“生產者/消費者”問題：

1. /*
2. * File : pc.cpp
3. *
4. * Title : Demo Producer/Consumer.
5. *
6. * Short : A solution to the producer consumer problem using pthreads.
7. * This is a simple FIFO pipe between two tasks. The primary problem is
8. * ensuring that the producer blocks if the FIFO is full, and the consumer
9. * blocks if it is empty, and avoiding data-races along the way. A secondary
10. * concern is that there is as little interference between the two tasks as
11. * possible.
12. *
13. * Author : Andrae Muys
14. *
15. * Date : 18 September 1997
16. */
18. #include <stdlib.h>
19. #include <pthread.h>
20. #include <assert.h>
22. #ifdef _WIN32
23. # include <windows.h>
24. # define SLEEP(ms) Sleep(ms)
25. #elif defined(LINUX)
26. # include <unistd.h>
27. # define SLEEP(ms) sleep(ms)
28. #endif
30. #define QUEUE_SIZE 10
31. #define LOOP 20
33. void* producer (void *args);
34. void* consumer (void *args);
36. typedef struct
37. {
38. int buf[QUEUE_SIZE];
39. long head, tail;
40. bool full, empty;
41. pthread_mutex_t *mutex;
42. pthread_cond_t *notFull, *notEmpty;
43. } Queue;
45. Queue* queueInit (void);
46. void queueDelete (Queue *q);
47. void queueAdd (Queue *q, int in);
48. void queueDel (Queue *q, int *out);
50. int main(int argc, char* argv[])
51. {
52. Queue* fifo = queueInit ();
53. assert(fifo != NULL);
55. pthread_t pro, con;
56. pthread_create (&pro, NULL, &producer, fifo);
57. pthread_create (&con, NULL, &consumer, fifo);
58. pthread_join (pro, NULL);
59. pthread_join (con, NULL);
60. queueDelete (fifo);
61. return 0;
62. }
64. void* producer (void *q)
65. {
66. Queue* fifo = (Queue *)q;
68. for (int i = 0; i < LOOP; i++)
69. {
70. // 臨界區操作：若隊列未滿，添加新數據
71. pthread_mutex_lock (fifo->mutex);
72. while (fifo->full)
73. {
74. printf ("producer: Queue FULL.\n");
75. pthread_cond_wait (fifo->notFull, fifo->mutex);
76. }
77. queueAdd (fifo, i);
78. pthread_mutex_unlock (fifo->mutex);
80. // 數據添加結束，發“隊列有數據”信號
81. pthread_cond_signal (fifo->notEmpty);
82. SLEEP (100);
83. }
85. // 與上面類似
86. for (int i = 0; i < LOOP; i++)
87. {
88. pthread_mutex_lock (fifo->mutex);
89. while (fifo->full)
90. {
91. printf ("producer: Queue FULL.\n");
92. pthread_cond_wait (fifo->notFull, fifo->mutex);
93. }
94. queueAdd (fifo, i);
95. pthread_mutex_unlock (fifo->mutex);
96. pthread_cond_signal (fifo->notEmpty);
97. SLEEP (200);
98. }
99. return (NULL);
100. }
102. void* consumer (void *q)
103. {
104. int d;
106. Queue *fifo = (Queue *)q;
108. for (int i = 0; i < LOOP; i++)
109. {
110. // 臨界區操作：若隊列不空，則取出數據
111. pthread_mutex_lock (fifo->mutex);
112. while (fifo->empty)
113. {
114. printf ("consumer: Queue EMPTY.\n");
115. pthread_cond_wait (fifo->notEmpty, fifo->mutex);
116. }
117. queueDel (fifo, &d);
118. pthread_mutex_unlock (fifo->mutex);
120. // 取完數據，發“隊列不滿”信號
121. pthread_cond_signal (fifo->notFull);
122. printf ("consumer: recieved %d.\n", d);
123. SLEEP(200);
124. }
126. // 與上面類似
127. for (int i = 0; i < LOOP; i++)
128. {
129. pthread_mutex_lock (fifo->mutex);
130. while (fifo->empty)
131. {
132. printf ("consumer: Queue EMPTY.\n");
133. pthread_cond_wait (fifo->notEmpty, fifo->mutex);
134. }
135. queueDel (fifo, &d);
136. pthread_mutex_unlock (fifo->mutex);
137. pthread_cond_signal (fifo->notFull);
138. printf ("consumer: recieved %d.\n", d);
139. SLEEP (50);
140. }
141. return (NULL);
142. }
144. Queue *queueInit (void)
145. {
146. Queue *q = (Queue *)malloc (sizeof (Queue));
147. if (q == NULL) return (NULL);
149. q->empty = true;
150. q->full = false;
151. q->head = 0;
152. q->tail = 0;
154. q->mutex = (pthread_mutex_t *) malloc (sizeof (pthread_mutex_t));
155. pthread_mutex_init (q->mutex, NULL);
156. q->notFull = (pthread_cond_t *) malloc (sizeof (pthread_cond_t));
157. pthread_cond_init (q->notFull, NULL);
158. q->notEmpty = (pthread_cond_t *) malloc (sizeof (pthread_cond_t));
159. pthread_cond_init (q->notEmpty, NULL);
161. return (q);
162. }
164. void queueDelete (Queue *q)
165. {
166. pthread_mutex_destroy (q->mutex);
167. free (q->mutex);
169. pthread_cond_destroy (q->notFull);
170. free (q->notFull);
172. pthread_cond_destroy (q->notEmpty);
173. free (q->notEmpty);
175. free (q);
176. }
178. void queueAdd (Queue *q, int in)
179. {
180. q->buf[q->tail] = in;
181. q->tail++;
182. if (q->tail == QUEUE_SIZE) // 循環隊列
183. q->tail = 0;
185. if (q->tail == q->head) // 添加數據時“觸頂”
186. q->full = true;
187. q->empty = false;
189. return;
190. }
192. void queueDel (Queue *q, int *out)
193. {
194. *out = q->buf[q->head];
195. q->head++;
196. if (q->head == QUEUE_SIZE) // 循環隊列
197. q->head = 0;
199. if (q->head == q->tail) // 取出數據時“觸底”
200. q->empty = true;
201. q->full = false;
203. return;
204. }