MPMCPipeline.h

Go to the documentation of this file.

/*
 * Copyright 2013-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 <utility>

#include <glog/logging.h>

#include <folly/Portability.h>
#include <folly/detail/MPMCPipelineDetail.h>

namespace folly {

template <class T, size_t Amp>
class MPMCPipelineStage;

template <class In, class... Stages>
class MPMCPipeline {
  typedef std::tuple<detail::PipelineStageInfo<Stages>...> StageInfos;
  typedef std::tuple<
      detail::MPMCPipelineStageImpl<In>,
      detail::MPMCPipelineStageImpl<
          typename detail::PipelineStageInfo<Stages>::value_type>...>
      StageTuple;
  static constexpr size_t kAmplification =
      detail::AmplificationProduct<StageInfos>::value;

  class TicketBaseDebug {
   public:
    TicketBaseDebug() noexcept : owner_(nullptr), value_(0xdeadbeeffaceb00c) {}
    TicketBaseDebug(TicketBaseDebug&& other) noexcept
        : owner_(std::exchange(other.owner_, nullptr)),
          value_(std::exchange(other.value_, 0xdeadbeeffaceb00c)) {}
    explicit TicketBaseDebug(MPMCPipeline* owner, uint64_t value) noexcept
        : owner_(owner), value_(value) {}
    void check_owner(MPMCPipeline* owner) const {
      CHECK(owner == owner_);
    }

    MPMCPipeline* owner_;
    uint64_t value_;
  };

  class TicketBaseNDebug {
   public:
    TicketBaseNDebug() = default;
    TicketBaseNDebug(TicketBaseNDebug&&) = default;
    explicit TicketBaseNDebug(MPMCPipeline*, uint64_t value) noexcept
        : value_(value) {}
    void check_owner(MPMCPipeline*) const {}

    uint64_t value_;
  };

  using TicketBase =
      std::conditional_t<kIsDebug, TicketBaseDebug, TicketBaseNDebug>;

 public:
  template <size_t Stage>
  class Ticket : TicketBase {
   public:
    ~Ticket() noexcept {
      CHECK_EQ(remainingUses_, 0) << "All tickets must be completely used!";
    }

    Ticket() noexcept : remainingUses_(0) {}

    Ticket(Ticket&& other) noexcept
        : TicketBase(static_cast<TicketBase&&>(other)),
          remainingUses_(std::exchange(other.remainingUses_, 0)) {}

    Ticket& operator=(Ticket&& other) noexcept {
      if (this != &other) {
        this->~Ticket();
        new (this) Ticket(std::move(other));
      }
      return *this;
    }

   private:
    friend class MPMCPipeline;
    size_t remainingUses_;

    Ticket(MPMCPipeline* owner, size_t amplification, uint64_t value) noexcept
        : TicketBase(owner, value * amplification),
          remainingUses_(amplification) {}

    uint64_t use(MPMCPipeline* owner) {
      CHECK_GT(remainingUses_--, 0);
      TicketBase::check_owner(owner);
      return TicketBase::value_++;
    }
  };

  MPMCPipeline() = default;

  template <class... Sizes>
  explicit MPMCPipeline(Sizes... sizes) : stages_(sizes...) {}

  template <class... Args>
  void blockingWrite(Args&&... args) {
    std::get<0>(stages_).blockingWrite(std::forward<Args>(args)...);
  }

  template <class... Args>
  bool write(Args&&... args) {
    return std::get<0>(stages_).write(std::forward<Args>(args)...);
  }

  template <size_t Stage>
  Ticket<Stage> blockingReadStage(
      typename std::tuple_element<Stage, StageTuple>::type::value_type& elem) {
    return Ticket<Stage>(
        this,
        std::tuple_element<Stage, StageInfos>::type::kAmplification,
        std::get<Stage>(stages_).blockingRead(elem));
  }

  template <size_t Stage>
  bool readStage(
      Ticket<Stage>& ticket,
      typename std::tuple_element<Stage, StageTuple>::type::value_type& elem) {
    uint64_t tval;
    if (!std::get<Stage>(stages_).readAndGetTicket(tval, elem)) {
      return false;
    }
    ticket = Ticket<Stage>(
        this,
        std::tuple_element<Stage, StageInfos>::type::kAmplification,
        tval);
    return true;
  }

  template <size_t Stage, class... Args>
  void blockingWriteStage(Ticket<Stage>& ticket, Args&&... args) {
    std::get<Stage + 1>(stages_).blockingWriteWithTicket(
        ticket.use(this), std::forward<Args>(args)...);
  }

  void blockingRead(typename std::tuple_element<sizeof...(Stages), StageTuple>::
                        type::value_type& elem) {
    std::get<sizeof...(Stages)>(stages_).blockingRead(elem);
  }

  bool read(typename std::tuple_element<sizeof...(Stages), StageTuple>::type::
                value_type& elem) {
    return std::get<sizeof...(Stages)>(stages_).read(elem);
  }

  ssize_t sizeGuess() const noexcept {
    return ssize_t(
        std::get<0>(stages_).writeCount() * kAmplification -
        std::get<sizeof...(Stages)>(stages_).readCount());
  }

 private:
  StageTuple stages_;
};

} // namespace folly