2 * Copyright 2017-present Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
19 #include <folly/Likely.h>
20 #include <folly/detail/Futex.h>
21 #include <folly/portability/Asm.h>
22 #include <folly/synchronization/WaitOptions.h>
24 #include <glog/logging.h>
30 /// SaturatingSemaphore is a flag that allows concurrent posting by
31 /// multiple posters and concurrent non-destructive waiting by
34 /// A SaturatingSemaphore allows one or more waiter threads to check,
35 /// spin, or block, indefinitely or with timeout, for a flag to be set
36 /// by one or more poster threads. By setting the flag, posters
37 /// announce to waiters (that may be already waiting or will check
38 /// the flag in the future) that some condition is true. Posts to an
39 /// already set flag are idempotent.
41 /// SaturatingSemaphore is called so because it behaves like a hybrid
42 /// binary/counted _semaphore_ with values zero and infinity, and
43 /// post() and wait() functions. It is called _saturating_ because one
44 /// post() is enough to set it to infinity and to satisfy any number
45 /// of wait()-s. Once set (to infinity) it remains unchanged by
46 /// subsequent post()-s and wait()-s, until it is reset() back to
49 /// The implementation of SaturatingSemaphore is based on that of
50 /// Baton. It includes no internal padding, and is only 4 bytes in
51 /// size. Any alignment or padding to avoid false sharing is up to
53 /// SaturatingSemaphore differs from Baton as follows:
54 /// - Baton allows at most one call to post(); this allows any number
56 /// - Baton allows at most one successful call to any wait variant;
57 /// this allows any number and concurrently.
59 /// Template parameter:
60 /// - bool MayBlock: If false, waiting operations spin only. If
61 /// true, timed and wait operations may block; adds an atomic
62 /// instruction to the critical path of posters.
65 /// WaitOptions contains optional per call setting for spin-max duration:
66 /// Calls to wait(), try_wait_until(), and try_wait_for() block only after the
67 /// passage of the spin-max period. The default spin-max duration is 10 usec.
68 /// The spin-max option is applicable only if MayBlock is true.
72 /// Returns true if the flag is set by a call to post, otherwise false.
73 /// Equivalent to try_wait, but available on const receivers.
77 /// Sets the flag and wakes all current waiters, i.e., causes all
78 /// concurrent calls to wait, try_wait_for, and try_wait_until to
81 /// WaitOptions opt = wait_options());
82 /// Waits for the flag to be set by a call to post.
84 /// Returns true if the flag is set by a call to post, otherwise false.
85 /// bool try_wait_until(
86 /// time_point& deadline,
87 /// WaitOptions& = wait_options());
88 /// Returns true if the flag is set by a call to post before the
89 /// deadline, otherwise false.
90 /// bool try_wait_for(
92 /// WaitOptions& = wait_options());
93 /// Returns true if the flag is set by a call to post before the
94 /// expiration of the specified duration, otherwise false.
98 /// SaturatingSemaphore</* MayBlock = */ true> f;
99 /// ASSERT_FALSE(f.try_wait());
100 /// ASSERT_FALSE(f.try_wait_until(
101 /// std::chrono::steady_clock::now() + std::chrono::microseconds(1)));
102 /// ASSERT_FALSE(f.try_wait_until(
103 /// std::chrono::steady_clock::now() + std::chrono::microseconds(1),
104 /// f.wait_options().spin_max(std::chrono::microseconds(1))));
108 /// f.wait(f.wait_options().spin_max(std::chrono::nanoseconds(100)));
109 /// ASSERT_TRUE(f.try_wait());
110 /// ASSERT_TRUE(f.try_wait_until(
111 /// std::chrono::steady_clock::now() + std::chrono::microseconds(1)));
114 /// ASSERT_FALSE(f.try_wait());
117 template <bool MayBlock, template <typename> class Atom = std::atomic>
118 class SaturatingSemaphore {
119 detail::Futex<Atom> state_;
121 enum State : uint32_t {
128 FOLLY_ALWAYS_INLINE static WaitOptions wait_options() {
133 constexpr SaturatingSemaphore() noexcept : state_(NOTREADY) {}
136 ~SaturatingSemaphore() {}
139 FOLLY_ALWAYS_INLINE bool ready() const noexcept {
140 return state_.load(std::memory_order_acquire) == READY;
144 void reset() noexcept {
145 state_.store(NOTREADY, std::memory_order_relaxed);
149 FOLLY_ALWAYS_INLINE void post() noexcept {
151 state_.store(READY, std::memory_order_release);
153 postFastWaiterMayBlock();
159 void wait(const WaitOptions& opt = wait_options()) noexcept {
160 try_wait_until(std::chrono::steady_clock::time_point::max(), opt);
164 FOLLY_ALWAYS_INLINE bool try_wait() noexcept {
168 /** try_wait_until */
169 template <typename Clock, typename Duration>
170 FOLLY_ALWAYS_INLINE bool try_wait_until(
171 const std::chrono::time_point<Clock, Duration>& deadline,
172 const WaitOptions& opt = wait_options()) noexcept {
173 if (LIKELY(try_wait())) {
176 return tryWaitSlow(deadline, opt);
180 template <class Rep, class Period>
181 FOLLY_ALWAYS_INLINE bool try_wait_for(
182 const std::chrono::duration<Rep, Period>& duration,
183 const WaitOptions& opt = wait_options()) noexcept {
184 if (LIKELY(try_wait())) {
187 auto deadline = std::chrono::steady_clock::now() + duration;
188 return tryWaitSlow(deadline, opt);
192 FOLLY_ALWAYS_INLINE void postFastWaiterMayBlock() noexcept {
193 uint32_t before = NOTREADY;
194 if (LIKELY(state_.compare_exchange_strong(
197 std::memory_order_release,
198 std::memory_order_relaxed))) {
201 postSlowWaiterMayBlock(before);
204 void postSlowWaiterMayBlock(uint32_t before) noexcept; // defined below
206 template <typename Clock, typename Duration>
208 const std::chrono::time_point<Clock, Duration>& deadline,
209 const WaitOptions& opt) noexcept; // defined below
213 /// Member function definitioons
216 /** postSlowWaiterMayBlock */
217 template <bool MayBlock, template <typename> class Atom>
218 FOLLY_NOINLINE void SaturatingSemaphore<MayBlock, Atom>::postSlowWaiterMayBlock(
219 uint32_t before) noexcept {
221 if (before == NOTREADY) {
222 if (state_.compare_exchange_strong(
225 std::memory_order_release,
226 std::memory_order_relaxed)) {
230 if (before == READY) { // Only if multiple posters
231 // The reason for not simply returning (without the following
232 // steps) is to prevent the following case:
235 // local1.post(); local2.post(); global.wait();
236 // global.post(); global.post(); global.reset();
238 // local1.try_wait() == true;
239 // local2.try_wait() == false;
241 // This following steps correspond to T2's global.post(), where
242 // global is already posted by T1.
244 // The following fence and load guarantee that T3 does not miss
245 // T2's prior stores, i.e., local2.post() in this example.
247 // The following case is prevented:
249 // Starting with local2 == NOTREADY and global == READY
252 // store READY to local2 // post store NOTREADY to global // reset
253 // seq_cst fenc seq_cst fence
254 // load READY from global // post load NOTREADY from local2 // try_wait
256 std::atomic_thread_fence(std::memory_order_seq_cst);
257 before = state_.load(std::memory_order_relaxed);
258 if (before == READY) {
263 DCHECK_EQ(before, BLOCKED);
264 if (state_.compare_exchange_strong(
267 std::memory_order_release,
268 std::memory_order_relaxed)) {
276 template <bool MayBlock, template <typename> class Atom>
277 template <typename Clock, typename Duration>
278 FOLLY_NOINLINE bool SaturatingSemaphore<MayBlock, Atom>::tryWaitSlow(
279 const std::chrono::time_point<Clock, Duration>& deadline,
280 const WaitOptions& opt) noexcept {
281 auto tbegin = Clock::now();
283 auto before = state_.load(std::memory_order_acquire);
284 if (before == READY) {
287 if (Clock::now() >= deadline) {
291 auto tnow = Clock::now();
293 // backward time discontinuity in Clock, revise spin_max starting point
296 auto dur = std::chrono::duration_cast<Duration>(tnow - tbegin);
297 if (dur >= opt.spin_max()) {
298 if (before == NOTREADY) {
299 if (!state_.compare_exchange_strong(
302 std::memory_order_relaxed,
303 std::memory_order_relaxed)) {
307 if (deadline == std::chrono::time_point<Clock, Duration>::max()) {
308 state_.futexWait(BLOCKED);
310 state_.futexWaitUntil(BLOCKED, deadline);
314 asm_volatile_pause();