2 * Copyright 2016 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.
23 #include <folly/detail/Futex.h>
24 #include <folly/portability/Asm.h>
25 #include <folly/portability/Unistd.h>
31 /// A TurnSequencer allows threads to order their execution according to
32 /// a monotonically increasing (with wraparound) "turn" value. The two
33 /// operations provided are to wait for turn T, and to move to the next
34 /// turn. Every thread that is waiting for T must have arrived before
35 /// that turn is marked completed (for MPMCQueue only one thread waits
36 /// for any particular turn, so this is trivially true).
38 /// TurnSequencer's state_ holds 26 bits of the current turn (shifted
39 /// left by 6), along with a 6 bit saturating value that records the
40 /// maximum waiter minus the current turn. Wraparound of the turn space
41 /// is expected and handled. This allows us to atomically adjust the
42 /// number of outstanding waiters when we perform a FUTEX_WAKE operation.
43 /// Compare this strategy to sem_t's separate num_waiters field, which
44 /// isn't decremented until after the waiting thread gets scheduled,
45 /// during which time more enqueues might have occurred and made pointless
48 /// TurnSequencer uses futex() directly. It is optimized for the
49 /// case that the highest awaited turn is 32 or less higher than the
50 /// current turn. We use the FUTEX_WAIT_BITSET variant, which lets
51 /// us embed 32 separate wakeup channels in a single futex. See
52 /// http://locklessinc.com/articles/futex_cheat_sheet for a description.
54 /// We only need to keep exact track of the delta between the current
55 /// turn and the maximum waiter for the 32 turns that follow the current
56 /// one, because waiters at turn t+32 will be awoken at turn t. At that
57 /// point they can then adjust the delta using the higher base. Since we
58 /// need to encode waiter deltas of 0 to 32 inclusive, we use 6 bits.
59 /// We actually store waiter deltas up to 63, since that might reduce
60 /// the number of CAS operations a tiny bit.
62 /// To avoid some futex() calls entirely, TurnSequencer uses an adaptive
63 /// spin cutoff before waiting. The overheads (and convergence rate)
64 /// of separately tracking the spin cutoff for each TurnSequencer would
65 /// be prohibitive, so the actual storage is passed in as a parameter and
66 /// updated atomically. This also lets the caller use different adaptive
67 /// cutoffs for different operations (read versus write, for example).
68 /// To avoid contention, the spin cutoff is only updated when requested
70 template <template<typename> class Atom>
71 struct TurnSequencer {
72 explicit TurnSequencer(const uint32_t firstTurn = 0) noexcept
73 : state_(encode(firstTurn << kTurnShift, 0))
76 /// Returns true iff a call to waitForTurn(turn, ...) won't block
77 bool isTurn(const uint32_t turn) const noexcept {
78 auto state = state_.load(std::memory_order_acquire);
79 return decodeCurrentSturn(state) == (turn << kTurnShift);
82 /// See tryWaitForTurn
83 /// Requires that `turn` is not a turn in the past.
84 void waitForTurn(const uint32_t turn,
85 Atom<uint32_t>& spinCutoff,
86 const bool updateSpinCutoff) noexcept {
87 bool success = tryWaitForTurn(turn, spinCutoff, updateSpinCutoff);
92 // Internally we always work with shifted turn values, which makes the
93 // truncation and wraparound work correctly. This leaves us bits at
94 // the bottom to store the number of waiters. We call shifted turns
95 // "sturns" inside this class.
97 /// Blocks the current thread until turn has arrived. If
98 /// updateSpinCutoff is true then this will spin for up to kMaxSpins tries
99 /// before blocking and will adjust spinCutoff based on the results,
100 /// otherwise it will spin for at most spinCutoff spins.
101 /// Returns true if the wait succeeded, false if the turn is in the past
102 /// or the absTime time value is not nullptr and is reached before the turn
104 template <class Clock = std::chrono::steady_clock,
105 class Duration = typename Clock::duration>
106 bool tryWaitForTurn(const uint32_t turn,
107 Atom<uint32_t>& spinCutoff,
108 const bool updateSpinCutoff,
109 const std::chrono::time_point<Clock, Duration>* absTime =
111 uint32_t prevThresh = spinCutoff.load(std::memory_order_relaxed);
112 const uint32_t effectiveSpinCutoff =
113 updateSpinCutoff || prevThresh == 0 ? kMaxSpins : prevThresh;
116 const uint32_t sturn = turn << kTurnShift;
117 for (tries = 0; ; ++tries) {
118 uint32_t state = state_.load(std::memory_order_acquire);
119 uint32_t current_sturn = decodeCurrentSturn(state);
120 if (current_sturn == sturn) {
124 // wrap-safe version of (current_sturn >= sturn)
125 if(sturn - current_sturn >= std::numeric_limits<uint32_t>::max() / 2) {
126 // turn is in the past
130 // the first effectSpinCutoff tries are spins, after that we will
131 // record ourself as a waiter and block with futexWait
132 if (tries < effectiveSpinCutoff) {
133 asm_volatile_pause();
137 uint32_t current_max_waiter_delta = decodeMaxWaitersDelta(state);
138 uint32_t our_waiter_delta = (sturn - current_sturn) >> kTurnShift;
140 if (our_waiter_delta <= current_max_waiter_delta) {
141 // state already records us as waiters, probably because this
142 // isn't our first time around this loop
145 new_state = encode(current_sturn, our_waiter_delta);
146 if (state != new_state &&
147 !state_.compare_exchange_strong(state, new_state)) {
153 state_.futexWaitUntil(new_state, *absTime, futexChannel(turn));
154 if (futexResult == FutexResult::TIMEDOUT) {
158 state_.futexWait(new_state, futexChannel(turn));
162 if (updateSpinCutoff || prevThresh == 0) {
163 // if we hit kMaxSpins then spinning was pointless, so the right
164 // spinCutoff is kMinSpins
166 if (tries >= kMaxSpins) {
169 // to account for variations, we allow ourself to spin 2*N when
170 // we think that N is actually required in order to succeed
171 target = std::min<uint32_t>(kMaxSpins,
172 std::max<uint32_t>(kMinSpins, tries * 2));
175 if (prevThresh == 0) {
177 spinCutoff.store(target);
179 // try once, keep moving if CAS fails. Exponential moving average
181 // Be careful that the quantity we add to prevThresh is signed.
182 spinCutoff.compare_exchange_weak(
183 prevThresh, prevThresh + int(target - prevThresh) / 8);
190 /// Unblocks a thread running waitForTurn(turn + 1)
191 void completeTurn(const uint32_t turn) noexcept {
192 uint32_t state = state_.load(std::memory_order_acquire);
194 assert(state == encode(turn << kTurnShift, decodeMaxWaitersDelta(state)));
195 uint32_t max_waiter_delta = decodeMaxWaitersDelta(state);
197 encode((turn + 1) << kTurnShift,
198 max_waiter_delta == 0 ? 0 : max_waiter_delta - 1);
199 if (state_.compare_exchange_strong(state, new_state)) {
200 if (max_waiter_delta != 0) {
201 state_.futexWake(std::numeric_limits<int>::max(),
202 futexChannel(turn + 1));
206 // failing compare_exchange_strong updates first arg to the value
207 // that caused the failure, so no need to reread state_
211 /// Returns the least-most significant byte of the current uncompleted
212 /// turn. The full 32 bit turn cannot be recovered.
213 uint8_t uncompletedTurnLSB() const noexcept {
214 return state_.load(std::memory_order_acquire) >> kTurnShift;
219 /// kTurnShift counts the bits that are stolen to record the delta
220 /// between the current turn and the maximum waiter. It needs to be big
221 /// enough to record wait deltas of 0 to 32 inclusive. Waiters more
222 /// than 32 in the future will be woken up 32*n turns early (since
223 /// their BITSET will hit) and will adjust the waiter count again.
224 /// We go a bit beyond and let the waiter count go up to 63, which
225 /// is free and might save us a few CAS
227 kWaitersMask = (1 << kTurnShift) - 1,
229 /// The minimum spin count that we will adaptively select
232 /// The maximum spin count that we will adaptively select, and the
233 /// spin count that will be used when probing to get a new data point
234 /// for the adaptation
238 /// This holds both the current turn, and the highest waiting turn,
239 /// stored as (current_turn << 6) | min(63, max(waited_turn - current_turn))
242 /// Returns the bitmask to pass futexWait or futexWake when communicating
243 /// about the specified turn
244 int futexChannel(uint32_t turn) const noexcept { return 1 << (turn & 31); }
246 uint32_t decodeCurrentSturn(uint32_t state) const noexcept {
247 return state & ~kWaitersMask;
250 uint32_t decodeMaxWaitersDelta(uint32_t state) const noexcept {
251 return state & kWaitersMask;
254 uint32_t encode(uint32_t currentSturn, uint32_t maxWaiterD) const noexcept {
255 return currentSturn | std::min(uint32_t{kWaitersMask}, maxWaiterD);
259 } // namespace detail