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   /// Read the value at the cursor.
 105   /// Returns true if the read succeeded, false otherwise. If the return
 106   /// value is false, dest is to be considered partially read and in an
 107   /// inconsistent state. Readers are advised to discard it.
 108   bool tryRead(T& dest, const Cursor& cursor) noexcept {
 109     return slots_[idx(cursor.ticket)].tryRead(dest, turn(cursor.ticket));
 110   }
 111
 112   /// Read the value at the cursor or block if the write has not occurred yet.
 113   /// Returns true if the read succeeded, false otherwise. If the return
 114   /// value is false, dest is to be considered partially read and in an
 115   /// inconsistent state. Readers are advised to discard it.
 116   bool waitAndTryRead(T& dest, const Cursor& cursor) noexcept {
 117     return slots_[idx(cursor.ticket)].waitAndTryRead(dest, turn(cursor.ticket));
 118   }
 119
 120   /// Returns a Cursor pointing to the first write that has not occurred yet.
 121   Cursor currentHead() noexcept {
 122     return Cursor(ticket_.load());
 123   }
 124
 125   /// Returns a Cursor pointing to a currently readable write.
 126   /// skipFraction is a value in the [0, 1] range indicating how far into the
 127   /// currently readable window to place the cursor. 0 means the
 128   /// earliest readable write, 1 means the latest readable write (if any).
 129   Cursor currentTail(double skipFraction = 0.0) noexcept {
 130     assert(skipFraction >= 0.0 && skipFraction <= 1.0);
 131     uint64_t ticket = ticket_.load();
 132
 133     uint64_t backStep = llround((1.0 - skipFraction) * capacity_);
 134
 135     // always try to move at least one step backward to something readable
 136     backStep = std::max<uint64_t>(1, backStep);
 137
 138     // can't go back more steps than we've taken
 139     backStep = std::min(ticket, backStep);
 140
 141     return Cursor(ticket - backStep);
 142   }
 143
 144   ~LockFreeRingBuffer() {
 145   }
 146
 147 private:
 148   const uint32_t capacity_;
 149
 150   const std::unique_ptr<detail::RingBufferSlot<T,Atom>[]> slots_;
 151
 152   Atom<uint64_t> ticket_;
 153
 154   uint32_t idx(uint64_t ticket) noexcept {
 155     return ticket % capacity_;
 156   }
 157
 158   uint32_t turn(uint64_t ticket) noexcept {
 159     return (uint32_t)(ticket / capacity_);
 160   }
 161 }; // LockFreeRingBuffer
 162
 163 namespace detail {
 164 template<typename T, template<typename> class Atom>
 165 class RingBufferSlot {
 166 public:
 167   explicit RingBufferSlot() noexcept
 168     : sequencer_()
 169     , data()
 170   {
 171   }
 172
 173   void write(const uint32_t turn, T& value) noexcept {
 174     Atom<uint32_t> cutoff(0);
 175     sequencer_.waitForTurn(turn * 2, cutoff, false);
 176
 177     // Change to an odd-numbered turn to indicate write in process
 178     sequencer_.completeTurn(turn * 2);
 179
 180     data = std::move(value);
 181     sequencer_.completeTurn(turn * 2 + 1);
 182     // At (turn + 1) * 2
 183   }
 184
 185   bool waitAndTryRead(T& dest, uint32_t turn) noexcept {
 186     uint32_t desired_turn = (turn + 1) * 2;
 187     Atom<uint32_t> cutoff(0);
 188     if(!sequencer_.tryWaitForTurn(desired_turn, cutoff, false)) {
 189       return false;
 190     }
 191     memcpy(&dest, &data, sizeof(T));
 192
 193     // if it's still the same turn, we read the value successfully
 194     return sequencer_.isTurn(desired_turn);
 195   }
 196
 197   bool tryRead(T& dest, uint32_t turn) noexcept {
 198     // The write that started at turn 0 ended at turn 2
 199     if (!sequencer_.isTurn((turn + 1) * 2)) {
 200       return false;
 201     }
 202     memcpy(&dest, &data, sizeof(T));
 203
 204     // if it's still the same turn, we read the value successfully
 205     return sequencer_.isTurn((turn + 1) * 2);
 206   }
 207
 208
 209 private:
 210   TurnSequencer<Atom> sequencer_;
 211   T data;
 212 }; // RingBufferSlot
 213
 214 } // namespace detail
 215
 216 } // namespace folly