connection: skip read activate call when reading from transport socket if the connection is read disabled

include/envoy/network/transport_socket.h

@@ -65,10 +65,10 @@ class TransportSocketCallbacks {
   virtual bool shouldDrainReadBuffer() PURE;
 
   /**
-   * Mark read buffer ready to read in the event loop. This is used when yielding following
-   * shouldDrainReadBuffer().
+   * Mark the transport socket as readable in order to force a read in a future iteration of the
+   * event loop. This is used when yielding following shouldDrainReadBuffer().
    */
-  virtual void setReadBufferReady() PURE;
+  virtual void setTransportSocketIsReadable() PURE;
 
   /**
    * Raise a connection event to the connection. This can be used by a secure socket (e.g. TLS)

source/common/network/connection_impl.cc

@@ -194,6 +194,19 @@ Connection::State ConnectionImpl::state() const {
 
 void ConnectionImpl::closeConnectionImmediately() { closeSocket(ConnectionEvent::LocalClose); }
 
+void ConnectionImpl::setTransportSocketIsReadable() {
+  // Remember that the transport requested read resumption, in case the resumption event is not
+  // scheduled immediately or is "lost" because read was disabled.
+  transport_wants_read_ = true;
+  // Only schedule a read activation if the connection is not read disabled to avoid spurious
+  // wakeups. When read disabled, the connection will not read from the transport, and limit
+  // dispatch to the current contents of the read buffer if its high-watermark is triggered and
+  // dispatch_buffered_data_ is set.
+  if (read_disable_count_ == 0) {
+    ioHandle().activateFileEvents(Event::FileReadyType::Read);
+  }
+}
+
 bool ConnectionImpl::filterChainWantsData() {
   return read_disable_count_ == 0 ||
          (read_disable_count_ == 1 && read_buffer_.highWatermarkTriggered());
@@ -358,18 +371,19 @@ void ConnectionImpl::readDisable(bool disable) {
     }
 
     if (filterChainWantsData() && (read_buffer_.length() > 0 || transport_wants_read_)) {
-      // If the read_buffer_ is not empty or transport_wants_read_ is true, the connection may be
-      // able to process additional bytes even if there is no data in the kernel to kick off the
-      // filter chain. Alternately if the read buffer has data the fd could be read disabled. To
-      // handle these cases, fake an event to make sure the buffered data in the read buffer or in
-      // transport socket internal buffers gets processed regardless and ensure that we dispatch it
-      // via onRead.
-
       // Sanity check: resumption with read_disable_count_ > 0 should only happen if the read
       // buffer's high watermark has triggered.
       ASSERT(read_buffer_.length() > 0 || read_disable_count_ == 0);
+
+      // If the read_buffer_ is not empty or transport_wants_read_ is true, the connection may be
+      // able to process additional bytes even if there is no data in the kernel to kick off the
+      // filter chain. Alternately the connection may need read resumption while read disabled and
+      // not registered for read events because the read buffer's high-watermark has triggered. To
+      // handle these cases, directly schedule a fake read event to make sure the buffered data in
+      // the read buffer or in transport socket internal buffers gets processed regardless and
+      // ensure that we dispatch it via onRead.
       dispatch_buffered_data_ = true;
-      setReadBufferReady();
+      ioHandle().activateFileEvents(Event::FileReadyType::Read);
     }
   }
 }
@@ -560,8 +574,9 @@ void ConnectionImpl::onReadReady() {
   ASSERT(!connecting_);
 
   // We get here while read disabled in two ways.
-  // 1) There was a call to setReadBufferReady(), for example if a raw buffer socket ceded due to
-  //    shouldDrainReadBuffer(). In this case we defer the event until the socket is read enabled.
+  // 1) There was a call to setTransportSocketIsReadable(), for example if a raw buffer socket ceded
+  //    due to shouldDrainReadBuffer(). In this case we defer the event until the socket is read
+  //    enabled.
   // 2) The consumer of connection data called readDisable(true), and instead of reading from the
   //    socket we simply need to dispatch already read data.
   if (read_disable_count_ != 0) {

source/common/network/connection_impl.h

@@ -119,10 +119,7 @@ class ConnectionImpl : public ConnectionImplBase, public TransportSocketCallback
   // 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() override {
-    transport_wants_read_ = true;
-    ioHandle().activateFileEvents(Event::FileReadyType::Read);
-  }
+  void setTransportSocketIsReadable() override;
   void flushWriteBuffer() override;
 
   // Obtain global next connection ID. This should only be used in tests.
@@ -200,10 +197,11 @@ class ConnectionImpl : public ConnectionImplBase, public TransportSocketCallback
   bool write_end_stream_ : 1;
   bool current_write_end_stream_ : 1;
   bool dispatch_buffered_data_ : 1;
-  // True if the most recent call to the transport socket's doRead method invoked setReadBufferReady
-  // to schedule read resumption after yielding due to shouldDrainReadBuffer(). When true,
-  // readDisable must schedule read resumption when read_disable_count_ == 0 to ensure that read
-  // resumption happens when remaining bytes are held in transport socket internal buffers.
+  // True if the most recent call to the transport socket's doRead method invoked
+  // setTransportSocketIsReadable to schedule read resumption after yielding due to
+  // shouldDrainReadBuffer(). When true, readDisable must schedule read resumption when
+  // read_disable_count_ == 0 to ensure that read resumption happens when remaining bytes are held
+  // in transport socket internal buffers.
   bool transport_wants_read_ : 1;
 };
 

source/common/network/raw_buffer_socket.cc

@@ -30,7 +30,7 @@ IoResult RawBufferSocket::doRead(Buffer::Instance& buffer) {
       }
       bytes_read += result.rc_;
       if (callbacks_->shouldDrainReadBuffer()) {
-        callbacks_->setReadBufferReady();
+        callbacks_->setTransportSocketIsReadable();
         break;
       }
     } else {

source/extensions/transport_sockets/alts/noop_transport_socket_callbacks.h

@@ -25,7 +25,7 @@ class NoOpTransportSocketCallbacks : public Network::TransportSocketCallbacks {
   /*
    * No-op for these two methods to hold back the callbacks.
    */
-  void setReadBufferReady() override {}
+  void setTransportSocketIsReadable() override {}
   void raiseEvent(Network::ConnectionEvent) override {}
   void flushWriteBuffer() override {}
 

source/extensions/transport_sockets/alts/tsi_socket.cc

@@ -149,7 +149,7 @@ Network::PostIoAction TsiSocket::doHandshakeNextDone(NextResultPtr&& next_result
   }
 
   if (raw_read_buffer_.length() > 0) {
-    callbacks_->setReadBufferReady();
+    callbacks_->setTransportSocketIsReadable();
   }
 
   // Try to write raw buffer when next call is done, even this is not in do[Read|Write] stack.

source/extensions/transport_sockets/tls/ssl_socket.cc

@@ -165,7 +165,7 @@ Network::IoResult SslSocket::doRead(Buffer::Instance& read_buffer) {
     if (slices_to_commit > 0) {
       read_buffer.commit(slices, slices_to_commit);
       if (callbacks_->shouldDrainReadBuffer()) {
-        callbacks_->setReadBufferReady();
+        callbacks_->setTransportSocketIsReadable();
         keep_reading = false;
       }
     }
@@ -301,7 +301,7 @@ void SslSocket::shutdownSsl() {
       if (rc == 0) {
         // See https://www.openssl.org/docs/manmaster/man3/SSL_shutdown.html
         // if return value is 0,  Call SSL_read() to do a bidirectional shutdown.
-        callbacks_->setReadBufferReady();
+        callbacks_->setTransportSocketIsReadable();
       }
     }
     ENVOY_CONN_LOG(debug, "SSL shutdown: rc={}", callbacks_->connection(), rc);

test/common/network/connection_impl_test.cc

@@ -1889,8 +1889,8 @@ TEST_F(MockTransportConnectionImplTest, ObjectDestructOrder) {
   file_ready_cb_(Event::FileReadyType::Read);
 }
 
-// Verify that read resumptions requested via setReadBufferReady() are scheduled once read is
-// re-enabled.
+// Verify that read resumptions requested via setTransportSocketIsReadable() are scheduled once read
+// is re-enabled.
 TEST_F(MockTransportConnectionImplTest, ReadBufferReadyResumeAfterReadDisable) {
   InSequence s;
 
@@ -1905,9 +1905,9 @@ TEST_F(MockTransportConnectionImplTest, ReadBufferReadyResumeAfterReadDisable) {
   EXPECT_CALL(*file_event_, activate(_)).Times(0);
   connection_->readDisable(false);
 
-  // setReadBufferReady triggers an immediate call to activate.
+  // setTransportSocketIsReadable triggers an immediate call to activate.
   EXPECT_CALL(*file_event_, activate(Event::FileReadyType::Read));
-  connection_->setReadBufferReady();
+  connection_->setTransportSocketIsReadable();
 
   // When processing a sequence of read disable/read enable, changes to the enabled event mask
   // happen only when the disable count transitions to/from 0.
@@ -1919,7 +1919,7 @@ TEST_F(MockTransportConnectionImplTest, ReadBufferReadyResumeAfterReadDisable) {
   connection_->readDisable(false);
   EXPECT_CALL(*file_event_, setEnabled(Event::FileReadyType::Read | Event::FileReadyType::Write));
   // Expect a read activation since there have been no transport doRead calls since the call to
-  // setReadBufferReady.
+  // setTransportSocketIsReadable.
   EXPECT_CALL(*file_event_, activate(Event::FileReadyType::Read));
   connection_->readDisable(false);
 
@@ -2023,19 +2023,19 @@ TEST_F(MockTransportConnectionImplTest, ResumeWhileAndAfterReadDisable) {
   connection_->enableHalfClose(true);
   connection_->addReadFilter(read_filter);
 
-  // Add some data to the read buffer and also call setReadBufferReady to mimic what transport
-  // sockets are expected to do when the read buffer high watermark is hit.
+  // Add some data to the read buffer and also call setTransportSocketIsReadable to mimic what
+  // transport sockets are expected to do when the read buffer high watermark is hit.
   EXPECT_CALL(*transport_socket_, doRead(_))
       .WillOnce(Invoke([this](Buffer::Instance& buffer) -> IoResult {
         buffer.add("0123456789");
-        connection_->setReadBufferReady();
+        connection_->setTransportSocketIsReadable();
         return {PostIoAction::KeepOpen, 10, false};
       }));
   // Expect a change to the event mask when hitting the read buffer high-watermark.
   EXPECT_CALL(*file_event_, setEnabled(Event::FileReadyType::Write));
-  // The setReadBufferReady call adds a spurious read activation.
-  // TODO(antoniovicente) Skip the read activate in setReadBufferReady when read_disable_count_ > 0.
-  EXPECT_CALL(*file_event_, activate(Event::FileReadyType::Read));
+  // The setTransportSocketIsReadable does not call activate because read_disable_count_ > 0 due to
+  // high-watermark.
+  EXPECT_CALL(*file_event_, activate(Event::FileReadyType::Read)).Times(0);
   EXPECT_CALL(*read_filter, onNewConnection()).WillOnce(Return(FilterStatus::Continue));
   EXPECT_CALL(*read_filter, onData(_, false)).WillOnce(Return(FilterStatus::Continue));
   file_ready_cb_(Event::FileReadyType::Read);
@@ -2061,7 +2061,7 @@ TEST_F(MockTransportConnectionImplTest, ResumeWhileAndAfterReadDisable) {
         data.drain(data.length());
         return FilterStatus::Continue;
       }));
-  // The buffer is fully drained. Expect a read activation because setReadBufferReady set
+  // The buffer is fully drained. Expect a read activation because setTransportSocketIsReadable set
   // transport_wants_read_ and no transport doRead calls have happened.
   EXPECT_CALL(*file_event_, setEnabled(Event::FileReadyType::Read | Event::FileReadyType::Write));
   EXPECT_CALL(*file_event_, activate(Event::FileReadyType::Read));

test/extensions/transport_sockets/alts/noop_transport_socket_callbacks_test.cc

@@ -24,17 +24,17 @@ class TestTransportSocketCallbacks : public Network::TransportSocketCallbacks {
   const Network::IoHandle& ioHandle() const override { return *io_handle_; }
   Network::Connection& connection() override { return connection_; }
   bool shouldDrainReadBuffer() override { return false; }
-  void setReadBufferReady() override { set_read_buffer_ready_ = true; }
+  void setTransportSocketIsReadable() override { transport_socket_is_readable_ = true; }
   void raiseEvent(Network::ConnectionEvent) override { event_raised_ = true; }
   void flushWriteBuffer() override { write_buffer_flushed_ = true; }
 
-  bool event_raised() const { return event_raised_; }
-  bool set_read_buffer_ready() const { return set_read_buffer_ready_; }
-  bool write_buffer_flushed() const { return write_buffer_flushed_; }
+  bool eventRaised() const { return event_raised_; }
+  bool transportSocketIsReadable() const { return transport_socket_is_readable_; }
+  bool writeBufferFlushed() const { return write_buffer_flushed_; }
 
 private:
   bool event_raised_{false};
-  bool set_read_buffer_ready_{false};
+  bool transport_socket_is_readable_{false};
   bool write_buffer_flushed_{false};
   Network::IoHandlePtr io_handle_;
   Network::Connection& connection_;
@@ -55,12 +55,12 @@ TEST_F(NoOpTransportSocketCallbacksTest, TestAllCallbacks) {
   EXPECT_EQ(&connection_, &wrapped_callbacks_.connection());
   EXPECT_FALSE(wrapped_callbacks_.shouldDrainReadBuffer());
 
-  wrapped_callbacks_.setReadBufferReady();
-  EXPECT_FALSE(wrapper_callbacks_.set_read_buffer_ready());
+  wrapped_callbacks_.setTransportSocketIsReadable();
+  EXPECT_FALSE(wrapper_callbacks_.transportSocketIsReadable());
   wrapped_callbacks_.raiseEvent(Network::ConnectionEvent::Connected);
-  EXPECT_FALSE(wrapper_callbacks_.event_raised());
+  EXPECT_FALSE(wrapper_callbacks_.eventRaised());
   wrapped_callbacks_.flushWriteBuffer();
-  EXPECT_FALSE(wrapper_callbacks_.write_buffer_flushed());
+  EXPECT_FALSE(wrapper_callbacks_.writeBufferFlushed());
 }
 
 } // namespace

test/mocks/network/mocks.h

@@ -474,7 +474,7 @@ class MockTransportSocketCallbacks : public TransportSocketCallbacks {
   MOCK_METHOD(const IoHandle&, ioHandle, (), (const));
   MOCK_METHOD(Connection&, connection, ());
   MOCK_METHOD(bool, shouldDrainReadBuffer, ());
-  MOCK_METHOD(void, setReadBufferReady, ());
+  MOCK_METHOD(void, setTransportSocketIsReadable, ());
   MOCK_METHOD(void, raiseEvent, (ConnectionEvent));
   MOCK_METHOD(void, flushWriteBuffer, ());