buffer: add method to extract front slice without copying

include/envoy/buffer/buffer.h

@@ -16,6 +16,7 @@
 #include "absl/container/inlined_vector.h"
 #include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
+#include "absl/types/span.h"
 
 namespace Envoy {
 namespace Buffer {
@@ -55,6 +56,30 @@ class BufferFragment {
   virtual void done() PURE;
 };
 
+/**
+ * A class to facilitate extracting buffer slices from a buffer instance.
+ */
+class SliceData {
+public:
+  virtual ~SliceData() = default;
+
+  /**
+   * @return true if the underlying slice data is mutable.
+   */
+  virtual bool isMutable() const PURE;
+  /**
+   * @return an immutable view of the slice data.
+   */
+  virtual absl::Span<const uint8_t> getData() const PURE;
+  /**
+   * @return a mutable view of the slice data, but only if isMutable() returns true.
+   * If isMutable() returns false then the mutable view will be {nullptr,0}.
+   */
+  virtual absl::Span<uint8_t> getMutableData() PURE;
+};
+
+using SliceDataPtr = std::unique_ptr<SliceData>;
+
 /**
  * A basic buffer abstraction.
  */
@@ -144,6 +169,14 @@ class Instance {
   virtual RawSliceVector
   getRawSlices(absl::optional<uint64_t> max_slices = absl::nullopt) const PURE;
 
+  /**
+   * Transfer ownership of the front slice to the caller. Must only be called if the
+   * buffer is not empty otherwise the implementation will have undefined behavior.
+   * @return pointer to SliceData object that wraps the front slice
+   */
+  virtual SliceDataPtr extractFrontSlice() PURE;
+  virtual SliceDataPtr extractMutableFrontSlice() PURE;
+
   /**
    * @return uint64_t the total length of the buffer (not necessarily contiguous in memory).
    */

source/common/buffer/buffer_impl.cc

@@ -199,6 +199,36 @@ RawSliceVector OwnedImpl::getRawSlices(absl::optional<uint64_t> max_slices) cons
   return raw_slices;
 }
 
+SliceDataPtr OwnedImpl::extractFrontSlice() {
+  RELEASE_ASSERT(length_ > 0, "Extract called on empty buffer");
+  // Remove zero byte fragments from the front of the queue to ensure
+  // that the extracted slice has data.
+  while (!slices_.empty() && slices_.front()->dataSize() == 0) {
+    slices_.pop_front();
+  }
+  ASSERT(!slices_.empty());
+  ASSERT(slices_.front());
+  auto slice = std::move(slices_.front());
+  length_ -= slice->dataSize();
+  slices_.pop_front();
+  return slice;
+}
+
+SliceDataPtr OwnedImpl::extractMutableFrontSlice() {
+  auto slice_data = extractFrontSlice();
+  if (!slice_data->isMutable()) {
+    // Create a mutable copy of the immutable slice data
+    auto immutable_data = slice_data->getData();
+    ASSERT(immutable_data.data() != nullptr);
+    auto mutable_slice = OwnedSlice::create(immutable_data.size());
+    auto copy_size = mutable_slice->append(immutable_data.data(), immutable_data.size());
+    ASSERT(copy_size == immutable_data.size());
+    return mutable_slice;
+  } else {
+    return slice_data;
+  }
+}
+
 uint64_t OwnedImpl::length() const {
 #ifndef NDEBUG
   // When running in debug mode, verify that the precomputed length matches the sum

source/common/buffer/buffer_impl.h

@@ -31,7 +31,7 @@ namespace Buffer {
  *                   |
  *                   data()
  */
-class Slice {
+class Slice : public SliceData {
 public:
   using Reservation = RawSlice;
 
@@ -41,6 +41,16 @@ class Slice {
     }
   }
 
+  // SliceData
+  bool isMutable() const override {
+    // Extracted slice data is immutable by default.
+    return false;
+  };
+  absl::Span<const uint8_t> getData() const override {
+    return {base_ + data_, reservable_ - data_};
+  };
+  absl::Span<uint8_t> getMutableData() override { return {nullptr, 0}; }
+
   /**
    * @return a pointer to the start of the usable content.
    */
@@ -117,10 +127,10 @@ class Slice {
    * @param reservation a reservation obtained from a previous call to reserve().
    *        If the reservation is not from this Slice, commit() will return false.
    *        If the caller is committing fewer bytes than provided by reserve(), it
-   *        should change the mem_ field of the reservation before calling commit().
+   *        should change the len_ field of the reservation before calling commit().
    *        For example, if a caller reserve()s 4KB to do a nonblocking socket read,
    *        and the read only returns two bytes, the caller should set
-   *        reservation.mem_ = 2 and then call `commit(reservation)`.
+   *        reservation.len_ = 2 and then call `commit(reservation)`.
    * @return whether the Reservation was successfully committed to the Slice.
    */
   bool commit(const Reservation& reservation) {
@@ -258,6 +268,10 @@ class OwnedSlice final : public Slice, public InlineStorage {
     return slice;
   }
 
+  // SliceData
+  bool isMutable() const override { return true; };
+  absl::Span<uint8_t> getMutableData() override { return {base_ + data_, reservable_ - data_}; }
+
 private:
   OwnedSlice(uint64_t size) : Slice(0, 0, size) { base_ = storage_; }
 
@@ -539,6 +553,8 @@ class OwnedImpl : public LibEventInstance {
   void copyOut(size_t start, uint64_t size, void* data) const override;
   void drain(uint64_t size) override;
   RawSliceVector getRawSlices(absl::optional<uint64_t> max_slices = absl::nullopt) const override;
+  SliceDataPtr extractFrontSlice() override;
+  SliceDataPtr extractMutableFrontSlice() override;
   uint64_t length() const override;
   void* linearize(uint32_t size) override;
   void move(Instance& rhs) override;
@@ -558,13 +574,13 @@ class OwnedImpl : public LibEventInstance {
    * @param data start of the content to copy.
    *
    */
-  void appendSliceForTest(const void* data, uint64_t size);
+  virtual void appendSliceForTest(const void* data, uint64_t size);
 
   /**
    * Create a new slice at the end of the buffer, and copy the supplied string into it.
    * @param data the string to append to the buffer.
    */
-  void appendSliceForTest(absl::string_view data);
+  virtual void appendSliceForTest(absl::string_view data);
 
   /**
    * Describe the in-memory representation of the slices in the buffer. For use

source/common/buffer/watermark_buffer.cc

@@ -51,6 +51,18 @@ void WatermarkBuffer::move(Instance& rhs, uint64_t length) {
   checkHighAndOverflowWatermarks();
 }
 
+SliceDataPtr WatermarkBuffer::extractFrontSlice() {
+  auto result = OwnedImpl::extractFrontSlice();
+  checkLowWatermark();
+  return result;
+}
+
+SliceDataPtr WatermarkBuffer::extractMutableFrontSlice() {
+  auto result = OwnedImpl::extractMutableFrontSlice();
+  checkLowWatermark();
+  return result;
+}
+
 Api::IoCallUint64Result WatermarkBuffer::read(Network::IoHandle& io_handle, uint64_t max_length) {
   Api::IoCallUint64Result result = OwnedImpl::read(io_handle, max_length);
   checkHighAndOverflowWatermarks();
@@ -69,6 +81,15 @@ Api::IoCallUint64Result WatermarkBuffer::write(Network::IoHandle& io_handle) {
   return result;
 }
 
+void WatermarkBuffer::appendSliceForTest(const void* data, uint64_t size) {
+  OwnedImpl::appendSliceForTest(data, size);
+  checkHighAndOverflowWatermarks();
+}
+
+void WatermarkBuffer::appendSliceForTest(absl::string_view data) {
+  appendSliceForTest(data.data(), data.size());
+}
+
 void WatermarkBuffer::setWatermarks(uint32_t low_watermark, uint32_t high_watermark) {
   ASSERT(low_watermark < high_watermark || (high_watermark == 0 && low_watermark == 0));
   uint32_t overflow_watermark_multiplier =

source/common/buffer/watermark_buffer.h

@@ -34,10 +34,14 @@ class WatermarkBuffer : public OwnedImpl {
   void drain(uint64_t size) override;
   void move(Instance& rhs) override;
   void move(Instance& rhs, uint64_t length) override;
+  SliceDataPtr extractFrontSlice() override;
+  SliceDataPtr extractMutableFrontSlice() override;
   Api::IoCallUint64Result read(Network::IoHandle& io_handle, uint64_t max_length) override;
   uint64_t reserve(uint64_t length, RawSlice* iovecs, uint64_t num_iovecs) override;
   Api::IoCallUint64Result write(Network::IoHandle& io_handle) override;
   void postProcess() override { checkLowWatermark(); }
+  void appendSliceForTest(const void* data, uint64_t size) override;
+  void appendSliceForTest(absl::string_view data) override;
 
   void setWatermarks(uint32_t watermark) { setWatermarks(watermark / 2, watermark); }
   void setWatermarks(uint32_t low_watermark, uint32_t high_watermark);

test/common/buffer/buffer_fuzz.cc

@@ -133,6 +133,10 @@ class StringBuffer : public Buffer::Instance {
     return mutableStart();
   }
 
+  Buffer::SliceDataPtr extractFrontSlice() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
+
+  Buffer::SliceDataPtr extractMutableFrontSlice() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
+
   void move(Buffer::Instance& rhs) override { move(rhs, rhs.length()); }
 
   void move(Buffer::Instance& rhs, uint64_t length) override {