uint64_t Slice::prepend(const void* data, uint64_t size) {

  const uint8_t* src = static_cast<const uint8_t*>(data);

  uint64_t copy_size;

  if (dataSize() == 0) {

    copy_size = std::min(size, reservableSize());

    reservable_ = capacity_;

    data_ = capacity_ - copy_size;

  } else {

    if (data_ == 0) {

      return 0;

    }

    copy_size = std::min(size, data_);

    data_ -= copy_size;

  }

  memcpy(base_ + data_, src + size - copy_size, copy_size); // NOLINT(safe-memcpy)

  return copy_size;

}

void OwnedImpl::addImpl(const void* data, uint64_t size) {

  const char* src = static_cast<const char*>(data);

  bool new_slice_needed = slices_.empty();

  while (size != 0) {

    if (new_slice_needed) {

      slices_.emplace_back(Slice(size, account_));

    }

    uint64_t copy_size = slices_.back().append(src, size);

    src += copy_size;

    size -= copy_size;

    length_ += copy_size;

    new_slice_needed = true;

  }

}

void OwnedImpl::addDrainTracker(std::function<void()> drain_tracker) {

  ASSERT(!slices_.empty());

  slices_.back().addDrainTracker(std::move(drain_tracker));

}

void OwnedImpl::bindAccount(BufferMemoryAccountSharedPtr account) {

  ASSERT(slices_.empty());

  account_ = std::move(account);

}

BufferMemoryAccountSharedPtr OwnedImpl::getAccountForTest() { return account_; }

void OwnedImpl::add(const void* data, uint64_t size) { addImpl(data, size); }

void OwnedImpl::addBufferFragment(BufferFragment& fragment) {

  length_ += fragment.size();

  slices_.emplace_back(fragment);

}

void OwnedImpl::add(absl::string_view data) { add(data.data(), data.size()); }

void OwnedImpl::add(const Instance& data) {

  ASSERT(&data != this);

  for (const RawSlice& slice : data.getRawSlices()) {

    add(slice.mem_, slice.len_);

  }

}

void OwnedImpl::prepend(absl::string_view data) {

  uint64_t size = data.size();

  bool new_slice_needed = slices_.empty();

  while (size != 0) {

    if (new_slice_needed) {

      slices_.emplace_front(Slice(size, account_));

    }

    uint64_t copy_size = slices_.front().prepend(data.data(), size);

    size -= copy_size;

    length_ += copy_size;

    new_slice_needed = true;

  }

}

void OwnedImpl::prepend(Instance& data) {

  ASSERT(&data != this);

  OwnedImpl& other = static_cast<OwnedImpl&>(data);

  while (!other.slices_.empty()) {

    uint64_t slice_size = other.slices_.back().dataSize();

    length_ += slice_size;

    slices_.emplace_front(std::move(other.slices_.back()));

    slices_.front().maybeChargeAccount(account_);

    other.slices_.pop_back();

    other.length_ -= slice_size;

  }

  other.postProcess();

}

void OwnedImpl::copyOut(size_t start, uint64_t size, void* data) const {

  uint64_t bytes_to_skip = start;

  uint8_t* dest = static_cast<uint8_t*>(data);

  for (const auto& slice : slices_) {

    if (size == 0) {

      break;

    }

    uint64_t data_size = slice.dataSize();

    if (data_size <= bytes_to_skip) {

      bytes_to_skip -= data_size;

      continue;

    }

    uint64_t copy_size = std::min(size, data_size - bytes_to_skip);

    memcpy(dest, slice.data() + bytes_to_skip, copy_size); // NOLINT(safe-memcpy)

    size -= copy_size;

    dest += copy_size;

    bytes_to_skip = 0;

  }

  ASSERT(size == 0);

}

                                    uint64_t num_slice) const {

  uint64_t total_length_to_read = std::min(size, this->length());

  uint64_t num_bytes_read = 0;

  uint64_t num_dest_slices_read = 0;

  uint64_t num_src_slices_read = 0;

  uint64_t dest_slice_offset = 0;

  uint64_t src_slice_offset = 0;

  while (num_dest_slices_read < num_slice && num_bytes_read < total_length_to_read) {

    const Slice& src_slice = slices_[num_src_slices_read];

    const Buffer::RawSlice& dest_slice = dest_slices[num_dest_slices_read];

    uint64_t left_to_read = total_length_to_read - num_bytes_read;

    uint64_t left_data_size_in_dst_slice = dest_slice.len_ - dest_slice_offset;

    uint64_t left_data_size_in_src_slice = src_slice.dataSize() - src_slice_offset;

    uint64_t length_to_copy =

        std::min(left_data_size_in_src_slice, std::min(left_data_size_in_dst_slice, left_to_read));

    memcpy(static_cast<uint8_t*>(dest_slice.mem_) + dest_slice_offset, // NOLINT(safe-memcpy)

           src_slice.data() + src_slice_offset, length_to_copy);

    src_slice_offset = src_slice_offset + length_to_copy;

    dest_slice_offset = dest_slice_offset + length_to_copy;

    if (src_slice_offset == src_slice.dataSize()) {

      num_src_slices_read++;

      src_slice_offset = 0;

    }

    if (dest_slice_offset == dest_slice.len_) {

      num_dest_slices_read++;

      dest_slice_offset = 0;

    }

    ASSERT(src_slice_offset <= src_slice.dataSize());

    ASSERT(dest_slice_offset <= dest_slice.len_);

    num_bytes_read += length_to_copy;

  }

  return num_bytes_read;

}

void OwnedImpl::drain(uint64_t size) { drainImpl(size); }

void OwnedImpl::drainImpl(uint64_t size) {

  while (size != 0) {

    if (slices_.empty()) {

      break;

    }

    uint64_t slice_size = slices_.front().dataSize();

    if (slice_size <= size) {

      slices_.pop_front();

      length_ -= slice_size;

      size -= slice_size;

    } else {

      slices_.front().drain(size);

      length_ -= size;

      size = 0;

    }

  }

  while (!slices_.empty() && slices_.front().dataSize() == 0) {

    slices_.pop_front();

  }

}

RawSliceVector OwnedImpl::getRawSlices(absl::optional<uint64_t> max_slices) const {

  uint64_t max_out = slices_.size();

  if (max_slices.has_value()) {

    max_out = std::min(max_out, max_slices.value());

  }

  RawSliceVector raw_slices;

  raw_slices.reserve(max_out);

  for (const auto& slice : slices_) {

    if (raw_slices.size() >= max_out) {

      break;

    }

    if (slice.dataSize() == 0) {

      continue;

    }

    raw_slices.emplace_back(

        RawSlice{const_cast<uint8_t*>(slice.data()), static_cast<size_t>(slice.dataSize())});

  }

  return raw_slices;

}

RawSlice OwnedImpl::frontSlice() const {

  for (const auto& slice : slices_) {

    if (slice.dataSize() > 0) {

      return RawSlice{const_cast<uint8_t*>(slice.data()),

                      static_cast<absl::Span<uint8_t>::size_type>(slice.dataSize())};

    }

  }

  return {nullptr, 0};

}

SliceDataPtr OwnedImpl::extractMutableFrontSlice() {

  RELEASE_ASSERT(length_ > 0, "Extract called on empty buffer");

  while (!slices_.empty() && slices_.front().dataSize() == 0) {

    slices_.pop_front();

  }

  ASSERT(!slices_.empty());

  auto slice = std::move(slices_.front());

  auto size = slice.dataSize();

  length_ -= size;

  slices_.pop_front();

  if (!slice.isMutable()) {

    Slice mutable_slice{size, nullptr};

    auto copy_size = mutable_slice.append(slice.data(), size);

    ASSERT(copy_size == size);

    return std::make_unique<SliceDataImpl>(std::move(mutable_slice));

  } else {

    slice.callAndClearDrainTrackersAndCharges();

    return std::make_unique<SliceDataImpl>(std::move(slice));

  }

}

uint64_t OwnedImpl::length() const {

  return length_;

}

void* OwnedImpl::linearize(uint32_t size) {

  RELEASE_ASSERT(size <= length(), "Linearize size exceeds buffer size");

  if (slices_.empty()) {

    return nullptr;

  }

  if (slices_[0].dataSize() < size) {

    Slice new_slice{size, account_};

    Slice::Reservation reservation = new_slice.reserve(size);

    ASSERT(reservation.mem_ != nullptr);

    ASSERT(reservation.len_ == size);

    copyOut(0, size, reservation.mem_);

    new_slice.commit(reservation);

    drainImpl(size);

    slices_.emplace_front(std::move(new_slice));

    length_ += size;

  }

  return slices_.front().data();

}

void OwnedImpl::coalesceOrAddSlice(Slice&& other_slice) {

  const uint64_t slice_size = other_slice.dataSize();

  if (other_slice.canCoalesce() && !slices_.empty() && slice_size < CopyThreshold &&

      slices_.back().reservableSize() >= slice_size) {

    addImpl(other_slice.data(), slice_size);

    other_slice.transferDrainTrackersTo(slices_.back());

  } else {

    other_slice.maybeChargeAccount(account_);

    slices_.emplace_back(std::move(other_slice));

    length_ += slice_size;

  }

}

void OwnedImpl::move(Instance& rhs) {

  ASSERT(&rhs != this);

  OwnedImpl& other = static_cast<OwnedImpl&>(rhs);

  while (!other.slices_.empty()) {

    const uint64_t slice_size = other.slices_.front().dataSize();

    coalesceOrAddSlice(std::move(other.slices_.front()));

    other.length_ -= slice_size;

    other.slices_.pop_front();

  }

  other.postProcess();

}

void OwnedImpl::move(Instance& rhs, uint64_t length) {

  move(rhs, length, /*reset_drain_trackers_and_accounting=*/false);

}

void OwnedImpl::move(Instance& rhs, uint64_t length, bool reset_drain_trackers_and_accounting) {

  ASSERT(&rhs != this);

  OwnedImpl& other = static_cast<OwnedImpl&>(rhs);

  while (length != 0 && !other.slices_.empty()) {

    const uint64_t slice_size = other.slices_.front().dataSize();

    const uint64_t copy_size = std::min(slice_size, length);

    if (copy_size == 0) {

    } else if (copy_size < slice_size) {

      add(other.slices_.front().data(), copy_size);

      other.slices_.front().drain(copy_size);

      other.length_ -= copy_size;

    } else {

      if (reset_drain_trackers_and_accounting) {

        other.slices_.front().callAndClearDrainTrackersAndCharges();

      }

      coalesceOrAddSlice(std::move(other.slices_.front()));

      other.slices_.pop_front();

      other.length_ -= slice_size;

    }

    length -= copy_size;

  }

  other.postProcess();

}

Reservation OwnedImpl::reserveForRead() {

  return reserveWithMaxLength(default_read_reservation_size_);

}

Reservation OwnedImpl::reserveWithMaxLength(uint64_t max_length) {

  Reservation reservation = Reservation::bufferImplUseOnlyConstruct(*this);

  if (max_length == 0) {

  while (!slices_.empty() && slices_.back().dataSize() == 0) {

  uint64_t bytes_remaining = max_length;

  uint64_t reserved = 0;

  auto& reservation_slices = reservation.bufferImplUseOnlySlices();

  auto slices_owner = std::make_unique<OwnedImplReservationSlicesOwnerMultiple>();

  uint64_t reservable_size = slices_.empty() ? 0 : slices_.back().reservableSize();

  if (reservable_size >= max_length || reservable_size >= (Slice::default_slice_size_ / 8)) {

    uint64_t reserve_size = std::min(reservable_size, bytes_remaining);

    RawSlice slice = slices_.back().reserve(reserve_size);

    reservation_slices.push_back(slice);

    slices_owner->owned_storages_.push_back({});

    bytes_remaining -= slice.len_;

    reserved += slice.len_;

  }

  while (bytes_remaining != 0 && reservation_slices.size() < reservation.MAX_SLICES_) {

    constexpr uint64_t size = Slice::default_slice_size_;

    if (size > bytes_remaining && reserved >= size) {

      break;

    }

    Slice::SizedStorage storage = slices_owner->newStorage();

    ASSERT(storage.len_ == size);

    const RawSlice raw_slice{storage.mem_.get(), size};

    slices_owner->owned_storages_.emplace_back(std::move(storage));

    reservation_slices.push_back(raw_slice);

    bytes_remaining -= std::min<uint64_t>(raw_slice.len_, bytes_remaining);

    reserved += raw_slice.len_;

  }

  ASSERT(reservation_slices.size() == slices_owner->owned_storages_.size());

  reservation.bufferImplUseOnlySlicesOwner() = std::move(slices_owner);

  reservation.bufferImplUseOnlySetLength(reserved);

  return reservation;

}

ReservationSingleSlice OwnedImpl::reserveSingleSlice(uint64_t length, bool separate_slice) {

  ReservationSingleSlice reservation = ReservationSingleSlice::bufferImplUseOnlyConstruct(*this);

  if (length == 0) {

    return reservation;

  }

  while (!slices_.empty() && slices_.back().dataSize() == 0) {

    slices_.pop_back();

  }

  auto& reservation_slice = reservation.bufferImplUseOnlySlice();

  auto slice_owner = std::make_unique<OwnedImplReservationSlicesOwnerSingle>();

  uint64_t reservable_size =

      (separate_slice || slices_.empty()) ? 0 : slices_.back().reservableSize();

  if (reservable_size >= length) {

    reservation_slice = slices_.back().reserve(length);

  } else {

    slice_owner->owned_storage_ = Slice::newStorage(length);

    ASSERT(slice_owner->owned_storage_.len_ >= length);

    reservation_slice = {slice_owner->owned_storage_.mem_.get(), static_cast<size_t>(length)};

  }

  reservation.bufferImplUseOnlySliceOwner() = std::move(slice_owner);

  return reservation;

}

                       ReservationSlicesOwnerPtr slices_owner_base) {

  if (length == 0) {

    return;

  }

  ASSERT(dynamic_cast<OwnedImplReservationSlicesOwner*>(slices_owner_base.get()) != nullptr);

  std::unique_ptr<OwnedImplReservationSlicesOwner> slices_owner(

      static_cast<OwnedImplReservationSlicesOwner*>(slices_owner_base.release()));

  absl::Span<Slice::SizedStorage> owned_storages = slices_owner->ownedStorages();

  ASSERT(slices.size() == owned_storages.size());

  uint64_t bytes_remaining = length;

  for (uint32_t i = 0; i < slices.size() && bytes_remaining > 0; i++) {

    slices[i].len_ = std::min<uint64_t>(slices[i].len_, bytes_remaining);

    if (auto& owned_storage = owned_storages[i]; owned_storage.mem_ != nullptr) {

      ASSERT(slices[i].len_ <= owned_storage.len_);

      slices_.emplace_back(Slice(std::move(owned_storage), slices[i].len_, account_));

    } else {

      bool success = slices_.back().commit<false>(slices[i]);

      ASSERT(success);

    }

    length_ += slices[i].len_;

    bytes_remaining -= slices[i].len_;

  }

}

ssize_t OwnedImpl::search(const void* data, uint64_t size, size_t start, size_t length) const {

  if (size == 0) {

    return (start <= length_) ? start : -1;

  }

  size_t left_to_search = length;

  if (0 == length) {

    left_to_search = length_ - start;

  }

  ssize_t offset = 0;

  const uint8_t* needle = static_cast<const uint8_t*>(data);

  for (size_t slice_index = 0; slice_index < slices_.size() && (left_to_search > 0);

       slice_index++) {

    const auto& slice = slices_[slice_index];

    uint64_t slice_size = slice.dataSize();

    if (slice_size <= start) {

      start -= slice_size;

      offset += slice_size;

      continue;

    }

    const uint8_t* slice_start = slice.data();

    const uint8_t* haystack = slice_start;

    const uint8_t* haystack_end = haystack + slice_size;

    haystack += start;

    while (haystack < haystack_end) {

      const size_t slice_search_limit =

          std::min(static_cast<size_t>(haystack_end - haystack), left_to_search);

      const uint8_t* first_byte_match =

          static_cast<const uint8_t*>(memchr(haystack, needle[0], slice_search_limit));

      if (first_byte_match == nullptr) {

        left_to_search -= slice_search_limit;

        break;

      }

      left_to_search -= static_cast<size_t>(first_byte_match - haystack + 1);

      const size_t saved_left_to_search = left_to_search;

      size_t i = 1;

      size_t match_index = slice_index;

      const uint8_t* match_next = first_byte_match + 1;

      const uint8_t* match_end = haystack_end;

      while ((i < size) && (0 < left_to_search)) {

        if (match_next >= match_end) {

          match_index++;

          if (match_index == slices_.size()) {

            break;

          }

          const auto& match_slice = slices_[match_index];

          match_next = match_slice.data();

          match_end = match_next + match_slice.dataSize();

          continue;

        }

        left_to_search--;

        if (*match_next++ != needle[i]) {

          break;

        }

        i++;

      }

      if (i == size) {

        return offset + (first_byte_match - slice_start);

      }

      haystack = first_byte_match + 1;

      left_to_search = saved_left_to_search;

    }

    start = 0;

    offset += slice_size;

  }

  return -1;

}

bool OwnedImpl::startsWith(absl::string_view data) const {

  if (length() < data.length()) {

    return false;

  }

  if (data.length() == 0) {

    return true;

  }

  const uint8_t* prefix = reinterpret_cast<const uint8_t*>(data.data());

  size_t size = data.length();

  for (const auto& slice : slices_) {

    uint64_t slice_size = slice.dataSize();

    const uint8_t* slice_start = slice.data();

    if (slice_size >= size) {

      return memcmp(prefix, slice_start, size) == 0;

    }

    if (memcmp(prefix, slice_start, slice_size) != 0) {

    prefix += slice_size;

    size -= slice_size;

  }

}

OwnedImpl::OwnedImpl() = default;

OwnedImpl::OwnedImpl(absl::string_view data) : OwnedImpl() { add(data); }

OwnedImpl::OwnedImpl(const Instance& data) : OwnedImpl() { add(data); }

OwnedImpl::OwnedImpl(const void* data, uint64_t size) : OwnedImpl() { add(data, size); }

OwnedImpl::OwnedImpl(BufferMemoryAccountSharedPtr account) : account_(std::move(account)) {}

std::string OwnedImpl::toString() const {

  std::string output;

  output.reserve(length());

  for (const RawSlice& slice : getRawSlices()) {

    output.append(static_cast<const char*>(slice.mem_), slice.len_);

  }

  return output;

}

void OwnedImpl::postProcess() {}

void OwnedImpl::appendSliceForTest(const void* data, uint64_t size) {

  slices_.emplace_back(Slice(size, account_));

  slices_.back().append(data, size);

  length_ += size;

}

void OwnedImpl::appendSliceForTest(absl::string_view data) {

  appendSliceForTest(data.data(), data.size());

}

std::vector<Slice::SliceRepresentation> OwnedImpl::describeSlicesForTest() const {

  std::vector<Slice::SliceRepresentation> slices;

  for (const auto& slice : slices_) {

    slices.push_back(slice.describeSliceForTest());

  }

  return slices;

}

size_t OwnedImpl::addFragments(absl::Span<const absl::string_view> fragments) {

  size_t total_size_to_copy = 0;

  for (const auto& fragment : fragments) {

    total_size_to_copy += fragment.size();

  }

  if (slices_.empty()) {

    slices_.emplace_back(Slice(total_size_to_copy, account_));

  }

  Slice& back = slices_.back();

  Slice::Reservation reservation = back.reserve(total_size_to_copy);

  uint8_t* mem = static_cast<uint8_t*>(reservation.mem_);

  if (reservation.len_ == total_size_to_copy) {

    for (const auto& fragment : fragments) {

      memcpy(mem, fragment.data(), fragment.size()); // NOLINT(safe-memcpy)

      mem += fragment.size();

    }

    back.commit<false>(reservation);

    length_ += total_size_to_copy;

  } else {

    size_t fragment_index = 0;

    uint64_t bytes_written_to_reservation = 0;

    while (fragment_index < fragments.size() &&

           bytes_written_to_reservation + fragments[fragment_index].size() <= reservation.len_) {

      const auto& fragment = fragments[fragment_index];

      memcpy(mem, fragment.data(), fragment.size()); // NOLINT(safe-memcpy)

      mem += fragment.size();

      bytes_written_to_reservation += fragment.size();

      fragment_index++;

    }

    if (bytes_written_to_reservation > 0) {

      back.commit<false>({reservation.mem_, bytes_written_to_reservation});

      length_ += bytes_written_to_reservation;

    }

    if (fragment_index < fragments.size()) {

      size_t remaining_size = 0;

      for (size_t i = fragment_index; i < fragments.size(); i++) {

        remaining_size += fragments[i].size();

      }

      slices_.emplace_back(Slice(remaining_size, account_));

      Slice& new_slice = slices_.back();

      Slice::Reservation new_reservation = new_slice.reserve(remaining_size);

      ASSERT(new_reservation.len_ == remaining_size);

      uint8_t* new_mem = static_cast<uint8_t*>(new_reservation.mem_);

      for (size_t i = fragment_index; i < fragments.size(); i++) {

        memcpy(new_mem, fragments[i].data(), fragments[i].size()); // NOLINT(safe-memcpy)

        new_mem += fragments[i].size();

      }

      new_slice.commit<false>(new_reservation);

      length_ += remaining_size;

    }

  }

  return total_size_to_copy;

}