Introduce listener TCP connection buffer configuration and implement …

docs/configuration/listeners/listeners.rst

@@ -52,6 +52,9 @@ use_original_dst
   destination port. If there is no listener associated with the original destination port, the
   connection is handled by the listener that receives it. Default is false.
 
+per_connection_buffer_limit_bytes
+  *(optional, integer)* Soft limit on size of the listener's new connection read and write buffers. If unspecified, an implementation defined default is applied (1MB).
+
 .. toctree::
   :hidden:
 

include/envoy/event/dispatcher.h

@@ -86,13 +86,16 @@ class Dispatcher {
    * (http://www.haproxy.org/download/1.5/doc/proxy-protocol.txt)
    * @param use_orig_dst if a connection was redirected to this port using iptables,
    *        allow the listener to hand it off to the listener associated to the original port
+   * @param per_connection_buffer_limit_bytes soft limit on size of the listener's new connection
+   *        read and write buffers.
    * @return Network::ListenerPtr a new listener that is owned by the caller.
    */
   virtual Network::ListenerPtr createListener(Network::ConnectionHandler& conn_handler,
                                               Network::ListenSocket& socket,
                                               Network::ListenerCallbacks& cb,
                                               Stats::Store& stats_store, bool bind_to_port,
-                                              bool use_proxy_proto, bool use_orig_dst) PURE;
+                                              bool use_proxy_proto, bool use_orig_dst,
+                                              size_t per_connection_buffer_limit_bytes) PURE;
 
   /**
    * Create a listener on a specific port.
@@ -108,14 +111,15 @@ class Dispatcher {
    * (http://www.haproxy.org/download/1.5/doc/proxy-protocol.txt)
    * @param use_orig_dst if a connection was redirected to this port using iptables,
    *        allow the listener to hand it off to the listener associated to the original port
+   * @param per_connection_buffer_limit_bytes soft limit on size of the listener's new connection
+   *        read and write buffers.
    * @return Network::ListenerPtr a new listener that is owned by the caller.
    */
-  virtual Network::ListenerPtr createSslListener(Network::ConnectionHandler& conn_handler,
-                                                 Ssl::ServerContext& ssl_ctx,
-                                                 Network::ListenSocket& socket,
-                                                 Network::ListenerCallbacks& cb,
-                                                 Stats::Store& stats_store, bool bind_to_port,
-                                                 bool use_proxy_proto, bool use_orig_dst) PURE;
+  virtual Network::ListenerPtr
+  createSslListener(Network::ConnectionHandler& conn_handler, Ssl::ServerContext& ssl_ctx,
+                    Network::ListenSocket& socket, Network::ListenerCallbacks& cb,
+                    Stats::Store& stats_store, bool bind_to_port, bool use_proxy_proto,
+                    bool use_orig_dst, size_t per_connection_buffer_limit_bytes) PURE;
 
   /**
    * Allocate a timer. @see Event::Timer for docs on how to use the timer.

include/envoy/network/connection.h

@@ -146,6 +146,12 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
    * are installed.
    */
   virtual void write(Buffer::Instance& data) PURE;
+
+  /**
+   * Set a soft limit on the size of the read buffer prior to flushing to further stages in the
+   * processing pipeline.
+   */
+  virtual void setReadBufferLimit(size_t limit) PURE;
 };
 
 typedef std::unique_ptr<Connection> ConnectionPtr;

include/envoy/network/connection_handler.h

@@ -28,9 +28,12 @@ class ConnectionHandler {
    * (http://www.haproxy.org/download/1.5/doc/proxy-protocol.txt)
    * @param use_orig_dst if a connection was redirected to this port using iptables,
    *        allow the listener to hand it off to the listener associated to the original port
+   * @param per_connection_buffer_limit_bytes soft limit on size of the listener's new connection
+   *        read and write buffers.
    */
   virtual void addListener(Network::FilterChainFactory& factory, Network::ListenSocket& socket,
-                           bool bind_to_port, bool use_proxy_proto, bool use_orig_dst) PURE;
+                           bool bind_to_port, bool use_proxy_proto, bool use_orig_dst,
+                           size_t per_connection_buffer_limit_bytes) PURE;
 
   /**
    * Adds listener to the handler.
@@ -43,10 +46,13 @@ class ConnectionHandler {
    * (http://www.haproxy.org/download/1.5/doc/proxy-protocol.txt)
    * @param use_orig_dst if a connection was redirected to this port using iptables,
    *        allow the listener to hand it off to the listener associated to the original port
+   * @param per_connection_buffer_limit_bytes soft limit on size of the listener's new connection
+   *        read and write buffers.
    */
   virtual void addSslListener(Network::FilterChainFactory& factory, Ssl::ServerContext& ssl_ctx,
                               Network::ListenSocket& socket, bool bind_to_port,
-                              bool use_proxy_proto, bool use_orig_dst) PURE;
+                              bool use_proxy_proto, bool use_orig_dst,
+                              size_t per_connection_buffer_limit_bytes) PURE;
 
   /**
    * Find a listener based on the provided listener port value.

include/envoy/server/configuration.h

@@ -50,6 +50,12 @@ class Listener {
    *         allow the listener to hand it off to the listener associated to the original port
    */
   virtual bool useOriginalDst() PURE;
+
+  /**
+   * @return size_t providing a soft limit on size of the listener's new connection read and write
+   *         buffers.
+   */
+  virtual size_t perConnectionBufferLimitBytes() PURE;
 };
 
 typedef std::unique_ptr<Listener> ListenerPtr;

source/common/event/dispatcher_impl.cc

@@ -83,20 +83,23 @@ Network::ListenerPtr DispatcherImpl::createListener(Network::ConnectionHandler&
                                                     Network::ListenSocket& socket,
                                                     Network::ListenerCallbacks& cb,
                                                     Stats::Store& stats_store, bool bind_to_port,
-                                                    bool use_proxy_proto, bool use_orig_dst) {
-  return Network::ListenerPtr{new Network::ListenerImpl(
-      conn_handler, *this, socket, cb, stats_store, bind_to_port, use_proxy_proto, use_orig_dst)};
+                                                    bool use_proxy_proto, bool use_orig_dst,
+                                                    size_t per_connection_buffer_limit_bytes) {
+  return Network::ListenerPtr{
+      new Network::ListenerImpl(conn_handler, *this, socket, cb, stats_store, bind_to_port,
+                                use_proxy_proto, use_orig_dst, per_connection_buffer_limit_bytes)};
 }
 
 Network::ListenerPtr DispatcherImpl::createSslListener(Network::ConnectionHandler& conn_handler,
                                                        Ssl::ServerContext& ssl_ctx,
                                                        Network::ListenSocket& socket,
                                                        Network::ListenerCallbacks& cb,
                                                        Stats::Store& stats_store, bool bind_to_port,
-                                                       bool use_proxy_proto, bool use_orig_dst) {
-  return Network::ListenerPtr{new Network::SslListenerImpl(conn_handler, *this, ssl_ctx, socket, cb,
-                                                           stats_store, bind_to_port,
-                                                           use_proxy_proto, use_orig_dst)};
+                                                       bool use_proxy_proto, bool use_orig_dst,
+                                                       size_t per_connection_buffer_limit_bytes) {
+  return Network::ListenerPtr{new Network::SslListenerImpl(
+      conn_handler, *this, ssl_ctx, socket, cb, stats_store, bind_to_port, use_proxy_proto,
+      use_orig_dst, per_connection_buffer_limit_bytes)};
 }
 
 TimerPtr DispatcherImpl::createTimer(TimerCb cb) { return TimerPtr{new TimerImpl(*this, cb)}; }

source/common/event/dispatcher_impl.h

@@ -37,12 +37,13 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>, public Dispatcher {
   Network::ListenerPtr createListener(Network::ConnectionHandler& conn_handler,
                                       Network::ListenSocket& socket, Network::ListenerCallbacks& cb,
                                       Stats::Store& stats_store, bool bind_to_port,
-                                      bool use_proxy_proto, bool use_orig_dst) override;
+                                      bool use_proxy_proto, bool use_orig_dst,
+                                      size_t per_connection_buffer_limit_bytes) override;
   Network::ListenerPtr createSslListener(Network::ConnectionHandler& conn_handler,
                                          Ssl::ServerContext& ssl_ctx, Network::ListenSocket& socket,
                                          Network::ListenerCallbacks& cb, Stats::Store& stats_store,
-                                         bool bind_to_port, bool use_proxy_proto,
-                                         bool use_orig_dst) override;
+                                         bool bind_to_port, bool use_proxy_proto, bool use_orig_dst,
+                                         size_t per_connection_buffer_limit_bytes) override;
   TimerPtr createTimer(TimerCb cb) override;
   void deferredDelete(DeferredDeletablePtr&& to_delete) override;
   void exit() override;

source/common/json/config_schemas.cc

@@ -42,7 +42,12 @@ const std::string Json::Schema::LISTENER_SCHEMA(R"EOF(
        "ssl_context" : {"$ref" : "#/definitions/ssl_context"},
        "bind_to_port" : {"type": "boolean"},
        "use_proxy_proto" : {"type" : "boolean"},
-       "use_original_dst" : {"type" : "boolean"}
+       "use_original_dst" : {"type" : "boolean"},
+       "per_connection_buffer_limit_bytes" : {
+         "type" : "integer",
+         "minimum" : 0,
+         "exclusiveMinimum" : true
+       }
     },
     "required": ["port", "filters"],
     "additionalProperties": false

source/common/network/connection_impl.cc

@@ -270,7 +270,7 @@ void ConnectionImpl::onFileEvent(uint32_t events) {
 }
 
 ConnectionImpl::IoResult ConnectionImpl::doReadFromSocket() {
-  PostIoAction action;
+  PostIoAction action = PostIoAction::KeepOpen;
   uint64_t bytes_read = 0;
   do {
     // 16K read is arbitrary. IIRC, libevent will currently clamp this to 4K. libevent will also
@@ -288,15 +288,17 @@ ConnectionImpl::IoResult ConnectionImpl::doReadFromSocket() {
     } else if (rc == -1) {
       // Remote error (might be no data).
       conn_log_trace("read error: {}", *this, errno);
-      if (errno == EAGAIN) {
-        action = PostIoAction::KeepOpen;
-      } else {
+      if (errno != EAGAIN) {
         action = PostIoAction::Close;
       }
 
       break;
     } else {
       bytes_read += rc;
+      if (shouldDrainReadBuffer()) {
+        setReadBufferReady();
+        break;
+      }
     }
   } while (true);
 

source/common/network/connection_impl.h

@@ -63,6 +63,7 @@ class ConnectionImpl : public virtual Connection,
   Ssl::Connection* ssl() override { return nullptr; }
   State state() override;
   void write(Buffer::Instance& data) override;
+  void setReadBufferLimit(size_t limit) override { read_buffer_limit_ = limit; }
 
   // Network::BufferSource
   Buffer::Instance& getReadBuffer() override { return read_buffer_; }
@@ -79,6 +80,16 @@ class ConnectionImpl : public virtual Connection,
   virtual void closeSocket(uint32_t close_type);
   void doConnect();
   void raiseEvents(uint32_t events);
+  // Should the read buffer be drained?
+  bool shouldDrainReadBuffer() {
+    return read_buffer_limit_ > 0 && read_buffer_.length() >= read_buffer_limit_;
+  }
+  // Mark read buffer ready to read in the event loop. This is used when yielding following
+  // shouldDrainReadBuffer().
+  // TODO(htuch): While this is the basis for also yielding to other connections to provide some
+  // fair sharing of CPU resources, the underlying event loop does not make any fairness guarantees.
+  // Reconsider how to make fairness happen.
+  void setReadBufferReady() { file_event_->activate(Event::FileReadyType::Read); }
 
   static const Address::InstancePtr null_local_address_;
 
@@ -87,6 +98,7 @@ class ConnectionImpl : public virtual Connection,
   Address::InstancePtr local_address_;
   Buffer::OwnedImpl read_buffer_;
   Buffer::OwnedImpl write_buffer_;
+  size_t read_buffer_limit_ = 0;
 
 private:
   // clang-format off

source/common/network/listener_impl.cc

@@ -63,10 +63,12 @@ void ListenerImpl::listenCallback(evconnlistener*, evutil_socket_t fd, sockaddr*
 ListenerImpl::ListenerImpl(Network::ConnectionHandler& conn_handler,
                            Event::DispatcherImpl& dispatcher, ListenSocket& socket,
                            ListenerCallbacks& cb, Stats::Store& stats_store, bool bind_to_port,
-                           bool use_proxy_proto, bool use_orig_dst)
+                           bool use_proxy_proto, bool use_orig_dst,
+                           size_t per_connection_buffer_limit_bytes)
     : connection_handler_(conn_handler), dispatcher_(dispatcher), socket_(socket), cb_(cb),
       bind_to_port_(bind_to_port), use_proxy_proto_(use_proxy_proto), proxy_protocol_(stats_store),
-      use_original_dst_(use_orig_dst), listener_(nullptr) {
+      use_original_dst_(use_orig_dst),
+      per_connection_buffer_limit_bytes_(per_connection_buffer_limit_bytes), listener_(nullptr) {
 
   if (bind_to_port_) {
     listener_.reset(
@@ -90,6 +92,7 @@ void ListenerImpl::errorCallback(evconnlistener*, void*) {
 void ListenerImpl::newConnection(int fd, Address::InstancePtr remote_address,
                                  Address::InstancePtr local_address) {
   ConnectionPtr new_connection(new ConnectionImpl(dispatcher_, fd, remote_address, local_address));
+  new_connection->setReadBufferLimit(per_connection_buffer_limit_bytes_);
   cb_.onNewConnection(std::move(new_connection));
 }
 
@@ -98,6 +101,7 @@ void SslListenerImpl::newConnection(int fd, Address::InstancePtr remote_address,
   ConnectionPtr new_connection(new Ssl::ConnectionImpl(dispatcher_, fd, remote_address,
                                                        local_address, ssl_ctx_,
                                                        Ssl::ConnectionImpl::InitialState::Server));
+  new_connection->setReadBufferLimit(per_connection_buffer_limit_bytes_);
   cb_.onNewConnection(std::move(new_connection));
 }
 

source/common/network/listener_impl.h

@@ -20,7 +20,8 @@ class ListenerImpl : public Listener {
 public:
   ListenerImpl(Network::ConnectionHandler& conn_handler, Event::DispatcherImpl& dispatcher,
                ListenSocket& socket, ListenerCallbacks& cb, Stats::Store& stats_store,
-               bool bind_to_port, bool use_proxy_proto, bool use_orig_dst);
+               bool bind_to_port, bool use_proxy_proto, bool use_orig_dst,
+               size_t per_connection_buffer_limit_bytes);
 
   /**
    * Accept/process a new connection.
@@ -47,6 +48,7 @@ class ListenerImpl : public Listener {
   const bool use_proxy_proto_;
   ProxyProtocol proxy_protocol_;
   const bool use_original_dst_;
+  const size_t per_connection_buffer_limit_bytes_;
 
 private:
   static void errorCallback(evconnlistener* listener, void* context);
@@ -60,9 +62,9 @@ class SslListenerImpl : public ListenerImpl {
   SslListenerImpl(Network::ConnectionHandler& conn_handler, Event::DispatcherImpl& dispatcher,
                   Ssl::Context& ssl_ctx, ListenSocket& socket, ListenerCallbacks& cb,
                   Stats::Store& stats_store, bool bind_to_port, bool use_proxy_proto,
-                  bool use_orig_dst)
+                  bool use_orig_dst, size_t per_connection_buffer_limit_bytes)
       : ListenerImpl(conn_handler, dispatcher, socket, cb, stats_store, bind_to_port,
-                     use_proxy_proto, use_orig_dst),
+                     use_proxy_proto, use_orig_dst, per_connection_buffer_limit_bytes),
         ssl_ctx_(ssl_ctx) {}
 
   // ListenerImpl

source/common/ssl/connection_impl.cc

@@ -79,6 +79,10 @@ Network::ConnectionImpl::IoResult ConnectionImpl::doReadFromSocket() {
 
     if (slices_to_commit > 0) {
       read_buffer_.commit(slices, slices_to_commit);
+      if (shouldDrainReadBuffer()) {
+        setReadBufferReady();
+        keep_reading = false;
+      }
     }
   }
 

source/server/configuration_impl.cc

@@ -129,6 +129,8 @@ MainImpl::ListenerConfig::ListenerConfig(MainImpl& parent, Json::Object& json)
   bind_to_port_ = json.getBoolean("bind_to_port", true);
   use_proxy_proto_ = json.getBoolean("use_proxy_proto", false);
   use_original_dst_ = json.getBoolean("use_original_dst", false);
+  per_connection_buffer_limit_bytes_ =
+      json.getInteger("per_connection_buffer_limit_bytes", 1024 * 1024);
 
   std::vector<Json::ObjectPtr> filters = json.getObjectArray("filters");
   for (size_t i = 0; i < filters.size(); i++) {

source/server/configuration_impl.h

@@ -95,6 +95,7 @@ class MainImpl : Logger::Loggable<Logger::Id::config>, public Main {
     Ssl::ServerContext* sslContext() override { return ssl_context_.get(); }
     bool useProxyProto() override { return use_proxy_proto_; }
     bool useOriginalDst() override { return use_original_dst_; }
+    size_t perConnectionBufferLimitBytes() override { return per_connection_buffer_limit_bytes_; }
 
     // Network::FilterChainFactory
     bool createFilterChain(Network::Connection& connection) override;
@@ -107,6 +108,7 @@ class MainImpl : Logger::Loggable<Logger::Id::config>, public Main {
     Ssl::ServerContextPtr ssl_context_;
     bool use_proxy_proto_{};
     bool use_original_dst_{};
+    size_t per_connection_buffer_limit_bytes_{};
     std::list<NetworkFilterFactoryCb> filter_factories_;
   };