LockFreeRingBuffer.h

Go to the documentation of this file.

/*
 * Copyright 2015-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <atomic>
#include <cmath>
#include <cstring>
#include <memory>
#include <type_traits>

#include <boost/noncopyable.hpp>

#include <folly/Portability.h>
#include <folly/Traits.h>
#include <folly/detail/TurnSequencer.h>
#include <folly/portability/Unistd.h>

namespace folly {
namespace detail {

template <typename T, template <typename> class Atom>
class RingBufferSlot;
} // namespace detail


template <typename T, template <typename> class Atom = std::atomic>
class LockFreeRingBuffer : boost::noncopyable {
  static_assert(
      std::is_nothrow_default_constructible<T>::value,
      "Element type must be nothrow default constructible");

  static_assert(
      folly::is_trivially_copyable<T>::value,
      "Element type must be trivially copyable");

 public:
  struct Cursor {
    explicit Cursor(uint64_t initialTicket) noexcept : ticket(initialTicket) {}

    bool moveForward(uint64_t steps = 1) noexcept {
      uint64_t prevTicket = ticket;
      ticket += steps;
      return prevTicket != ticket;
    }

    bool moveBackward(uint64_t steps = 1) noexcept {
      uint64_t prevTicket = ticket;
      if (steps > ticket) {
        ticket = 0;
      } else {
        ticket -= steps;
      }
      return prevTicket != ticket;
    }

   protected: // for test visibility reasons
    uint64_t ticket;
    friend class LockFreeRingBuffer;
  };

  explicit LockFreeRingBuffer(uint32_t capacity) noexcept
      : capacity_(capacity),
        slots_(new detail::RingBufferSlot<T, Atom>[capacity]),
        ticket_(0) {}

  void write(T& value) noexcept {
    uint64_t ticket = ticket_.fetch_add(1);
    slots_[idx(ticket)].write(turn(ticket), value);
  }

  Cursor writeAndGetCursor(T& value) noexcept {
    uint64_t ticket = ticket_.fetch_add(1);
    slots_[idx(ticket)].write(turn(ticket), value);
    return Cursor(ticket);
  }

  bool tryRead(T& dest, const Cursor& cursor) noexcept {
    return slots_[idx(cursor.ticket)].tryRead(dest, turn(cursor.ticket));
  }

  bool waitAndTryRead(T& dest, const Cursor& cursor) noexcept {
    return slots_[idx(cursor.ticket)].waitAndTryRead(dest, turn(cursor.ticket));
  }

  Cursor currentHead() noexcept {
    return Cursor(ticket_.load());
  }

  Cursor currentTail(double skipFraction = 0.0) noexcept {
    assert(skipFraction >= 0.0 && skipFraction <= 1.0);
    uint64_t ticket = ticket_.load();

    uint64_t backStep = llround((1.0 - skipFraction) * capacity_);

    // always try to move at least one step backward to something readable
    backStep = std::max<uint64_t>(1, backStep);

    // can't go back more steps than we've taken
    backStep = std::min(ticket, backStep);

    return Cursor(ticket - backStep);
  }

  ~LockFreeRingBuffer() {}

 private:
  const uint32_t capacity_;

  const std::unique_ptr<detail::RingBufferSlot<T, Atom>[]> slots_;

  Atom<uint64_t> ticket_;

  uint32_t idx(uint64_t ticket) noexcept {
    return ticket % capacity_;
  }

  uint32_t turn(uint64_t ticket) noexcept {
    return (uint32_t)(ticket / capacity_);
  }
}; // LockFreeRingBuffer

namespace detail {
template <typename T, template <typename> class Atom>
class RingBufferSlot {
 public:
  explicit RingBufferSlot() noexcept : sequencer_(), data() {}

  void write(const uint32_t turn, T& value) noexcept {
    Atom<uint32_t> cutoff(0);
    sequencer_.waitForTurn(turn * 2, cutoff, false);

    // Change to an odd-numbered turn to indicate write in process
    sequencer_.completeTurn(turn * 2);

    data = std::move(value);
    sequencer_.completeTurn(turn * 2 + 1);
    // At (turn + 1) * 2
  }

  bool waitAndTryRead(T& dest, uint32_t turn) noexcept {
    uint32_t desired_turn = (turn + 1) * 2;
    Atom<uint32_t> cutoff(0);
    if (sequencer_.tryWaitForTurn(desired_turn, cutoff, false) !=
        TurnSequencer<Atom>::TryWaitResult::SUCCESS) {
      return false;
    }
    memcpy(&dest, &data, sizeof(T));

    // if it's still the same turn, we read the value successfully
    return sequencer_.isTurn(desired_turn);
  }

  bool tryRead(T& dest, uint32_t turn) noexcept {
    // The write that started at turn 0 ended at turn 2
    if (!sequencer_.isTurn((turn + 1) * 2)) {
      return false;
    }
    memcpy(&dest, &data, sizeof(T));

    // if it's still the same turn, we read the value successfully
    return sequencer_.isTurn((turn + 1) * 2);
  }

 private:
  TurnSequencer<Atom> sequencer_;
  T data;
}; // RingBufferSlot

} // namespace detail

} // namespace folly