pipe.h

Go to the documentation of this file.

/*
 * Copyright (c) 2023, Carnegie Mellon University
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted (subject to the limitations in the disclaimer
 * below) provided that the following conditions are met:
 *
 *      * Redistributions of source code must retain the above copyright notice,
 *      this list of conditions and the following disclaimer.
 *
 *      * Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *
 *      * Neither the name of the copyright holder nor the names of its
 *      contributors may be used to endorse or promote products derived from
 *      this software without specific prior written permission.
 *
 * NO EXPRESS OR IMPLIED LICENSES TO ANY PARTY'S PATENT RIGHTS ARE GRANTED BY
 * THIS LICENSE. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
 * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT
 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
 
#ifndef SOFTWARE_INCLUDE_ENSO_PIPE_H_
#define SOFTWARE_INCLUDE_ENSO_PIPE_H_
 
#include <enso/consts.h>
#include <enso/helpers.h>
#include <enso/internals.h>
 
#include <array>
#include <cassert>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <vector>
 
namespace enso {
 
class RxPipe;
class TxPipe;
class RxTxPipe;
 
class PktIterator;
class PeekPktIterator;
 
uint32_t external_peek_next_batch_from_queue(
    struct RxEnsoPipeInternal* enso_pipe,
    struct NotificationBufPair* notification_buf_pair, void** buf);
 
class Device {
 public:
  static std::unique_ptr<Device> Create(
      const std::string& pcie_addr = "",
      const std::string& huge_page_prefix = "") noexcept;
 
  Device(const Device&) = delete;
  Device& operator=(const Device&) = delete;
  Device(Device&&) = delete;
  Device& operator=(Device&&) = delete;
 
  ~Device();
 
  RxPipe* AllocateRxPipe(bool fallback = false) noexcept;
 
  TxPipe* AllocateTxPipe(uint8_t* buf = nullptr) noexcept;
 
  int GetNbFallbackQueues() noexcept;
 
  RxTxPipe* AllocateRxTxPipe(bool fallback = false) noexcept;
 
  RxPipe* NextRxPipeToRecv();
 
  RxTxPipe* NextRxTxPipeToRecv();
 
  void ProcessCompletions();
 
  int EnableTimeStamping(uint8_t offset = kDefaultRttOffset);
 
  int DisableTimeStamping();
 
  int EnableRateLimiting(uint16_t num, uint16_t den);
 
  int DisableRateLimiting();
 
  int EnableRoundRobin();
 
  int DisableRoundRobin();
 
  int GetRoundRobinStatus() noexcept;
 
  int ApplyConfig(struct TxNotification* config_notification);
 
 private:
  struct TxPendingRequest {
    uint32_t pipe_id;
    uint32_t nb_bytes;
  };
 
  Device(const std::string& pcie_addr, std::string huge_page_prefix) noexcept
      : kPcieAddr(pcie_addr) {
#ifndef NDEBUG
    std::cerr << "Warning: assertions are enabled. Performance may be affected."
              << std::endl;
#endif  // NDEBUG
    if (huge_page_prefix == "") {
      huge_page_prefix = std::string(kHugePageDefaultPrefix);
    }
    huge_page_prefix_ = huge_page_prefix;
  }
 
  int Init() noexcept;
 
  void Send(uint32_t tx_enso_pipe_id, uint64_t phys_addr, uint32_t nb_bytes);
 
  friend class RxPipe;
  friend class TxPipe;
  friend class RxTxPipe;
 
  const std::string kPcieAddr;
 
  struct NotificationBufPair notification_buf_pair_;
  int16_t core_id_;
  uint16_t bdf_;
  std::string huge_page_prefix_;
 
  std::vector<RxPipe*> rx_pipes_;
  std::vector<TxPipe*> tx_pipes_;
  std::vector<RxTxPipe*> rx_tx_pipes_;
 
  std::array<RxPipe*, kMaxNbFlows> rx_pipes_map_ = {};
  std::array<RxTxPipe*, kMaxNbFlows> rx_tx_pipes_map_ = {};
 
  int32_t next_pipe_id_ = -1;
 
  uint32_t tx_pr_head_ = 0;
  uint32_t tx_pr_tail_ = 0;
  std::array<TxPendingRequest, kMaxPendingTxRequests + 1> tx_pending_requests_;
  static constexpr uint32_t kPendingTxRequestsBufMask = kMaxPendingTxRequests;
  static_assert((kMaxPendingTxRequests & (kMaxPendingTxRequests + 1)) == 0,
                "kMaxPendingTxRequests + 1 must be a power of 2");
};
 
class RxPipe {
 public:
  template <typename T>
  class MessageBatch {
   public:
    constexpr MessageBatch() : MessageBatch(nullptr, 0, 0, nullptr) {}
 
    MessageBatch(const MessageBatch&) = default;
    MessageBatch(MessageBatch&&) = default;
 
    MessageBatch& operator=(const MessageBatch&) = default;
    MessageBatch& operator=(MessageBatch&&) = default;
 
    constexpr T begin() { return T(buf_, message_limit_, this); }
    constexpr T begin() const { return T(buf_, message_limit_, this); }
 
    constexpr T end() {
      return T(buf_ + available_bytes_, message_limit_, this);
    }
    constexpr T end() const {
      return T(buf_ + available_bytes_, message_limit_, this);
    }
 
    uint32_t processed_bytes() const { return processed_bytes_; }
 
    inline void NotifyProcessedBytes(uint32_t nb_bytes) {
      processed_bytes_ += nb_bytes;
    }
 
    uint32_t available_bytes() const { return available_bytes_; }
 
    int32_t message_limit() const { return message_limit_; }
 
    uint8_t* buf() const { return buf_; }
 
   private:
    constexpr MessageBatch(uint8_t* buf, uint32_t available_bytes,
                           int32_t message_limit, RxPipe* pipe)
        : available_bytes_(available_bytes),
          message_limit_(message_limit),
          buf_(buf),
          pipe_(pipe) {}
 
    friend class RxPipe;
    friend T;
 
    uint32_t available_bytes_;
    int32_t message_limit_;
    uint8_t* buf_;
    uint32_t processed_bytes_ = 0;
    RxPipe* pipe_;
  };
 
  RxPipe(const RxPipe&) = delete;
  RxPipe& operator=(const RxPipe&) = delete;
  RxPipe(RxPipe&&) = delete;
  RxPipe& operator=(RxPipe&&) = delete;
 
  int Bind(uint16_t dst_port, uint16_t src_port, uint32_t dst_ip,
           uint32_t src_ip, uint32_t protocol);
 
  uint32_t Recv(uint8_t** buf, uint32_t max_nb_bytes);
 
  uint32_t Peek(uint8_t** buf, uint32_t max_nb_bytes);
 
  constexpr void ConfirmBytes(uint32_t nb_bytes) {
    uint32_t rx_tail = internal_rx_pipe_.rx_tail;
    rx_tail = (rx_tail + nb_bytes / 64) % kEnsoPipeSize;
    internal_rx_pipe_.rx_tail = rx_tail;
  }
 
  constexpr uint32_t capacity() const {
    uint32_t rx_head = internal_rx_pipe_.rx_head;
    uint32_t rx_tail = internal_rx_pipe_.rx_tail;
    return ((rx_head - rx_tail) % kEnsoPipeSize) * 64;
  }
 
  template <typename T>
  constexpr MessageBatch<T> RecvMessages(int32_t max_nb_messages = -1) {
    uint8_t* buf = nullptr;
    uint32_t recv = Peek(&buf, ~0);
    return MessageBatch<T>((uint8_t*)buf, recv, max_nb_messages, this);
  }
 
  inline MessageBatch<PktIterator> RecvPkts(int32_t max_nb_pkts = -1) {
    return RecvMessages<PktIterator>(max_nb_pkts);
  }
 
  inline MessageBatch<PeekPktIterator> PeekPkts(int32_t max_nb_pkts = -1) {
    return RecvMessages<PeekPktIterator>(max_nb_pkts);
  }
 
  void Prefetch();
 
  void Free(uint32_t nb_bytes);
 
  void Clear();
 
  inline uint8_t* buf() const { return (uint8_t*)internal_rx_pipe_.buf; }
 
  inline enso_pipe_id_t id() const { return id_; }
 
  inline void* context() const { return context_; }
 
  inline void set_context(void* new_context) { context_ = new_context; }
 
  static constexpr uint32_t kQuantumSize = 64;
 
  static constexpr uint32_t kMaxCapacity = kEnsoPipeSize * 64 - kQuantumSize;
 
 private:
  explicit RxPipe(Device* device) noexcept
      : notification_buf_pair_(&(device->notification_buf_pair_)) {}
 
  ~RxPipe();
 
  int Init(bool fallback) noexcept;
 
  void SetAsNextPipe() noexcept { next_pipe_ = true; }
 
  friend class Device;
 
  bool next_pipe_ = false;  
  enso_pipe_id_t id_;       
  void* context_;
  struct RxEnsoPipeInternal internal_rx_pipe_;
  struct NotificationBufPair* notification_buf_pair_;
};
 
class TxPipe {
 public:
  TxPipe(const TxPipe&) = delete;
  TxPipe& operator=(const TxPipe&) = delete;
  TxPipe(TxPipe&&) = delete;
  TxPipe& operator=(TxPipe&&) = delete;
 
  uint8_t* AllocateBuf(uint32_t target_capacity = 0) {
    ExtendBufToTarget(target_capacity);
    return buf_ + app_begin_;
  }
 
  inline void SendAndFree(uint32_t nb_bytes) {
    uint64_t phys_addr = buf_phys_addr_ + app_begin_;
    assert(nb_bytes <= kMaxCapacity);
    assert(nb_bytes / kQuantumSize * kQuantumSize == nb_bytes);
 
    app_begin_ = (app_begin_ + nb_bytes) & kBufMask;
 
    device_->Send(kId, phys_addr, nb_bytes);
  }
 
  inline uint32_t TryExtendBuf() {
    device_->ProcessCompletions();
    return capacity();
  }
 
  inline uint32_t ExtendBufToTarget(uint32_t target_capacity) {
    uint32_t _capacity = capacity();
    assert(target_capacity <= kMaxCapacity);
    while (_capacity < target_capacity) {
      _capacity = TryExtendBuf();
    }
    return _capacity;
  }
 
  inline uint32_t capacity() const {
    return (app_end_ - app_begin_ - 1) & kBufMask;
  }
 
  inline uint32_t pending_transmission() const {
    return kMaxCapacity - ((app_end_ - app_begin_) & kBufMask);
  }
 
  inline uint8_t* buf() const { return buf_; }
 
  inline enso_pipe_id_t id() const { return kId; }
 
  inline void* context() const { return context_; }
 
  inline void set_context(void* new_context) { context_ = new_context; }
 
  static constexpr uint32_t kQuantumSize = 64;
 
  static constexpr uint32_t kMaxCapacity = kEnsoPipeSize * 64 - kQuantumSize;
 
 private:
  explicit TxPipe(uint32_t id, Device* device, uint8_t* buf = nullptr) noexcept
      : kId(id), device_(device), buf_(buf), internal_buf_(buf == nullptr) {}
 
  ~TxPipe();
 
  int Init() noexcept;
 
  inline void NotifyCompletion(uint32_t nb_bytes) {
    app_end_ = (app_end_ + nb_bytes) & kBufMask;
  }
 
  inline std::string GetHugePageFilePath() const {
    return device_->huge_page_prefix_ + std::string(kHugePagePathPrefix) +
           std::to_string(kId);
  }
 
  friend class Device;
 
  const enso_pipe_id_t kId;  
  void* context_;
  Device* device_;
  uint8_t* buf_;
  bool internal_buf_;       // If true, the buffer is allocated internally.
  uint32_t app_begin_ = 0;  // The next byte to be sent.
  uint32_t app_end_ = 0;    // The next byte to be allocated.
  uint64_t buf_phys_addr_;
 
  static constexpr uint32_t kBufMask = (kMaxCapacity + kQuantumSize) - 1;
  static_assert((kBufMask & (kBufMask + 1)) == 0,
                "(kBufMask + 1) must be a power of 2");
 
  // Buffer layout:
  //                                     | app_begin_          | app_end_
  //                                     v                     v
  //    +---+------------------------+---+---------------------+--------+
  //    |   |  Waiting transmission  |   |  Available to user  |        |
  //    +---+------------------------+---+---------------------+--------+
  //        ^                        ^
  //        | hw_begin_              | hw_end_
  //
  // The area available to the user is between `app_begin_` and `app_end_`.
  // The area between `hw_begin_` and `hw_end_` is waiting to be transmitted.
  //
  // SendAndFree() will advance `app_begin_` by `nb_bytes_`, reducing the
  // available space to the user. The size of the available region can be
  // obtained by calling `capacity()`.
  //
  // To reclaim space, the user must call `Extend()`, which will check for
  // completions and potentially advance the `hw_begin_`. We can then advance
  // `app_end_` to match `hw_begin_`, increasing the available space to the
  // user.
  //
  // All the buffer pointers (i.e., app_begin_, app_end_, hw_begin_, hw_end_)
  // are in units of 64 bytes. The buffer itself is a circular buffer, so
  // `app_end_` can be smaller than `app_begin_`.
  //
  // Currently app_begin_ == hw_end_ and app_end_ == hw_begin_. But this may
  // change in the future. One reason it might be useful to change this is that
  // currently `Device::Send()` blocks when the notification buffer is full. If
  // we make sending non-blocking, it may be useful to let the hw_*_ and app_*_
  // pointers diverge.
};
 
class RxTxPipe {
 public:
  RxTxPipe(const RxTxPipe&) = delete;
  RxTxPipe& operator=(const RxTxPipe&) = delete;
  RxTxPipe(RxTxPipe&&) = delete;
  RxTxPipe& operator=(RxTxPipe&&) = delete;
 
  inline int Bind(uint16_t dst_port, uint16_t src_port, uint32_t dst_ip,
                  uint32_t src_ip, uint32_t protocol) {
    return rx_pipe_->Bind(dst_port, src_port, dst_ip, src_ip, protocol);
  }
 
  inline uint32_t Recv(uint8_t** buf, uint32_t max_nb_bytes) {
    device_->ProcessCompletions();
    return rx_pipe_->Recv(buf, max_nb_bytes);
  }
 
  inline uint32_t Peek(uint8_t** buf, uint32_t max_nb_bytes) {
    device_->ProcessCompletions();
    return rx_pipe_->Peek(buf, max_nb_bytes);
  }
 
  inline void ConfirmBytes(uint32_t nb_bytes) {
    rx_pipe_->ConfirmBytes(nb_bytes);
  }
 
  template <typename T>
  inline RxPipe::MessageBatch<T> RecvMessages(int32_t max_nb_messages = -1) {
    device_->ProcessCompletions();
    return rx_pipe_->RecvMessages<T>(max_nb_messages);
  }
 
  inline RxPipe::MessageBatch<PktIterator> RecvPkts(int32_t max_nb_pkts = -1) {
    device_->ProcessCompletions();
    return rx_pipe_->RecvPkts(max_nb_pkts);
  }
 
  inline RxPipe::MessageBatch<PeekPktIterator> PeekPkts(
      int32_t max_nb_pkts = -1) {
    device_->ProcessCompletions();
    return rx_pipe_->PeekPkts(max_nb_pkts);
  }
 
  inline void Prefetch() { rx_pipe_->Prefetch(); }
 
  inline void SendAndFree(uint32_t nb_bytes) {
    tx_pipe_->SendAndFree(nb_bytes);
    last_tx_pipe_capacity_ -= nb_bytes;
  }
 
  inline void ProcessCompletions() {
    uint32_t new_capacity = tx_pipe_->capacity();
 
    // If the capacity has increased, we need to free up space in the RX pipe.
    if (new_capacity > last_tx_pipe_capacity_) {
      rx_pipe_->Free(new_capacity - last_tx_pipe_capacity_);
      last_tx_pipe_capacity_ = new_capacity;
    }
  }
 
  inline uint8_t* buf() const { return rx_pipe_->buf(); }
 
  inline enso_pipe_id_t rx_id() const { return rx_pipe_->id(); }
 
  inline enso_pipe_id_t tx_id() const { return tx_pipe_->id(); }
 
  inline void* context() const { return rx_pipe_->context(); }
 
  inline void set_context(void* new_context) {
    rx_pipe_->set_context(new_context);
  }
 
  static constexpr uint32_t kQuantumSize = RxPipe::kQuantumSize;
 
  static constexpr uint32_t kMaxCapacity = RxPipe::kMaxCapacity;
 
  static_assert(RxPipe::kQuantumSize == TxPipe::kQuantumSize,
                "Quantum size mismatch");
  static_assert(RxPipe::kMaxCapacity == TxPipe::kMaxCapacity,
                "Max capacity mismatch");
 
 private:
  static constexpr uint32_t kCompletionsThreshold = kEnsoPipeSize * 64 / 2;
 
  explicit RxTxPipe(Device* device) noexcept : device_(device) {}
 
  ~RxTxPipe() = default;
 
  int Init(bool fallback) noexcept;
 
  friend class Device;
 
  Device* device_;
  RxPipe* rx_pipe_;
  TxPipe* tx_pipe_;
  uint32_t last_tx_pipe_capacity_;
};
 
template <typename T>
class MessageIteratorBase {
 public:
  constexpr uint8_t* operator*() { return addr_; }
 
  /*
   * This is a bit ugly but, as far as I know, there is no way to check for the
   * end of a range-based for loop without overloading the != operator.
   *
   * The issue is that it does not work as expected for actual inequality check.
   */
  constexpr bool operator!=(const MessageIteratorBase& other) const {
    return (missing_messages_ != 0) && (next_addr_ <= other.addr_);
  }
 
  constexpr T& operator++() {
    T* child = static_cast<T*>(this);
 
    uint32_t nb_bytes = next_addr_ - addr_;
 
    child->OnAdvanceMessage(nb_bytes);
 
    addr_ = next_addr_;
    next_addr_ = child->GetNextMessage(addr_);
 
    --missing_messages_;
    batch_->NotifyProcessedBytes(nb_bytes);
 
    return *child;
  }
 
 protected:
  inline MessageIteratorBase()
      : addr_(nullptr),
        missing_messages_(0),
        batch_(nullptr),
        next_addr_(nullptr) {}
 
  MessageIteratorBase(const MessageIteratorBase&) = default;
  MessageIteratorBase& operator=(const MessageIteratorBase&) = default;
  MessageIteratorBase(MessageIteratorBase&&) = default;
  MessageIteratorBase& operator=(MessageIteratorBase&&) = default;
 
  inline MessageIteratorBase(uint8_t* addr, int32_t message_limit,
                             RxPipe::MessageBatch<T>* batch)
      : addr_(addr),
        missing_messages_(message_limit),
        batch_(batch),
        next_addr_(static_cast<T*>(this)->GetNextMessage(addr)) {}
 
  uint8_t* addr_;
  int32_t missing_messages_;
  RxPipe::MessageBatch<T>* batch_;
  uint8_t* next_addr_;
};
 
class PktIterator : public MessageIteratorBase<PktIterator> {
 public:
  inline PktIterator() : MessageIteratorBase() {}
 
  inline PktIterator(uint8_t* addr, int32_t message_limit,
                     RxPipe::MessageBatch<PktIterator>* batch)
      : MessageIteratorBase(addr, message_limit, batch) {}
 
  PktIterator(const PktIterator&) = default;
  PktIterator& operator=(const PktIterator&) = default;
  PktIterator(PktIterator&&) = default;
  PktIterator& operator=(PktIterator&&) = default;
 
  _enso_always_inline uint8_t* GetNextMessage(uint8_t* current_message) {
    return get_next_pkt(current_message);
  }
 
  constexpr void OnAdvanceMessage(uint32_t nb_bytes) {
    batch_->pipe_->ConfirmBytes(nb_bytes);
  }
};
 
class PeekPktIterator : public MessageIteratorBase<PeekPktIterator> {
 public:
  inline PeekPktIterator(uint8_t* addr, int32_t message_limit,
                         RxPipe::MessageBatch<PeekPktIterator>* batch)
      : MessageIteratorBase(addr, message_limit, batch) {}
 
  _enso_always_inline uint8_t* GetNextMessage(uint8_t* current_message) {
    return get_next_pkt(current_message);
  }
 
  constexpr void OnAdvanceMessage([[maybe_unused]] uint32_t nb_bytes) {}
};
 
}  // namespace enso
 
#endif  // SOFTWARE_INCLUDE_ENSO_PIPE_H_