[Refactor] Store the DCP conn handler in its own variable

ep-engine used to store the "conn handler" object which represents
the DCP object in the "engine-specific" part of the cookie, but
that field is also used by other commands for different things
(a sync-write would use it to keep the cas; compaction use it
to store a setting that it is running etc). If any of these
commands would be received on a DCP connection they would
clear the engine-specific part in the cookie when they
complete and none of the methods in the DcpIface would work.

This patch change that logic to store the conn handler in
the Connection object.

Change-Id: I9850b1b30881868a9ba1731cea387f1014368fb9
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/140956
Tested-by: Build Bot <[email protected]>
Reviewed-by: Paolo Cocchi <[email protected]>

Author: trondn

auditd/tests/testauditd.cc

@@ -47,6 +47,14 @@ static bool ready = false;
 
 class AuditMockServerCookieApi : public ServerCookieIface {
 public:
+    void setDcpConnHandler(gsl::not_null<const void*> cookie,
+                           DcpConnHandlerIface* handler) override {
+        throw std::runtime_error("Not implemented");
+    }
+    DcpConnHandlerIface* getDcpConnHandler(
+            gsl::not_null<const void*> cookie) override {
+        throw std::runtime_error("Not implemented");
+    }
     void store_engine_specific(gsl::not_null<const void*> cookie,
                                void* engine_data) override {
         throw std::runtime_error("Not implemented");

daemon/connection.h

@@ -553,6 +553,14 @@ class Connection : public DcpMessageProducersIface {
      */
     bool processServerEvents();
 
+    DcpConnHandlerIface* getDcpConnHandlerIface() const {
+        return dcpConnHandlerIface.load(std::memory_order_acquire);
+    }
+
+    void setDcpConnHandlerIface(DcpConnHandlerIface* handler) {
+        dcpConnHandlerIface.store(handler, std::memory_order_release);
+    }
+
     /**
      * Set the name of the connected agent
      */
@@ -869,6 +877,9 @@ class Connection : public DcpMessageProducersIface {
     /** Name of the local socket if known */
     const std::string sockname;
 
+    /// The stored DCP Connection Interface
+    std::atomic<DcpConnHandlerIface*> dcpConnHandlerIface{nullptr};
+
     /**
      * The connections' priority.
      * atomic to allow read (from DCP stats) without acquiring any

daemon/server_api.cc

@@ -118,6 +118,16 @@ struct ServerDocumentApi : public ServerDocumentIface {
 };
 
 struct ServerCookieApi : public ServerCookieIface {
+    void setDcpConnHandler(gsl::not_null<const void*> cookie,
+                           DcpConnHandlerIface* handler) override {
+        getCookie(cookie).getConnection().setDcpConnHandlerIface(handler);
+    }
+
+    DcpConnHandlerIface* getDcpConnHandler(
+            gsl::not_null<const void*> cookie) override {
+        return getCookie(cookie).getConnection().getDcpConnHandlerIface();
+    }
+
     void store_engine_specific(gsl::not_null<const void*> void_cookie,
                                void* engine_data) override {
         const auto* cc = reinterpret_cast<const Cookie*>(void_cookie.get());

engines/ep/src/connhandler.h

@@ -75,7 +75,7 @@ struct ConnCounter {
     size_t      conn_queueItemOnDisk;
 };
 
-class ConnHandler {
+class ConnHandler : public DcpConnHandlerIface {
 public:
     /// The maximum length of a DCP stat name
     static constexpr size_t MaxDcpStatNameLength = 47;
@@ -92,7 +92,7 @@ class ConnHandler {
                 const void* c,
                 std::string name);
 
-    virtual ~ConnHandler();
+    ~ConnHandler() override;
 
     virtual ENGINE_ERROR_CODE addStream(uint32_t opaque,
                                         Vbid vbucket,

engines/ep/src/ep_engine.cc

@@ -1986,6 +1986,18 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::releaseCookie(const void *cookie)
     return serverApi->cookie->release(cookie);
 }
 
+void EventuallyPersistentEngine::setDcpConnHandler(
+        const void* cookie, DcpConnHandlerIface* handler) {
+    NonBucketAllocationGuard guard;
+    serverApi->cookie->setDcpConnHandler(cookie, handler);
+}
+
+DcpConnHandlerIface* EventuallyPersistentEngine::getDcpConnHandler(
+        const void* cookie) {
+    NonBucketAllocationGuard guard;
+    return serverApi->cookie->getDcpConnHandler(cookie);
+}
+
 void EventuallyPersistentEngine::storeEngineSpecific(const void* cookie,
                                                      void* engine_data) {
     NonBucketAllocationGuard guard;
@@ -6139,14 +6151,14 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::dcpOpen(
     (void) seqno;
     std::string connName{stream_name};
 
-    if (getEngineSpecific(cookie) != nullptr) {
+    auto* handler = getConnHandler(cookie);
+    if (handler) {
         EP_LOG_WARN(
-                "Cannot open DCP connection as another"
-                " connection exists on the same socket");
+                "Cannot open DCP connection as another connection exists on "
+                "the same socket");
         return ENGINE_DISCONNECT;
     }
 
-    ConnHandler *handler = nullptr;
     if (flags & (cb::mcbp::request::DcpOpenPayload::Producer |
                  cb::mcbp::request::DcpOpenPayload::Notifier)) {
         if (flags & cb::mcbp::request::DcpOpenPayload::PiTR) {
@@ -6202,8 +6214,7 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::dcpOpen(
                 "cannot be reserved");
         return ENGINE_DISCONNECT;
     }
-
-    storeEngineSpecific(cookie, handler);
+    setDcpConnHandler(cookie, handler);
 
     return ENGINE_SUCCESS;
 }
@@ -6222,15 +6233,22 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::dcpAddStream(const void* cookie,
 
 ConnHandler* EventuallyPersistentEngine::getConnHandler(const void* cookie,
                                                         bool logNonExistent) {
-    void* specific = getEngineSpecific(cookie);
-    auto* handler = reinterpret_cast<ConnHandler*>(specific);
-    if (!handler && logNonExistent) {
+    auto* iface = getDcpConnHandler(cookie);
+    if (iface) {
+        auto* handler = dynamic_cast<ConnHandler*>(iface);
+        if (handler) {
+            return handler;
+        }
+    }
+
+    if (logNonExistent) {
         auto li = serverApi->cookie->get_log_info(cookie);
         EP_LOG_WARN("{}: Invalid streaming connection: cookie:{}",
                     li.first,
                     cb::to_hex(uint64_t(cookie)));
     }
-    return handler;
+
+    return nullptr;
 }
 
 void EventuallyPersistentEngine::handleDisconnect(const void *cookie) {

engines/ep/src/ep_engine.h

@@ -533,6 +533,9 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
     ENGINE_ERROR_CODE reserveCookie(const void *cookie);
     ENGINE_ERROR_CODE releaseCookie(const void *cookie);
 
+    void setDcpConnHandler(const void* cookie, DcpConnHandlerIface* handler);
+    DcpConnHandlerIface* getDcpConnHandler(const void* cookie);
+
     void storeEngineSpecific(const void* cookie, void* engine_data);
 
     void* getEngineSpecific(const void* cookie);

engines/ep/tests/module_tests/dcp_test.cc

@@ -80,7 +80,14 @@ class WrappedServerCookieIface : public ServerCookieIface {
     ~WrappedServerCookieIface() override {
         get_mock_server_api()->cookie = wrapped;
     }
-
+    void setDcpConnHandler(gsl::not_null<const void*> cookie,
+                           DcpConnHandlerIface* handler) override {
+        wrapped->setDcpConnHandler(cookie, handler);
+    }
+    DcpConnHandlerIface* getDcpConnHandler(
+            gsl::not_null<const void*> cookie) override {
+        return wrapped->getDcpConnHandler(cookie);
+    }
     void store_engine_specific(gsl::not_null<const void*> cookie,
                                void* engine_data) override {
         wrapped->store_engine_specific(cookie, engine_data);

include/memcached/dcp.h

@@ -41,6 +41,11 @@ enum class id : uint32_t;
 enum class version : uint8_t;
 } // namespace mcbp::systemevent
 
+class DcpConnHandlerIface {
+public:
+    virtual ~DcpConnHandlerIface() = default;
+};
+
 /**
  * The message producers are used by the engine's DCP producer
  * to add messages into the DCP stream.  Please look at the full

include/memcached/server_cookie_iface.h

@@ -31,12 +31,35 @@ namespace cb::mcbp {
 class Request;
 } // namespace cb::mcbp
 
+class DcpConnHandlerIface;
+
 /**
  * Commands to operate on a specific cookie.
  */
 struct ServerCookieIface {
     virtual ~ServerCookieIface() = default;
 
+    /**
+     * Set the DCP connection handler to be used for the connection the
+     * provided cookie belongs to.
+     *
+     * @param cookie The cookie provided by the core for the operation
+     * @param handler The new handler (may be nullptr to clear the handler)
+     */
+    virtual void setDcpConnHandler(gsl::not_null<const void*> cookie,
+                                   DcpConnHandlerIface* handler) = 0;
+
+    /**
+     * Get the DCP connection handler for the connection the provided
+     * cookie belongs to
+     *
+     * @param cookie The cookie provided by the core for the operation
+     * @return The handler stored for the connection (may be nullptr if
+     *         none is specified)
+     */
+    virtual DcpConnHandlerIface* getDcpConnHandler(
+            gsl::not_null<const void*> cookie) = 0;
+
     /**
      * Store engine-specific session data on the given cookie.
      *

programs/engine_testapp/mock_cookie.h

@@ -26,6 +26,8 @@
 #include <mutex>
 #include <string>
 
+class DcpConnHandlerIface;
+
 struct MockCookie : cb::tracing::Traceable {
     /**
      * Create a new cookie which isn't bound to an engine. This cookie won't
@@ -59,6 +61,7 @@ struct MockCookie : cb::tracing::Traceable {
     uint64_t num_processed_notifications{};
     std::string authenticatedUser{"nobody"};
     in_port_t parent_port{666};
+    DcpConnHandlerIface* connHandlerIface = nullptr;
 
     void validate() const;