1/*
2 * Copyright (C) 2015-2016 Apple Inc. All rights reserved.
3 *
4 * Redistribution and use in source and binary forms, with or without
5 * modification, are permitted provided that the following conditions
6 * are met:
7 * 1. Redistributions of source code must retain the above copyright
8 * notice, this list of conditions and the following disclaimer.
9 * 2. Redistributions in binary form must reproduce the above copyright
10 * notice, this list of conditions and the following disclaimer in the
11 * documentation and/or other materials provided with the distribution.
12 *
13 * THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS CONTRIBUTORS ``AS IS''
14 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
15 * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
16 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL APPLE INC. OR ITS CONTRIBUTORS
17 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
18 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
19 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
20 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
21 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
22 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
23 * THE POSSIBILITY OF SUCH DAMAGE.
24 */
25
26#include "config.h"
27#include <mutex>
28#include <thread>
29#include <wtf/Condition.h>
30#include <wtf/DataLog.h>
31#include <wtf/Deque.h>
32#include <wtf/Lock.h>
33#include <wtf/StringPrintStream.h>
34#include <wtf/Threading.h>
35#include <wtf/Vector.h>
36
37namespace TestWebKitAPI {
38
39namespace {
40
41const bool verbose = false;
42
43enum NotifyStyle {
44 AlwaysNotifyOne,
45 TacticallyNotifyAll
46};
47
48template<typename Functor>
49void wait(Condition& condition, std::unique_lock<Lock>& locker, const Functor& predicate, Seconds timeout)
50{
51 if (timeout == Seconds::infinity())
52 condition.wait(locker, predicate);
53 else {
54 // This tests timeouts in the sense that it verifies that we can call wait() again after a
55 // timeout happened. That's a non-trivial piece of functionality since upon timeout the
56 // ParkingLot has to remove us from the queue.
57 while (!predicate())
58 condition.waitFor(locker, timeout, predicate);
59 }
60}
61
62void notify(NotifyStyle notifyStyle, Condition& condition, bool shouldNotify)
63{
64 switch (notifyStyle) {
65 case AlwaysNotifyOne:
66 condition.notifyOne();
67 break;
68 case TacticallyNotifyAll:
69 if (shouldNotify)
70 condition.notifyAll();
71 break;
72 }
73}
74
75void runTest(
76 unsigned numProducers,
77 unsigned numConsumers,
78 unsigned maxQueueSize,
79 unsigned numMessagesPerProducer,
80 NotifyStyle notifyStyle,
81 Seconds timeout = Seconds::infinity(),
82 Seconds delay = 0_s)
83{
84 Deque<unsigned> queue;
85 bool shouldContinue = true;
86 Lock lock;
87 Condition emptyCondition;
88 Condition fullCondition;
89
90 Vector<Ref<Thread>> consumerThreads;
91 Vector<Ref<Thread>> producerThreads;
92
93 Vector<unsigned> received;
94 Lock receivedLock;
95
96 for (unsigned i = numConsumers; i--;) {
97 consumerThreads.append(Thread::create(
98 "Consumer thread",
99 [&] () {
100 for (;;) {
101 unsigned result;
102 unsigned shouldNotify = false;
103 {
104 std::unique_lock<Lock> locker(lock);
105 wait(
106 emptyCondition, locker,
107 [&] () {
108 if (verbose)
109 dataLog(toString(Thread::current(), ": Checking consumption predicate with shouldContinue = ", shouldContinue, ", queue.size() == ", queue.size(), "\n"));
110 return !shouldContinue || !queue.isEmpty();
111 },
112 timeout);
113 if (!shouldContinue && queue.isEmpty())
114 return;
115 shouldNotify = queue.size() == maxQueueSize;
116 result = queue.takeFirst();
117 }
118 notify(notifyStyle, fullCondition, shouldNotify);
119
120 {
121 std::lock_guard<Lock> locker(receivedLock);
122 received.append(result);
123 }
124 }
125 }));
126 }
127
128 sleep(delay);
129
130 for (unsigned i = numProducers; i--;) {
131 producerThreads.append(Thread::create(
132 "Producer Thread",
133 [&] () {
134 for (unsigned i = 0; i < numMessagesPerProducer; ++i) {
135 bool shouldNotify = false;
136 {
137 std::unique_lock<Lock> locker(lock);
138 wait(
139 fullCondition, locker,
140 [&] () {
141 if (verbose)
142 dataLog(toString(Thread::current(), ": Checking production predicate with shouldContinue = ", shouldContinue, ", queue.size() == ", queue.size(), "\n"));
143 return queue.size() < maxQueueSize;
144 },
145 timeout);
146 shouldNotify = queue.isEmpty();
147 queue.append(i);
148 }
149 notify(notifyStyle, emptyCondition, shouldNotify);
150 }
151 }));
152 }
153
154 for (auto& thread : producerThreads)
155 thread->waitForCompletion();
156
157 {
158 std::lock_guard<Lock> locker(lock);
159 shouldContinue = false;
160 }
161 emptyCondition.notifyAll();
162
163 for (auto& thread : consumerThreads)
164 thread->waitForCompletion();
165
166 EXPECT_EQ(numProducers * numMessagesPerProducer, received.size());
167 std::sort(received.begin(), received.end());
168 for (unsigned messageIndex = 0; messageIndex < numMessagesPerProducer; ++messageIndex) {
169 for (unsigned producerIndex = 0; producerIndex < numProducers; ++producerIndex)
170 EXPECT_EQ(messageIndex, received[messageIndex * numProducers + producerIndex]);
171 }
172}
173
174} // anonymous namespace
175
176TEST(WTF_Condition, OneProducerOneConsumerOneSlot)
177{
178 runTest(1, 1, 1, 100000, TacticallyNotifyAll);
179}
180
181TEST(WTF_Condition, OneProducerOneConsumerOneSlotTimeout)
182{
183 runTest(
184 1, 1, 1, 100000, TacticallyNotifyAll,
185 Seconds::fromMilliseconds(10),
186 Seconds(1));
187}
188
189TEST(WTF_Condition, OneProducerOneConsumerHundredSlots)
190{
191 runTest(1, 1, 100, 1000000, TacticallyNotifyAll);
192}
193
194TEST(WTF_Condition, TenProducersOneConsumerOneSlot)
195{
196 runTest(10, 1, 1, 10000, TacticallyNotifyAll);
197}
198
199TEST(WTF_Condition, TenProducersOneConsumerHundredSlotsNotifyAll)
200{
201 runTest(10, 1, 100, 10000, TacticallyNotifyAll);
202}
203
204TEST(WTF_Condition, TenProducersOneConsumerHundredSlotsNotifyOne)
205{
206 runTest(10, 1, 100, 10000, AlwaysNotifyOne);
207}
208
209TEST(WTF_Condition, OneProducerTenConsumersOneSlot)
210{
211 runTest(1, 10, 1, 10000, TacticallyNotifyAll);
212}
213
214TEST(WTF_Condition, OneProducerTenConsumersHundredSlotsNotifyAll)
215{
216 runTest(1, 10, 100, 100000, TacticallyNotifyAll);
217}
218
219TEST(WTF_Condition, OneProducerTenConsumersHundredSlotsNotifyOne)
220{
221 runTest(1, 10, 100, 100000, AlwaysNotifyOne);
222}
223
224TEST(WTF_Condition, TenProducersTenConsumersOneSlot)
225{
226 runTest(10, 10, 1, 50000, TacticallyNotifyAll);
227}
228
229TEST(WTF_Condition, TenProducersTenConsumersHundredSlotsNotifyAll)
230{
231 runTest(10, 10, 100, 50000, TacticallyNotifyAll);
232}
233
234TEST(WTF_Condition, TenProducersTenConsumersHundredSlotsNotifyOne)
235{
236 runTest(10, 10, 100, 50000, AlwaysNotifyOne);
237}
238
239TEST(WTF_Condition, TimeoutTimesOut)
240{
241 Lock lock;
242 Condition condition;
243
244 lock.lock();
245 bool result = condition.waitFor(
246 lock, Seconds::fromMilliseconds(10), [] () -> bool { return false; });
247 lock.unlock();
248
249 EXPECT_FALSE(result);
250}
251
252} // namespace TestWebKitAPI
253
254