1
/* Example code for Think OS.
2

3
Copyright 2015 Allen Downey
4
License: Creative Commons Attribution-ShareAlike 3.0
5

6
*/
7

8
#include <stdio.h>
9
#include <stdlib.h>
10
#include <assert.h>
11
#include <pthread.h>
12
#include <semaphore.h>
13
#include "utils.h"
14
#include "sem.h"
15

16
#define NUM_CHILDREN 2
17
#define NUM_ITEMS 127
18

19
// QUEUE
20

21
typedef struct {
22
  int *array;
23
  int length;
24
  int next_in;
25
  int next_out;
26
  Semaphore *mutex;
27
  Semaphore *items;
28
  Semaphore *spaces;
29
} Queue;
30

31
Queue *make_queue(int length)
32
{
33
  Queue *queue = (Queue *) malloc(sizeof(Queue));
34
  queue->length = length;
35
  queue->array = (int *) malloc(length * sizeof(int));
36
  queue->next_in = 0;
37
  queue->next_out = 0;
38
  queue->mutex = make_semaphore(1);
39
  queue->items = make_semaphore(0);
40
  queue->spaces = make_semaphore(length-1);
41
  return queue;
42
}
43

44
int queue_incr(Queue *queue, int i)
45
{
46
  return (i+1) % queue->length;
47
}
48

49
int queue_empty(Queue *queue)
50
{
51
  // queue is empty if next_in and next_out are the same
52
  int res = (queue->next_in == queue->next_out);
53
  return res;
54
}
55

56
int queue_full(Queue *queue)
57
{
58
  // queue is full if incrementing next_in lands on next_out
59
  int res = (queue_incr(queue, queue->next_in) == queue->next_out);
60
  return res;
61
}
62

63
void queue_push(Queue *queue, int item) {
64
  semaphore_wait(queue->spaces);
65
  semaphore_wait(queue->mutex);
66

67
  queue->array[queue->next_in] = item;
68
  queue->next_in = queue_incr(queue, queue->next_in);
69

70
  semaphore_signal(queue->mutex);
71
  semaphore_signal(queue->items);
72
}
73

74
int queue_pop(Queue *queue) {
75
  semaphore_wait(queue->items);
76
  semaphore_wait(queue->mutex);
77
  
78
  int item = queue->array[queue->next_out];
79
  queue->next_out = queue_incr(queue, queue->next_out);
80

81
  semaphore_signal(queue->mutex);
82
  semaphore_signal(queue->spaces);
83

84
  return item;
85
}
86

87
// SHARED
88

89
typedef struct {
90
  Queue *queue;
91
} Shared;
92

93
Shared *make_shared()
94
{
95
  Shared *shared = check_malloc(sizeof(Shared));
96
  shared->queue = make_queue(128);
97
  return shared;
98
}
99

100
// THREAD
101

102
pthread_t make_thread(void *(*entry)(void *), Shared *shared)
103
{
104
  int ret;
105
  pthread_t thread;
106

107
  ret = pthread_create(&thread, NULL, entry, (void *) shared);
108
  if (ret != 0) {
109
      perror_exit("pthread_create failed");
110
  }
111
  return thread;
112
}
113

114
void join_thread(pthread_t thread)
115
{
116
  int ret = pthread_join(thread, NULL);
117
  if (ret == -1) {
118
      perror_exit("pthread_join failed");
119
  }
120
}
121

122
// PRODUCER-CONSUMER
123

124
void *producer_entry(void *arg)
125
{
126
  int i;
127
  Shared *shared = (Shared *) arg;
128
  for (i=0; i<NUM_ITEMS; i++) {
129
    printf("adding item %d\n", i);
130
    queue_push(shared->queue, i);
131
  }
132
  pthread_exit(NULL);
133
}
134

135
void *consumer_entry(void *arg)
136
{
137
  int i;
138
  int item;
139
  Shared *shared = (Shared *) arg;
140

141
  for (i=0; i<NUM_ITEMS; i++) {
142
    item = queue_pop(shared->queue);
143
    printf("consuming item %d\n", item);
144
  }
145
  pthread_exit(NULL);
146
}
147

148
// TEST CODE
149

150
void queue_test()
151
{
152
  int i;
153
  int item;
154
  int length = 128;
155

156
  Queue *queue = make_queue(length);
157
  assert(queue_empty(queue));
158
  for (i=0; i<length-1; i++) {
159
    queue_push(queue, i);
160
  }
161
  assert(queue_full(queue));
162
  for (i=0; i<10; i++) {
163
    item = queue_pop(queue);
164
    assert(i == item);
165
  }
166
  assert(!queue_empty(queue));
167
  assert(!queue_full(queue));
168
  for (i=0; i<10; i++) {
169
    queue_push(queue, i);
170
  }
171
  assert(queue_full(queue));
172
  for (i=0; i<10; i++) {
173
    item = queue_pop(queue);
174
  }  
175
  assert(item == 19);
176
}
177

178
int main()
179
{
180
  int i;
181
  pthread_t child[NUM_CHILDREN];
182

183
  Shared *shared = make_shared();
184

185
  child[0] = make_thread(producer_entry, shared);
186
  child[1] = make_thread(consumer_entry, shared);
187

188
  for (i=0; i<NUM_CHILDREN; i++) {
189
    join_thread(child[i]);
190
  }
191

192
  return 0;
193
}
194

195