folly/experimental/LockFreeRingBuffer.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 <atomic>
  20 #include <boost/noncopyable.hpp>
  21 #include <cmath>
  22 #include <memory>
  23 #include <string.h>
  24 #include <type_traits>
  25
  26 #include <folly/Portability.h>
  27 #include <folly/detail/TurnSequencer.h>
  28 #include <folly/portability/TypeTraits.h>
  29 #include <folly/portability/Unistd.h>
  30
  31 namespace folly {
  32 namespace detail {
  33
  34 template<typename T,
  35          template<typename> class Atom>
  36 class RingBufferSlot;
  37 } // namespace detail
  38
  39 /// LockFreeRingBuffer<T> is a fixed-size, concurrent ring buffer with the
  40 /// following semantics:
  41 ///
  42 ///  1. Writers cannot block on other writers UNLESS they are <capacity> writes
  43 ///     apart from each other (writing to the same slot after a wrap-around)
  44 ///  2. Writers cannot block on readers
  45 ///  3. Readers can wait for writes that haven't occurred yet
  46 ///  4. Readers can detect if they are lagging behind
  47 ///
  48 /// In this sense, reads from this buffer are best-effort but writes
  49 /// are guaranteed.
  50 ///
  51 /// Another way to think about this is as an unbounded stream of writes. The
  52 /// buffer contains the last <capacity> writes but readers can attempt to read
  53 /// any part of the stream, even outside this window. The read API takes a
  54 /// Cursor that can point anywhere in this stream of writes. Reads from the
  55 /// "future" can optionally block but reads from the "past" will always fail.
  56 ///
  57
  58 template<typename T, template<typename> class Atom = std::atomic>
  59 class LockFreeRingBuffer: boost::noncopyable {
  60
  61    static_assert(std::is_nothrow_default_constructible<T>::value,
  62        "Element type must be nothrow default constructible");
  63
  64    static_assert(FOLLY_IS_TRIVIALLY_COPYABLE(T),
  65        "Element type must be trivially copyable");
  66
  67 public:
  68   /// Opaque pointer to a past or future write.
  69   /// Can be moved relative to its current location but not in absolute terms.
  70   struct Cursor {
  71     explicit Cursor(uint64_t initialTicket) noexcept : ticket(initialTicket) {}
  72
  73     /// Returns true if this cursor now points to a different
  74     /// write, false otherwise.
  75     bool moveForward(uint64_t steps = 1) noexcept {
  76       uint64_t prevTicket = ticket;
  77       ticket += steps;
  78       return prevTicket != ticket;
  79     }
  80
  81     /// Returns true if this cursor now points to a previous
  82     /// write, false otherwise.
  83     bool moveBackward(uint64_t steps = 1) noexcept {
  84       uint64_t prevTicket = ticket;
  85       if (steps > ticket) {
  86         ticket = 0;
  87       } else {
  88         ticket -= steps;
  89       }
  90       return prevTicket != ticket;
  91     }
  92
  93   protected: // for test visibility reasons
  94     uint64_t ticket;
  95     friend class LockFreeRingBuffer;
  96   };
  97
  98   explicit LockFreeRingBuffer(uint32_t capacity) noexcept
  99     : capacity_(capacity)
 100     , slots_(new detail::RingBufferSlot<T,Atom>[capacity])
 101     , ticket_(0)
 102   {}
 103
 104   /// Perform a single write of an object of type T.
 105   /// Writes can block iff a previous writer has not yet completed a write
 106   /// for the same slot (before the most recent wrap-around).
 107   void write(T& value) noexcept {
 108     uint64_t ticket = ticket_.fetch_add(1);
 109     slots_[idx(ticket)].write(turn(ticket), value);
 110   }
 111
 112   /// Perform a single write of an object of type T.
 113   /// Writes can block iff a previous writer has not yet completed a write
 114   /// for the same slot (before the most recent wrap-around).
 115   /// Returns a Cursor pointing to the just-written T.
 116   Cursor writeAndGetCursor(T& value) noexcept {
 117     uint64_t ticket = ticket_.fetch_add(1);
 118     slots_[idx(ticket)].write(turn(ticket), value);
 119     return Cursor(ticket);
 120   }
 121
 122   /// Read the value at the cursor.
 123   /// Returns true if the read succeeded, false otherwise. If the return
 124   /// value is false, dest is to be considered partially read and in an
 125   /// inconsistent state. Readers are advised to discard it.
 126   bool tryRead(T& dest, const Cursor& cursor) noexcept {
 127     return slots_[idx(cursor.ticket)].tryRead(dest, turn(cursor.ticket));
 128   }
 129
 130   /// Read the value at the cursor or block if the write has not occurred yet.
 131   /// Returns true if the read succeeded, false otherwise. If the return
 132   /// value is false, dest is to be considered partially read and in an
 133   /// inconsistent state. Readers are advised to discard it.
 134   bool waitAndTryRead(T& dest, const Cursor& cursor) noexcept {
 135     return slots_[idx(cursor.ticket)].waitAndTryRead(dest, turn(cursor.ticket));
 136   }
 137
 138   /// Returns a Cursor pointing to the first write that has not occurred yet.
 139   Cursor currentHead() noexcept {
 140     return Cursor(ticket_.load());
 141   }
 142
 143   /// Returns a Cursor pointing to a currently readable write.
 144   /// skipFraction is a value in the [0, 1] range indicating how far into the
 145   /// currently readable window to place the cursor. 0 means the
 146   /// earliest readable write, 1 means the latest readable write (if any).
 147   Cursor currentTail(double skipFraction = 0.0) noexcept {
 148     assert(skipFraction >= 0.0 && skipFraction <= 1.0);
 149     uint64_t ticket = ticket_.load();
 150
 151     uint64_t backStep = llround((1.0 - skipFraction) * capacity_);
 152
 153     // always try to move at least one step backward to something readable
 154     backStep = std::max<uint64_t>(1, backStep);
 155
 156     // can't go back more steps than we've taken
 157     backStep = std::min(ticket, backStep);
 158
 159     return Cursor(ticket - backStep);
 160   }
 161
 162   ~LockFreeRingBuffer() {
 163   }
 164
 165 private:
 166   const uint32_t capacity_;
 167
 168   const std::unique_ptr<detail::RingBufferSlot<T,Atom>[]> slots_;
 169
 170   Atom<uint64_t> ticket_;
 171
 172   uint32_t idx(uint64_t ticket) noexcept {
 173     return ticket % capacity_;
 174   }
 175
 176   uint32_t turn(uint64_t ticket) noexcept {
 177     return (uint32_t)(ticket / capacity_);
 178   }
 179 }; // LockFreeRingBuffer
 180
 181 namespace detail {
 182 template<typename T, template<typename> class Atom>
 183 class RingBufferSlot {
 184 public:
 185   explicit RingBufferSlot() noexcept
 186     : sequencer_()
 187     , data()
 188   {
 189   }
 190
 191   void write(const uint32_t turn, T& value) noexcept {
 192     Atom<uint32_t> cutoff(0);
 193     sequencer_.waitForTurn(turn * 2, cutoff, false);
 194
 195     // Change to an odd-numbered turn to indicate write in process
 196     sequencer_.completeTurn(turn * 2);
 197
 198     data = std::move(value);
 199     sequencer_.completeTurn(turn * 2 + 1);
 200     // At (turn + 1) * 2
 201   }
 202
 203   bool waitAndTryRead(T& dest, uint32_t turn) noexcept {
 204     uint32_t desired_turn = (turn + 1) * 2;
 205     Atom<uint32_t> cutoff(0);
 206     if (sequencer_.tryWaitForTurn(desired_turn, cutoff, false) !=
 207         TurnSequencer<Atom>::TryWaitResult::SUCCESS) {
 208       return false;
 209     }
 210     memcpy(&dest, &data, sizeof(T));
 211
 212     // if it's still the same turn, we read the value successfully
 213     return sequencer_.isTurn(desired_turn);
 214   }
 215
 216   bool tryRead(T& dest, uint32_t turn) noexcept {
 217     // The write that started at turn 0 ended at turn 2
 218     if (!sequencer_.isTurn((turn + 1) * 2)) {
 219       return false;
 220     }
 221     memcpy(&dest, &data, sizeof(T));
 222
 223     // if it's still the same turn, we read the value successfully
 224     return sequencer_.isTurn((turn + 1) * 2);
 225   }
 226
 227
 228 private:
 229   TurnSequencer<Atom> sequencer_;
 230   T data;
 231 }; // RingBufferSlot
 232
 233 } // namespace detail
 234
 235 } // namespace folly