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
|
Condition variables are used to synchronize threads on the values of data,
rather than on the mere access to data (for which plain mutex-objects can be
used). Using condition variables, a thread simply sleeps until it is notified
by another thread. In a producer-consumer type of program this is usually
accomplished like this:
verb( consumer loop:
- wait until there's an item in store,
then reduce the number of stored items
- remove the item from the store
- increment the number of available storage locations
- do something with the retrieved item
producer loop:
- produce the next item
- wait until there's room to store the item,
then reduce the number of available storage locations
- store the item
- increment the number of stored items)
It is important that the two storage administrative tasks (registering the
number of available items and available storage locations) are either
performed by the client or by the producer. For the consumer `waiting' means:
itemization(
it() Get a lock on the variable containing the actual count
it() As long as the count is zero: wait, releasing the lock until another
thread has increased the count, then re-acquire the lock.
it() Reduce the count
it() Release the lock.
)
This scheme is implemented in a class ti(Semaphore), offering members
tt(wait) and tt(notify_all). For a more extensive discussion of semaphores see
em(i(Tanenbaum, A.S.)) (2016)
i(Structured Computer Organization), Pearson Prentice-Hall.
As a brief summary: semaphores restrict the number of threads that can access
a resource of limited size. It ensures that the number of threads that add
items to the resource (the producers) can never exceed the resource's maximum
size, or it ensures that the number of threads that retrieve items from the
resource (the consumers) can never exceed the resource's current size. Thus,
in a producer/consumer design two semaphores are used: one to control access
to the resource by the producers, and one to control access to the resource by
the consumers.
For example, say we have ten producing threads, as well as ten consumers, and
a lockable queue that must not grow bigger than 1000 items.
Producers try to push one item at a time; consumers try to pop one.
The data member containing the actual count is called tt(d_available). It is
protected by tt(mutex d_mutex). In addition a tt(condition_variable
d_condition) is defined:
verbinclude(-s4 //data examples/events.cc)
The waiting process is implemented through its member function tt(wait):
verbinclude(-nNs4 //wait examples/events.cc)
In line 5 tt(d_condition.wait) releases the lock. It waits until receiving
a notification, and re-acquires the lock just before returning. Consequently,
tt(wait's) code always has complete and unique control over tt(d_available).
What about notifying a waiting thread? This is handled in lines 4 and
5 of the member function tt(notify_all):
verbinclude(-nNs4 //notify_all examples/events.cc)
At line 4 tt(d_available) is always incremented; by using a postfix
increment it can simultaneously be tested for being zero. If it was initially
zero then tt(d_available) is now one. A thread waiting until tt(d_available)
exceeds zero may now continue. A waiting thread is notified by calling
tt(d_condition.notify_one). In situations where multiple threads are waiting
`ti(notify_all)' can also be used.
Using the facilities of the class tt(Semaphore) whose constructor expects
an initial value of its tt(d_available) data member, the classic
consumer-producer paradigm can now be implemented using
multi-threading+footnote(A more elaborate example of the producer-consumer
program is found in the tt(yo/threading/examples/events.cc) file in the
bf(C++) Annotations's source archive):
verb( Semaphore available(10);
Semaphore filled(0);
std::queue<size_t> itemQueue;
std::mutex qMutex;
void consumer()
{
while (true)
{
filled.wait();
// mutex lock the queue:
{
std::lock_guard lg(qMutex);
size_t item = itemQueue.front();
itemQueue.pop();
}
available.notify_all();
process(item); // not implemented here
}
}
void producer()
{
size_t item = 0;
while (true)
{
++item;
available.wait();
// mutex lock the queue with multiple consumers
{
std::lock_guard lg(qMutex);
itemQueue.push(item);
}
filled.notify_all();
}
}
int main()
{
thread consume(consumer);
thread produce(producer);
consume.join();
produce.join();
})
Note that a tt(mutex) is used to avoid simultaneous access to the queue by
multiple threads. Consider the situation where the queue contains 5 items: in
that situation the semaphores allow the consumer and the producer to access
the queue, but to avoid currupting the queue only one of them may modify the
queue at a time. This is realized by both threads obtaining the tt(std::mutex
qMutex) lock before modifying the queue.
|
