listener: remove the peek from the listener filters

envoy/buffer/buffer.h

@@ -34,6 +34,17 @@ struct RawSlice {
   bool operator!=(const RawSlice& rhs) const { return !(*this == rhs); }
 };
 
+/**
+ * A const raw memory data slice including the location and length.
+ */
+struct ConstRawSlice {
+  const void* mem_ = nullptr;
+  size_t len_ = 0;
+
+  bool operator==(const RawSlice& rhs) const { return mem_ == rhs.mem_ && len_ == rhs.len_; }
+  bool operator!=(const RawSlice& rhs) const { return !(*this == rhs); }
+};
+
 using RawSliceVector = absl::InlinedVector<RawSlice, 16>;
 
 /**

envoy/network/BUILD

@@ -86,6 +86,7 @@ envoy_cc_library(
     hdrs = ["filter.h"],
     deps = [
         ":listen_socket_interface",
+        ":listener_filter_buffer_interface",
         ":transport_socket_interface",
         "//envoy/buffer:buffer_interface",
         "//envoy/stream_info:stream_info_interface",
@@ -147,6 +148,14 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "listener_filter_buffer_interface",
+    hdrs = ["listener_filter_buffer.h"],
+    deps = [
+        "//envoy/buffer:buffer_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "transport_socket_interface",
     hdrs = ["transport_socket.h"],

envoy/network/filter.h

@@ -4,6 +4,7 @@
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/network/listen_socket.h"
+#include "envoy/network/listener_filter_buffer.h"
 #include "envoy/network/transport_socket.h"
 #include "envoy/stream_info/stream_info.h"
 #include "envoy/upstream/host_description.h"
@@ -330,6 +331,20 @@ class ListenerFilter {
    * @return status used by the filter manager to manage further filter iteration.
    */
   virtual FilterStatus onAccept(ListenerFilterCallbacks& cb) PURE;
+
+  /**
+   * Called when data read from the connection. If the filter chain doesn't get
+   * enough data, the filter chain can be stopped, then waiting for more data.
+   * @param buffer the buffer of data.
+   * @return status used by the filter manager to manage further filter iteration.
+   */
+  virtual FilterStatus onData(Network::ListenerFilterBuffer& buffer) PURE;
+
+  /**
+   * Return the size of data the filter want to inspect from the connection.
+   * @return the size of data inspect from the connection. 0 means filter needn't any data.
+   */
+  virtual size_t maxReadBytes() const PURE;
 };
 
 using ListenerFilterPtr = std::unique_ptr<ListenerFilter>;

envoy/network/listener_filter_buffer.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <memory>
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Network {
+
+/**
+ * Interface for ListenerFilterBuffer
+ */
+class ListenerFilterBuffer {
+public:
+  virtual ~ListenerFilterBuffer() = default;
+
+  /**
+   * Return a single const raw slice to the buffer of the data.
+   * @return a Buffer::ConstRawSlice pointed to raw buffer.
+   */
+  virtual const Buffer::ConstRawSlice rawSlice() const PURE;
+
+  /**
+   * Drain the data from the beginning of the buffer.
+   * @param length the length of data to drain.
+   * @return return True if the drain success, otherwise return False.
+   */
+  virtual bool drain(uint64_t length) PURE;
+
+  /**
+   * Return the length of data in the buffer
+   * @return length The length of data in the buffer.
+   */
+  virtual uint64_t length() const PURE;
+};
+
+} // namespace Network
+} // namespace Envoy

source/common/event/file_event_impl.cc

@@ -139,7 +139,7 @@ void FileEventImpl::registerEventIfEmulatedEdge(uint32_t event) {
       auto new_event_mask = enabled_events_ | event;
       if (event & FileReadyType::Read && (enabled_events_ & FileReadyType::Closed)) {
         // We never ask for both early close and read at the same time.
-        new_event_mask = new_event_mask & ~FileReadyType::Read;
+        //new_event_mask = new_event_mask & ~FileReadyType::Read;
       }
       updateEvents(new_event_mask);
     }

source/common/network/BUILD

@@ -232,6 +232,17 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "listener_filter_buffer_lib",
+    srcs = ["listener_filter_buffer_impl.cc"],
+    hdrs = ["listener_filter_buffer_impl.h"],
+    deps = [
+        "//envoy/network:io_handle_interface",
+        "//envoy/network:listener_filter_buffer_interface",
+        "//source/common/buffer:buffer_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "listener_lib",
     srcs = [

source/common/network/listener_filter_buffer_impl.cc

@@ -0,0 +1,96 @@
+#include "source/common/network/listener_filter_buffer_impl.h"
+
+#include <string>
+
+namespace Envoy {
+namespace Network {
+
+ListenerFilterBufferImpl::ListenerFilterBufferImpl(IoHandle& io_handle,
+                                                   Event::Dispatcher& dispatcher,
+                                                   ListenerFilterBufferOnCloseCb close_cb,
+                                                   ListenerFilterBufferOnDataCb on_data_cb,
+                                                   uint64_t buffer_size)
+    : io_handle_(io_handle), dispatcher_(dispatcher), on_close_cb_(close_cb),
+      on_data_cb_(on_data_cb), buffer_(std::make_unique<uint8_t[]>(buffer_size)),
+      base_(buffer_.get()), buffer_size_(buffer_size) {
+  // If the buffer_size not greater than 0, it means that doesn't expect any data.
+  ASSERT(buffer_size > 0);
+
+  io_handle_.initializeFileEvent(
+      dispatcher_, [this](uint32_t events) { onFileEvent(events); },
+      Event::PlatformDefaultTriggerType, Event::FileReadyType::Read);
+}
+
+const Buffer::ConstRawSlice ListenerFilterBufferImpl::rawSlice() const {
+  Buffer::ConstRawSlice slice;
+  slice.mem_ = base_;
+  slice.len_ = data_size_;
+  return slice;
+}
+
+bool ListenerFilterBufferImpl::drain(uint64_t length) {
+  if (length == 0) {
+    return true;
+  }
+
+  ASSERT(length <= data_size_);
+
+  uint64_t read_size = 0;
+  while (1) {
+    auto result = io_handle_.recv(base_, length - read_size, 0);
+    ENVOY_LOG(trace, "recv returned: {}", result.return_value_);
+
+    // The socket buffer is expected to have the data. so the
+    // recv doesn't expected to fail.
+    ASSERT(result.ok());
+
+    read_size += result.return_value_;
+    if (read_size < length) {
+      continue;
+    }
+    base_ += length;
+    data_size_ -= length;
+    buffer_size_ -= length;
+    break;
+  }
+  return true;
+}
+
+PeekState ListenerFilterBufferImpl::peekFromSocket() {
+  const auto result = io_handle_.recv(base_, buffer_size_, MSG_PEEK);
+  ENVOY_LOG(trace, "recv returned: {}", result.return_value_);
+
+  if (!result.ok()) {
+    if (result.err_->getErrorCode() == Api::IoError::IoErrorCode::Again) {
+      ENVOY_LOG(trace, "recv return try again");
+      return PeekState::Again;
+    }
+    ENVOY_LOG(debug, "recv failed: {}: {}", result.err_->getErrorCode(),
+              result.err_->getErrorDetails());
+    return PeekState::Error;
+  }
+  // Remote closed
+  if (result.return_value_ == 0) {
+    ENVOY_LOG(debug, "recv failed: remote closed");
+    return PeekState::Error;
+  }
+  data_size_ = result.return_value_;
+  ASSERT(data_size_ <= buffer_size_);
+
+  return PeekState::Done;
+}
+
+void ListenerFilterBufferImpl::onFileEvent(uint32_t events) {
+  ENVOY_LOG(trace, "onFileEvent: {}", events);
+
+  auto state = peekFromSocket();
+  if (state == PeekState::Done) {
+    on_data_cb_();
+  } else if (state == PeekState::Error) {
+    on_close_cb_();
+  }
+  // Did nothing for `Api::IoError::IoErrorCode::Again`
+}
+
+} // namespace Network
+} // namespace Envoy

source/common/network/listener_filter_buffer_impl.h

@@ -0,0 +1,68 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/network/io_handle.h"
+#include "envoy/network/listener_filter_buffer.h"
+
+#include "source/common/buffer/buffer_impl.h"
+
+namespace Envoy {
+namespace Network {
+
+using ListenerFilterBufferOnCloseCb = std::function<void()>;
+using ListenerFilterBufferOnDataCb = std::function<void()>;
+
+enum class PeekState {
+  // Peek data status successful.
+  Done,
+  // Need to try again.
+  Again,
+  // Error to peek data.
+  Error
+};
+
+class ListenerFilterBufferImpl : public ListenerFilterBuffer, Logger::Loggable<Logger::Id::filter> {
+public:
+  ListenerFilterBufferImpl(IoHandle& io_handle, Event::Dispatcher& dispatcher,
+                           ListenerFilterBufferOnCloseCb close_cb,
+                           ListenerFilterBufferOnDataCb on_data_cb, uint64_t buffer_size);
+
+  // ListenerFilterBuffer
+  const Buffer::ConstRawSlice rawSlice() const override;
+  // ListenerFilterBuffer
+  bool drain(uint64_t length) override;
+  // ListenerFilterBuffer
+  uint64_t length() const override { return data_size_; }
+
+  /**
+   * Trigger the data peek from the socket.
+   */
+  PeekState peekFromSocket();
+
+  void reset() { io_handle_.resetFileEvents(); }
+
+private:
+  void onFileEvent(uint32_t events);
+
+  IoHandle& io_handle_;
+  Event::Dispatcher& dispatcher_;
+  ListenerFilterBufferOnCloseCb on_close_cb_;
+  ListenerFilterBufferOnDataCb on_data_cb_;
+
+  // The buffer for the data peeked from the socket.
+  std::unique_ptr<uint8_t[]> buffer_;
+  // The start of buffer.
+  uint8_t* base_;
+  // The size of buffer;
+  uint64_t buffer_size_;
+  // The size of valid data.
+  uint64_t data_size_{0};
+};
+
+using ListenerFilterBufferImplPtr = std::unique_ptr<ListenerFilterBufferImpl>;
+
+} // namespace Network
+} // namespace Envoy

source/common/network/utility.cc

@@ -346,12 +346,14 @@ Address::InstanceConstSharedPtr Utility::getIpv6LoopbackAddress() {
                          new Address::Ipv6Instance("::1", 0, nullptr));
 }
 
-Address::InstanceConstSharedPtr Utility::getIpv4AnyAddress(uint32_t port) {
-  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, new Address::Ipv4Instance(port));
+Address::InstanceConstSharedPtr Utility::getIpv4AnyAddress() {
+  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,
+                         new Address::Ipv4Instance(static_cast<uint32_t>(0)));
 }
 
-Address::InstanceConstSharedPtr Utility::getIpv6AnyAddress(uint32_t port) {
-  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, new Address::Ipv6Instance(port));
+Address::InstanceConstSharedPtr Utility::getIpv6AnyAddress() {
+  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,
+                         new Address::Ipv6Instance(static_cast<uint32_t>(0)));
 }
 
 const std::string& Utility::getIpv4CidrCatchAllAddress() {

source/common/network/utility.h

@@ -246,20 +246,18 @@ class Utility {
   static Address::InstanceConstSharedPtr getIpv6LoopbackAddress();
 
   /**
-   * @param port to be included in address, the default is 0.
    * @return Address::InstanceConstSharedPtr an address that represents the IPv4 wildcard address
    *         (i.e. "0.0.0.0"). Used during binding to indicate that incoming connections to any
    *         local IPv4 address are to be accepted.
    */
-  static Address::InstanceConstSharedPtr getIpv4AnyAddress(uint32_t port = 0);
+  static Address::InstanceConstSharedPtr getIpv4AnyAddress();
 
   /**
-   * @param port to be included in address, the default is 0.
    * @return Address::InstanceConstSharedPtr an address that represents the IPv6 wildcard address
    *         (i.e. "::"). Used during binding to indicate that incoming connections to any local
    *         IPv6 address are to be accepted.
    */
-  static Address::InstanceConstSharedPtr getIpv6AnyAddress(uint32_t port = 0);
+  static Address::InstanceConstSharedPtr getIpv6AnyAddress();
 
   /**
    * @return the IPv4 CIDR catch-all address (0.0.0.0/0).

source/common/network/win32_socket_handle_impl.h

@@ -44,6 +44,30 @@ class Win32SocketHandleImpl : public IoSocketHandleImpl {
 
 private:
   void reEnableEventBasedOnIOResult(const Api::IoCallUint64Result& result, uint32_t event);
+
+  // On Windows we use the MSG_PEEK on recv instead of peeking the socket
+  // we drain the socket to memory. Subsequent read calls need to read
+  // first from the class buffer and then go to the underlying socket.
+  // Implement the peek logic of recv for readability purposes
+  Api::IoCallUint64Result emulatePeek(void* buffer, size_t length);
+
+  /**
+   * Drain the socket into `peek_buffer_`.
+   * @param length is the desired length of data drained into the `peek_buffer_`.
+   * @return the actual length of data drained into the `peek_buffer_`.
+   */
+  Api::IoCallUint64Result drainToPeekBuffer(size_t length);
+
+  // Useful functions to read from the peek buffer based on
+  // the signatures of readv/read/recv OS socket functions.
+  Api::IoCallUint64Result readFromPeekBuffer(void* buffer, size_t length);
+  Api::IoCallUint64Result readFromPeekBuffer(Buffer::Instance& buffer, size_t length);
+  Api::IoCallUint64Result readvFromPeekBuffer(uint64_t max_length, Buffer::RawSlice* slices,
+                                              uint64_t num_slice);
+  Api::IoCallUint64Result peekFromPeekBuffer(void* buffer, size_t length);
+
+  // For windows mimic MSG_PEEK
+  Buffer::OwnedImpl peek_buffer_;
 };
 } // namespace Network
 } // namespace Envoy