1 // Copyright 2019 The Marl Authors.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 #ifndef marl_condition_variable_h
16 #define marl_condition_variable_h
17
18 #include "containers.h"
19 #include "debug.h"
20 #include "memory.h"
21 #include "mutex.h"
22 #include "scheduler.h"
23 #include "tsa.h"
24
25 #include <atomic>
26 #include <condition_variable>
27
28 namespace marl {
29
30 // ConditionVariable is a synchronization primitive that can be used to block
31 // one or more fibers or threads, until another fiber or thread modifies a
32 // shared variable (the condition) and notifies the ConditionVariable.
33 //
34 // If the ConditionVariable is blocked on a thread with a Scheduler bound, the
35 // thread will work on other tasks until the ConditionVariable is unblocked.
36 class ConditionVariable {
37 public:
38 MARL_NO_EXPORT inline ConditionVariable(
39 Allocator* allocator = Allocator::Default);
40
41 // notify_one() notifies and potentially unblocks one waiting fiber or thread.
42 MARL_NO_EXPORT inline void notify_one();
43
44 // notify_all() notifies and potentially unblocks all waiting fibers and/or
45 // threads.
46 MARL_NO_EXPORT inline void notify_all();
47
48 // wait() blocks the current fiber or thread until the predicate is satisfied
49 // and the ConditionVariable is notified.
50 template <typename Predicate>
51 MARL_NO_EXPORT inline void wait(marl::lock& lock, Predicate&& pred);
52
53 // wait_for() blocks the current fiber or thread until the predicate is
54 // satisfied, and the ConditionVariable is notified, or the timeout has been
55 // reached. Returns false if pred still evaluates to false after the timeout
56 // has been reached, otherwise true.
57 template <typename Rep, typename Period, typename Predicate>
58 MARL_NO_EXPORT inline bool wait_for(
59 marl::lock& lock,
60 const std::chrono::duration<Rep, Period>& duration,
61 Predicate&& pred);
62
63 // wait_until() blocks the current fiber or thread until the predicate is
64 // satisfied, and the ConditionVariable is notified, or the timeout has been
65 // reached. Returns false if pred still evaluates to false after the timeout
66 // has been reached, otherwise true.
67 template <typename Clock, typename Duration, typename Predicate>
68 MARL_NO_EXPORT inline bool wait_until(
69 marl::lock& lock,
70 const std::chrono::time_point<Clock, Duration>& timeout,
71 Predicate&& pred);
72
73 private:
74 ConditionVariable(const ConditionVariable&) = delete;
75 ConditionVariable(ConditionVariable&&) = delete;
76 ConditionVariable& operator=(const ConditionVariable&) = delete;
77 ConditionVariable& operator=(ConditionVariable&&) = delete;
78
79 marl::mutex mutex;
80 containers::list<Scheduler::Fiber*> waiting;
81 std::condition_variable condition;
82 std::atomic<int> numWaiting = {0};
83 std::atomic<int> numWaitingOnCondition = {0};
84 };
85
ConditionVariable(Allocator * allocator)86 ConditionVariable::ConditionVariable(
87 Allocator* allocator /* = Allocator::Default */)
88 : waiting(allocator) {}
89
notify_one()90 void ConditionVariable::notify_one() {
91 if (numWaiting == 0) {
92 return;
93 }
94 {
95 marl::lock lock(mutex);
96 if (waiting.size() > 0) {
97 (*waiting.begin())->notify(); // Only wake one fiber.
98 }
99 }
100 if (numWaitingOnCondition > 0) {
101 condition.notify_one();
102 }
103 }
104
notify_all()105 void ConditionVariable::notify_all() {
106 if (numWaiting == 0) {
107 return;
108 }
109 {
110 marl::lock lock(mutex);
111 for (auto fiber : waiting) {
112 fiber->notify();
113 }
114 }
115 if (numWaitingOnCondition > 0) {
116 condition.notify_all();
117 }
118 }
119
120 template <typename Predicate>
wait(marl::lock & lock,Predicate && pred)121 void ConditionVariable::wait(marl::lock& lock, Predicate&& pred) {
122 if (pred()) {
123 return;
124 }
125 numWaiting++;
126 if (auto fiber = Scheduler::Fiber::current()) {
127 // Currently executing on a scheduler fiber.
128 // Yield to let other tasks run that can unblock this fiber.
129 mutex.lock();
130 auto it = waiting.emplace_front(fiber);
131 mutex.unlock();
132
133 fiber->wait(lock, pred);
134
135 mutex.lock();
136 waiting.erase(it);
137 mutex.unlock();
138 } else {
139 // Currently running outside of the scheduler.
140 // Delegate to the std::condition_variable.
141 numWaitingOnCondition++;
142 lock.wait(condition, pred);
143 numWaitingOnCondition--;
144 }
145 numWaiting--;
146 }
147
148 template <typename Rep, typename Period, typename Predicate>
wait_for(marl::lock & lock,const std::chrono::duration<Rep,Period> & duration,Predicate && pred)149 bool ConditionVariable::wait_for(
150 marl::lock& lock,
151 const std::chrono::duration<Rep, Period>& duration,
152 Predicate&& pred) {
153 return wait_until(lock, std::chrono::system_clock::now() + duration, pred);
154 }
155
156 template <typename Clock, typename Duration, typename Predicate>
wait_until(marl::lock & lock,const std::chrono::time_point<Clock,Duration> & timeout,Predicate && pred)157 bool ConditionVariable::wait_until(
158 marl::lock& lock,
159 const std::chrono::time_point<Clock, Duration>& timeout,
160 Predicate&& pred) {
161 if (pred()) {
162 return true;
163 }
164
165 if (auto fiber = Scheduler::Fiber::current()) {
166 numWaiting++;
167
168 // Currently executing on a scheduler fiber.
169 // Yield to let other tasks run that can unblock this fiber.
170 mutex.lock();
171 auto it = waiting.emplace_front(fiber);
172 mutex.unlock();
173
174 auto res = fiber->wait(lock, timeout, pred);
175
176 mutex.lock();
177 waiting.erase(it);
178 mutex.unlock();
179
180 numWaiting--;
181 return res;
182 }
183
184 // Currently running outside of the scheduler.
185 // Delegate to the std::condition_variable.
186 numWaiting++;
187 numWaitingOnCondition++;
188 auto res = lock.wait_until(condition, timeout, pred);
189 numWaitingOnCondition--;
190 numWaiting--;
191 return res;
192 }
193
194 } // namespace marl
195
196 #endif // marl_condition_variable_h
197