MB-66385: Reduce the number of notifications

Change the behavior of Connection::triggerCallback to only
schedule a callback if none will be generated automatically
due to data transfer (when all data is moved to kernelspace)

Change-Id: Id6b8222cc34ad462977230014bbaa590ac4fa480
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/226752
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Jim Walker <jim@couchbase.com>

Author: trondn

daemon/connection.cc

@@ -865,10 +865,11 @@ void Connection::tryToProgressDcpStream() {
             more = false;
         }
     }
-    if (more && (numEvents == 0 || exceededTimeslice)) {
-        // We used the entire timeslice... schedule a new one
-        triggerCallback();
-    }
+    // There is no need to try to request a "trigger" to continue the
+    // state machine because we would either have data in the send queue
+    // causing a callback to arrive (no matter if more is true or false)
+    // If numevents we would either have data in the output buffer or
+    // progressing the input stream would already have set up the callback
 }
 
 void Connection::processBlockedSendQueue(
@@ -917,7 +918,12 @@ void Connection::processNotifiedCookie(Cookie& cookie, cb::engine_errc status) {
                     return ptr == cookie.get();
                 });
             }
-            triggerCallback();
+
+            if (!isPacketAvailable()) {
+                enableReadEvent();
+            } else {
+                triggerCallback();
+            }
         } else if (std::chrono::steady_clock::now() > current_timeslice_end) {
             ++yields;
         }
@@ -1336,7 +1342,7 @@ bool Connection::maybeInitiateShutdown(const std::string_view reason) {
                  {"reason", reason});
     setTerminationReason(std::move(message));
     shutdown();
-    triggerCallback();
+    triggerCallback(true);
     return true;
 }
 
@@ -1516,7 +1522,7 @@ bool Connection::signalIfIdle() {
     }
 
     if (state != State::immediate_close) {
-        triggerCallback();
+        triggerCallback(true);
         return true;
     }
     return false;

daemon/connection.h

@@ -645,8 +645,16 @@ class Connection : public ConnectionIface, public DcpMessageProducersIface {
      * Trigger a callback from libevent for the connection at some time
      * in the future (as part of the event dispatch loop) so that the
      * connection may continue its command execution.
+     * By default a new callback is only scheduled if there isn't data
+     * in the send queue as that will generate a callback once the
+     * data is in the send queue. In some cases we would want the
+     * callback to be scheduled anyway (for instance when trying to shut
+     * down a connection which isn't reading its socket). To do so
+     * set force to true
+     *
+     * @param force force a new EV_READ callback to be triggered
      */
-    virtual void triggerCallback() = 0;
+    virtual void triggerCallback(bool force = false) = 0;
 
     /// Check if DCP should use the write buffer for the message or if it
     /// should use an IOVector to do so

daemon/connection_libevent.cc

@@ -340,7 +340,11 @@ void LibeventConnection::ssl_read_callback(bufferevent* bev, void* ctx) {
     LibeventConnection::read_callback(bev, ctx);
 }
 
-void LibeventConnection::triggerCallback() {
+void LibeventConnection::triggerCallback(bool force) {
+    if (!force && getSendQueueSize() != 0) {
+        // The framework will send a notification once the data is sent
+        return;
+    }
     constexpr auto opt = BEV_TRIG_IGNORE_WATERMARKS | BEV_TRIG_DEFER_CALLBACKS;
     bufferevent_trigger(bev.get(), EV_READ, opt);
 }

daemon/connection_libevent.h

@@ -37,7 +37,7 @@ class LibeventConnection : public Connection {
     void nextPacket() override;
     cb::const_byte_buffer getAvailableBytes() const override;
     size_t getSendQueueSize() const override;
-    void triggerCallback() override;
+    void triggerCallback(bool force) override;
     void disableReadEvent() override;
     void enableReadEvent() override;
 

protocol/mcbp/mcbp_fuzz_test.cc

@@ -56,7 +56,7 @@ class FuzzConnection : public Connection {
     cb::const_byte_buffer getAvailableBytes() const override {
         throw std::runtime_error("FuzzConnection: Not implemented");
     }
-    void triggerCallback() override {
+    void triggerCallback(bool) override {
         throw std::runtime_error("FuzzConnection: Not implemented");
     }
 

tests/mcbp/mcbp_mock_connection.h

@@ -37,7 +37,7 @@ class McbpMockConnection : public Connection {
     cb::const_byte_buffer getAvailableBytes() const override {
         throw std::runtime_error("MockConnection: Not implemented");
     }
-    void triggerCallback() override {
+    void triggerCallback(bool) override {
         throw std::runtime_error("MockConnection: Not implemented");
     }
 

tests/testapp/testapp_dcp.cc

@@ -181,8 +181,10 @@ TEST_P(DcpTest, DcpStreamStats) {
             << "dcp stats: " << stats.dump(2);
 }
 
-/// Verify that we log unclean DCP disconnects
-TEST_P(DcpTest, MB60706) {
+/// Verify that we log unclean DCP disconnects. Disabled as part of the
+/// change to reduce the number of notifications (this changes the scheduling
+/// order in the test)
+TEST_P(DcpTest, DISABLED_MB60706) {
     nlohmann::json json;
 
     std::string value(2048 * 1024, 'a');