folly/experimental/LockFreeRingBuffer.h

   1 /*
   2  * Copyright 2015 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 <iostream>
  22 #include <cmath>
  23 #include <memory>
  24 #include <string.h>
  25 #include <type_traits>
  26 #include <unistd.h>
  27
  28 #include <folly/detail/TurnSequencer.h>
  29 #include <folly/Portability.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     void moveForward(uint64_t steps = 1) noexcept {
  74       ticket += steps;
  75     }
  76
  77     void moveBackward(uint64_t steps = 1) noexcept {
  78       if (steps > ticket) {
  79         ticket = 0;
  80       } else {
  81         ticket -= steps;
  82       }
  83     }
  84
  85   protected: // for test visibility reasons
  86     uint64_t ticket;
  87     friend class LockFreeRingBuffer;
  88   };
  89
  90   explicit LockFreeRingBuffer(uint32_t capacity) noexcept
  91     : capacity_(capacity)
  92     , slots_(new detail::RingBufferSlot<T,Atom>[capacity])
  93     , ticket_(0)
  94   {}
  95
  96   /// Perform a single write of an object of type T.
  97   /// Writes can block iff a previous writer has not yet completed a write
  98   /// for the same slot (before the most recent wrap-around).
  99   void write(T& value) noexcept {
 100     uint64_t ticket = ticket_.fetch_add(1);
 101     slots_[idx(ticket)].write(turn(ticket), value);
 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   /// Returns a Cursor pointing to the just-written T.
 108   Cursor writeAndGetCursor(T& value) noexcept {
 109     uint64_t ticket = ticket_.fetch_add(1);
 110     slots_[idx(ticket)].write(turn(ticket), value);
 111     return Cursor(ticket);
 112   }
 113
 114   /// Read the value at the cursor.
 115   /// Returns true if the read succeeded, false otherwise. If the return
 116   /// value is false, dest is to be considered partially read and in an
 117   /// inconsistent state. Readers are advised to discard it.
 118   bool tryRead(T& dest, const Cursor& cursor) noexcept {
 119     return slots_[idx(cursor.ticket)].tryRead(dest, turn(cursor.ticket));
 120   }
 121
 122   /// Read the value at the cursor or block if the write has not occurred yet.
 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 waitAndTryRead(T& dest, const Cursor& cursor) noexcept {
 127     return slots_[idx(cursor.ticket)].waitAndTryRead(dest, turn(cursor.ticket));
 128   }
 129
 130   /// Returns a Cursor pointing to the first write that has not occurred yet.
 131   Cursor currentHead() noexcept {
 132     return Cursor(ticket_.load());
 133   }
 134
 135   /// Returns a Cursor pointing to a currently readable write.
 136   /// skipFraction is a value in the [0, 1] range indicating how far into the
 137   /// currently readable window to place the cursor. 0 means the
 138   /// earliest readable write, 1 means the latest readable write (if any).
 139   Cursor currentTail(double skipFraction = 0.0) noexcept {
 140     assert(skipFraction >= 0.0 && skipFraction <= 1.0);
 141     uint64_t ticket = ticket_.load();
 142
 143     uint64_t backStep = llround((1.0 - skipFraction) * capacity_);
 144
 145     // always try to move at least one step backward to something readable
 146     backStep = std::max<uint64_t>(1, backStep);
 147
 148     // can't go back more steps than we've taken
 149     backStep = std::min(ticket, backStep);
 150
 151     return Cursor(ticket - backStep);
 152   }
 153
 154   ~LockFreeRingBuffer() {
 155   }
 156
 157 private:
 158   const uint32_t capacity_;
 159
 160   const std::unique_ptr<detail::RingBufferSlot<T,Atom>[]> slots_;
 161
 162   Atom<uint64_t> ticket_;
 163
 164   uint32_t idx(uint64_t ticket) noexcept {
 165     return ticket % capacity_;
 166   }
 167
 168   uint32_t turn(uint64_t ticket) noexcept {
 169     return (uint32_t)(ticket / capacity_);
 170   }
 171 }; // LockFreeRingBuffer
 172
 173 namespace detail {
 174 template<typename T, template<typename> class Atom>
 175 class RingBufferSlot {
 176 public:
 177   explicit RingBufferSlot() noexcept
 178     : sequencer_()
 179     , data()
 180   {
 181   }
 182
 183   void write(const uint32_t turn, T& value) noexcept {
 184     Atom<uint32_t> cutoff(0);
 185     sequencer_.waitForTurn(turn * 2, cutoff, false);
 186
 187     // Change to an odd-numbered turn to indicate write in process
 188     sequencer_.completeTurn(turn * 2);
 189
 190     data = std::move(value);
 191     sequencer_.completeTurn(turn * 2 + 1);
 192     // At (turn + 1) * 2
 193   }
 194
 195   bool waitAndTryRead(T& dest, uint32_t turn) noexcept {
 196     uint32_t desired_turn = (turn + 1) * 2;
 197     Atom<uint32_t> cutoff(0);
 198     if(!sequencer_.tryWaitForTurn(desired_turn, cutoff, false)) {
 199       return false;
 200     }
 201     memcpy(&dest, &data, sizeof(T));
 202
 203     // if it's still the same turn, we read the value successfully
 204     return sequencer_.isTurn(desired_turn);
 205   }
 206
 207   bool tryRead(T& dest, uint32_t turn) noexcept {
 208     // The write that started at turn 0 ended at turn 2
 209     if (!sequencer_.isTurn((turn + 1) * 2)) {
 210       return false;
 211     }
 212     memcpy(&dest, &data, sizeof(T));
 213
 214     // if it's still the same turn, we read the value successfully
 215     return sequencer_.isTurn((turn + 1) * 2);
 216   }
 217
 218
 219 private:
 220   TurnSequencer<Atom> sequencer_;
 221   T data;
 222 }; // RingBufferSlot
 223
 224 } // namespace detail
 225
 226 } // namespace folly