diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
index 688cf3d22577e..686b2da44469e 100644
--- a/docs/root/intro/version_history.rst
+++ b/docs/root/intro/version_history.rst
@@ -37,6 +37,7 @@ Version history
 * mysql: added a MySQL proxy filter that is capable of parsing SQL queries over MySQL wire protocol. Refer to ::ref:`MySQL proxy<config_network_filters_mysql_proxy>` for more details.
 * http: added :ref:`max request headers size <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.max_request_headers_kb>`. The default behaviour is unchanged.
 * http: added modifyDecodingBuffer/modifyEncodingBuffer to allow modifying the buffered request/response data.
+* performance: new buffer implementation (disabled by default; to test it, add "--use-libevent-buffers 0" to the command-line arguments when starting Envoy).
 * http: added encodeComplete/decodeComplete. These are invoked at the end of the stream, after all data has been encoded/decoded respectively. Default implementation is a no-op.
 * redis: added :ref:`hashtagging <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.enable_hashtagging>` to guarantee a given key's upstream.
 * redis: added :ref:`latency stats <config_network_filters_redis_proxy_per_command_stats>` for commands.
diff --git a/include/envoy/server/options.h b/include/envoy/server/options.h
index 905a218f9cc3f..354a1bc53e6f5 100644
--- a/include/envoy/server/options.h
+++ b/include/envoy/server/options.h
@@ -174,6 +174,11 @@ class Options {
    */
   virtual bool mutexTracingEnabled() const PURE;
 
+  /**
+   * @return whether to use the old libevent evbuffer-based Buffer implementation.
+   */
+  virtual bool libeventBufferEnabled() const PURE;
+
   /**
    * @return bool indicating whether cpuset size should determine the number of worker threads.
    */
diff --git a/source/common/buffer/buffer_impl.cc b/source/common/buffer/buffer_impl.cc
index d0f15d8154119..88253bdd124b4 100644
--- a/source/common/buffer/buffer_impl.cc
+++ b/source/common/buffer/buffer_impl.cc
@@ -11,25 +11,43 @@
 namespace Envoy {
 namespace Buffer {
 
-// RawSlice is the same structure as evbuffer_iovec. This was put into place to avoid leaking
-// libevent into most code since we will likely replace evbuffer with our own implementation at
-// some point. However, we can avoid a bunch of copies since the structure is the same.
-static_assert(sizeof(RawSlice) == sizeof(evbuffer_iovec), "RawSlice != evbuffer_iovec");
-static_assert(offsetof(RawSlice, mem_) == offsetof(evbuffer_iovec, iov_base),
-              "RawSlice != evbuffer_iovec");
-static_assert(offsetof(RawSlice, len_) == offsetof(evbuffer_iovec, iov_len),
-              "RawSlice != evbuffer_iovec");
-
-void OwnedImpl::add(const void* data, uint64_t size) { evbuffer_add(buffer_.get(), data, size); }
+void OwnedImpl::add(const void* data, uint64_t size) {
+  if (old_impl_) {
+    evbuffer_add(buffer_.get(), data, size);
+  } else {
+    const char* src = static_cast<const char*>(data);
+    bool new_slice_needed = slices_.empty();
+    while (size != 0) {
+      if (new_slice_needed) {
+        slices_.emplace_back(OwnedSlice::create(size));
+      }
+      uint64_t copy_size = slices_.back()->append(src, size);
+      src += copy_size;
+      size -= copy_size;
+      length_ += copy_size;
+      new_slice_needed = true;
+    }
+  }
+}
 
 void OwnedImpl::addBufferFragment(BufferFragment& fragment) {
-  evbuffer_add_reference(
-      buffer_.get(), fragment.data(), fragment.size(),
-      [](const void*, size_t, void* arg) { static_cast<BufferFragment*>(arg)->done(); }, &fragment);
+  if (old_impl_) {
+    evbuffer_add_reference(
+        buffer_.get(), fragment.data(), fragment.size(),
+        [](const void*, size_t, void* arg) { static_cast<BufferFragment*>(arg)->done(); },
+        &fragment);
+  } else {
+    length_ += fragment.size();
+    slices_.emplace_back(std::make_unique<UnownedSlice>(fragment));
+  }
 }
 
 void OwnedImpl::add(absl::string_view data) {
-  evbuffer_add(buffer_.get(), data.data(), data.size());
+  if (old_impl_) {
+    evbuffer_add(buffer_.get(), data.data(), data.size());
+  } else {
+    add(data.data(), data.size());
+  }
 }
 
 void OwnedImpl::add(const Instance& data) {
@@ -43,84 +61,302 @@ void OwnedImpl::add(const Instance& data) {
 }
 
 void OwnedImpl::prepend(absl::string_view data) {
-  // Prepending an empty string seems to mess up libevent internally.
-  // evbuffer_prepend doesn't have a check for empty (unlike
-  // evbuffer_prepend_buffer which does). This then results in an allocation of
-  // an empty chain, which causes problems with a following move/append. This
-  // only seems to happen the original buffer was created via
-  // addBufferFragment(), this forces the code execution path in
-  // evbuffer_prepend related to immutable buffers.
-  if (data.size() == 0) {
-    return;
+  if (old_impl_) {
+    // Prepending an empty string seems to mess up libevent internally.
+    // evbuffer_prepend doesn't have a check for empty (unlike
+    // evbuffer_prepend_buffer which does). This then results in an allocation of
+    // an empty chain, which causes problems with a following move/append. This
+    // only seems to happen the original buffer was created via
+    // addBufferFragment(), this forces the code execution path in
+    // evbuffer_prepend related to immutable buffers.
+    if (data.size() == 0) {
+      return;
+    }
+    evbuffer_prepend(buffer_.get(), data.data(), data.size());
+  } else {
+    uint64_t size = data.size();
+    bool new_slice_needed = slices_.empty();
+    while (size != 0) {
+      if (new_slice_needed) {
+        slices_.emplace_front(OwnedSlice::create(size));
+      }
+      uint64_t copy_size = slices_.front()->prepend(data.data(), size);
+      size -= copy_size;
+      length_ += copy_size;
+      new_slice_needed = true;
+    }
   }
-  evbuffer_prepend(buffer_.get(), data.data(), data.size());
 }
 
 void OwnedImpl::prepend(Instance& data) {
   ASSERT(&data != this);
-  int rc =
-      evbuffer_prepend_buffer(buffer_.get(), static_cast<LibEventInstance&>(data).buffer().get());
-  ASSERT(rc == 0);
-  ASSERT(data.length() == 0);
-  static_cast<LibEventInstance&>(data).postProcess();
+  ASSERT(isSameBufferImpl(data));
+  // See the comments in move() for why we do the static_cast.
+  if (old_impl_) {
+    ASSERT(dynamic_cast<LibEventInstance*>(&data) != nullptr);
+    int rc =
+        evbuffer_prepend_buffer(buffer_.get(), static_cast<LibEventInstance&>(data).buffer().get());
+    ASSERT(rc == 0);
+    ASSERT(data.length() == 0);
+    static_cast<LibEventInstance&>(data).postProcess();
+  } else {
+    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()));
+      other.slices_.pop_back();
+      other.length_ -= slice_size;
+    }
+    other.postProcess();
+  }
 }
 
 void OwnedImpl::commit(RawSlice* iovecs, uint64_t num_iovecs) {
-  int rc =
-      evbuffer_commit_space(buffer_.get(), reinterpret_cast<evbuffer_iovec*>(iovecs), num_iovecs);
-  ASSERT(rc == 0);
-}
+  if (old_impl_) {
+    int rc =
+        evbuffer_commit_space(buffer_.get(), reinterpret_cast<evbuffer_iovec*>(iovecs), num_iovecs);
+    ASSERT(rc == 0);
+  } else {
+    if (num_iovecs == 0) {
+      return;
+    }
+    // Find the slices in the buffer that correspond to the iovecs:
+    // First, scan backward from the end of the buffer to find the last slice containing
+    // any content. Reservations are made from the end of the buffer, and out-of-order commits
+    // aren't supported, so any slices before this point cannot match the iovecs being committed.
+    ssize_t slice_index = static_cast<ssize_t>(slices_.size()) - 1;
+    while (slice_index >= 0 && slices_[slice_index]->dataSize() == 0) {
+      slice_index--;
+    }
+    if (slice_index < 0) {
+      // There was no slice containing any data, so rewind the iterator at the first slice.
+      slice_index = 0;
+    }
 
-void OwnedImpl::copyOut(size_t start, uint64_t size, void* data) const {
-  ASSERT(start + size <= length());
+    // Next, scan forward and attempt to match the slices against iovecs.
+    uint64_t num_slices_committed = 0;
+    while (num_slices_committed < num_iovecs) {
+      if (slices_[slice_index]->commit(iovecs[num_slices_committed])) {
+        length_ += iovecs[num_slices_committed].len_;
+        num_slices_committed++;
+      }
+      slice_index++;
+      if (slice_index == static_cast<ssize_t>(slices_.size())) {
+        break;
+      }
+    }
 
-  evbuffer_ptr start_ptr;
-  int rc = evbuffer_ptr_set(buffer_.get(), &start_ptr, start, EVBUFFER_PTR_SET);
-  ASSERT(rc != -1);
+    ASSERT(num_slices_committed > 0);
+  }
+}
 
-  ev_ssize_t copied = evbuffer_copyout_from(buffer_.get(), &start_ptr, data, size);
-  ASSERT(static_cast<uint64_t>(copied) == size);
+void OwnedImpl::copyOut(size_t start, uint64_t size, void* data) const {
+  if (old_impl_) {
+    ASSERT(start + size <= length());
+
+    evbuffer_ptr start_ptr;
+    int rc = evbuffer_ptr_set(buffer_.get(), &start_ptr, start, EVBUFFER_PTR_SET);
+    ASSERT(rc != -1);
+
+    ev_ssize_t copied = evbuffer_copyout_from(buffer_.get(), &start_ptr, data, size);
+    ASSERT(static_cast<uint64_t>(copied) == size);
+  } else {
+    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) {
+        // The offset where the caller wants to start copying is after the end of this slice,
+        // so just skip over this slice completely.
+        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);
+      size -= copy_size;
+      dest += copy_size;
+      // Now that we've started copying, there are no bytes left to skip over. If there
+      // is any more data to be copied, the next iteration can start copying from the very
+      // beginning of the next slice.
+      bytes_to_skip = 0;
+    }
+    ASSERT(size == 0);
+  }
 }
 
 void OwnedImpl::drain(uint64_t size) {
-  ASSERT(size <= length());
-  int rc = evbuffer_drain(buffer_.get(), size);
-  ASSERT(rc == 0);
+  if (old_impl_) {
+    ASSERT(size <= length());
+    int rc = evbuffer_drain(buffer_.get(), size);
+    ASSERT(rc == 0);
+  } else {
+    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;
+      }
+    }
+  }
 }
 
 uint64_t OwnedImpl::getRawSlices(RawSlice* out, uint64_t out_size) const {
-  return evbuffer_peek(buffer_.get(), -1, nullptr, reinterpret_cast<evbuffer_iovec*>(out),
-                       out_size);
+  if (old_impl_) {
+    return evbuffer_peek(buffer_.get(), -1, nullptr, reinterpret_cast<evbuffer_iovec*>(out),
+                         out_size);
+  } else {
+    uint64_t num_slices = 0;
+    for (const auto& slice : slices_) {
+      if (slice->dataSize() == 0) {
+        continue;
+      }
+      if (num_slices < out_size) {
+        out[num_slices].mem_ = slice->data();
+        out[num_slices].len_ = slice->dataSize();
+      }
+      // Per the definition of getRawSlices in include/envoy/buffer/buffer.h, we need to return
+      // the total number of slices needed to access all the data in the buffer, which can be
+      // larger than out_size. So we keep iterating and counting non-empty slices here, even
+      // if all the caller-supplied slices have been filled.
+      num_slices++;
+    }
+    return num_slices;
+  }
 }
 
-uint64_t OwnedImpl::length() const { return evbuffer_get_length(buffer_.get()); }
+uint64_t OwnedImpl::length() const {
+  if (old_impl_) {
+    return evbuffer_get_length(buffer_.get());
+  } else {
+#ifndef NDEBUG
+    // When running in debug mode, verify that the precomputed length matches the sum
+    // of the lengths of the slices.
+    uint64_t length = 0;
+    for (const auto& slice : slices_) {
+      length += slice->dataSize();
+    }
+    ASSERT(length == length_);
+#endif
+
+    return length_;
+  }
+}
 
 void* OwnedImpl::linearize(uint32_t size) {
-  ASSERT(size <= length());
-  void* const ret = evbuffer_pullup(buffer_.get(), size);
-  RELEASE_ASSERT(ret != nullptr || size == 0,
-                 "Failure to linearize may result in buffer overflow by the caller.");
-  return ret;
+  RELEASE_ASSERT(size <= length(), "Linearize size exceeds buffer size");
+  if (old_impl_) {
+    void* const ret = evbuffer_pullup(buffer_.get(), size);
+    RELEASE_ASSERT(ret != nullptr || size == 0,
+                   "Failure to linearize may result in buffer overflow by the caller.");
+    return ret;
+  } else {
+    if (slices_.empty()) {
+      return nullptr;
+    }
+    uint64_t linearized_size = 0;
+    uint64_t num_slices_to_linearize = 0;
+    for (const auto& slice : slices_) {
+      num_slices_to_linearize++;
+      linearized_size += slice->dataSize();
+      if (linearized_size >= size) {
+        break;
+      }
+    }
+    if (num_slices_to_linearize > 1) {
+      auto new_slice = OwnedSlice::create(linearized_size);
+      uint64_t bytes_copied = 0;
+      Slice::Reservation reservation = new_slice->reserve(linearized_size);
+      ASSERT(reservation.mem_ != nullptr);
+      ASSERT(reservation.len_ == linearized_size);
+      auto dest = static_cast<uint8_t*>(reservation.mem_);
+      do {
+        uint64_t data_size = slices_.front()->dataSize();
+        memcpy(dest, slices_.front()->data(), data_size);
+        bytes_copied += data_size;
+        dest += data_size;
+        slices_.pop_front();
+      } while (bytes_copied < linearized_size);
+      ASSERT(dest == static_cast<const uint8_t*>(reservation.mem_) + linearized_size);
+      new_slice->commit(reservation);
+      slices_.emplace_front(std::move(new_slice));
+    }
+    return slices_.front()->data();
+  }
 }
 
 void OwnedImpl::move(Instance& rhs) {
   ASSERT(&rhs != this);
-  // We do the static cast here because in practice we only have one buffer implementation right
-  // now and this is safe. Using the evbuffer move routines require having access to both evbuffers.
-  // This is a reasonable compromise in a high performance path where we want to maintain an
-  // abstraction in case we get rid of evbuffer later.
-  int rc = evbuffer_add_buffer(buffer_.get(), static_cast<LibEventInstance&>(rhs).buffer().get());
-  ASSERT(rc == 0);
-  static_cast<LibEventInstance&>(rhs).postProcess();
+  ASSERT(isSameBufferImpl(rhs));
+  if (old_impl_) {
+    // We do the static cast here because in practice we only have one buffer implementation right
+    // now and this is safe. Using the evbuffer move routines require having access to both
+    // evbuffers. This is a reasonable compromise in a high performance path where we want to
+    // maintain an abstraction in case we get rid of evbuffer later.
+    ASSERT(dynamic_cast<LibEventInstance*>(&rhs) != nullptr);
+    int rc = evbuffer_add_buffer(buffer_.get(), static_cast<LibEventInstance&>(rhs).buffer().get());
+    ASSERT(rc == 0);
+    static_cast<LibEventInstance&>(rhs).postProcess();
+  } else {
+    // We do the static cast here because in practice we only have one buffer implementation right
+    // now and this is safe. This is a reasonable compromise in a high performance path where we
+    // want to maintain an abstraction.
+    OwnedImpl& other = static_cast<OwnedImpl&>(rhs);
+    while (!other.slices_.empty()) {
+      const uint64_t slice_size = other.slices_.front()->dataSize();
+      slices_.emplace_back(std::move(other.slices_.front()));
+      other.slices_.pop_front();
+      length_ += slice_size;
+      other.length_ -= slice_size;
+    }
+    other.postProcess();
+  }
 }
 
 void OwnedImpl::move(Instance& rhs, uint64_t length) {
   ASSERT(&rhs != this);
-  // See move() above for why we do the static cast.
-  int rc = evbuffer_remove_buffer(static_cast<LibEventInstance&>(rhs).buffer().get(), buffer_.get(),
-                                  length);
-  ASSERT(static_cast<uint64_t>(rc) == length);
-  static_cast<LibEventInstance&>(rhs).postProcess();
+  ASSERT(isSameBufferImpl(rhs));
+  if (old_impl_) {
+    // See move() above for why we do the static cast.
+    int rc = evbuffer_remove_buffer(static_cast<LibEventInstance&>(rhs).buffer().get(),
+                                    buffer_.get(), length);
+    ASSERT(static_cast<uint64_t>(rc) == length);
+    static_cast<LibEventInstance&>(rhs).postProcess();
+  } else {
+    // See move() above for why we do the static cast.
+    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) {
+        other.slices_.pop_front();
+      } else if (copy_size < slice_size) {
+        // TODO(brian-pane) add reference-counting to allow slices to share their storage
+        // and eliminate the copy for this partial-slice case?
+        add(other.slices_.front()->data(), copy_size);
+        other.slices_.front()->drain(copy_size);
+        other.length_ -= copy_size;
+      } else {
+        slices_.emplace_back(std::move(other.slices_.front()));
+        other.slices_.pop_front();
+        length_ += slice_size;
+        other.length_ -= slice_size;
+      }
+      length -= copy_size;
+    }
+    other.postProcess();
+  }
 }
 
 Api::IoCallUint64Result OwnedImpl::read(Network::IoHandle& io_handle, uint64_t max_length) {
@@ -131,8 +367,10 @@ Api::IoCallUint64Result OwnedImpl::read(Network::IoHandle& io_handle, uint64_t m
   RawSlice slices[MaxSlices];
   const uint64_t num_slices = reserve(max_length, slices, MaxSlices);
   Api::IoCallUint64Result result = io_handle.readv(max_length, slices, num_slices);
-  if (result.ok()) {
-    // Read succeeded.
+  if (old_impl_) {
+    if (!result.ok()) {
+      return result;
+    }
     uint64_t num_slices_to_commit = 0;
     uint64_t bytes_to_commit = result.rc_;
     ASSERT(bytes_to_commit <= max_length);
@@ -145,28 +383,153 @@ Api::IoCallUint64Result OwnedImpl::read(Network::IoHandle& io_handle, uint64_t m
     }
     ASSERT(num_slices_to_commit <= num_slices);
     commit(slices, num_slices_to_commit);
+  } else {
+    uint64_t bytes_to_commit = result.ok() ? result.rc_ : 0;
+    ASSERT(bytes_to_commit <= max_length);
+    for (uint64_t i = 0; i < num_slices; i++) {
+      slices[i].len_ = std::min(slices[i].len_, static_cast<size_t>(bytes_to_commit));
+      bytes_to_commit -= slices[i].len_;
+    }
+    commit(slices, num_slices);
   }
   return result;
 }
 
 uint64_t OwnedImpl::reserve(uint64_t length, RawSlice* iovecs, uint64_t num_iovecs) {
-  ASSERT(length > 0);
-  int ret = evbuffer_reserve_space(buffer_.get(), length, reinterpret_cast<evbuffer_iovec*>(iovecs),
-                                   num_iovecs);
-  RELEASE_ASSERT(ret >= 1, "Failure to allocate may result in callers writing to uninitialized "
-                           "memory, buffer overflows, etc");
-  return static_cast<uint64_t>(ret);
+  if (num_iovecs == 0 || length == 0) {
+    return 0;
+  }
+  if (old_impl_) {
+    int ret = evbuffer_reserve_space(buffer_.get(), length,
+                                     reinterpret_cast<evbuffer_iovec*>(iovecs), num_iovecs);
+    RELEASE_ASSERT(ret >= 1, "Failure to allocate may result in callers writing to uninitialized "
+                             "memory, buffer overflows, etc");
+    return static_cast<uint64_t>(ret);
+  } else {
+    // Check whether there are any empty slices with reservable space at the end of the buffer.
+    size_t first_reservable_slice = slices_.size();
+    while (first_reservable_slice > 0) {
+      if (slices_[first_reservable_slice - 1]->reservableSize() == 0) {
+        break;
+      }
+      first_reservable_slice--;
+      if (slices_[first_reservable_slice]->dataSize() != 0) {
+        // There is some content in this slice, so anything in front of it is nonreservable.
+        break;
+      }
+    }
+
+    // Having found the sequence of reservable slices at the back of the buffer, reserve
+    // as much space as possible from each one.
+    uint64_t num_slices_used = 0;
+    uint64_t bytes_remaining = length;
+    size_t slice_index = first_reservable_slice;
+    while (slice_index < slices_.size() && bytes_remaining != 0 && num_slices_used < num_iovecs) {
+      auto& slice = slices_[slice_index];
+      const uint64_t reservation_size = std::min(slice->reservableSize(), bytes_remaining);
+      if (num_slices_used + 1 == num_iovecs && reservation_size < bytes_remaining) {
+        // There is only one iovec left, and this next slice does not have enough space to
+        // complete the reservation. Stop iterating, with last one iovec still unpopulated,
+        // so the code following this loop can allocate a new slice to hold the rest of the
+        // reservation.
+        break;
+      }
+      iovecs[num_slices_used] = slice->reserve(reservation_size);
+      bytes_remaining -= iovecs[num_slices_used].len_;
+      num_slices_used++;
+      slice_index++;
+    }
+
+    // If needed, allocate one more slice at the end to provide the remainder of the reservation.
+    if (bytes_remaining != 0) {
+      slices_.emplace_back(OwnedSlice::create(bytes_remaining));
+      iovecs[num_slices_used] = slices_.back()->reserve(bytes_remaining);
+      bytes_remaining -= iovecs[num_slices_used].len_;
+      num_slices_used++;
+    }
+
+    ASSERT(num_slices_used <= num_iovecs);
+    ASSERT(bytes_remaining == 0);
+    return num_slices_used;
+  }
 }
 
 ssize_t OwnedImpl::search(const void* data, uint64_t size, size_t start) const {
-  evbuffer_ptr start_ptr;
-  if (-1 == evbuffer_ptr_set(buffer_.get(), &start_ptr, start, EVBUFFER_PTR_SET)) {
+  if (old_impl_) {
+    evbuffer_ptr start_ptr;
+    if (-1 == evbuffer_ptr_set(buffer_.get(), &start_ptr, start, EVBUFFER_PTR_SET)) {
+      return -1;
+    }
+
+    evbuffer_ptr result_ptr =
+        evbuffer_search(buffer_.get(), static_cast<const char*>(data), size, &start_ptr);
+    return result_ptr.pos;
+  } else {
+    // This implementation uses the same search algorithm as evbuffer_search(), a naive
+    // scan that requires O(M*N) comparisons in the worst case.
+    // TODO(brian-pane): replace this with a more efficient search if it shows up
+    // prominently in CPU profiling.
+    if (size == 0) {
+      return (start <= length_) ? start : -1;
+    }
+    ssize_t offset = 0;
+    const uint8_t* needle = static_cast<const uint8_t*>(data);
+    for (size_t slice_index = 0; slice_index < slices_.size(); 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) {
+        // Search within this slice for the first byte of the needle.
+        const uint8_t* first_byte_match =
+            static_cast<const uint8_t*>(memchr(haystack, needle[0], haystack_end - haystack));
+        if (first_byte_match == nullptr) {
+          break;
+        }
+        // After finding a match for the first byte of the needle, check whether the following
+        // bytes in the buffer match the remainder of the needle. Note that the match can span
+        // two or more slices.
+        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) {
+          if (match_next >= match_end) {
+            // We've hit the end of this slice, so continue checking against the next slice.
+            match_index++;
+            if (match_index == slices_.size()) {
+              // We've hit the end of the entire buffer.
+              break;
+            }
+            const auto& match_slice = slices_[match_index];
+            match_next = match_slice->data();
+            match_end = match_next + match_slice->dataSize();
+            continue;
+          }
+          if (*match_next++ != needle[i]) {
+            break;
+          }
+          i++;
+        }
+        if (i == size) {
+          // Successful match of the entire needle.
+          return offset + (first_byte_match - slice_start);
+        }
+        // If this wasn't a successful match, start scanning again at the next byte.
+        haystack = first_byte_match + 1;
+      }
+      start = 0;
+      offset += slice_size;
+    }
     return -1;
   }
-
-  evbuffer_ptr result_ptr =
-      evbuffer_search(buffer_.get(), static_cast<const char*>(data), size, &start_ptr);
-  return result_ptr.pos;
 }
 
 Api::IoCallUint64Result OwnedImpl::write(Network::IoHandle& io_handle) {
@@ -180,7 +543,11 @@ Api::IoCallUint64Result OwnedImpl::write(Network::IoHandle& io_handle) {
   return result;
 }
 
-OwnedImpl::OwnedImpl() : buffer_(evbuffer_new()) {}
+OwnedImpl::OwnedImpl() : old_impl_(use_old_impl_) {
+  if (old_impl_) {
+    buffer_ = evbuffer_new();
+  }
+}
 
 OwnedImpl::OwnedImpl(absl::string_view data) : OwnedImpl() { add(data); }
 
@@ -205,5 +572,33 @@ std::string OwnedImpl::toString() const {
   return output;
 }
 
+void OwnedImpl::postProcess() {}
+
+void OwnedImpl::appendSliceForTest(const void* data, uint64_t size) {
+  if (old_impl_) {
+    OwnedImpl rhs(data, size);
+    move(rhs);
+  } else {
+    slices_.emplace_back(OwnedSlice::create(data, size));
+    length_ += size;
+  }
+}
+
+void OwnedImpl::appendSliceForTest(absl::string_view data) {
+  appendSliceForTest(data.data(), data.size());
+}
+
+void OwnedImpl::useOldImpl(bool use_old_impl) { use_old_impl_ = use_old_impl; }
+
+bool OwnedImpl::isSameBufferImpl(const Instance& rhs) const {
+  const OwnedImpl* other = dynamic_cast<const OwnedImpl*>(&rhs);
+  if (other == nullptr) {
+    return false;
+  }
+  return usesOldImpl() == other->usesOldImpl();
+}
+
+bool OwnedImpl::use_old_impl_ = false;
+
 } // namespace Buffer
 } // namespace Envoy
diff --git a/source/common/buffer/buffer_impl.h b/source/common/buffer/buffer_impl.h
index dd48199513b9d..b09c0773ac8bc 100644
--- a/source/common/buffer/buffer_impl.h
+++ b/source/common/buffer/buffer_impl.h
@@ -1,17 +1,428 @@
 #pragma once
 
+#include <algorithm>
 #include <cstdint>
+#include <deque>
 #include <string>
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/network/io_handle.h"
 
+#include "common/common/assert.h"
 #include "common/common/non_copyable.h"
 #include "common/event/libevent.h"
 
 namespace Envoy {
 namespace Buffer {
 
+/**
+ * A Slice manages a contiguous block of bytes.
+ * The block is arranged like this:
+ *                   |<- data_size() -->|<- reservable_size() ->|
+ * +-----------------+------------------+-----------------------+
+ * | Drained         | Data             | Reservable            |
+ * | Unused space    | Usable content   | New content can be    |
+ * | that formerly   |                  | added here with       |
+ * | was in the Data |                  | reserve()/commit()    |
+ * | section         |                  |                       |
+ * +-----------------+------------------+-----------------------+
+ *                   ^
+ *                   |
+ *                   data()
+ */
+class Slice {
+public:
+  using Reservation = RawSlice;
+
+  virtual ~Slice() = default;
+
+  /**
+   * @return a pointer to the start of the usable content.
+   */
+  const uint8_t* data() const { return base_ + data_; }
+
+  /**
+   * @return a pointer to the start of the usable content.
+   */
+  uint8_t* data() { return base_ + data_; }
+
+  /**
+   * @return the size in bytes of the usable content.
+   */
+  uint64_t dataSize() const { return reservable_ - data_; }
+
+  /**
+   * Remove the first `size` bytes of usable content. Runs in O(1) time.
+   * @param size number of bytes to remove. If greater than data_size(), the result is undefined.
+   */
+  void drain(uint64_t size) {
+    ASSERT(data_ + size <= reservable_);
+    data_ += size;
+    if (data_ == reservable_ && !reservation_outstanding_) {
+      // There is no more content in the slice, and there is no outstanding reservation,
+      // so reset the Data section to the start of the slice to facilitate reuse.
+      data_ = reservable_ = 0;
+    }
+  }
+
+  /**
+   * @return the number of bytes available to be reserve()d.
+   * @note If reserve() has been called without a corresponding commit(), this method
+   *       should return 0.
+   * @note Read-only implementations of Slice should return zero from this method.
+   */
+  uint64_t reservableSize() const {
+    if (reservation_outstanding_) {
+      return 0;
+    }
+    return capacity_ - reservable_;
+  }
+
+  /**
+   * Reserve `size` bytes that the caller can populate with content. The caller SHOULD then
+   * call commit() to add the newly populated content from the Reserved section to the Data
+   * section.
+   * @note If there is already an outstanding reservation (i.e., a reservation obtained
+   *       from reserve() that has not been released by calling commit()), this method will
+   *       return {nullptr, 0}.
+   * @param size the number of bytes to reserve. The Slice implementation MAY reserve
+   *        fewer bytes than requested (for example, if it doesn't have enough room in the
+   *        Reservable section to fulfill the whole request).
+   * @return a tuple containing the address of the start of resulting reservation and the
+   *         reservation size in bytes. If the address is null, the reservation failed.
+   * @note Read-only implementations of Slice should return {nullptr, 0} from this method.
+   */
+  Reservation reserve(uint64_t size) {
+    if (reservation_outstanding_ || size == 0) {
+      return {nullptr, 0};
+    }
+    uint64_t available_size = capacity_ - reservable_;
+    if (available_size == 0) {
+      return {nullptr, 0};
+    }
+    uint64_t reservation_size = std::min(size, available_size);
+    void* reservation = &(base_[reservable_]);
+    reservation_outstanding_ = true;
+    return {reservation, reservation_size};
+  }
+
+  /**
+   * Commit a Reservation that was previously obtained from a call to reserve().
+   * The Reservation's size is added to the Data section.
+   * @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().
+   *        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)`.
+   * @return whether the Reservation was successfully committed to the Slice.
+   */
+  bool commit(const Reservation& reservation) {
+    if (static_cast<const uint8_t*>(reservation.mem_) != base_ + reservable_ ||
+        reservable_ + reservation.len_ > capacity_ || reservable_ >= capacity_) {
+      // The reservation is not from this OwnedSlice.
+      return false;
+    }
+    ASSERT(reservation_outstanding_);
+    reservable_ += reservation.len_;
+    reservation_outstanding_ = false;
+    return true;
+  }
+
+  /**
+   * Copy as much of the supplied data as possible to the end of the slice.
+   * @param data start of the data to copy.
+   * @param size number of bytes to copy.
+   * @return number of bytes copied (may be a smaller than size, may even be zero).
+   */
+  uint64_t append(const void* data, uint64_t size) {
+    if (reservation_outstanding_) {
+      return 0;
+    }
+    uint64_t copy_size = std::min(size, reservableSize());
+    uint8_t* dest = base_ + reservable_;
+    reservable_ += copy_size;
+    // NOLINTNEXTLINE(clang-analyzer-core.NullDereference)
+    memcpy(dest, data, copy_size);
+    return copy_size;
+  }
+
+  /**
+   * Copy as much of the supplied data as possible to the front of the slice.
+   * If only part of the data will fit in the slice, the bytes from the _end_ are
+   * copied.
+   * @param data start of the data to copy.
+   * @param size number of bytes to copy.
+   * @return number of bytes copied (may be a smaller than size, may even be zero).
+   */
+  uint64_t prepend(const void* data, uint64_t size) {
+    if (reservation_outstanding_) {
+      return 0;
+    }
+    const uint8_t* src = static_cast<const uint8_t*>(data);
+    uint64_t copy_size;
+    if (dataSize() == 0) {
+      // There is nothing in the slice, so put the data at the very end in case the caller
+      // later tries to prepend anything else in front of it.
+      copy_size = std::min(size, reservableSize());
+      reservable_ = capacity_;
+      data_ = capacity_ - copy_size;
+    } else {
+      if (data_ == 0) {
+        // There is content in the slice, and no space in front of it to write anything.
+        return 0;
+      }
+      // Write into the space in front of the slice's current content.
+      copy_size = std::min(size, data_);
+      data_ -= copy_size;
+    }
+    memcpy(base_ + data_, src + size - copy_size, copy_size);
+    return copy_size;
+  }
+
+protected:
+  Slice(uint64_t data, uint64_t reservable, uint64_t capacity)
+      : data_(data), reservable_(reservable), capacity_(capacity) {}
+
+  /** Start of the slice - subclasses must set this */
+  uint8_t* base_{nullptr};
+
+  /** Offset in bytes from the start of the slice to the start of the Data section */
+  uint64_t data_;
+
+  /** Offset in bytes from the start of the slice to the start of the Reservable section */
+  uint64_t reservable_;
+
+  /** Total number of bytes in the slice */
+  uint64_t capacity_;
+
+  /** Whether reserve() has been called without a corresponding commit(). */
+  bool reservation_outstanding_{false};
+};
+
+using SlicePtr = std::unique_ptr<Slice>;
+
+class OwnedSlice : public Slice {
+public:
+  /**
+   * Create an empty OwnedSlice.
+   * @param capacity number of bytes of space the slice should have.
+   * @return an OwnedSlice with at least the specified capacity.
+   */
+  static SlicePtr create(uint64_t capacity) {
+    uint64_t slice_capacity = sliceSize(capacity);
+    return SlicePtr(new (slice_capacity) OwnedSlice(slice_capacity));
+  }
+
+  /**
+   * Create an OwnedSlice and initialize it with a copy of the supplied copy.
+   * @param data the content to copy into the slice.
+   * @param size length of the content.
+   * @return an OwnedSlice containing a copy of the content, which may (dependent on
+   *         the internal implementation) have a nonzero amount of reservable space at the end.
+   */
+  static SlicePtr create(const void* data, uint64_t size) {
+    uint64_t slice_capacity = sliceSize(size);
+    std::unique_ptr<OwnedSlice> slice(new (slice_capacity) OwnedSlice(slice_capacity));
+    memcpy(slice->base_, data, size);
+    slice->reservable_ = size;
+    return slice;
+  }
+
+  // Custom delete operator to keep C++14 from using the global operator delete(void*, size_t),
+  // which would result in the compiler error:
+  // "exception cleanup for this placement new selects non-placement operator delete"
+  static void operator delete(void* address) { ::operator delete(address); }
+
+private:
+  static void* operator new(size_t object_size, size_t data_size) {
+    return ::operator new(object_size + data_size);
+  }
+
+  OwnedSlice(uint64_t size) : Slice(0, 0, size) { base_ = storage_; }
+
+  /**
+   * Compute a slice size big enough to hold a specified amount of data.
+   * @param data_size the minimum amount of data the slice must be able to store, in bytes.
+   * @return a recommended slice size, in bytes.
+   */
+  static uint64_t sliceSize(uint64_t data_size) {
+    static constexpr uint64_t PageSize = 4096;
+    const uint64_t num_pages = (sizeof(OwnedSlice) + data_size + PageSize - 1) / PageSize;
+    return num_pages * PageSize - sizeof(OwnedSlice);
+  }
+
+  uint8_t storage_[];
+};
+
+/**
+ * Queue of SlicePtr that supports efficient read and write access to both
+ * the front and the back of the queue.
+ * @note This class has similar properties to std::deque<T>. The reason for using
+ *       a custom deque implementation is that benchmark testing during development
+ *       revealed that std::deque was too slow to reach performance parity with the
+ *       prior evbuffer-based buffer implementation.
+ */
+class SliceDeque {
+public:
+  SliceDeque() : ring_(inline_ring_), capacity_(InlineRingCapacity) {}
+
+  SliceDeque(SliceDeque&& rhs) noexcept {
+    // This custom move constructor is needed so that ring_ will be updated properly.
+    std::move(rhs.inline_ring_, rhs.inline_ring_ + InlineRingCapacity, inline_ring_);
+    external_ring_ = std::move(rhs.external_ring_);
+    ring_ = (external_ring_ != nullptr) ? external_ring_.get() : inline_ring_;
+    start_ = rhs.start_;
+    size_ = rhs.size_;
+    capacity_ = rhs.capacity_;
+  }
+
+  SliceDeque& operator=(SliceDeque&& rhs) noexcept {
+    // This custom assignment move operator is needed so that ring_ will be updated properly.
+    std::move(rhs.inline_ring_, rhs.inline_ring_ + InlineRingCapacity, inline_ring_);
+    external_ring_ = std::move(rhs.external_ring_);
+    ring_ = (external_ring_ != nullptr) ? external_ring_.get() : inline_ring_;
+    start_ = rhs.start_;
+    size_ = rhs.size_;
+    capacity_ = rhs.capacity_;
+    return *this;
+  }
+
+  void emplace_back(SlicePtr&& slice) {
+    growRing();
+    size_t index = internalIndex(size_);
+    ring_[index] = std::move(slice);
+    size_++;
+  }
+
+  void emplace_front(SlicePtr&& slice) {
+    growRing();
+    start_ = (start_ == 0) ? capacity_ - 1 : start_ - 1;
+    ring_[start_] = std::move(slice);
+    size_++;
+  }
+
+  bool empty() const { return size() == 0; }
+  size_t size() const { return size_; }
+
+  SlicePtr& front() { return ring_[start_]; }
+  const SlicePtr& front() const { return ring_[start_]; }
+  SlicePtr& back() { return ring_[internalIndex(size_ - 1)]; }
+  const SlicePtr& back() const { return ring_[internalIndex(size_ - 1)]; }
+
+  SlicePtr& operator[](size_t i) { return ring_[internalIndex(i)]; }
+  const SlicePtr& operator[](size_t i) const { return ring_[internalIndex(i)]; }
+
+  void pop_front() {
+    if (size() == 0) {
+      return;
+    }
+    front() = SlicePtr();
+    size_--;
+    start_++;
+    if (start_ == capacity_) {
+      start_ = 0;
+    }
+  }
+
+  void pop_back() {
+    if (size() == 0) {
+      return;
+    }
+    back() = SlicePtr();
+    size_--;
+  }
+
+  /**
+   * Forward const iterator for SliceDeque.
+   * @note this implementation currently supports the minimum functionality needed to support
+   *       the `for (const auto& slice : slice_deque)` idiom.
+   */
+  class ConstIterator {
+  public:
+    const SlicePtr& operator*() { return deque_[index_]; }
+
+    ConstIterator operator++() {
+      index_++;
+      return *this;
+    }
+
+    bool operator!=(const ConstIterator& rhs) const {
+      return &deque_ != &rhs.deque_ || index_ != rhs.index_;
+    }
+
+    friend class SliceDeque;
+
+  private:
+    ConstIterator(const SliceDeque& deque, size_t index) : deque_(deque), index_(index) {}
+    const SliceDeque& deque_;
+    size_t index_;
+  };
+
+  ConstIterator begin() const noexcept { return ConstIterator(*this, 0); }
+
+  ConstIterator end() const noexcept { return ConstIterator(*this, size_); }
+
+private:
+  constexpr static size_t InlineRingCapacity = 8;
+
+  size_t internalIndex(size_t index) const {
+    size_t internal_index = start_ + index;
+    if (internal_index >= capacity_) {
+      internal_index -= capacity_;
+      ASSERT(internal_index < capacity_);
+    }
+    return internal_index;
+  }
+
+  void growRing() {
+    if (size_ < capacity_) {
+      return;
+    }
+    const size_t new_capacity = capacity_ * 2;
+    auto new_ring = std::make_unique<SlicePtr[]>(new_capacity);
+    for (size_t i = 0; i < new_capacity; i++) {
+      ASSERT(new_ring[i] == nullptr);
+    }
+    size_t src = start_;
+    size_t dst = 0;
+    for (size_t i = 0; i < size_; i++) {
+      new_ring[dst++] = std::move(ring_[src++]);
+      if (src == capacity_) {
+        src = 0;
+      }
+    }
+    for (size_t i = 0; i < capacity_; i++) {
+      ASSERT(ring_[i].get() == nullptr);
+    }
+    external_ring_.swap(new_ring);
+    ring_ = external_ring_.get();
+    start_ = 0;
+    capacity_ = new_capacity;
+  }
+
+  SlicePtr inline_ring_[InlineRingCapacity];
+  std::unique_ptr<SlicePtr[]> external_ring_;
+  SlicePtr* ring_; // points to start of either inline or external ring.
+  size_t start_{0};
+  size_t size_{0};
+  size_t capacity_;
+};
+
+class UnownedSlice : public Slice {
+public:
+  UnownedSlice(BufferFragment& fragment)
+      : Slice(0, fragment.size(), fragment.size()), fragment_(fragment) {
+    base_ = static_cast<uint8_t*>(const_cast<void*>(fragment.data()));
+  }
+
+  ~UnownedSlice() override { fragment_.done(); }
+
+private:
+  BufferFragment& fragment_;
+};
+
 /**
  * An implementation of BufferFragment where a releasor callback is called when the data is
  * no longer needed.
@@ -56,10 +467,34 @@ class LibEventInstance : public Instance {
 };
 
 /**
- * Wraps an allocated and owned evbuffer.
+ * Wrapper for uint64_t that asserts upon integer overflow and underflow.
+ */
+class OverflowDetectingUInt64 {
+public:
+  operator uint64_t() const { return value_; }
+
+  OverflowDetectingUInt64& operator+=(uint64_t size) {
+    uint64_t new_value = value_ + size;
+    RELEASE_ASSERT(new_value >= value_, "64-bit unsigned integer overflowed");
+    value_ = new_value;
+    return *this;
+  }
+
+  OverflowDetectingUInt64& operator-=(uint64_t size) {
+    RELEASE_ASSERT(value_ >= size, "unsigned integer underflowed");
+    value_ -= size;
+    return *this;
+  }
+
+private:
+  uint64_t value_{0};
+};
+
+/**
+ * Wraps an allocated and owned buffer.
  *
- * Note that due to the internals of move() accessing buffer(), OwnedImpl is not
- * compatible with non-LibEventInstance buffers.
+ * Note that due to the internals of move(), OwnedImpl is not
+ * compatible with non-OwnedImpl buffers.
  */
 class OwnedImpl : public LibEventInstance {
 public:
@@ -68,7 +503,7 @@ class OwnedImpl : public LibEventInstance {
   OwnedImpl(const Instance& data);
   OwnedImpl(const void* data, uint64_t size);
 
-  // LibEventInstance
+  // Buffer::Instance
   void add(const void* data, uint64_t size) override;
   void addBufferFragment(BufferFragment& fragment) override;
   void add(absl::string_view data) override;
@@ -87,12 +522,63 @@ class OwnedImpl : public LibEventInstance {
   uint64_t reserve(uint64_t length, RawSlice* iovecs, uint64_t num_iovecs) override;
   ssize_t search(const void* data, uint64_t size, size_t start) const override;
   Api::IoCallUint64Result write(Network::IoHandle& io_handle) override;
-  void postProcess() override {}
   std::string toString() const override;
 
+  // LibEventInstance
   Event::Libevent::BufferPtr& buffer() override { return buffer_; }
+  virtual void postProcess() override;
+
+  /**
+   * Create a new slice at the end of the buffer, and copy the supplied content into it.
+   * @param data start of the content to copy.
+   *
+   */
+  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);
+
+  // Support for choosing the buffer implementation at runtime.
+  // TODO(brian-pane) remove this once the new implementation has been
+  // running in production for a while.
+
+  /** @return whether this buffer uses the old evbuffer-based implementation. */
+  bool usesOldImpl() const { return old_impl_; }
+
+  /**
+   * @param use_old_impl whether to use the evbuffer-based implementation for new buffers
+   * @warning Except for testing code, this method should be called at most once per process,
+   *          before any OwnedImpl objects are created. The reason is that it is unsafe to
+   *          mix and match buffers with different implementations. The move() method,
+   *          in particular, only works if the source and destination objects are using
+   *          the same destination.
+   */
+  static void useOldImpl(bool use_old_impl);
 
 private:
+  /**
+   * @param rhs another buffer
+   * @return whether the rhs buffer is also an instance of OwnedImpl (or a subclass) that
+   *         uses the same internal implementation as this buffer.
+   */
+  bool isSameBufferImpl(const Instance& rhs) const;
+
+  /** Whether to use the old evbuffer implementation when constructing new OwnedImpl objects. */
+  static bool use_old_impl_;
+
+  /** Whether this buffer uses the old evbuffer implementation. */
+  bool old_impl_;
+
+  /** Ring buffer of slices. */
+  SliceDeque slices_;
+
+  /** Sum of the dataSize of all slices. */
+  OverflowDetectingUInt64 length_;
+
+  /** Used when old_impl_==true */
   Event::Libevent::BufferPtr buffer_;
 };
 
diff --git a/source/common/http/http2/metadata_encoder.h b/source/common/http/http2/metadata_encoder.h
index af292f98a633a..7e27444204463 100644
--- a/source/common/http/http2/metadata_encoder.h
+++ b/source/common/http/http2/metadata_encoder.h
@@ -7,6 +7,7 @@
 #include "envoy/http/codec.h"
 
 #include "common/buffer/buffer_impl.h"
+#include "common/common/c_smart_ptr.h"
 #include "common/common/logger.h"
 
 #include "nghttp2/nghttp2.h"
diff --git a/source/server/BUILD b/source/server/BUILD
index 3943a76618b28..a64c247463e98 100644
--- a/source/server/BUILD
+++ b/source/server/BUILD
@@ -297,6 +297,7 @@ envoy_cc_library(
         "//include/envoy/upstream:cluster_manager_interface",
         "//source/common/access_log:access_log_manager_lib",
         "//source/common/api:api_lib",
+        "//source/common/buffer:buffer_lib",
         "//source/common/common:logger_lib",
         "//source/common/common:mutex_tracer_lib",
         "//source/common/common:utility_lib",
diff --git a/source/server/options_impl.cc b/source/server/options_impl.cc
index 92e05e3fa3190..5f6a199136d65 100644
--- a/source/server/options_impl.cc
+++ b/source/server/options_impl.cc
@@ -122,6 +122,10 @@ OptionsImpl::OptionsImpl(int argc, const char* const* argv,
   TCLAP::SwitchArg cpuset_threads(
       "", "cpuset-threads", "Get the default # of worker threads from cpuset size", cmd, false);
 
+  TCLAP::ValueArg<bool> use_libevent_buffer("", "use-libevent-buffers",
+                                            "Use the original libevent buffer implementation",
+                                            false, true, "bool", cmd);
+
   cmd.setExceptionHandling(false);
   try {
     cmd.parse(argc, argv);
@@ -158,6 +162,7 @@ OptionsImpl::OptionsImpl(int argc, const char* const* argv,
 
   mutex_tracing_enabled_ = enable_mutex_tracing.getValue();
 
+  libevent_buffer_enabled_ = use_libevent_buffer.getValue();
   cpuset_threads_ = cpuset_threads.getValue();
 
   log_level_ = default_log_level;
@@ -323,6 +328,7 @@ OptionsImpl::OptionsImpl(const std::string& service_cluster, const std::string&
       service_cluster_(service_cluster), service_node_(service_node), service_zone_(service_zone),
       file_flush_interval_msec_(10000), drain_time_(600), parent_shutdown_time_(900),
       mode_(Server::Mode::Serve), max_stats_(ENVOY_DEFAULT_MAX_STATS), hot_restart_disabled_(false),
-      signal_handling_enabled_(true), mutex_tracing_enabled_(false), cpuset_threads_(false) {}
+      signal_handling_enabled_(true), mutex_tracing_enabled_(false), cpuset_threads_(false),
+      libevent_buffer_enabled_(false) {}
 
 } // namespace Envoy
diff --git a/source/server/options_impl.h b/source/server/options_impl.h
index b23faf2c34bbb..08e8e13927c3e 100644
--- a/source/server/options_impl.h
+++ b/source/server/options_impl.h
@@ -107,6 +107,7 @@ class OptionsImpl : public Server::Options, protected Logger::Loggable<Logger::I
   bool hotRestartDisabled() const override { return hot_restart_disabled_; }
   bool signalHandlingEnabled() const override { return signal_handling_enabled_; }
   bool mutexTracingEnabled() const override { return mutex_tracing_enabled_; }
+  bool libeventBufferEnabled() const override { return libevent_buffer_enabled_; }
   virtual Server::CommandLineOptionsPtr toCommandLineOptions() const override;
   void parseComponentLogLevels(const std::string& component_log_levels);
   bool cpusetThreadsEnabled() const override { return cpuset_threads_; }
@@ -141,6 +142,7 @@ class OptionsImpl : public Server::Options, protected Logger::Loggable<Logger::I
   bool signal_handling_enabled_;
   bool mutex_tracing_enabled_;
   bool cpuset_threads_;
+  bool libevent_buffer_enabled_;
   uint32_t count_;
 };
 
diff --git a/source/server/server.cc b/source/server/server.cc
index 62e9ae99a902f..b0d36829729a8 100644
--- a/source/server/server.cc
+++ b/source/server/server.cc
@@ -20,6 +20,7 @@
 
 #include "common/api/api_impl.h"
 #include "common/api/os_sys_calls_impl.h"
+#include "common/buffer/buffer_impl.h"
 #include "common/common/mutex_tracer_impl.h"
 #include "common/common/utility.h"
 #include "common/common/version.h"
@@ -220,6 +221,12 @@ void InstanceImpl::initialize(const Options& options,
             Registry::FactoryRegistry<
                 Configuration::UpstreamTransportSocketConfigFactory>::allFactoryNames());
 
+  // Enable the selected buffer implementation (old libevent evbuffer version or new native
+  // version) early in the initialization, before any buffers can be created.
+  Buffer::OwnedImpl::useOldImpl(options.libeventBufferEnabled());
+  ENVOY_LOG(info, "buffer implementation: {}",
+            Buffer::OwnedImpl().usesOldImpl() ? "old (libevent)" : "new");
+
   // Handle configuration that needs to take place prior to the main configuration load.
   InstanceUtil::loadBootstrapConfig(bootstrap_, options, api());
   bootstrap_config_update_time_ = time_source_.systemTime();
diff --git a/test/common/buffer/BUILD b/test/common/buffer/BUILD
index ee5036cd3e727..40c4953d43507 100644
--- a/test/common/buffer/BUILD
+++ b/test/common/buffer/BUILD
@@ -53,9 +53,11 @@ envoy_cc_test(
     name = "owned_impl_test",
     srcs = ["owned_impl_test.cc"],
     deps = [
+        ":utility_lib",
         "//source/common/buffer:buffer_lib",
         "//source/common/network:io_socket_handle_lib",
         "//test/mocks/api:api_mocks",
+        "//test/test_common:logging_lib",
         "//test/test_common:threadsafe_singleton_injector_lib",
     ],
 )
@@ -64,6 +66,7 @@ envoy_cc_test(
     name = "watermark_buffer_test",
     srcs = ["watermark_buffer_test.cc"],
     deps = [
+        ":utility_lib",
         "//source/common/buffer:buffer_lib",
         "//source/common/buffer:watermark_buffer_lib",
         "//source/common/network:io_socket_handle_lib",
@@ -74,6 +77,7 @@ envoy_cc_test(
     name = "zero_copy_input_stream_test",
     srcs = ["zero_copy_input_stream_test.cc"],
     deps = [
+        ":utility_lib",
         "//source/common/buffer:zero_copy_input_stream_lib",
     ],
 )
diff --git a/test/common/buffer/buffer_fuzz_test.cc b/test/common/buffer/buffer_fuzz_test.cc
index 6928e9c217f3d..5b9225860742d 100644
--- a/test/common/buffer/buffer_fuzz_test.cc
+++ b/test/common/buffer/buffer_fuzz_test.cc
@@ -273,7 +273,7 @@ uint32_t bufferAction(Context& ctxt, char insert_value, uint32_t max_alloc, Buff
     Buffer::RawSlice slices[1];
     const uint64_t slices_used = target_buffer.getRawSlices(slices, 1);
     if (linearize_size > 0) {
-      FUZZ_ASSERT(slices_used == 1);
+      FUZZ_ASSERT(slices_used >= 1);
       FUZZ_ASSERT(slices[0].len_ >= linearize_size);
     }
     break;
diff --git a/test/common/buffer/buffer_test.cc b/test/common/buffer/buffer_test.cc
index c70ea116bdfbc..5df98389460ab 100644
--- a/test/common/buffer/buffer_test.cc
+++ b/test/common/buffer/buffer_test.cc
@@ -14,9 +14,269 @@ namespace Envoy {
 namespace Buffer {
 namespace {
 
-TEST(BufferHelperTest, PeekI8) {
+class DummySlice : public Slice {
+public:
+  DummySlice(const std::string& data, const std::function<void()>& deletion_callback)
+      : Slice(0, data.size(), data.size()), deletion_callback_(deletion_callback) {
+    base_ = reinterpret_cast<uint8_t*>(const_cast<char*>(data.c_str()));
+  }
+  ~DummySlice() override {
+    if (deletion_callback_ != nullptr) {
+      deletion_callback_();
+    }
+  }
+
+private:
+  const std::function<void()> deletion_callback_;
+};
+
+class OwnedSliceTest : public testing::Test {
+protected:
+  static void expectReservationSuccess(const Slice::Reservation& reservation, const Slice& slice,
+                                       uint64_t reservation_size) {
+    EXPECT_NE(nullptr, reservation.mem_);
+    EXPECT_EQ(static_cast<const uint8_t*>(slice.data()) + slice.dataSize(), reservation.mem_);
+    EXPECT_EQ(reservation_size, reservation.len_);
+    EXPECT_EQ(0, slice.reservableSize());
+  }
+
+  static void expectReservationFailure(const Slice::Reservation& reservation, const Slice& slice,
+                                       uint64_t reservable_size) {
+    EXPECT_EQ(nullptr, reservation.mem_);
+    EXPECT_EQ(0, reservation.mem_);
+    EXPECT_EQ(reservable_size, slice.reservableSize());
+  }
+
+  static void expectCommitSuccess(bool committed, const Slice& slice, uint64_t data_size,
+                                  uint64_t reservable_size) {
+    EXPECT_TRUE(committed);
+    EXPECT_EQ(data_size, slice.dataSize());
+    EXPECT_EQ(reservable_size, slice.reservableSize());
+  }
+};
+
+bool sliceMatches(const SlicePtr& slice, const std::string& expected) {
+  return slice != nullptr && slice->dataSize() == expected.size() &&
+         memcmp(slice->data(), expected.data(), expected.size()) == 0;
+}
+
+TEST_F(OwnedSliceTest, Create) {
+  static constexpr uint64_t Sizes[] = {0, 1, 64, 4096 - sizeof(OwnedSlice), 65535};
+  for (const auto size : Sizes) {
+    auto slice = OwnedSlice::create(size);
+    EXPECT_NE(nullptr, slice->data());
+    EXPECT_EQ(0, slice->dataSize());
+    EXPECT_LE(size, slice->reservableSize());
+  }
+}
+
+TEST_F(OwnedSliceTest, ReserveCommit) {
+  auto slice = OwnedSlice::create(100);
+  const uint64_t initial_capacity = slice->reservableSize();
+  EXPECT_LE(100, initial_capacity);
+
+  {
+    // Verify that a zero-byte reservation is rejected.
+    Slice::Reservation reservation = slice->reserve(0);
+    expectReservationFailure(reservation, *slice, initial_capacity);
+  }
+
+  {
+    // Create a reservation smaller than the reservable size.
+    // It should reserve the exact number of bytes requested.
+    Slice::Reservation reservation = slice->reserve(10);
+    expectReservationSuccess(reservation, *slice, 10);
+
+    // Request a second reservation while the first reservation remains uncommitted.
+    // This should fail.
+    EXPECT_EQ(0, slice->reservableSize());
+    Slice::Reservation reservation2 = slice->reserve(1);
+    expectReservationFailure(reservation2, *slice, 0);
+
+    // Commit the entire reserved size.
+    bool committed = slice->commit(reservation);
+    expectCommitSuccess(committed, *slice, 10, initial_capacity - 10);
+
+    // Verify that a reservation can only be committed once.
+    EXPECT_FALSE(slice->commit(reservation));
+  }
+
+  {
+    // Request another reservation, and commit only part of it.
+    Slice::Reservation reservation = slice->reserve(10);
+    expectReservationSuccess(reservation, *slice, 10);
+    reservation.len_ = 5;
+    bool committed = slice->commit(reservation);
+    expectCommitSuccess(committed, *slice, 15, initial_capacity - 15);
+  }
+
+  {
+    // Request another reservation, and commit only part of it.
+    Slice::Reservation reservation = slice->reserve(10);
+    expectReservationSuccess(reservation, *slice, 10);
+    reservation.len_ = 5;
+    bool committed = slice->commit(reservation);
+    expectCommitSuccess(committed, *slice, 20, initial_capacity - 20);
+  }
+
+  {
+    // Request another reservation, and commit zero bytes of it.
+    // This should clear the reservation.
+    Slice::Reservation reservation = slice->reserve(10);
+    expectReservationSuccess(reservation, *slice, 10);
+    reservation.len_ = 0;
+    bool committed = slice->commit(reservation);
+    expectCommitSuccess(committed, *slice, 20, initial_capacity - 20);
+  }
+
+  {
+    // Try to commit a reservation from the wrong slice, and verify that the slice rejects it.
+    Slice::Reservation reservation = slice->reserve(10);
+    expectReservationSuccess(reservation, *slice, 10);
+    auto other_slice = OwnedSlice::create(100);
+    Slice::Reservation other_reservation = other_slice->reserve(10);
+    expectReservationSuccess(other_reservation, *other_slice, 10);
+    EXPECT_FALSE(slice->commit(other_reservation));
+    EXPECT_FALSE(other_slice->commit(reservation));
+
+    // Commit the reservations to the proper slices to clear them.
+    reservation.len_ = 0;
+    bool committed = slice->commit(reservation);
+    EXPECT_TRUE(committed);
+    other_reservation.len_ = 0;
+    committed = other_slice->commit(other_reservation);
+    EXPECT_TRUE(committed);
+  }
+
+  {
+    // Try to reserve more space than is available in the slice.
+    uint64_t reservable_size = slice->reservableSize();
+    Slice::Reservation reservation = slice->reserve(reservable_size + 1);
+    expectReservationSuccess(reservation, *slice, reservable_size);
+    bool committed = slice->commit(reservation);
+    expectCommitSuccess(committed, *slice, initial_capacity, 0);
+  }
+
+  {
+    // Now that the view has no more reservable space, verify that it rejects
+    // subsequent reservation requests.
+    Slice::Reservation reservation = slice->reserve(1);
+    expectReservationFailure(reservation, *slice, 0);
+  }
+}
+
+TEST_F(OwnedSliceTest, Drain) {
+  // Create a slice and commit all the available space.
+  auto slice = OwnedSlice::create(100);
+  Slice::Reservation reservation = slice->reserve(slice->reservableSize());
+  bool committed = slice->commit(reservation);
+  EXPECT_TRUE(committed);
+  EXPECT_EQ(0, slice->reservableSize());
+
+  // Drain some data from the front of the view and verify that the data start moves accordingly.
+  const uint8_t* original_data = static_cast<const uint8_t*>(slice->data());
+  uint64_t original_size = slice->dataSize();
+  slice->drain(0);
+  EXPECT_EQ(original_data, slice->data());
+  EXPECT_EQ(original_size, slice->dataSize());
+  slice->drain(10);
+  EXPECT_EQ(original_data + 10, slice->data());
+  EXPECT_EQ(original_size - 10, slice->dataSize());
+  slice->drain(50);
+  EXPECT_EQ(original_data + 60, slice->data());
+  EXPECT_EQ(original_size - 60, slice->dataSize());
+
+  // Drain all the remaining data.
+  slice->drain(slice->dataSize());
+  EXPECT_EQ(0, slice->dataSize());
+  EXPECT_EQ(original_size, slice->reservableSize());
+}
+
+TEST(UnownedSliceTest, CreateDelete) {
+  constexpr char input[] = "hello world";
+  bool release_callback_called = false;
+  BufferFragmentImpl fragment(
+      input, sizeof(input) - 1,
+      [&release_callback_called](const void*, size_t, const BufferFragmentImpl*) {
+        release_callback_called = true;
+      });
+  auto slice = std::make_unique<UnownedSlice>(fragment);
+  EXPECT_EQ(11, slice->dataSize());
+  EXPECT_EQ(0, slice->reservableSize());
+  EXPECT_EQ(0, memcmp(slice->data(), input, slice->dataSize()));
+  EXPECT_FALSE(release_callback_called);
+  slice.reset(nullptr);
+  EXPECT_TRUE(release_callback_called);
+}
+
+TEST(SliceDequeTest, CreateDelete) {
+  bool slice1_deleted = false;
+  bool slice2_deleted = false;
+  bool slice3_deleted = false;
+
+  {
+    // Create an empty deque.
+    SliceDeque slices;
+    EXPECT_TRUE(slices.empty());
+    EXPECT_EQ(0, slices.size());
+
+    // Append a view to the deque.
+    const std::string slice1 = "slice1";
+    slices.emplace_back(
+        std::make_unique<DummySlice>(slice1, [&slice1_deleted]() { slice1_deleted = true; }));
+    EXPECT_FALSE(slices.empty());
+    ASSERT_EQ(1, slices.size());
+    EXPECT_FALSE(slice1_deleted);
+    EXPECT_TRUE(sliceMatches(slices.front(), slice1));
+
+    // Append another view to the deque, and verify that both views are accessible.
+    const std::string slice2 = "slice2";
+    slices.emplace_back(
+        std::make_unique<DummySlice>(slice2, [&slice2_deleted]() { slice2_deleted = true; }));
+    EXPECT_FALSE(slices.empty());
+    ASSERT_EQ(2, slices.size());
+    EXPECT_FALSE(slice1_deleted);
+    EXPECT_FALSE(slice2_deleted);
+    EXPECT_TRUE(sliceMatches(slices.front(), slice1));
+    EXPECT_TRUE(sliceMatches(slices.back(), slice2));
+
+    // Prepend a view to the deque, to exercise the ring buffer wraparound case.
+    const std::string slice3 = "slice3";
+    slices.emplace_front(
+        std::make_unique<DummySlice>(slice3, [&slice3_deleted]() { slice3_deleted = true; }));
+    EXPECT_FALSE(slices.empty());
+    ASSERT_EQ(3, slices.size());
+    EXPECT_FALSE(slice1_deleted);
+    EXPECT_FALSE(slice2_deleted);
+    EXPECT_FALSE(slice3_deleted);
+    EXPECT_TRUE(sliceMatches(slices.front(), slice3));
+    EXPECT_TRUE(sliceMatches(slices.back(), slice2));
+
+    // Remove the first view from the deque, and verify that its slice is deleted.
+    slices.pop_front();
+    EXPECT_FALSE(slices.empty());
+    ASSERT_EQ(2, slices.size());
+    EXPECT_FALSE(slice1_deleted);
+    EXPECT_FALSE(slice2_deleted);
+    EXPECT_TRUE(slice3_deleted);
+    EXPECT_TRUE(sliceMatches(slices.front(), slice1));
+    EXPECT_TRUE(sliceMatches(slices.back(), slice2));
+  }
+
+  EXPECT_TRUE(slice1_deleted);
+  EXPECT_TRUE(slice2_deleted);
+  EXPECT_TRUE(slice3_deleted);
+}
+
+class BufferHelperTest : public BufferImplementationParamTest {};
+
+INSTANTIATE_TEST_CASE_P(BufferHelperTest, BufferHelperTest,
+                        testing::ValuesIn({BufferImplementation::Old, BufferImplementation::New}));
+
+TEST_P(BufferHelperTest, PeekI8) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 0xFE});
     EXPECT_EQ(buffer.peekInt<int8_t>(), 0);
     EXPECT_EQ(buffer.peekInt<int8_t>(0), 0);
@@ -27,19 +287,22 @@ TEST(BufferHelperTest, PeekI8) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekInt<int8_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeByte(0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekInt<int8_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEI16) {
+TEST_P(BufferHelperTest, PeekLEI16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<int16_t>(), 0x0100);
     EXPECT_EQ(buffer.peekLEInt<int16_t>(0), 0x0100);
@@ -51,19 +314,22 @@ TEST(BufferHelperTest, PeekLEI16) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int16_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 2, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int16_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEI32) {
+TEST_P(BufferHelperTest, PeekLEI32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<int32_t>(), 0x03020100);
     EXPECT_EQ(buffer.peekLEInt<int32_t>(0), 0x03020100);
@@ -74,19 +340,22 @@ TEST(BufferHelperTest, PeekLEI32) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int32_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 4, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int32_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEI64) {
+TEST_P(BufferHelperTest, PeekLEI64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<int64_t>(), 0x0706050403020100);
     EXPECT_EQ(buffer.peekLEInt<int64_t>(0), 0x0706050403020100);
@@ -106,6 +375,7 @@ TEST(BufferHelperTest, PeekLEI64) {
   {
     // signed
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0xFF, 0xFF, 0xFF, 0x00, 0xFF, 0xFE, 0xFF, 0xFF});
     EXPECT_EQ((buffer.peekLEInt<int64_t, 2>()), -1);
     EXPECT_EQ((buffer.peekLEInt<int64_t, 2>(2)), 255);  // 0x00FF
@@ -115,6 +385,7 @@ TEST(BufferHelperTest, PeekLEI64) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF});
     EXPECT_THROW_WITH_MESSAGE(
         (buffer.peekLEInt<int64_t, sizeof(int64_t)>(buffer.length() - sizeof(int64_t) + 1)),
@@ -123,19 +394,22 @@ TEST(BufferHelperTest, PeekLEI64) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int64_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 8, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<int64_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEU16) {
+TEST_P(BufferHelperTest, PeekLEU16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<uint16_t>(), 0x0100);
     EXPECT_EQ(buffer.peekLEInt<uint16_t>(0), 0x0100);
@@ -146,19 +420,22 @@ TEST(BufferHelperTest, PeekLEU16) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint16_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 2, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint16_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEU32) {
+TEST_P(BufferHelperTest, PeekLEU32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<uint32_t>(), 0x03020100);
     EXPECT_EQ(buffer.peekLEInt<uint32_t>(0), 0x03020100);
@@ -169,19 +446,22 @@ TEST(BufferHelperTest, PeekLEU32) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint32_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 4, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint32_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekLEU64) {
+TEST_P(BufferHelperTest, PeekLEU64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekLEInt<uint64_t>(), 0x0706050403020100);
     EXPECT_EQ(buffer.peekLEInt<uint64_t>(0), 0x0706050403020100);
@@ -192,19 +472,22 @@ TEST(BufferHelperTest, PeekLEU64) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint64_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 8, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekLEInt<uint64_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEI16) {
+TEST_P(BufferHelperTest, PeekBEI16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<int16_t>(), 1);
     EXPECT_EQ(buffer.peekBEInt<int16_t>(0), 1);
@@ -216,19 +499,22 @@ TEST(BufferHelperTest, PeekBEI16) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int16_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 2, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int16_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEI32) {
+TEST_P(BufferHelperTest, PeekBEI32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<int32_t>(), 0x00010203);
     EXPECT_EQ(buffer.peekBEInt<int32_t>(0), 0x00010203);
@@ -239,19 +525,22 @@ TEST(BufferHelperTest, PeekBEI32) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int32_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 4, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int32_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEI64) {
+TEST_P(BufferHelperTest, PeekBEI64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<int64_t>(), 0x0001020304050607);
     EXPECT_EQ(buffer.peekBEInt<int64_t>(0), 0x0001020304050607);
@@ -270,6 +559,7 @@ TEST(BufferHelperTest, PeekBEI64) {
   {
     // signed
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0xFF, 0xFF, 0xFF, 0x00, 0xFF, 0xFF, 0xFF, 0xFE});
     EXPECT_EQ((buffer.peekBEInt<int64_t, 2>()), -1);
     EXPECT_EQ((buffer.peekBEInt<int64_t, 2>(2)), -256); // 0xFF00
@@ -279,6 +569,7 @@ TEST(BufferHelperTest, PeekBEI64) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF});
     EXPECT_THROW_WITH_MESSAGE(
         (buffer.peekBEInt<int64_t, sizeof(int64_t)>(buffer.length() - sizeof(int64_t) + 1)),
@@ -287,19 +578,22 @@ TEST(BufferHelperTest, PeekBEI64) {
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int64_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 8, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<int64_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEU16) {
+TEST_P(BufferHelperTest, PeekBEU16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<uint16_t>(), 1);
     EXPECT_EQ(buffer.peekBEInt<uint16_t>(0), 1);
@@ -310,19 +604,22 @@ TEST(BufferHelperTest, PeekBEU16) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint16_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 2, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint16_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEU32) {
+TEST_P(BufferHelperTest, PeekBEU32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<uint32_t>(), 0x00010203);
     EXPECT_EQ(buffer.peekBEInt<uint32_t>(0), 0x00010203);
@@ -333,19 +630,22 @@ TEST(BufferHelperTest, PeekBEU32) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint32_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 4, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint32_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, PeekBEU64) {
+TEST_P(BufferHelperTest, PeekBEU64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
     EXPECT_EQ(buffer.peekBEInt<uint64_t>(), 0x0001020304050607);
     EXPECT_EQ(buffer.peekBEInt<uint64_t>(0), 0x0001020304050607);
@@ -356,18 +656,21 @@ TEST(BufferHelperTest, PeekBEU64) {
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint64_t>(0), EnvoyException, "buffer underflow");
   }
 
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     addRepeated(buffer, 8, 0);
     EXPECT_THROW_WITH_MESSAGE(buffer.peekBEInt<uint64_t>(1), EnvoyException, "buffer underflow");
   }
 }
 
-TEST(BufferHelperTest, DrainI8) {
+TEST_P(BufferHelperTest, DrainI8) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 0xFE});
   EXPECT_EQ(buffer.drainInt<int8_t>(), 0);
   EXPECT_EQ(buffer.drainInt<int8_t>(), 1);
@@ -375,8 +678,9 @@ TEST(BufferHelperTest, DrainI8) {
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainLEI16) {
+TEST_P(BufferHelperTest, DrainLEI16) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainLEInt<int16_t>(), 0x0100);
   EXPECT_EQ(buffer.drainLEInt<int16_t>(), 0x0302);
@@ -384,40 +688,45 @@ TEST(BufferHelperTest, DrainLEI16) {
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainLEI32) {
+TEST_P(BufferHelperTest, DrainLEI32) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainLEInt<int32_t>(), 0x03020100);
   EXPECT_EQ(buffer.drainLEInt<int32_t>(), -1);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainLEI64) {
+TEST_P(BufferHelperTest, DrainLEI64) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainLEInt<int64_t>(), 0x0706050403020100);
   EXPECT_EQ(buffer.drainLEInt<int64_t>(), -1);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainLEU32) {
+TEST_P(BufferHelperTest, DrainLEU32) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainLEInt<uint32_t>(), 0x03020100);
   EXPECT_EQ(buffer.drainLEInt<uint32_t>(), 0xFFFFFFFF);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainLEU64) {
+TEST_P(BufferHelperTest, DrainLEU64) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainLEInt<uint64_t>(), 0x0706050403020100);
   EXPECT_EQ(buffer.drainLEInt<uint64_t>(), 0xFFFFFFFFFFFFFFFF);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainBEI16) {
+TEST_P(BufferHelperTest, DrainBEI16) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainBEInt<int16_t>(), 1);
   EXPECT_EQ(buffer.drainBEInt<int16_t>(), 0x0203);
@@ -425,40 +734,45 @@ TEST(BufferHelperTest, DrainBEI16) {
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainBEI32) {
+TEST_P(BufferHelperTest, DrainBEI32) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainBEInt<int32_t>(), 0x00010203);
   EXPECT_EQ(buffer.drainBEInt<int32_t>(), -1);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainBEI64) {
+TEST_P(BufferHelperTest, DrainBEI64) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainBEInt<int64_t>(), 0x0001020304050607);
   EXPECT_EQ(buffer.drainBEInt<int64_t>(), -1);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainBEU32) {
+TEST_P(BufferHelperTest, DrainBEU32) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainBEInt<uint32_t>(), 0x00010203);
   EXPECT_EQ(buffer.drainBEInt<uint32_t>(), 0xFFFFFFFF);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, DrainBEU64) {
+TEST_P(BufferHelperTest, DrainBEU64) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   addSeq(buffer, {0, 1, 2, 3, 4, 5, 6, 7, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF});
   EXPECT_EQ(buffer.drainBEInt<uint64_t>(), 0x0001020304050607);
   EXPECT_EQ(buffer.drainBEInt<uint64_t>(), 0xFFFFFFFFFFFFFFFF);
   EXPECT_EQ(buffer.length(), 0);
 }
 
-TEST(BufferHelperTest, WriteI8) {
+TEST_P(BufferHelperTest, WriteI8) {
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.writeByte(-128);
   buffer.writeByte(-1);
   buffer.writeByte(0);
@@ -468,229 +782,269 @@ TEST(BufferHelperTest, WriteI8) {
   EXPECT_EQ(std::string("\x80\xFF\0\x1\x7F", 5), buffer.toString());
 }
 
-TEST(BufferHelperTest, WriteLEI16) {
+TEST_P(BufferHelperTest, WriteLEI16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int16_t>(std::numeric_limits<int16_t>::min());
     EXPECT_EQ(std::string("\0\x80", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int16_t>(0);
     EXPECT_EQ(std::string("\0\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int16_t>(1);
     EXPECT_EQ(std::string("\x1\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int16_t>(std::numeric_limits<int16_t>::max());
     EXPECT_EQ("\xFF\x7F", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteLEU16) {
+TEST_P(BufferHelperTest, WriteLEU16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint16_t>(0);
     EXPECT_EQ(std::string("\0\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint16_t>(1);
     EXPECT_EQ(std::string("\x1\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint16_t>(static_cast<uint16_t>(std::numeric_limits<int16_t>::max()) + 1);
     EXPECT_EQ(std::string("\0\x80", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint16_t>(std::numeric_limits<uint16_t>::max());
     EXPECT_EQ("\xFF\xFF", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteLEI32) {
+TEST_P(BufferHelperTest, WriteLEI32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int32_t>(std::numeric_limits<int32_t>::min());
     EXPECT_EQ(std::string("\0\0\0\x80", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int32_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int32_t>(1);
     EXPECT_EQ(std::string("\x1\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int32_t>(std::numeric_limits<int32_t>::max());
     EXPECT_EQ("\xFF\xFF\xFF\x7F", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteLEU32) {
+TEST_P(BufferHelperTest, WriteLEU32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint32_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint32_t>(1);
     EXPECT_EQ(std::string("\x1\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint32_t>(static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) + 1);
     EXPECT_EQ(std::string("\0\0\0\x80", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<uint32_t>(std::numeric_limits<uint32_t>::max());
     EXPECT_EQ("\xFF\xFF\xFF\xFF", buffer.toString());
   }
 }
-TEST(BufferHelperTest, WriteLEI64) {
+TEST_P(BufferHelperTest, WriteLEI64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int64_t>(std::numeric_limits<int64_t>::min());
     EXPECT_EQ(std::string("\0\0\0\0\0\0\0\x80", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int64_t>(1);
     EXPECT_EQ(std::string("\x1\0\0\0\0\0\0\0", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int64_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0\0\0\0\0", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeLEInt<int64_t>(std::numeric_limits<int64_t>::max());
     EXPECT_EQ("\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x7F", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteBEI16) {
+TEST_P(BufferHelperTest, WriteBEI16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int16_t>(std::numeric_limits<int16_t>::min());
     EXPECT_EQ(std::string("\x80\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int16_t>(0);
     EXPECT_EQ(std::string("\0\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int16_t>(1);
     EXPECT_EQ(std::string("\0\x1", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int16_t>(std::numeric_limits<int16_t>::max());
     EXPECT_EQ("\x7F\xFF", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteBEU16) {
+TEST_P(BufferHelperTest, WriteBEU16) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint16_t>(0);
     EXPECT_EQ(std::string("\0\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint16_t>(1);
     EXPECT_EQ(std::string("\0\x1", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint16_t>(static_cast<uint16_t>(std::numeric_limits<int16_t>::max()) + 1);
     EXPECT_EQ(std::string("\x80\0", 2), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint16_t>(std::numeric_limits<uint16_t>::max());
     EXPECT_EQ("\xFF\xFF", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteBEI32) {
+TEST_P(BufferHelperTest, WriteBEI32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int32_t>(std::numeric_limits<int32_t>::min());
     EXPECT_EQ(std::string("\x80\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int32_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int32_t>(1);
     EXPECT_EQ(std::string("\0\0\0\x1", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int32_t>(std::numeric_limits<int32_t>::max());
     EXPECT_EQ("\x7F\xFF\xFF\xFF", buffer.toString());
   }
 }
 
-TEST(BufferHelperTest, WriteBEU32) {
+TEST_P(BufferHelperTest, WriteBEU32) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint32_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint32_t>(1);
     EXPECT_EQ(std::string("\0\0\0\x1", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint32_t>(static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) + 1);
     EXPECT_EQ(std::string("\x80\0\0\0", 4), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<uint32_t>(std::numeric_limits<uint32_t>::max());
     EXPECT_EQ("\xFF\xFF\xFF\xFF", buffer.toString());
   }
 }
-TEST(BufferHelperTest, WriteBEI64) {
+TEST_P(BufferHelperTest, WriteBEI64) {
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int64_t>(std::numeric_limits<int64_t>::min());
     EXPECT_EQ(std::string("\x80\0\0\0\0\0\0\0\0", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int64_t>(1);
     EXPECT_EQ(std::string("\0\0\0\0\0\0\0\x1", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int64_t>(0);
     EXPECT_EQ(std::string("\0\0\0\0\0\0\0\0", 8), buffer.toString());
   }
   {
     Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
     buffer.writeBEInt<int64_t>(std::numeric_limits<int64_t>::max());
     EXPECT_EQ("\x7F\xFF\xFF\xFF\xFF\xFF\xFF\xFF", buffer.toString());
   }
diff --git a/test/common/buffer/owned_impl_test.cc b/test/common/buffer/owned_impl_test.cc
index c4cc2b248c1b8..07a658bf80bd5 100644
--- a/test/common/buffer/owned_impl_test.cc
+++ b/test/common/buffer/owned_impl_test.cc
@@ -3,7 +3,9 @@
 #include "common/buffer/buffer_impl.h"
 #include "common/network/io_socket_handle_impl.h"
 
+#include "test/common/buffer/utility.h"
 #include "test/mocks/api/mocks.h"
+#include "test/test_common/logging.h"
 #include "test/test_common/threadsafe_singleton_injector.h"
 
 #include "absl/strings/str_cat.h"
@@ -17,15 +19,31 @@ namespace Envoy {
 namespace Buffer {
 namespace {
 
-class OwnedImplTest : public testing::Test {
+class OwnedImplTest : public BufferImplementationParamTest {
 public:
   bool release_callback_called_ = false;
+
+protected:
+  static void clearReservation(Buffer::RawSlice* iovecs, uint64_t num_iovecs, OwnedImpl& buffer) {
+    for (uint64_t i = 0; i < num_iovecs; i++) {
+      iovecs[i].len_ = 0;
+    }
+    buffer.commit(iovecs, num_iovecs);
+  }
+
+  static void commitReservation(Buffer::RawSlice* iovecs, uint64_t num_iovecs, OwnedImpl& buffer) {
+    buffer.commit(iovecs, num_iovecs);
+  }
 };
 
-TEST_F(OwnedImplTest, AddBufferFragmentNoCleanup) {
+INSTANTIATE_TEST_CASE_P(OwnedImplTest, OwnedImplTest,
+                        testing::ValuesIn({BufferImplementation::Old, BufferImplementation::New}));
+
+TEST_P(OwnedImplTest, AddBufferFragmentNoCleanup) {
   char input[] = "hello world";
   BufferFragmentImpl frag(input, 11, nullptr);
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.addBufferFragment(frag);
   EXPECT_EQ(11, buffer.length());
 
@@ -33,12 +51,13 @@ TEST_F(OwnedImplTest, AddBufferFragmentNoCleanup) {
   EXPECT_EQ(0, buffer.length());
 }
 
-TEST_F(OwnedImplTest, AddBufferFragmentWithCleanup) {
+TEST_P(OwnedImplTest, AddBufferFragmentWithCleanup) {
   char input[] = "hello world";
   BufferFragmentImpl frag(input, 11, [this](const void*, size_t, const BufferFragmentImpl*) {
     release_callback_called_ = true;
   });
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.addBufferFragment(frag);
   EXPECT_EQ(11, buffer.length());
 
@@ -51,7 +70,7 @@ TEST_F(OwnedImplTest, AddBufferFragmentWithCleanup) {
   EXPECT_TRUE(release_callback_called_);
 }
 
-TEST_F(OwnedImplTest, AddBufferFragmentDynamicAllocation) {
+TEST_P(OwnedImplTest, AddBufferFragmentDynamicAllocation) {
   char input_stack[] = "hello world";
   char* input = new char[11];
   std::copy(input_stack, input_stack + 11, input);
@@ -64,6 +83,7 @@ TEST_F(OwnedImplTest, AddBufferFragmentDynamicAllocation) {
       });
 
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.addBufferFragment(*frag);
   EXPECT_EQ(11, buffer.length());
 
@@ -76,19 +96,59 @@ TEST_F(OwnedImplTest, AddBufferFragmentDynamicAllocation) {
   EXPECT_TRUE(release_callback_called_);
 }
 
-TEST_F(OwnedImplTest, Prepend) {
-  std::string suffix = "World!", prefix = "Hello, ";
+TEST_P(OwnedImplTest, Add) {
+  const std::string string1 = "Hello, ", string2 = "World!";
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
+
+  buffer.add(string1);
+  EXPECT_EQ(string1.size(), buffer.length());
+  EXPECT_EQ(string1, buffer.toString());
+
+  buffer.add(string2);
+  EXPECT_EQ(string1.size() + string2.size(), buffer.length());
+  EXPECT_EQ(string1 + string2, buffer.toString());
+
+  // Append a large string that will only partially fit in the space remaining
+  // at the end of the buffer.
+  std::string big_suffix;
+  big_suffix.reserve(16385);
+  for (unsigned i = 0; i < 16; i++) {
+    big_suffix += std::string(1024, 'A' + i);
+  }
+  big_suffix.push_back('-');
+  buffer.add(big_suffix);
+  EXPECT_EQ(string1.size() + string2.size() + big_suffix.size(), buffer.length());
+  EXPECT_EQ(string1 + string2 + big_suffix, buffer.toString());
+}
+
+TEST_P(OwnedImplTest, Prepend) {
+  const std::string suffix = "World!", prefix = "Hello, ";
+  Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.add(suffix);
   buffer.prepend(prefix);
 
   EXPECT_EQ(suffix.size() + prefix.size(), buffer.length());
   EXPECT_EQ(prefix + suffix, buffer.toString());
+
+  // Prepend a large string that will only partially fit in the space remaining
+  // at the front of the buffer.
+  std::string big_prefix;
+  big_prefix.reserve(16385);
+  for (unsigned i = 0; i < 16; i++) {
+    big_prefix += std::string(1024, 'A' + i);
+  }
+  big_prefix.push_back('-');
+  buffer.prepend(big_prefix);
+  EXPECT_EQ(big_prefix.size() + prefix.size() + suffix.size(), buffer.length());
+  EXPECT_EQ(big_prefix + prefix + suffix, buffer.toString());
 }
 
-TEST_F(OwnedImplTest, PrependToEmptyBuffer) {
+TEST_P(OwnedImplTest, PrependToEmptyBuffer) {
   std::string data = "Hello, World!";
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.prepend(data);
 
   EXPECT_EQ(data.size(), buffer.length());
@@ -100,11 +160,13 @@ TEST_F(OwnedImplTest, PrependToEmptyBuffer) {
   EXPECT_EQ(data, buffer.toString());
 }
 
-TEST_F(OwnedImplTest, PrependBuffer) {
+TEST_P(OwnedImplTest, PrependBuffer) {
   std::string suffix = "World!", prefix = "Hello, ";
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   buffer.add(suffix);
   Buffer::OwnedImpl prefixBuffer;
+  verifyImplementation(buffer);
   prefixBuffer.add(prefix);
 
   buffer.prepend(prefixBuffer);
@@ -114,11 +176,12 @@ TEST_F(OwnedImplTest, PrependBuffer) {
   EXPECT_EQ(0, prefixBuffer.length());
 }
 
-TEST_F(OwnedImplTest, Write) {
+TEST_P(OwnedImplTest, Write) {
   Api::MockOsSysCalls os_sys_calls;
   TestThreadsafeSingletonInjector<Api::OsSysCallsImpl> os_calls(&os_sys_calls);
 
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   Network::IoSocketHandleImpl io_handle;
   buffer.add("example");
   EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{7, 0}));
@@ -164,11 +227,12 @@ TEST_F(OwnedImplTest, Write) {
   EXPECT_EQ(0, buffer.length());
 }
 
-TEST_F(OwnedImplTest, Read) {
+TEST_P(OwnedImplTest, Read) {
   Api::MockOsSysCalls os_sys_calls;
   TestThreadsafeSingletonInjector<Api::OsSysCallsImpl> os_calls(&os_sys_calls);
 
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   Network::IoSocketHandleImpl io_handle;
   EXPECT_CALL(os_sys_calls, readv(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{0, 0}));
   Api::IoCallUint64Result result = buffer.read(io_handle, 100);
@@ -194,8 +258,119 @@ TEST_F(OwnedImplTest, Read) {
   EXPECT_EQ(0, buffer.length());
 }
 
-TEST_F(OwnedImplTest, ToString) {
+TEST_P(OwnedImplTest, ReserveCommit) {
+  // This fragment will later be added to the buffer. It is declared in an enclosing scope to
+  // ensure it is not destructed until after the buffer is.
+  const std::string input = "Hello, world";
+  BufferFragmentImpl fragment(input.c_str(), input.size(), nullptr);
+
+  {
+    Buffer::OwnedImpl buffer;
+    verifyImplementation(buffer);
+
+    // A zero-byte reservation should fail.
+    static constexpr uint64_t NumIovecs = 16;
+    Buffer::RawSlice iovecs[NumIovecs];
+    uint64_t num_reserved = buffer.reserve(0, iovecs, NumIovecs);
+    EXPECT_EQ(0, num_reserved);
+    clearReservation(iovecs, num_reserved, buffer);
+    EXPECT_EQ(0, buffer.length());
+
+    // Test and commit a small reservation. This should succeed.
+    num_reserved = buffer.reserve(1, iovecs, NumIovecs);
+    EXPECT_EQ(1, num_reserved);
+    // The implementation might provide a bigger reservation than requested.
+    EXPECT_LE(1, iovecs[0].len_);
+    iovecs[0].len_ = 1;
+    commitReservation(iovecs, num_reserved, buffer);
+    EXPECT_EQ(1, buffer.length());
+
+    // The remaining tests validate internal optimizations of the new deque-of-slices
+    // implementation, so they're not valid for the old evbuffer implementation.
+    if (buffer.usesOldImpl()) {
+      return;
+    }
+
+    // Request a reservation that fits in the remaining space at the end of the last slice.
+    num_reserved = buffer.reserve(1, iovecs, NumIovecs);
+    EXPECT_EQ(1, num_reserved);
+    EXPECT_LE(1, iovecs[0].len_);
+    iovecs[0].len_ = 1;
+    const void* slice1 = iovecs[0].mem_;
+    clearReservation(iovecs, num_reserved, buffer);
+
+    // Request a reservation that is too large to fit in the remaining space at the end of
+    // the last slice, and allow the buffer to use only one slice. This should result in the
+    // creation of a new slice within the buffer.
+    num_reserved = buffer.reserve(4096 - sizeof(OwnedSlice), iovecs, 1);
+    const void* slice2 = iovecs[0].mem_;
+    EXPECT_EQ(1, num_reserved);
+    EXPECT_NE(slice1, slice2);
+    clearReservation(iovecs, num_reserved, buffer);
+
+    // Request the same size reservation, but allow the buffer to use multiple slices. This
+    // should result in the buffer splitting the reservation between its last two slices.
+    num_reserved = buffer.reserve(4096 - sizeof(OwnedSlice), iovecs, NumIovecs);
+    EXPECT_EQ(2, num_reserved);
+    EXPECT_EQ(slice1, iovecs[0].mem_);
+    EXPECT_EQ(slice2, iovecs[1].mem_);
+    clearReservation(iovecs, num_reserved, buffer);
+
+    // Request a reservation that too big to fit in the existing slices. This should result
+    // in the creation of a third slice.
+    num_reserved = buffer.reserve(8192, iovecs, NumIovecs);
+    EXPECT_EQ(3, num_reserved);
+    EXPECT_EQ(slice1, iovecs[0].mem_);
+    EXPECT_EQ(slice2, iovecs[1].mem_);
+    const void* slice3 = iovecs[2].mem_;
+    clearReservation(iovecs, num_reserved, buffer);
+
+    // Append a fragment to the buffer, and then request a small reservation. The buffer
+    // should make a new slice to satisfy the reservation; it cannot safely use any of
+    // the previously seen slices, because they are no longer at the end of the buffer.
+    buffer.addBufferFragment(fragment);
+    EXPECT_EQ(13, buffer.length());
+    num_reserved = buffer.reserve(1, iovecs, NumIovecs);
+    EXPECT_EQ(1, num_reserved);
+    EXPECT_NE(slice1, iovecs[0].mem_);
+    EXPECT_NE(slice2, iovecs[0].mem_);
+    EXPECT_NE(slice3, iovecs[0].mem_);
+    commitReservation(iovecs, num_reserved, buffer);
+    EXPECT_EQ(14, buffer.length());
+  }
+}
+
+TEST_P(OwnedImplTest, Search) {
+  // Populate a buffer with a string split across many small slices, to
+  // exercise edge cases in the search implementation.
+  static const char* Inputs[] = {"ab", "a", "", "aaa", "b", "a", "aaa", "ab", "a"};
   Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
+  for (const auto& input : Inputs) {
+    buffer.appendSliceForTest(input);
+  }
+  EXPECT_STREQ("abaaaabaaaaaba", buffer.toString().c_str());
+
+  EXPECT_EQ(-1, buffer.search("c", 1, 0));
+  EXPECT_EQ(0, buffer.search("", 0, 0));
+  EXPECT_EQ(buffer.length(), buffer.search("", 0, buffer.length()));
+  EXPECT_EQ(-1, buffer.search("", 0, buffer.length() + 1));
+  EXPECT_EQ(0, buffer.search("a", 1, 0));
+  EXPECT_EQ(1, buffer.search("b", 1, 1));
+  EXPECT_EQ(2, buffer.search("a", 1, 1));
+  EXPECT_EQ(0, buffer.search("abaa", 4, 0));
+  EXPECT_EQ(2, buffer.search("aaaa", 4, 0));
+  EXPECT_EQ(2, buffer.search("aaaa", 4, 1));
+  EXPECT_EQ(2, buffer.search("aaaa", 4, 2));
+  EXPECT_EQ(7, buffer.search("aaaaab", 6, 0));
+  EXPECT_EQ(0, buffer.search("abaaaabaaaaaba", 14, 0));
+  EXPECT_EQ(12, buffer.search("ba", 2, 10));
+  EXPECT_EQ(-1, buffer.search("abaaaabaaaaabaa", 15, 0));
+}
+
+TEST_P(OwnedImplTest, ToString) {
+  Buffer::OwnedImpl buffer;
+  verifyImplementation(buffer);
   EXPECT_EQ("", buffer.toString());
   auto append = [&buffer](absl::string_view str) { buffer.add(str.data(), str.size()); };
   append("Hello, ");
@@ -209,10 +384,33 @@ TEST_F(OwnedImplTest, ToString) {
   EXPECT_EQ(absl::StrCat("Hello, world!" + long_string), buffer.toString());
 }
 
+TEST_P(OwnedImplTest, AppendSliceForTest) {
+  static constexpr size_t NumInputs = 3;
+  static constexpr const char* Inputs[] = {"one", "2", "", "four", ""};
+  Buffer::OwnedImpl buffer;
+  RawSlice slices[NumInputs];
+  EXPECT_EQ(0, buffer.getRawSlices(slices, NumInputs));
+  for (const auto& input : Inputs) {
+    buffer.appendSliceForTest(input);
+  }
+  // getRawSlices will only return the 3 slices with nonzero length.
+  EXPECT_EQ(3, buffer.getRawSlices(slices, NumInputs));
+
+  auto expectSlice = [](const RawSlice& slice, const char* expected) {
+    size_t length = strlen(expected);
+    EXPECT_EQ(length, slice.len_);
+    EXPECT_EQ(0, memcmp(slice.mem_, expected, length));
+  };
+
+  expectSlice(slices[0], "one");
+  expectSlice(slices[1], "2");
+  expectSlice(slices[2], "four");
+}
+
 // Regression test for oss-fuzz issue
 // https://bugs.chromium.org/p/oss-fuzz/issues/detail?id=13263, where prepending
 // an empty buffer resulted in a corrupted libevent internal state.
-TEST_F(OwnedImplTest, PrependEmpty) {
+TEST_P(OwnedImplTest, PrependEmpty) {
   Buffer::OwnedImpl buf;
   Buffer::OwnedImpl other_buf;
   char input[] = "foo";
@@ -226,6 +424,19 @@ TEST_F(OwnedImplTest, PrependEmpty) {
   EXPECT_EQ(0, buf.length());
 }
 
+TEST(OverflowDetectingUInt64, Arithmetic) {
+  Logger::StderrSinkDelegate stderr_sink(Logger::Registry::getSink()); // For coverage build.
+  OverflowDetectingUInt64 length;
+  length += 1;
+  length -= 1;
+  length -= 0;
+  EXPECT_DEATH(length -= 1, "underflow");
+  uint64_t half = uint64_t(1) << 63;
+  length += half;
+  length += (half - 1); // length is now 2^64 - 1
+  EXPECT_DEATH(length += 1, "overflow");
+}
+
 } // namespace
 } // namespace Buffer
 } // namespace Envoy
diff --git a/test/common/buffer/utility.h b/test/common/buffer/utility.h
index 868427de465ad..30723ccbe8b6a 100644
--- a/test/common/buffer/utility.h
+++ b/test/common/buffer/utility.h
@@ -10,6 +10,36 @@ namespace Envoy {
 namespace Buffer {
 namespace {
 
+/** Used to specify which OwnedImpl implementation to test. */
+enum class BufferImplementation {
+  Old, // original evbuffer-based version
+  New  // new deque-of-slices version
+};
+
+/**
+ * Base class for tests that are parameterized based on BufferImplementation.
+ */
+class BufferImplementationParamTest : public testing::TestWithParam<BufferImplementation> {
+protected:
+  BufferImplementationParamTest() {
+    OwnedImpl::useOldImpl(GetParam() == BufferImplementation::Old);
+  }
+
+  virtual ~BufferImplementationParamTest() {}
+
+  /** Verify that a buffer has been constructed using the expected implementation. */
+  void verifyImplementation(const OwnedImpl& buffer) {
+    switch (GetParam()) {
+    case BufferImplementation::Old:
+      ASSERT_TRUE(buffer.usesOldImpl());
+      break;
+    case BufferImplementation::New:
+      ASSERT_FALSE(buffer.usesOldImpl());
+      break;
+    }
+  }
+};
+
 inline void addRepeated(Buffer::Instance& buffer, int n, int8_t value) {
   for (int i = 0; i < n; i++) {
     buffer.add(&value, 1);
diff --git a/test/common/buffer/watermark_buffer_test.cc b/test/common/buffer/watermark_buffer_test.cc
index 4cad003f1364c..571a72a859580 100644
--- a/test/common/buffer/watermark_buffer_test.cc
+++ b/test/common/buffer/watermark_buffer_test.cc
@@ -4,6 +4,8 @@
 #include "common/buffer/watermark_buffer.h"
 #include "common/network/io_socket_handle_impl.h"
 
+#include "test/common/buffer/utility.h"
+
 #include "gtest/gtest.h"
 
 namespace Envoy {
@@ -12,9 +14,12 @@ namespace {
 
 const char TEN_BYTES[] = "0123456789";
 
-class WatermarkBufferTest : public testing::Test {
+class WatermarkBufferTest : public BufferImplementationParamTest {
 public:
-  WatermarkBufferTest() { buffer_.setWatermarks(5, 10); }
+  WatermarkBufferTest() {
+    verifyImplementation(buffer_);
+    buffer_.setWatermarks(5, 10);
+  }
 
   Buffer::WatermarkBuffer buffer_{[&]() -> void { ++times_low_watermark_called_; },
                                   [&]() -> void { ++times_high_watermark_called_; }};
@@ -22,9 +27,12 @@ class WatermarkBufferTest : public testing::Test {
   uint32_t times_high_watermark_called_{0};
 };
 
-TEST_F(WatermarkBufferTest, TestWatermark) { ASSERT_EQ(10, buffer_.highWatermark()); }
+INSTANTIATE_TEST_CASE_P(WatermarkBufferTest, WatermarkBufferTest,
+                        testing::ValuesIn({BufferImplementation::Old, BufferImplementation::New}));
+
+TEST_P(WatermarkBufferTest, TestWatermark) { ASSERT_EQ(10, buffer_.highWatermark()); }
 
-TEST_F(WatermarkBufferTest, CopyOut) {
+TEST_P(WatermarkBufferTest, CopyOut) {
   buffer_.add("hello world");
   std::array<char, 5> out;
   buffer_.copyOut(0, out.size(), out.data());
@@ -37,7 +45,7 @@ TEST_F(WatermarkBufferTest, CopyOut) {
   buffer_.copyOut(4, 0, out.data());
 }
 
-TEST_F(WatermarkBufferTest, AddChar) {
+TEST_P(WatermarkBufferTest, AddChar) {
   buffer_.add(TEN_BYTES, 10);
   EXPECT_EQ(0, times_high_watermark_called_);
   buffer_.add("a", 1);
@@ -45,7 +53,7 @@ TEST_F(WatermarkBufferTest, AddChar) {
   EXPECT_EQ(11, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, AddString) {
+TEST_P(WatermarkBufferTest, AddString) {
   buffer_.add(std::string(TEN_BYTES));
   EXPECT_EQ(0, times_high_watermark_called_);
   buffer_.add(std::string("a"));
@@ -53,7 +61,7 @@ TEST_F(WatermarkBufferTest, AddString) {
   EXPECT_EQ(11, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, AddBuffer) {
+TEST_P(WatermarkBufferTest, AddBuffer) {
   OwnedImpl first(TEN_BYTES);
   buffer_.add(first);
   EXPECT_EQ(0, times_high_watermark_called_);
@@ -63,7 +71,7 @@ TEST_F(WatermarkBufferTest, AddBuffer) {
   EXPECT_EQ(11, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, Prepend) {
+TEST_P(WatermarkBufferTest, Prepend) {
   std::string suffix = "World!", prefix = "Hello, ";
 
   buffer_.add(suffix);
@@ -73,7 +81,7 @@ TEST_F(WatermarkBufferTest, Prepend) {
   EXPECT_EQ(suffix.size() + prefix.size(), buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, PrependToEmptyBuffer) {
+TEST_P(WatermarkBufferTest, PrependToEmptyBuffer) {
   std::string suffix = "World!", prefix = "Hello, ";
 
   buffer_.prepend(suffix);
@@ -89,7 +97,7 @@ TEST_F(WatermarkBufferTest, PrependToEmptyBuffer) {
   EXPECT_EQ(suffix.size() + prefix.size(), buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, PrependBuffer) {
+TEST_P(WatermarkBufferTest, PrependBuffer) {
   std::string suffix = "World!", prefix = "Hello, ";
 
   uint32_t prefix_buffer_low_watermark_hits{0};
@@ -110,7 +118,7 @@ TEST_F(WatermarkBufferTest, PrependBuffer) {
   EXPECT_EQ(0, prefixBuffer.length());
 }
 
-TEST_F(WatermarkBufferTest, Commit) {
+TEST_P(WatermarkBufferTest, Commit) {
   buffer_.add(TEN_BYTES, 10);
   EXPECT_EQ(0, times_high_watermark_called_);
   RawSlice out;
@@ -122,7 +130,7 @@ TEST_F(WatermarkBufferTest, Commit) {
   EXPECT_EQ(20, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, Drain) {
+TEST_P(WatermarkBufferTest, Drain) {
   // Draining from above to below the low watermark does nothing if the high
   // watermark never got hit.
   buffer_.add(TEN_BYTES, 10);
@@ -145,7 +153,7 @@ TEST_F(WatermarkBufferTest, Drain) {
   EXPECT_EQ(2, times_high_watermark_called_);
 }
 
-TEST_F(WatermarkBufferTest, MoveFullBuffer) {
+TEST_P(WatermarkBufferTest, MoveFullBuffer) {
   buffer_.add(TEN_BYTES, 10);
   OwnedImpl data("a");
 
@@ -155,7 +163,7 @@ TEST_F(WatermarkBufferTest, MoveFullBuffer) {
   EXPECT_EQ(11, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, MoveOneByte) {
+TEST_P(WatermarkBufferTest, MoveOneByte) {
   buffer_.add(TEN_BYTES, 9);
   OwnedImpl data("ab");
 
@@ -168,7 +176,7 @@ TEST_F(WatermarkBufferTest, MoveOneByte) {
   EXPECT_EQ(11, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, WatermarkFdFunctions) {
+TEST_P(WatermarkBufferTest, WatermarkFdFunctions) {
   int pipe_fds[2] = {0, 0};
   ASSERT_EQ(0, pipe(pipe_fds));
 
@@ -201,7 +209,7 @@ TEST_F(WatermarkBufferTest, WatermarkFdFunctions) {
   EXPECT_EQ(20, buffer_.length());
 }
 
-TEST_F(WatermarkBufferTest, MoveWatermarks) {
+TEST_P(WatermarkBufferTest, MoveWatermarks) {
   buffer_.add(TEN_BYTES, 9);
   EXPECT_EQ(0, times_high_watermark_called_);
   buffer_.setWatermarks(1, 9);
@@ -225,7 +233,7 @@ TEST_F(WatermarkBufferTest, MoveWatermarks) {
   EXPECT_EQ(2, times_low_watermark_called_);
 }
 
-TEST_F(WatermarkBufferTest, GetRawSlices) {
+TEST_P(WatermarkBufferTest, GetRawSlices) {
   buffer_.add(TEN_BYTES, 10);
 
   RawSlice slices[2];
@@ -237,7 +245,7 @@ TEST_F(WatermarkBufferTest, GetRawSlices) {
   EXPECT_EQ(data_pointer, slices[0].mem_);
 }
 
-TEST_F(WatermarkBufferTest, Search) {
+TEST_P(WatermarkBufferTest, Search) {
   buffer_.add(TEN_BYTES, 10);
 
   EXPECT_EQ(1, buffer_.search(&TEN_BYTES[1], 2, 0));
@@ -245,7 +253,7 @@ TEST_F(WatermarkBufferTest, Search) {
   EXPECT_EQ(-1, buffer_.search(&TEN_BYTES[1], 2, 5));
 }
 
-TEST_F(WatermarkBufferTest, MoveBackWithWatermarks) {
+TEST_P(WatermarkBufferTest, MoveBackWithWatermarks) {
   int high_watermark_buffer1 = 0;
   int low_watermark_buffer1 = 0;
   Buffer::WatermarkBuffer buffer1{[&]() -> void { ++low_watermark_buffer1; },
diff --git a/test/common/buffer/zero_copy_input_stream_test.cc b/test/common/buffer/zero_copy_input_stream_test.cc
index 69070f1cc0262..bd747ed20a674 100644
--- a/test/common/buffer/zero_copy_input_stream_test.cc
+++ b/test/common/buffer/zero_copy_input_stream_test.cc
@@ -1,16 +1,19 @@
 #include "common/buffer/buffer_impl.h"
 #include "common/buffer/zero_copy_input_stream_impl.h"
 
+#include "test/common/buffer/utility.h"
+
 #include "gtest/gtest.h"
 
 namespace Envoy {
 namespace Buffer {
 namespace {
 
-class ZeroCopyInputStreamTest : public testing::Test {
+class ZeroCopyInputStreamTest : public BufferImplementationParamTest {
 public:
   ZeroCopyInputStreamTest() {
     Buffer::OwnedImpl buffer{"abcd"};
+    verifyImplementation(buffer);
     stream_.move(buffer);
   }
 
@@ -21,21 +24,23 @@ class ZeroCopyInputStreamTest : public testing::Test {
   int size_;
 };
 
-TEST_F(ZeroCopyInputStreamTest, Move) {
+TEST_P(ZeroCopyInputStreamTest, Move) {
   Buffer::OwnedImpl buffer{"abcd"};
+  verifyImplementation(buffer);
   stream_.move(buffer);
 
   EXPECT_EQ(0, buffer.length());
 }
 
-TEST_F(ZeroCopyInputStreamTest, Next) {
+TEST_P(ZeroCopyInputStreamTest, Next) {
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_EQ(4, size_);
   EXPECT_EQ(0, memcmp(slice_data_.data(), data_, size_));
 }
 
-TEST_F(ZeroCopyInputStreamTest, TwoSlices) {
+TEST_P(ZeroCopyInputStreamTest, TwoSlices) {
   Buffer::OwnedImpl buffer("efgh");
+  verifyImplementation(buffer);
 
   stream_.move(buffer);
 
@@ -47,7 +52,7 @@ TEST_F(ZeroCopyInputStreamTest, TwoSlices) {
   EXPECT_EQ(0, memcmp("efgh", data_, size_));
 }
 
-TEST_F(ZeroCopyInputStreamTest, BackUp) {
+TEST_P(ZeroCopyInputStreamTest, BackUp) {
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_EQ(4, size_);
 
@@ -60,7 +65,7 @@ TEST_F(ZeroCopyInputStreamTest, BackUp) {
   EXPECT_EQ(4, stream_.ByteCount());
 }
 
-TEST_F(ZeroCopyInputStreamTest, BackUpFull) {
+TEST_P(ZeroCopyInputStreamTest, BackUpFull) {
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_EQ(4, size_);
 
@@ -71,13 +76,13 @@ TEST_F(ZeroCopyInputStreamTest, BackUpFull) {
   EXPECT_EQ(4, stream_.ByteCount());
 }
 
-TEST_F(ZeroCopyInputStreamTest, ByteCount) {
+TEST_P(ZeroCopyInputStreamTest, ByteCount) {
   EXPECT_EQ(0, stream_.ByteCount());
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_EQ(4, stream_.ByteCount());
 }
 
-TEST_F(ZeroCopyInputStreamTest, Finish) {
+TEST_P(ZeroCopyInputStreamTest, Finish) {
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_TRUE(stream_.Next(&data_, &size_));
   EXPECT_EQ(0, size_);
diff --git a/test/mocks/server/mocks.h b/test/mocks/server/mocks.h
index ec26b4bcbc2f6..8c874f1b20d91 100644
--- a/test/mocks/server/mocks.h
+++ b/test/mocks/server/mocks.h
@@ -79,6 +79,7 @@ class MockOptions : public Options {
   MOCK_CONST_METHOD0(hotRestartDisabled, bool());
   MOCK_CONST_METHOD0(signalHandlingEnabled, bool());
   MOCK_CONST_METHOD0(mutexTracingEnabled, bool());
+  MOCK_CONST_METHOD0(libeventBufferEnabled, bool());
   MOCK_CONST_METHOD0(cpusetThreadsEnabled, bool());
   MOCK_CONST_METHOD0(toCommandLineOptions, Server::CommandLineOptionsPtr());
 
diff --git a/test/server/options_impl_test.cc b/test/server/options_impl_test.cc
index 31902e992add7..41ea4494c3b94 100644
--- a/test/server/options_impl_test.cc
+++ b/test/server/options_impl_test.cc
@@ -93,6 +93,7 @@ TEST_F(OptionsImplTest, All) {
   EXPECT_EQ(std::chrono::seconds(60), options->drainTime());
   EXPECT_EQ(std::chrono::seconds(90), options->parentShutdownTime());
   EXPECT_EQ(true, options->hotRestartDisabled());
+  EXPECT_EQ(true, options->libeventBufferEnabled());
   EXPECT_EQ(true, options->cpusetThreadsEnabled());
 
   options = createOptionsImpl("envoy --mode init_only");
@@ -216,11 +217,12 @@ TEST_F(OptionsImplTest, OptionsAreInSyncWithProto) {
   Server::CommandLineOptionsPtr command_line_options = options->toCommandLineOptions();
   // Failure of this condition indicates that the server_info proto is not in sync with the options.
   // If an option is added/removed, please update server_info proto as well to keep it in sync.
-  // Currently the following 3 options are not defined in proto, hence the count differs by 3.
-  // 2. version        - default TCLAP argument.
-  // 3. help           - default TCLAP argument.
-  // 4. ignore_rest    - default TCLAP argument.
-  EXPECT_EQ(options->count() - 3, command_line_options->GetDescriptor()->field_count());
+  // Currently the following 4 options are not defined in proto, hence the count differs by 5.
+  // 1. version        - default TCLAP argument.
+  // 2. help           - default TCLAP argument.
+  // 3. ignore_rest    - default TCLAP argument.
+  // 4. use-libevent-buffers  - short-term override for rollout of new buffer implementation.
+  EXPECT_EQ(options->count() - 4, command_line_options->GetDescriptor()->field_count());
 }
 
 TEST_F(OptionsImplTest, BadCliOption) {
diff --git a/test/test_common/utility.cc b/test/test_common/utility.cc
index ef145bda07111..2a901599656e3 100644
--- a/test/test_common/utility.cc
+++ b/test/test_common/utility.cc
@@ -98,23 +98,30 @@ bool TestUtility::buffersEqual(const Buffer::Instance& lhs, const Buffer::Instan
     return false;
   }
 
+  // Check whether the two buffers contain the same content. It is valid for the content
+  // to be arranged differently in the buffers. For example, lhs could have one slice
+  // containing 10 bytes while rhs has ten slices containing one byte each.
   uint64_t lhs_num_slices = lhs.getRawSlices(nullptr, 0);
   uint64_t rhs_num_slices = rhs.getRawSlices(nullptr, 0);
-  if (lhs_num_slices != rhs_num_slices) {
-    return false;
-  }
-
   STACK_ARRAY(lhs_slices, Buffer::RawSlice, lhs_num_slices);
   lhs.getRawSlices(lhs_slices.begin(), lhs_num_slices);
   STACK_ARRAY(rhs_slices, Buffer::RawSlice, rhs_num_slices);
   rhs.getRawSlices(rhs_slices.begin(), rhs_num_slices);
-  for (size_t i = 0; i < lhs_num_slices; i++) {
-    if (lhs_slices[i].len_ != rhs_slices[i].len_) {
-      return false;
-    }
-
-    if (0 != memcmp(lhs_slices[i].mem_, rhs_slices[i].mem_, lhs_slices[i].len_)) {
-      return false;
+  size_t rhs_slice = 0;
+  size_t rhs_offset = 0;
+  for (size_t lhs_slice = 0; lhs_slice < lhs_num_slices; lhs_slice++) {
+    for (size_t lhs_offset = 0; lhs_offset < lhs_slices[lhs_slice].len_; lhs_offset++) {
+      while (rhs_offset >= rhs_slices[rhs_slice].len_) {
+        rhs_slice++;
+        ASSERT(rhs_slice < rhs_num_slices);
+        rhs_offset = 0;
+      }
+      auto lhs_str = static_cast<const uint8_t*>(lhs_slices[lhs_slice].mem_);
+      auto rhs_str = static_cast<const uint8_t*>(rhs_slices[rhs_slice].mem_);
+      if (lhs_str[lhs_offset] != rhs_str[rhs_offset]) {
+        return false;
+      }
+      rhs_offset++;
     }
   }
 
diff --git a/test/test_common/utility.h b/test/test_common/utility.h
index 0c4a251918878..06ec1e60e1a82 100644
--- a/test/test_common/utility.h
+++ b/test/test_common/utility.h
@@ -120,7 +120,8 @@ class TestUtility {
    * Compare 2 buffers.
    * @param lhs supplies buffer 1.
    * @param rhs supplies buffer 2.
-   * @return TRUE if the buffers are equal, false if not.
+   * @return TRUE if the buffers contain equal content
+   *         (i.e., if lhs.toString() == rhs.toString()), false if not.
    */
   static bool buffersEqual(const Buffer::Instance& lhs, const Buffer::Instance& rhs);
 
diff --git a/test/test_common/utility_test.cc b/test/test_common/utility_test.cc
index 40083fe339469..2d2bd0743864b 100644
--- a/test/test_common/utility_test.cc
+++ b/test/test_common/utility_test.cc
@@ -26,4 +26,36 @@ TEST(headerMapEqualIgnoreOrder, NotEqual) {
   Http::TestHeaderMapImpl rhs{{":method", "GET"}, {":authority", "host"}};
   EXPECT_FALSE(TestUtility::headerMapEqualIgnoreOrder(lhs, rhs));
 }
+
+TEST(buffersEqual, Aligned) {
+  Buffer::OwnedImpl buffer1, buffer2;
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+
+  buffer1.appendSliceForTest("hello");
+  EXPECT_FALSE(TestUtility::buffersEqual(buffer1, buffer2));
+  buffer2.appendSliceForTest("hello");
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+
+  buffer1.appendSliceForTest(", world");
+  EXPECT_FALSE(TestUtility::buffersEqual(buffer1, buffer2));
+  buffer2.appendSliceForTest(", world");
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+}
+
+TEST(buffersEqual, NonAligned) {
+  Buffer::OwnedImpl buffer1, buffer2;
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+
+  buffer1.appendSliceForTest("hello");
+  EXPECT_FALSE(TestUtility::buffersEqual(buffer1, buffer2));
+  buffer2.appendSliceForTest("hello");
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+
+  buffer1.appendSliceForTest(", ");
+  buffer1.appendSliceForTest("world");
+  EXPECT_FALSE(TestUtility::buffersEqual(buffer1, buffer2));
+  buffer2.appendSliceForTest(", world");
+  EXPECT_TRUE(TestUtility::buffersEqual(buffer1, buffer2));
+}
+
 } // namespace Envoy
diff --git a/tools/spelling_dictionary.txt b/tools/spelling_dictionary.txt
index 34a034702ad1a..c882d75bd4e3f 100644
--- a/tools/spelling_dictionary.txt
+++ b/tools/spelling_dictionary.txt
@@ -149,6 +149,7 @@ NBF
 NDEBUG
 NGHTTP
 NOLINT
+NOLINTNEXTLINE
 NS
 NUL
 Nilsson
@@ -364,6 +365,7 @@ deflateInit
 deletable
 deleter
 delim
+deque
 dereference
 dereferences
 dereferencing
@@ -620,6 +622,7 @@ reparse
 reperform
 repicked
 repo
+reservable
 resize
 resized
 resizes
@@ -719,6 +722,7 @@ unordered
 unowned
 unparented
 unpause
+unpopulated
 unprotect
 unref
 unreferenced