perf: simplify connection buffer stats (#244)

We spend a considerable amount of time computing instantaneous
connection buffer stats and firing callbacks. Currently we only use
this information for stats purposes. This commit makes the buffer
stats eventually consistent and much more streamlined. Eventually,
we can provide direct accessors vs. callbacks if we need to inspect
buffer information in order to apply back pressure.

include/envoy/buffer/buffer.h

@@ -12,13 +12,6 @@ struct RawSlice {
   uint64_t len_;
 };
 
-/**
- * Buffer change callback.
- * @param old_size supplies the size of the buffer prior to the change.
- * @param data supplies how much data was added or removed.
- */
-typedef std::function<void(uint64_t old_size, int64_t delta)> Callback;
-
 /**
  * A basic buffer abstraction.
  */
@@ -118,13 +111,6 @@ class Instance {
    */
   virtual ssize_t search(const void* data, uint64_t size, size_t start) const PURE;
 
-  /**
-   * Set a buffer change callback. Only a single callback can be set at a time. The callback
-   * is invoked inline with buffer changes.
-   * @param callback supplies the callback to set. Pass nullptr to clear the callback.
-   */
-  virtual void setCallback(Callback callback) PURE;
-
   /**
    * Write the buffer out to a file descriptor.
    * @param fd supplies the descriptor to write to.

include/envoy/network/connection.h

@@ -34,14 +34,6 @@ class ConnectionCallbacks {
 public:
   virtual ~ConnectionCallbacks() {}
 
-  /**
-   * Callback for connection buffer changes.
-   * @param type supplies which buffer has changed.
-   * @param old_size supplies the original size of the buffer.
-   * @param delta supplies how much data was added or removed from the buffer.
-   */
-  virtual void onBufferChange(ConnectionBufferType type, uint64_t old_size, int64_t delta) PURE;
-
   /**
    * Callback for connection events.
    * @param events supplies the ConnectionEvent events that occurred as a bitmask.
@@ -64,6 +56,13 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
 public:
   enum class State { Open, Closing, Closed };
 
+  struct BufferStats {
+    Stats::Counter& read_total_;
+    Stats::Gauge& read_current_;
+    Stats::Counter& write_total_;
+    Stats::Gauge& write_current_;
+  };
+
   virtual ~Connection() {}
 
   /**
@@ -116,6 +115,13 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
    */
   virtual const std::string& remoteAddress() PURE;
 
+  /**
+   * Set the buffer stats to update when the connection's read/write buffers change. Note that
+   * for performance reasons these stats are eventually consistent and may not always accurately
+   * represent the buffer contents at any given point in time.
+   */
+  virtual void setBufferStats(const BufferStats& stats) PURE;
+
   /**
    * @return the SSL connection data if this is an SSL connection, or nullptr if it is not.
    */

source/common/buffer/buffer_impl.cc

@@ -6,10 +6,6 @@
 
 namespace Buffer {
 
-const evbuffer_cb_func OwnedImpl::buffer_cb_ =
-    [](evbuffer*, const evbuffer_cb_info* info, void* arg)
-        -> void { static_cast<OwnedImpl*>(arg)->onBufferChange(*info); };
-
 void OwnedImpl::add(const void* data, uint64_t size) { evbuffer_add(buffer_.get(), data, size); }
 
 void OwnedImpl::add(const std::string& data) {
@@ -78,10 +74,6 @@ void OwnedImpl::move(Instance& rhs, uint64_t length) {
   UNREFERENCED_PARAMETER(rc);
 }
 
-void OwnedImpl::onBufferChange(const evbuffer_cb_info& info) {
-  cb_(info.orig_size, info.n_added - info.n_deleted);
-}
-
 int OwnedImpl::read(int fd, uint64_t max_length) {
   return evbuffer_read(buffer_.get(), fd, max_length);
 }
@@ -108,17 +100,6 @@ ssize_t OwnedImpl::search(const void* data, uint64_t size, size_t start) const {
   return result_ptr.pos;
 }
 
-void OwnedImpl::setCallback(Callback callback) {
-  ASSERT(!callback || !cb_);
-  if (callback) {
-    evbuffer_add_cb(buffer_.get(), buffer_cb_, this);
-    cb_ = callback;
-  } else {
-    evbuffer_remove_cb(buffer_.get(), buffer_cb_, this);
-    cb_ = nullptr;
-  }
-}
-
 int OwnedImpl::write(int fd) { return evbuffer_write(buffer_.get(), fd); }
 
 OwnedImpl::OwnedImpl() : buffer_(evbuffer_new()) {}

source/common/buffer/buffer_impl.h

@@ -4,10 +4,6 @@
 
 #include "common/event/libevent.h"
 
-// Forward decls to avoid leaking libevent headers to rest of program.
-struct evbuffer_cb_info;
-typedef void (*evbuffer_cb_func)(evbuffer* buffer, const evbuffer_cb_info* info, void* arg);
-
 namespace Buffer {
 
 /**
@@ -34,17 +30,10 @@ class OwnedImpl : public Instance {
   int read(int fd, uint64_t max_length) override;
   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;
-  void setCallback(Callback callback) override;
   int write(int fd) override;
 
 private:
-  void onBufferChange(const evbuffer_cb_info& info);
-
-  static const evbuffer_cb_func buffer_cb_; // Static callback used for all evbuffer callbacks.
-                                            // This allows us to add/remove by value.
-
   Event::Libevent::BufferPtr buffer_;
-  Callback cb_; // The per buffer callback. Invoked via the buffer_cb_ static thunk.
 };
 
 } // Buffer

source/common/filter/auth/client_ssl.h

@@ -110,7 +110,6 @@ class Instance : public Network::ReadFilter, public Network::ConnectionCallbacks
   }
 
   // Network::ConnectionCallbacks
-  void onBufferChange(Network::ConnectionBufferType, uint64_t, int64_t) override {}
   void onEvent(uint32_t events) override;
 
 private:

source/common/filter/ratelimit.h

@@ -74,7 +74,6 @@ class Instance : public Network::ReadFilter,
   }
 
   // Network::ConnectionCallbacks
-  void onBufferChange(Network::ConnectionBufferType, uint64_t, int64_t) override {}
   void onEvent(uint32_t events) override;
 
   // RateLimit::RequestCallbacks

source/common/filter/tcp_proxy.cc

@@ -45,6 +45,16 @@ TcpProxyStats TcpProxyConfig::generateStats(const std::string& name, Stats::Stor
                               POOL_GAUGE_PREFIX(store, final_prefix))};
 }
 
+void TcpProxy::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
+  read_callbacks_ = &callbacks;
+  conn_log_info("new tcp proxy session", read_callbacks_->connection());
+  read_callbacks_->connection().addConnectionCallbacks(downstream_callbacks_);
+  read_callbacks_->connection().setBufferStats({config_->stats().downstream_cx_rx_bytes_total_,
+                                                config_->stats().downstream_cx_rx_bytes_buffered_,
+                                                config_->stats().downstream_cx_tx_bytes_total_,
+                                                config_->stats().downstream_cx_tx_bytes_buffered_});
+}
+
 Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
   Upstream::ResourceManager& upstream_cluster_resource_manager =
       cluster_manager_.get(config_->clusterName())
@@ -68,6 +78,11 @@ Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
 
   upstream_connection_->addReadFilter(upstream_callbacks_);
   upstream_connection_->addConnectionCallbacks(*upstream_callbacks_);
+  upstream_connection_->setBufferStats(
+      {read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_rx_bytes_total_,
+       read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_rx_bytes_buffered_,
+       read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_tx_bytes_total_,
+       read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_tx_bytes_buffered_});
   upstream_connection_->connect();
   upstream_connection_->noDelay(true);
 
@@ -103,18 +118,6 @@ Network::FilterStatus TcpProxy::onData(Buffer::Instance& data) {
   return Network::FilterStatus::StopIteration;
 }
 
-void TcpProxy::onDownstreamBufferChange(Network::ConnectionBufferType type, uint64_t,
-                                        int64_t delta) {
-  if (type == Network::ConnectionBufferType::Write) {
-    if (delta > 0) {
-      config_->stats().downstream_cx_tx_bytes_total_.add(delta);
-      config_->stats().downstream_cx_tx_bytes_buffered_.add(delta);
-    } else {
-      config_->stats().downstream_cx_tx_bytes_buffered_.sub(std::abs(delta));
-    }
-  }
-}
-
 void TcpProxy::onDownstreamEvent(uint32_t event) {
   if ((event & Network::ConnectionEvent::RemoteClose ||
        event & Network::ConnectionEvent::LocalClose) &&
@@ -127,18 +130,6 @@ void TcpProxy::onDownstreamEvent(uint32_t event) {
   }
 }
 
-void TcpProxy::onUpstreamBufferChange(Network::ConnectionBufferType type, uint64_t, int64_t delta) {
-  if (type == Network::ConnectionBufferType::Write) {
-    if (delta > 0) {
-      read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_tx_bytes_total_.add(delta);
-      read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_tx_bytes_buffered_.add(delta);
-    } else {
-      read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_tx_bytes_buffered_.sub(
-          std::abs(delta));
-    }
-  }
-}
-
 void TcpProxy::onUpstreamData(Buffer::Instance& data) {
   read_callbacks_->connection().write(data);
   ASSERT(0 == data.length());

source/common/filter/tcp_proxy.h

@@ -17,6 +17,8 @@ namespace Filter {
  */
 // clang-format off
 #define ALL_TCP_PROXY_STATS(COUNTER, GAUGE)                                                        \
+  COUNTER(downstream_cx_rx_bytes_total)                                                            \
+  GAUGE  (downstream_cx_rx_bytes_buffered)                                                         \
   COUNTER(downstream_cx_tx_bytes_total)                                                            \
   GAUGE  (downstream_cx_tx_bytes_buffered)
 // clang-format on
@@ -61,22 +63,13 @@ class TcpProxy : public Network::ReadFilter, Logger::Loggable<Logger::Id::filter
   // Network::ReadFilter
   Network::FilterStatus onData(Buffer::Instance& data) override;
   Network::FilterStatus onNewConnection() override { return initializeUpstreamConnection(); }
-  void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override {
-    read_callbacks_ = &callbacks;
-    conn_log_info("new tcp proxy session", read_callbacks_->connection());
-    read_callbacks_->connection().addConnectionCallbacks(downstream_callbacks_);
-  }
+  void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override;
 
 private:
   struct DownstreamCallbacks : public Network::ConnectionCallbacks {
     DownstreamCallbacks(TcpProxy& parent) : parent_(parent) {}
 
     // Network::ConnectionCallbacks
-    void onBufferChange(Network::ConnectionBufferType type, uint64_t old_size,
-                        int64_t delta) override {
-      parent_.onDownstreamBufferChange(type, old_size, delta);
-    }
-
     void onEvent(uint32_t event) override { parent_.onDownstreamEvent(event); }
 
     TcpProxy& parent_;
@@ -87,11 +80,6 @@ class TcpProxy : public Network::ReadFilter, Logger::Loggable<Logger::Id::filter
     UpstreamCallbacks(TcpProxy& parent) : parent_(parent) {}
 
     // Network::ConnectionCallbacks
-    void onBufferChange(Network::ConnectionBufferType type, uint64_t old_size,
-                        int64_t delta) override {
-      parent_.onUpstreamBufferChange(type, old_size, delta);
-    }
-
     void onEvent(uint32_t event) override { parent_.onUpstreamEvent(event); }
 
     // Network::ReadFilter
@@ -105,10 +93,7 @@ class TcpProxy : public Network::ReadFilter, Logger::Loggable<Logger::Id::filter
 
   Network::FilterStatus initializeUpstreamConnection();
   void onConnectTimeout();
-  void onDownstreamBufferChange(Network::ConnectionBufferType type, uint64_t old_size,
-                                int64_t delta);
   void onDownstreamEvent(uint32_t event);
-  void onUpstreamBufferChange(Network::ConnectionBufferType type, uint64_t old_size, int64_t delta);
   void onUpstreamData(Buffer::Instance& data);
   void onUpstreamEvent(uint32_t event);
 

source/common/http/codec_client.h

@@ -99,6 +99,10 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
    */
   StreamEncoder& newStream(StreamDecoder& response_decoder);
 
+  void setBufferStats(const Network::Connection::BufferStats& stats) {
+    connection_->setBufferStats(stats);
+  }
+
   void setCodecClientCallbacks(CodecClientCallbacks& callbacks) {
     codec_client_callbacks_ = &callbacks;
   }
@@ -180,7 +184,6 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
   void onData(Buffer::Instance& data);
 
   // Network::ConnectionCallbacks
-  void onBufferChange(Network::ConnectionBufferType, uint64_t, int64_t) override {}
   void onEvent(uint32_t events) override;
 
   std::list<ActiveRequestPtr> active_requests_;

source/common/http/conn_manager_impl.cc

@@ -57,6 +57,11 @@ void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCal
         [this]() -> void { onIdleTimeout(); });
     idle_timer_->enableTimer(config_.idleTimeout().value());
   }
+
+  read_callbacks_->connection().setBufferStats({stats_.named_.downstream_cx_rx_bytes_total_,
+                                                stats_.named_.downstream_cx_rx_bytes_buffered_,
+                                                stats_.named_.downstream_cx_tx_bytes_total_,
+                                                stats_.named_.downstream_cx_tx_bytes_buffered_});
 }
 
 ConnectionManagerImpl::~ConnectionManagerImpl() {
@@ -189,14 +194,6 @@ Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data) {
   return Network::FilterStatus::StopIteration;
 }
 
-void ConnectionManagerImpl::onBufferChange(Network::ConnectionBufferType type, uint64_t,
-                                           int64_t delta) {
-  Network::Utility::updateBufferStats(type, delta, stats_.named_.downstream_cx_rx_bytes_total_,
-                                      stats_.named_.downstream_cx_rx_bytes_buffered_,
-                                      stats_.named_.downstream_cx_tx_bytes_total_,
-                                      stats_.named_.downstream_cx_tx_bytes_buffered_);
-}
-
 void ConnectionManagerImpl::resetAllStreams() {
   while (!streams_.empty()) {
     // Mimic a downstream reset in this case.

source/common/http/conn_manager_impl.h

@@ -215,8 +215,6 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   StreamDecoder& newStream(StreamEncoder& response_encoder) override;
 
   // Network::ConnectionCallbacks
-  void onBufferChange(Network::ConnectionBufferType type, uint64_t old_size,
-                      int64_t delta) override;
   void onEvent(uint32_t events) override;
 
 private:

source/common/http/http1/conn_pool.cc

@@ -276,6 +276,11 @@ ConnPoolImpl::ActiveClient::ActiveClient(ConnPoolImpl& parent)
   conn_length_ = parent_.host_->cluster().stats().upstream_cx_length_ms_.allocateSpan();
   connect_timer_->enableTimer(parent_.host_->cluster().connectTimeout());
   parent_.host_->cluster().resourceManager(parent_.priority_).connections().inc();
+
+  codec_client_->setBufferStats({parent_.host_->cluster().stats().upstream_cx_rx_bytes_total_,
+                                 parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
+                                 parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
+                                 parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_});
 }
 
 ConnPoolImpl::ActiveClient::~ActiveClient() {
@@ -285,15 +290,6 @@ ConnPoolImpl::ActiveClient::~ActiveClient() {
   parent_.host_->cluster().resourceManager(parent_.priority_).connections().dec();
 }
 
-void ConnPoolImpl::ActiveClient::onBufferChange(Network::ConnectionBufferType type, uint64_t,
-                                                int64_t delta) {
-  Network::Utility::updateBufferStats(
-      type, delta, parent_.host_->cluster().stats().upstream_cx_rx_bytes_total_,
-      parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
-      parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
-      parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_);
-}
-
 void ConnPoolImpl::ActiveClient::onConnectTimeout() {
   // We just close the client at this point. This will result in both a timeout and a connect
   // failure and will fold into all the normal connect failure logic.

source/common/http/http1/conn_pool.h

@@ -70,8 +70,6 @@ class ConnPoolImpl : Logger::Loggable<Logger::Id::pool>, public ConnectionPool::
     void onConnectTimeout();
 
     // Network::ConnectionCallbacks
-    void onBufferChange(Network::ConnectionBufferType type, uint64_t old_size,
-                        int64_t delta) override;
     void onEvent(uint32_t events) override { parent_.onConnectionEvent(*this, events); }
 
     ConnPoolImpl& parent_;

source/common/http/http2/conn_pool.cc

@@ -226,6 +226,11 @@ ConnPoolImpl::ActiveClient::ActiveClient(ConnPoolImpl& parent)
   parent_.host_->cluster().stats().upstream_cx_active_.inc();
   parent_.host_->cluster().stats().upstream_cx_http2_total_.inc();
   conn_length_ = parent_.host_->cluster().stats().upstream_cx_length_ms_.allocateSpan();
+
+  client_->setBufferStats({parent_.host_->cluster().stats().upstream_cx_rx_bytes_total_,
+                           parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
+                           parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
+                           parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_});
 }
 
 ConnPoolImpl::ActiveClient::~ActiveClient() {
@@ -234,15 +239,6 @@ ConnPoolImpl::ActiveClient::~ActiveClient() {
   conn_length_->complete();
 }
 
-void ConnPoolImpl::ActiveClient::onBufferChange(Network::ConnectionBufferType type, uint64_t,
-                                                int64_t delta) {
-  Network::Utility::updateBufferStats(
-      type, delta, parent_.host_->cluster().stats().upstream_cx_rx_bytes_total_,
-      parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
-      parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
-      parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_);
-}
-
 CodecClientPtr ProdConnPoolImpl::createCodecClient(Upstream::Host::CreateConnectionData& data) {
   CodecClientStats stats{host_->cluster().stats().upstream_cx_protocol_error_};
   CodecClientPtr codec{new CodecClientProd(CodecClient::Type::HTTP2, std::move(data.connection_),