// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/storage/lib/buffer/ring_buffer.h"

#include <zircon/status.h>

#include <algorithm>
#include <utility>
#include <vector>

namespace storage {

zx_status_t internal::RingBufferState::Reserve(uint64_t blocks, RingBufferReservation* out) {
  if (blocks == 0) {
    return ZX_ERR_INVALID_ARGS;
  }
  size_t destination_offset = 0;
  {
    std::lock_guard lock(lock_);
    if (!IsSpaceAvailableLocked(blocks)) {
      return ZX_ERR_NO_SPACE;
    }
    destination_offset = (reserved_start_ + reserved_length_) % capacity();
    reserved_length_ += blocks;
  }
  *out = RingBufferReservation(this, destination_offset, blocks);
  return ZX_OK;
}

void internal::RingBufferState::Free(const RingBufferReservation& reservation) {
  std::lock_guard lock(lock_);
  ZX_DEBUG_ASSERT_MSG(reservation.length() <= reserved_length_,
                      "Attempting to free more blocks than available");

  // To perform optimally, the RingBuffer is expected to operate as a FIFO allocator,
  // where requests are freed in the same order they are allocated.
  // Under these conditions, reservations are always freed from the back
  // of the reserved portion of the buffer, and freeing is a simple update
  // of "start and length". However, if the reservations are not freed in this order,
  // they are put into a "pending list", where freeing the reservation is delayed
  // until we *can* release a reservation in-order.
  //
  // Since the common case of execution is in-order releasing, we use a simple vector for
  // tracking allocatable space. This performs poorly for a high volume of out-of-order
  // frees, but performs reasonably well when out-of-order operation freeing is relatively rare.

  if (reserved_start_ != reservation.start()) {
    // Freeing reservation out-of-order.
    //
    // Ensure "pending_free_" stays sorted by releasing order.
    for (auto it = pending_free_.begin(); it != pending_free_.end(); ++it) {
      // Underflow here is OK, so long as results are unsigned.
      static_assert(std::is_unsigned_v<decltype(reservation.start() - reserved_start_)> &&
                    std::is_unsigned_v<decltype(it->start - reserved_start_)>);
      if (reservation.start() - reserved_start_ < it->start - reserved_start_) {
        pending_free_.insert(it,
                             Range{.start = reservation.start(), .length = reservation.length()});
        return;
      }
    }
    pending_free_.push_back(Range{.start = reservation.start(), .length = reservation.length()});
    return;
  }

  CompleteFreeLocked(reservation.start(), reservation.length());

  while (!pending_free_.empty()) {
    // We have already ensured "pending_free_" is sorted by start index.
    //
    // This means that we can try releasing previously freed operations
    // in-order.
    if (pending_free_[0].start != reserved_start_) {
      return;
    }
    CompleteFreeLocked(pending_free_[0].start, pending_free_[0].length);
    pending_free_.erase(pending_free_.begin());
  }
}

bool internal::RingBufferState::IsSpaceAvailableLocked(size_t blocks) const {
  return reserved_length_ + blocks <= capacity();
}

void internal::RingBufferState::CompleteFreeLocked(size_t start, size_t blocks) {
  ZX_DEBUG_ASSERT_MSG(start == reserved_start_, "Freeing out-of-order");

  size_t decommit_blocks = std::min(blocks, capacity() - reserved_start_);
  ZX_ASSERT(buffer_.vmo().op_range(ZX_VMO_OP_DECOMMIT, reserved_start_ * buffer_.BlockSize(),
                                   decommit_blocks * buffer_.BlockSize(), nullptr, 0) == ZX_OK);

  if (decommit_blocks < blocks) {
    // Wrapping around the circular buffer.
    decommit_blocks = blocks - decommit_blocks;
    ZX_ASSERT(buffer_.vmo().op_range(ZX_VMO_OP_DECOMMIT, 0, decommit_blocks * buffer_.BlockSize(),
                                     nullptr, 0) == ZX_OK);
  }

  reserved_start_ = (reserved_start_ + blocks) % capacity();
  reserved_length_ -= blocks;
}

RingBufferReservation::RingBufferReservation(internal::RingBufferState* buffer, size_t start,
                                             size_t length)
    : buffer_(buffer), view_(buffer->buffer(), start, length) {}

RingBufferReservation::RingBufferReservation(RingBufferReservation&& other)
    : buffer_(other.buffer_), view_(other.view_) {
  other.buffer_ = nullptr;
  other.Reset();
  ZX_DEBUG_ASSERT(!other.Reserved());
}

RingBufferReservation& RingBufferReservation::operator=(RingBufferReservation&& other) {
  if (&other == this) {
    return *this;
  }
  Reset();
  buffer_ = other.buffer_;
  view_ = other.view_;
  other.buffer_ = nullptr;
  other.Reset();
  ZX_DEBUG_ASSERT(!other.Reserved());
  return *this;
}

RingBufferReservation::~RingBufferReservation() { Reset(); }

void RingBufferReservation::Reset() {
  if (Reserved()) {
    buffer_->Free(*this);
  }
  buffer_ = nullptr;
  view_ = BlockBufferView();
  ZX_DEBUG_ASSERT(!Reserved());
}

zx::result<size_t> RingBufferReservation::CopyRequests(
    cpp20::span<const storage::UnbufferedOperation> in_operations, size_t offset,
    std::vector<storage::BufferedOperation>* out_operations) {
  ZX_DEBUG_ASSERT_MSG(Reserved(), "Copying to invalid reservation");
  out_operations->reserve(out_operations->size() + in_operations.size());

  ZX_DEBUG_ASSERT_MSG(offset + BlockCount(in_operations) <= length(),
                      "Copying requests into a buffer beyond limit of prior reservation");

  const size_t capacity = buffer_->capacity();
  // Offset into this reservation.
  size_t reservation_offset = offset;
  // Offset into the target ring buffer.
  size_t ring_buffer_offset = (start() + reservation_offset) % capacity;
  size_t done = 0;

  for (const auto& in_operation : in_operations) {
    // Read parameters of the current request.
    ZX_DEBUG_ASSERT_MSG(in_operation.op.type == storage::OperationType::kWrite,
                        "RingBuffer only accepts write requests");
    size_t vmo_offset = in_operation.op.vmo_offset;
    size_t dev_offset = in_operation.op.dev_offset;
    const size_t vmo_len = in_operation.op.length;
    ZX_DEBUG_ASSERT_MSG(vmo_len > 0, "Attempting to buffer empty request");

    // Calculate the offset/length we will need to write into the buffer.
    // Avoid writing beyond the end of the RingBuffer.
    size_t buf_len = std::min(vmo_len, capacity - ring_buffer_offset);

    // Verify that the length is valid.
    ZX_DEBUG_ASSERT_MSG(buf_len > 0, "Attempting to write zero-length request into buffer");

    // Write data from the vmo into the buffer.
    void* ptr = Data(reservation_offset);

    const uint8_t* data = static_cast<const uint8_t*>(in_operation.data);
    const zx::unowned_vmo& vmo = in_operation.vmo;

    if (data != nullptr) {
      data += vmo_offset * buffer_->BlockSize();
      memcpy(ptr, data, buf_len * buffer_->BlockSize());
      data += buf_len * buffer_->BlockSize();
    } else {
      zx_status_t status =
          vmo->read(ptr, vmo_offset * buffer_->BlockSize(), buf_len * buffer_->BlockSize());
      if (status != ZX_OK) {
        return zx::error(status);
      }
    }

    storage::BufferedOperation out_op;
    out_op.vmoid = vmoid();
    out_op.op.type = in_operation.op.type;
    out_op.op.vmo_offset = ring_buffer_offset;
    out_op.op.dev_offset = dev_offset;
    out_op.op.length = buf_len;
    out_op.op.trace_flow_id = in_operation.op.trace_flow_id;
    out_operations->push_back(out_op);

    ring_buffer_offset = (ring_buffer_offset + buf_len) % capacity;
    reservation_offset += buf_len;

    if (buf_len != vmo_len) {
      // We wrapped around; write what remains from this request.
      vmo_offset += buf_len;
      dev_offset += buf_len;
      buf_len = vmo_len - buf_len;
      ZX_DEBUG_ASSERT(buf_len > 0);

      ptr = Data(reservation_offset);

      if (data) {
        memcpy(ptr, data, buf_len * buffer_->BlockSize());
      } else {
        if (zx_status_t status =
                vmo->read(ptr, vmo_offset * buffer_->BlockSize(), buf_len * buffer_->BlockSize());
            status != ZX_OK) {
          return zx::error(status);
        }
      }

      ring_buffer_offset = (ring_buffer_offset + buf_len) % capacity;
      reservation_offset += buf_len;

      // Insert the "new" request, which is the latter half of the last request
      storage::BufferedOperation out_op;
      out_op.vmoid = vmoid();
      out_op.op.type = in_operation.op.type;
      out_op.op.vmo_offset = 0;
      out_op.op.dev_offset = dev_offset;
      out_op.op.length = buf_len;
      out_op.op.trace_flow_id = in_operation.op.trace_flow_id;
      out_operations->push_back(out_op);
    }

    done += vmo_len;
  }

  return zx::ok(done);
}

vmoid_t RingBufferReservation::vmoid() const {
  ZX_DEBUG_ASSERT(Reserved());
  return view_.vmoid();
}

void* RingBufferReservation::Data(size_t index) {
  ZX_DEBUG_ASSERT(Reserved());
  return view_.Data(index);
}

const void* RingBufferReservation::Data(size_t index) const {
  ZX_DEBUG_ASSERT(Reserved());
  return view_.Data(index);
}

zx_status_t RingBuffer::Create(VmoidRegistry* vmoid_registry, size_t blocks, uint32_t block_size,
                               const char* label, std::unique_ptr<RingBuffer>* out) {
  VmoBuffer buffer;
  zx_status_t status = buffer.Initialize(vmoid_registry, blocks, block_size, label);
  if (status != ZX_OK) {
    return status;
  }

  *out = std::make_unique<RingBuffer>(std::move(buffer));
  return ZX_OK;
}

RingBufferRequests::RingBufferRequests(std::vector<storage::BufferedOperation> requests,
                                       RingBufferReservation reservation)
    : requests_(std::move(requests)), reservation_(std::move(reservation)) {}

}  // namespace storage