[Refactor] Add scheduleDcpStep to server cookie iface

This is a step on the way to decouple the tight binding between
the cookie used in DCP open and the cookie used to hold the
connection in the engine.

Change-Id: Ib300a0dcd409d651511f602af35ee559fb79adc1
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/141088
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: trondn

auditd/tests/testauditd.cc

@@ -89,6 +89,10 @@ class AuditMockServerCookieApi : public ServerCookieIface {
         ready = true;
         cond.notify_one();
     }
+    void scheduleDcpStep(gsl::not_null<const void*> cookie) override {
+        throw std::runtime_error("Not implemented");
+    }
+
     ENGINE_ERROR_CODE reserve(gsl::not_null<const void*> cookie) override {
         throw std::runtime_error("Not implemented");
     }

daemon/server_api.cc

@@ -187,6 +187,33 @@ struct ServerCookieApi : public ServerCookieIface {
         ::notify_io_complete(cookie, status);
     }
 
+    void scheduleDcpStep(gsl::not_null<const void*> cookie_) override {
+        auto& cookie = getCookie(cookie_);
+        auto& connection = cookie.getConnection();
+        if (!connection.isDCP()) {
+            LOG_WARNING(
+                    "scheduleDcpStep: Must only be called with a DCP "
+                    "connection: {}",
+                    connection.toJSON().dump());
+            throw std::logic_error(
+                    "scheduleDcpStep(): Provided cookie is not bound to a "
+                    "connection set up for DCP");
+        }
+
+        if (cookie.getRefcount() == 0) {
+            LOG_WARNING(
+                    "scheduleDcpStep: DCP connection did not reserve the "
+                    "cookie: {}",
+                    cookie.getConnection().toJSON().dump());
+            throw std::logic_error("scheduleDcpStep: cookie must be reserved!");
+        }
+
+        // @todo refactor so that notify_io_complete can ensure that it is
+        //       only notified in a blocked state (and for this case we
+        //       notify if it isn't already notified)
+        ::notify_io_complete(&cookie, ENGINE_SUCCESS);
+    }
+
     ENGINE_ERROR_CODE reserve(gsl::not_null<const void*> void_cookie) override {
         getCookie(void_cookie).incrementRefcount();
         return ENGINE_SUCCESS;

engines/ep/src/connmap.cc

@@ -131,7 +131,7 @@ void ConnMap::notifyPausedConnection(const std::shared_ptr<ConnHandler>& conn) {
     {
         LockHolder rlh(releaseLock);
         if (conn.get() && conn->isPaused() && conn->isReserved()) {
-            engine.notifyIOComplete(conn->getCookie(), ENGINE_SUCCESS);
+            engine.scheduleDcpStep(conn->getCookie());
         }
     }
 }

engines/ep/src/dcp/dcpconnmap.cc

@@ -402,7 +402,7 @@ void DcpConnMap::manageConnections() {
 
     for (auto& it : toNotify) {
         if (it.get() && it->isReserved()) {
-            engine.notifyIOComplete(it->getCookie(), ENGINE_SUCCESS);
+            engine.scheduleDcpStep(it->getCookie());
         }
     }
 

engines/ep/src/ep_engine.cc

@@ -6059,6 +6059,12 @@ void EventuallyPersistentEngine::notifyIOComplete(const void* cookie,
     }
 }
 
+void EventuallyPersistentEngine::scheduleDcpStep(
+        gsl::not_null<const void*> cookie) {
+    NonBucketAllocationGuard guard;
+    serverApi->cookie->scheduleDcpStep(cookie);
+}
+
 ENGINE_ERROR_CODE EventuallyPersistentEngine::getRandomKey(
         const void* cookie,
         const cb::mcbp::Request& request,

engines/ep/src/ep_engine.h

@@ -529,6 +529,7 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
     void setDCPPriority(const void* cookie, ConnectionPriority priority);
 
     void notifyIOComplete(const void* cookie, ENGINE_ERROR_CODE status);
+    void scheduleDcpStep(gsl::not_null<const void*> cookie);
 
     ENGINE_ERROR_CODE reserveCookie(const void *cookie);
     ENGINE_ERROR_CODE releaseCookie(const void *cookie);

engines/ep/tests/module_tests/dcp_test.cc

@@ -120,6 +120,9 @@ class WrappedServerCookieIface : public ServerCookieIface {
                             ENGINE_ERROR_CODE status) override {
         return wrapped->notify_io_complete(cookie, status);
     }
+    void scheduleDcpStep(gsl::not_null<const void*> cookie) override {
+        wrapped->scheduleDcpStep(cookie);
+    }
     ENGINE_ERROR_CODE reserve(gsl::not_null<const void*> cookie) override {
         return wrapped->reserve(cookie);
     }
@@ -1601,8 +1604,7 @@ TEST_P(ConnectionTest, test_mb20716_connmap_notify_on_delete) {
     size_t notify_count = 0;
     class MockServerCookieApi : public WrappedServerCookieIface {
     public:
-        void notify_io_complete(gsl::not_null<const void*> cookie,
-                                ENGINE_ERROR_CODE status) override {
+        void scheduleDcpStep(gsl::not_null<const void*> cookie) override {
             auto* notify_ptr = reinterpret_cast<size_t*>(
                     wrapped->get_engine_specific(cookie));
             (*notify_ptr)++;
@@ -1661,8 +1663,7 @@ TEST_P(ConnectionTest, test_mb20716_connmap_notify_on_delete_consumer) {
 
     class MockServerCookieApi : public WrappedServerCookieIface {
     public:
-        void notify_io_complete(gsl::not_null<const void*> cookie,
-                                ENGINE_ERROR_CODE status) override {
+        void scheduleDcpStep(gsl::not_null<const void*> cookie) override {
             auto* notify_ptr = reinterpret_cast<size_t*>(
                     get_mock_server_api()->cookie->get_engine_specific(cookie));
             (*notify_ptr)++;

include/memcached/server_cookie_iface.h

@@ -156,6 +156,15 @@ struct ServerCookieIface {
     virtual void notify_io_complete(gsl::not_null<const void*> cookie,
                                     ENGINE_ERROR_CODE status) = 0;
 
+    /**
+     * Request the core to schedule a new call to dcp_step() as soon as
+     * possible as the underlying engine has data to send.
+     *
+     * @param cookie cookie representing the connection (MUST be a DCP
+     *               connection)
+     */
+    virtual void scheduleDcpStep(gsl::not_null<const void*> cookie) = 0;
+
     /**
      * Notify the core that we're holding on to this cookie for
      * future use. (The core guarantees it will not invalidate the

programs/engine_testapp/mock_server.cc

@@ -423,6 +423,10 @@ struct MockServerCookieApi : public ServerCookieIface {
         c->num_io_notifications++;
         c->cond.notify_all();
     }
+
+    void scheduleDcpStep(gsl::not_null<const void*> cookie) override {
+        notify_io_complete(cookie, ENGINE_SUCCESS);
+    }
 };
 
 ServerApi* get_mock_server_api() {