folly/Baton.h

   1 /*
   2  * Copyright 2017 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 <assert.h>
  20 #include <errno.h>
  21 #include <stdint.h>
  22 #include <atomic>
  23 #include <thread>
  24
  25 #include <folly/detail/Futex.h>
  26 #include <folly/detail/MemoryIdler.h>
  27 #include <folly/portability/Asm.h>
  28
  29 namespace folly {
  30
  31 /// A Baton allows a thread to block once and be awoken. The single
  32 /// poster version (with SinglePoster == true) captures a single
  33 /// handoff, and during its lifecycle (from construction/reset to
  34 /// destruction/reset) a baton must either be post()ed and wait()ed
  35 /// exactly once each, or not at all.
  36 ///
  37 /// The multi-poster version (SinglePoster == false) allows multiple
  38 /// concurrent handoff attempts, the first of which completes the
  39 /// handoff and the rest if any are idempotent.
  40 ///
  41 /// Baton includes no internal padding, and is only 4 bytes in size.
  42 /// Any alignment or padding to avoid false sharing is up to the user.
  43 ///
  44 /// This is basically a stripped-down semaphore that supports (only a
  45 /// single call to sem_post, when SinglePoster == true) and a single
  46 /// call to sem_wait.
  47 ///
  48 /// The non-blocking version (Blocking == false) provides more speed
  49 /// by using only load acquire and store release operations in the
  50 /// critical path, at the cost of disallowing blocking and timing out.
  51 ///
  52 /// The current posix semaphore sem_t isn't too bad, but this provides
  53 /// more a bit more speed, inlining, smaller size, a guarantee that
  54 /// the implementation won't change, and compatibility with
  55 /// DeterministicSchedule.  By having a much more restrictive
  56 /// lifecycle we can also add a bunch of assertions that can help to
  57 /// catch race conditions ahead of time.
  58 template <
  59     template <typename> class Atom = std::atomic,
  60     bool SinglePoster = true, //  single vs multiple posters
  61     bool Blocking = true> // blocking vs spinning
  62 struct Baton {
  63   constexpr Baton() : state_(INIT) {}
  64
  65   Baton(Baton const&) = delete;
  66   Baton& operator=(Baton const&) = delete;
  67
  68   /// It is an error to destroy a Baton on which a thread is currently
  69   /// wait()ing.  In practice this means that the waiter usually takes
  70   /// responsibility for destroying the Baton.
  71   ~Baton() {
  72     // The docblock for this function says that it can't be called when
  73     // there is a concurrent waiter.  We assume a strong version of this
  74     // requirement in which the caller must _know_ that this is true, they
  75     // are not allowed to be merely lucky.  If two threads are involved,
  76     // the destroying thread must actually have synchronized with the
  77     // waiting thread after wait() returned.  To convey causality the the
  78     // waiting thread must have used release semantics and the destroying
  79     // thread must have used acquire semantics for that communication,
  80     // so we are guaranteed to see the post-wait() value of state_,
  81     // which cannot be WAITING.
  82     //
  83     // Note that since we only care about a single memory location,
  84     // the only two plausible memory orders here are relaxed and seq_cst.
  85     assert(state_.load(std::memory_order_relaxed) != WAITING);
  86   }
  87
  88   /// Equivalent to destroying the Baton and creating a new one.  It is
  89   /// a bug to call this while there is a waiting thread, so in practice
  90   /// the waiter will be the one that resets the baton.
  91   void reset() {
  92     // See ~Baton for a discussion about why relaxed is okay here
  93     assert(state_.load(std::memory_order_relaxed) != WAITING);
  94
  95     // We use a similar argument to justify the use of a relaxed store
  96     // here.  Since both wait() and post() are required to be called
  97     // only once per lifetime, no thread can actually call those methods
  98     // correctly after a reset() unless it synchronizes with the thread
  99     // that performed the reset().  If a post() or wait() on another thread
 100     // didn't synchronize, then regardless of what operation we performed
 101     // here there would be a race on proper use of the Baton's spec
 102     // (although not on any particular load and store).  Put another way,
 103     // we don't need to synchronize here because anybody that might rely
 104     // on such synchronization is required by the baton rules to perform
 105     // an additional synchronization that has the desired effect anyway.
 106     //
 107     // There is actually a similar argument to be made about the
 108     // constructor, in which the fenceless constructor initialization
 109     // of state_ is piggybacked on whatever synchronization mechanism
 110     // distributes knowledge of the Baton's existence
 111     state_.store(INIT, std::memory_order_relaxed);
 112   }
 113
 114   /// Causes wait() to wake up.  For each lifetime of a Baton (where a
 115   /// lifetime starts at construction or reset() and ends at
 116   /// destruction or reset()) there can be at most one call to post(),
 117   /// in the single poster version.  Any thread may call post().
 118   void post() {
 119     if (!Blocking) {
 120       /// Non-blocking version
 121       ///
 122       assert([&] {
 123         auto state = state_.load(std::memory_order_relaxed);
 124         return (state == INIT || state == EARLY_DELIVERY);
 125       }());
 126       state_.store(EARLY_DELIVERY, std::memory_order_release);
 127       return;
 128     }
 129
 130     /// Blocking versions
 131     ///
 132     if (SinglePoster) {
 133       /// Single poster version
 134       ///
 135       uint32_t before = state_.load(std::memory_order_acquire);
 136
 137       assert(before == INIT || before == WAITING || before == TIMED_OUT);
 138
 139       if (before == INIT &&
 140           state_.compare_exchange_strong(before, EARLY_DELIVERY)) {
 141         return;
 142       }
 143
 144       assert(before == WAITING || before == TIMED_OUT);
 145
 146       if (before == TIMED_OUT) {
 147         return;
 148       }
 149
 150       assert(before == WAITING);
 151       state_.store(LATE_DELIVERY, std::memory_order_release);
 152       state_.futexWake(1);
 153     } else {
 154       /// Multi-poster version
 155       ///
 156       while (true) {
 157         uint32_t before = state_.load(std::memory_order_acquire);
 158
 159         if (before == INIT &&
 160             state_.compare_exchange_strong(before, EARLY_DELIVERY)) {
 161           return;
 162         }
 163
 164         if (before == TIMED_OUT) {
 165           return;
 166         }
 167
 168         if (before == EARLY_DELIVERY || before == LATE_DELIVERY) {
 169           // The reason for not simply returning (without the following
 170           // atomic operation) is to avoid the following case:
 171           //
 172           //  T1:             T2:             T3:
 173           //  local1.post();  local2.post();  global.wait();
 174           //  global.post();  global.post();  local1.try_wait() == true;
 175           //                                  local2.try_wait() == false;
 176           //
 177           if (state_.fetch_add(0) != before) {
 178             continue;
 179           }
 180           return;
 181         }
 182
 183         assert(before == WAITING);
 184         if (!state_.compare_exchange_weak(before, LATE_DELIVERY)) {
 185           continue;
 186         }
 187         state_.futexWake(1);
 188         return;
 189       }
 190     }
 191   }
 192
 193   /// Waits until post() has been called in the current Baton lifetime.
 194   /// May be called at most once during a Baton lifetime (construction
 195   /// |reset until destruction|reset).  If post is called before wait in
 196   /// the current lifetime then this method returns immediately.
 197   ///
 198   /// The restriction that there can be at most one wait() per lifetime
 199   /// could be relaxed somewhat without any perf or size regressions,
 200   /// but by making this condition very restrictive we can provide better
 201   /// checking in debug builds.
 202   void wait() {
 203     if (spinWaitForEarlyDelivery()) {
 204       assert(state_.load(std::memory_order_acquire) == EARLY_DELIVERY);
 205       return;
 206     }
 207
 208     if (!Blocking) {
 209       while (!try_wait()) {
 210         std::this_thread::yield();
 211       }
 212       return;
 213     }
 214
 215     // guess we have to block :(
 216     uint32_t expected = INIT;
 217     if (!state_.compare_exchange_strong(expected, WAITING)) {
 218       // CAS failed, last minute reprieve
 219       assert(expected == EARLY_DELIVERY);
 220       return;
 221     }
 222
 223     while (true) {
 224       detail::MemoryIdler::futexWait(state_, WAITING);
 225
 226       // state_ is the truth even if FUTEX_WAIT reported a matching
 227       // FUTEX_WAKE, since we aren't using type-stable storage and we
 228       // don't guarantee reuse.  The scenario goes like this: thread
 229       // A's last touch of a Baton is a call to wake(), which stores
 230       // LATE_DELIVERY and gets an unlucky context switch before delivering
 231       // the corresponding futexWake.  Thread B sees LATE_DELIVERY
 232       // without consuming a futex event, because it calls futexWait
 233       // with an expected value of WAITING and hence doesn't go to sleep.
 234       // B returns, so the Baton's memory is reused and becomes another
 235       // Baton (or a reuse of this one).  B calls futexWait on the new
 236       // Baton lifetime, then A wakes up and delivers a spurious futexWake
 237       // to the same memory location.  B's futexWait will then report a
 238       // consumed wake event even though state_ is still WAITING.
 239       //
 240       // It would be possible to add an extra state_ dance to communicate
 241       // that the futexWake has been sent so that we can be sure to consume
 242       // it before returning, but that would be a perf and complexity hit.
 243       uint32_t s = state_.load(std::memory_order_acquire);
 244       assert(s == WAITING || s == LATE_DELIVERY);
 245
 246       if (s == LATE_DELIVERY) {
 247         return;
 248       }
 249       // retry
 250     }
 251   }
 252
 253   /// Similar to wait, but with a timeout. The thread is unblocked if the
 254   /// timeout expires.
 255   /// Note: Only a single call to timed_wait/wait is allowed during a baton's
 256   /// life-cycle (from construction/reset to destruction/reset). In other
 257   /// words, after timed_wait the caller can't invoke wait/timed_wait/try_wait
 258   /// again on the same baton without resetting it.
 259   ///
 260   /// @param  deadline      Time until which the thread can block
 261   /// @return               true if the baton was posted to before timeout,
 262   ///                       false otherwise
 263   template <typename Clock, typename Duration = typename Clock::duration>
 264   bool timed_wait(const std::chrono::time_point<Clock,Duration>& deadline) {
 265     static_assert(Blocking, "Non-blocking Baton does not support timed wait.");
 266
 267     if (spinWaitForEarlyDelivery()) {
 268       assert(state_.load(std::memory_order_acquire) == EARLY_DELIVERY);
 269       return true;
 270     }
 271
 272     // guess we have to block :(
 273     uint32_t expected = INIT;
 274     if (!state_.compare_exchange_strong(expected, WAITING)) {
 275       // CAS failed, last minute reprieve
 276       assert(expected == EARLY_DELIVERY);
 277       return true;
 278     }
 279
 280     while (true) {
 281       auto rv = state_.futexWaitUntil(WAITING, deadline);
 282       if (rv == folly::detail::FutexResult::TIMEDOUT) {
 283         state_.store(TIMED_OUT, std::memory_order_release);
 284         return false;
 285       }
 286
 287       uint32_t s = state_.load(std::memory_order_acquire);
 288       assert(s == WAITING || s == LATE_DELIVERY);
 289       if (s == LATE_DELIVERY) {
 290         return true;
 291       }
 292     }
 293   }
 294
 295   /// Similar to timed_wait, but with a duration.
 296   template <typename Clock = std::chrono::steady_clock, typename Duration>
 297   bool timed_wait(const Duration& duration) {
 298     auto deadline = Clock::now() + duration;
 299     return timed_wait(deadline);
 300   }
 301
 302   /// Similar to wait, but doesn't block the thread if it hasn't been posted.
 303   ///
 304   /// try_wait has the following semantics:
 305   /// - It is ok to call try_wait any number times on the same baton until
 306   ///   try_wait reports that the baton has been posted.
 307   /// - It is ok to call timed_wait or wait on the same baton if try_wait
 308   ///   reports that baton hasn't been posted.
 309   /// - If try_wait indicates that the baton has been posted, it is invalid to
 310   ///   call wait, try_wait or timed_wait on the same baton without resetting
 311   ///
 312   /// @return       true if baton has been posted, false othewise
 313   bool try_wait() const {
 314     auto s = state_.load(std::memory_order_acquire);
 315     assert(s == INIT || s == EARLY_DELIVERY);
 316     return s == EARLY_DELIVERY;
 317   }
 318
 319  private:
 320   enum State : uint32_t {
 321     INIT = 0,
 322     EARLY_DELIVERY = 1,
 323     WAITING = 2,
 324     LATE_DELIVERY = 3,
 325     TIMED_OUT = 4
 326   };
 327
 328   enum {
 329     // Must be positive.  If multiple threads are actively using a
 330     // higher-level data structure that uses batons internally, it is
 331     // likely that the post() and wait() calls happen almost at the same
 332     // time.  In this state, we lose big 50% of the time if the wait goes
 333     // to sleep immediately.  On circa-2013 devbox hardware it costs about
 334     // 7 usec to FUTEX_WAIT and then be awoken (half the t/iter as the
 335     // posix_sem_pingpong test in BatonTests).  We can improve our chances
 336     // of EARLY_DELIVERY by spinning for a bit, although we have to balance
 337     // this against the loss if we end up sleeping any way.  Spins on this
 338     // hw take about 7 nanos (all but 0.5 nanos is the pause instruction).
 339     // We give ourself 300 spins, which is about 2 usec of waiting.  As a
 340     // partial consolation, since we are using the pause instruction we
 341     // are giving a speed boost to the colocated hyperthread.
 342     PreBlockAttempts = 300,
 343   };
 344
 345   // Spin for "some time" (see discussion on PreBlockAttempts) waiting
 346   // for a post.
 347   //
 348   // @return       true if we received an early delivery during the wait,
 349   //               false otherwise. If the function returns true then
 350   //               state_ is guaranteed to be EARLY_DELIVERY
 351   bool spinWaitForEarlyDelivery() {
 352
 353     static_assert(PreBlockAttempts > 0,
 354         "isn't this assert clearer than an uninitialized variable warning?");
 355     for (int i = 0; i < PreBlockAttempts; ++i) {
 356       if (try_wait()) {
 357         // hooray!
 358         return true;
 359       }
 360       // The pause instruction is the polite way to spin, but it doesn't
 361       // actually affect correctness to omit it if we don't have it.
 362       // Pausing donates the full capabilities of the current core to
 363       // its other hyperthreads for a dozen cycles or so
 364       asm_volatile_pause();
 365     }
 366
 367     return false;
 368   }
 369
 370   detail::Futex<Atom> state_;
 371 };
 372
 373 } // namespace folly