folly/synchronization/SaturatingSemaphore.h

   1 /*
   2  * Copyright 2017-present Facebook, Inc.
   3  *
   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
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  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.
  15  */
  16
  17 #pragma once
  18
  19 #include <folly/Likely.h>
  20 #include <folly/detail/Futex.h>
  21 #include <folly/portability/Asm.h>
  22 #include <folly/synchronization/WaitOptions.h>
  23
  24 #include <glog/logging.h>
  25
  26 #include <atomic>
  27
  28 namespace folly {
  29
  30 /// SaturatingSemaphore is a flag that allows concurrent posting by
  31 /// multiple posters and concurrent non-destructive waiting by
  32 /// multiple waiters.
  33 ///
  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.
  40 ///
  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
  47 /// zero.
  48 ///
  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
  52 /// the user.
  53 /// SaturatingSemaphore differs from Baton as follows:
  54 /// - Baton allows at most one call to post(); this allows any number
  55 ///   and concurrently.
  56 /// - Baton allows at most one successful call to any wait variant;
  57 ///   this allows any number and concurrently.
  58 ///
  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.
  63 ///
  64 /// Wait options:
  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.
  69 ///
  70 /// Functions:
  71 ///   bool ready():
  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.
  74 ///   void reset();
  75 ///     Clears the flag.
  76 ///   void post();
  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
  79 ///     return.
  80 ///   void wait(
  81 ///       WaitOptions opt = wait_options());
  82 ///     Waits for the flag to be set by a call to post.
  83 ///   bool try_wait();
  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(
  91 ///       duration&,
  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.
  95 ///
  96 /// Usage:
  97 /// @code
  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))));
 105 /// f.post();
 106 /// f.post();
 107 /// f.wait();
 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)));
 112 /// f.wait();
 113 /// f.reset();
 114 /// ASSERT_FALSE(f.try_wait());
 115 /// @endcode
 116
 117 template <bool MayBlock, template <typename> class Atom = std::atomic>
 118 class SaturatingSemaphore {
 119   detail::Futex<Atom> state_;
 120
 121   enum State : uint32_t {
 122     NOTREADY = 0,
 123     READY = 1,
 124     BLOCKED = 2,
 125   };
 126
 127  public:
 128   FOLLY_ALWAYS_INLINE static WaitOptions wait_options() {
 129     return {};
 130   }
 131
 132   /** constructor */
 133   constexpr SaturatingSemaphore() noexcept : state_(NOTREADY) {}
 134
 135   /** destructor */
 136   ~SaturatingSemaphore() {}
 137
 138   /** ready */
 139   FOLLY_ALWAYS_INLINE bool ready() const noexcept {
 140     return state_.load(std::memory_order_acquire) == READY;
 141   }
 142
 143   /** reset */
 144   void reset() noexcept {
 145     state_.store(NOTREADY, std::memory_order_relaxed);
 146   }
 147
 148   /** post */
 149   FOLLY_ALWAYS_INLINE void post() noexcept {
 150     if (!MayBlock) {
 151       state_.store(READY, std::memory_order_release);
 152     } else {
 153       postFastWaiterMayBlock();
 154     }
 155   }
 156
 157   /** wait */
 158   FOLLY_ALWAYS_INLINE
 159   void wait(const WaitOptions& opt = wait_options()) noexcept {
 160     try_wait_until(std::chrono::steady_clock::time_point::max(), opt);
 161   }
 162
 163   /** try_wait */
 164   FOLLY_ALWAYS_INLINE bool try_wait() noexcept {
 165     return ready();
 166   }
 167
 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())) {
 174       return true;
 175     }
 176     return tryWaitSlow(deadline, opt);
 177   }
 178
 179   /** try_wait_for */
 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())) {
 185       return true;
 186     }
 187     auto deadline = std::chrono::steady_clock::now() + duration;
 188     return tryWaitSlow(deadline, opt);
 189   }
 190
 191  private:
 192   FOLLY_ALWAYS_INLINE void postFastWaiterMayBlock() noexcept {
 193     uint32_t before = NOTREADY;
 194     if (LIKELY(state_.compare_exchange_strong(
 195             before,
 196             READY,
 197             std::memory_order_release,
 198             std::memory_order_relaxed))) {
 199       return;
 200     }
 201     postSlowWaiterMayBlock(before);
 202   }
 203
 204   void postSlowWaiterMayBlock(uint32_t before) noexcept; // defined below
 205
 206   template <typename Clock, typename Duration>
 207   bool tryWaitSlow(
 208       const std::chrono::time_point<Clock, Duration>& deadline,
 209       const WaitOptions& opt) noexcept; // defined below
 210 };
 211
 212 ///
 213 /// Member function definitioons
 214 ///
 215
 216 /** postSlowWaiterMayBlock */
 217 template <bool MayBlock, template <typename> class Atom>
 218 FOLLY_NOINLINE void SaturatingSemaphore<MayBlock, Atom>::postSlowWaiterMayBlock(
 219     uint32_t before) noexcept {
 220   while (true) {
 221     if (before == NOTREADY) {
 222       if (state_.compare_exchange_strong(
 223               before,
 224               READY,
 225               std::memory_order_release,
 226               std::memory_order_relaxed)) {
 227         return;
 228       }
 229     }
 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:
 233       //
 234       //  T1:             T2:             T3:
 235       //  local1.post();  local2.post();  global.wait();
 236       //  global.post();  global.post();  global.reset();
 237       //                                  seq_cst fence
 238       //                                  local1.try_wait() == true;
 239       //                                  local2.try_wait() == false;
 240       //
 241       // This following steps correspond to T2's global.post(), where
 242       // global is already posted by T1.
 243       //
 244       // The following fence and load guarantee that T3 does not miss
 245       // T2's prior stores, i.e., local2.post() in this example.
 246       //
 247       // The following case is prevented:
 248       //
 249       // Starting with local2 == NOTREADY and global == READY
 250       //
 251       // T2:                              T3:
 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
 255       //
 256       std::atomic_thread_fence(std::memory_order_seq_cst);
 257       before = state_.load(std::memory_order_relaxed);
 258       if (before == READY) {
 259         return;
 260       }
 261       continue;
 262     }
 263     DCHECK_EQ(before, BLOCKED);
 264     if (state_.compare_exchange_strong(
 265             before,
 266             READY,
 267             std::memory_order_release,
 268             std::memory_order_relaxed)) {
 269       state_.futexWake();
 270       return;
 271     }
 272   }
 273 }
 274
 275 /** tryWaitSlow */
 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();
 282   while (true) {
 283     auto before = state_.load(std::memory_order_acquire);
 284     if (before == READY) {
 285       return true;
 286     }
 287     if (Clock::now() >= deadline) {
 288       return false;
 289     }
 290     if (MayBlock) {
 291       auto tnow = Clock::now();
 292       if (tnow < tbegin) {
 293         // backward time discontinuity in Clock, revise spin_max starting point
 294         tbegin = tnow;
 295       }
 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(
 300                   before,
 301                   BLOCKED,
 302                   std::memory_order_relaxed,
 303                   std::memory_order_relaxed)) {
 304             continue;
 305           }
 306         }
 307         if (deadline == std::chrono::time_point<Clock, Duration>::max()) {
 308           state_.futexWait(BLOCKED);
 309         } else {
 310           state_.futexWaitUntil(BLOCKED, deadline);
 311         }
 312       }
 313     }
 314     asm_volatile_pause();
 315   }
 316 }
 317
 318 } // namespace folly