MB-33873: Revert "MB-33739: Use reference in ConnMap (and DcpConnMap) for vbConns"

This reverts commit 9c9825d.

Change-Id: Iaf6f5aef53dfc56178641c11e5acc2671d4a35c3
Reviewed-on: http://review.couchbase.org/108152
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/src/connmap.cc

@@ -82,11 +82,16 @@ class ConnManager : public GlobalTask {
     size_t snoozeTime;
 };
 
-ConnMap::ConnMap(EventuallyPersistentEngine& theEngine)
-    : vbConnLocks(vbConnLockNum),
-      vbConns(theEngine.getConfiguration().getMaxVbuckets()),
-      engine(theEngine),
-      connNotifier_(nullptr) {
+ConnMap::ConnMap(EventuallyPersistentEngine &theEngine)
+    :  vbConnLocks(vbConnLockNum),
+       engine(theEngine),
+       connNotifier_(nullptr) {
+
+    Configuration &config = engine.getConfiguration();
+    size_t max_vbs = config.getMaxVbuckets();
+    for (size_t i = 0; i < max_vbs; ++i) {
+        vbConns.push_back(std::list<std::weak_ptr<ConnHandler>>());
+    }
 }
 
 void ConnMap::initialize() {
@@ -153,17 +158,27 @@ void ConnMap::processPendingNotifications() {
     }
 }
 
-void ConnMap::addVBConnByVBId(ConnHandler& conn, Vbid vbid) {
+void ConnMap::addVBConnByVBId(std::shared_ptr<ConnHandler> conn, Vbid vbid) {
+    if (!conn.get()) {
+        return;
+    }
+
     size_t lock_num = vbid.get() % vbConnLockNum;
     std::lock_guard<std::mutex> lh(vbConnLocks[lock_num]);
-    vbConns[vbid.get()].emplace_back(conn);
+    vbConns[vbid.get()].emplace_back(std::move(conn));
 }
 
 void ConnMap::removeVBConnByVBId_UNLOCKED(const void* connCookie, Vbid vbid) {
-    auto& vb_conns = vbConns[vbid.get()];
-    vb_conns.remove_if([connCookie](ConnHandler& conn) {
-        return connCookie == conn.getCookie();
-    });
+    std::list<std::weak_ptr<ConnHandler>>& vb_conns = vbConns[vbid.get()];
+    for (auto itr = vb_conns.begin(); itr != vb_conns.end(); ++itr) {
+        auto connection = (*itr).lock();
+        // Erase if we cannot lock, or if the cookie matches
+        if (!connection ||
+            (connection && connCookie == connection->getCookie())) {
+            vb_conns.erase(itr);
+            break;
+        }
+    }
 }
 
 void ConnMap::removeVBConnByVBId(const void* connCookie, Vbid vbid) {

engines/ep/src/connmap.h

@@ -67,10 +67,10 @@ class ConnMap {
     /**
      * Adds the given connection to the set of connections associated
      * with the given vbucket.
-     * @param conn Connection to add to the set.
+     * @param conn Connection to add to the set. Refcount is retained.
      * @param vbid vBucket to add to.
      */
-    void addVBConnByVBId(ConnHandler& conn, Vbid vbid);
+    void addVBConnByVBId(std::shared_ptr<ConnHandler> conn, Vbid vbid);
 
     void removeVBConnByVBId_UNLOCKED(const void* connCookie, Vbid vbid);
 
@@ -122,7 +122,7 @@ class ConnMap {
     CookieToConnectionMap map_;
 
     std::vector<std::mutex> vbConnLocks;
-    std::vector<std::list<std::reference_wrapper<ConnHandler>>> vbConns;
+    std::vector<std::list<std::weak_ptr<ConnHandler>>> vbConns;
 
     /* Handle to the engine who owns us */
     EventuallyPersistentEngine &engine;

engines/ep/src/dcp/consumer.cc

@@ -1817,7 +1817,7 @@ void DcpConsumer::setDisconnect() {
 void DcpConsumer::registerStream(std::shared_ptr<PassiveStream> stream) {
     auto vbid = stream->getVBucket();
     streams.insert({vbid, stream});
-    engine_.getDcpConnMap().addVBConnByVBId(*this, vbid);
+    engine_.getDcpConnMap().addVBConnByVBId(shared_from_this(), vbid);
 }
 
 void DcpConsumer::removeStream(Vbid vbid) {

engines/ep/src/dcp/dcpconnmap.cc

@@ -249,10 +249,12 @@ bool DcpConnMap::handleSlowStream(Vbid vbid, const CheckpointCursor* cursor) {
     size_t lock_num = vbid.get() % vbConnLockNum;
     std::lock_guard<std::mutex> lh(vbConnLocks[lock_num]);
 
-    for (const auto& connection : vbConns[vbid.get()]) {
-        auto* producer = dynamic_cast<DcpProducer*>(&connection.get());
-
-        // Check that this connection is actually a producer
+    for (const auto& weakPtr : vbConns[vbid.get()]) {
+        auto connection = weakPtr.lock();
+        if (!connection) {
+            continue;
+        }
+        auto* producer = dynamic_cast<DcpProducer*>(connection.get());
         if (producer && producer->handleSlowStream(vbid, cursor)) {
             return true;
         }
@@ -396,20 +398,28 @@ void DcpConnMap::removeVBConnections(DcpProducer& prod) {
         size_t lock_num = vbid.get() % vbConnLockNum;
         std::lock_guard<std::mutex> lh(vbConnLocks[lock_num]);
         auto& vb_conns = vbConns[vbid.get()];
-        auto* cookie = prod.getCookie();
-        vb_conns.remove_if([cookie](ConnHandler& conn) {
-            return cookie == (conn.getCookie());
-        });
+        for (auto itr = vb_conns.begin(); itr != vb_conns.end(); ++itr) {
+            auto connection = (*itr).lock();
+            // Erase if we cannot lock, or if the cookie matches
+            if (!connection ||
+                (connection && prod.getCookie() == connection->getCookie())) {
+                vb_conns.erase(itr);
+                break;
+            }
+        }
     }
 }
 
 void DcpConnMap::notifyVBConnections(Vbid vbid, uint64_t bySeqno) {
     size_t lock_num = vbid.get() % vbConnLockNum;
     std::lock_guard<std::mutex> lh(vbConnLocks[lock_num]);
 
-    for (auto& connection : vbConns[vbid.get()]) {
-        auto* producer = dynamic_cast<DcpProducer*>(&connection.get());
-        // Check that this connection is actually a producer
+    for (auto& weakPtr : vbConns[vbid.get()]) {
+        auto connection = weakPtr.lock();
+        if (!connection) {
+            continue;
+        }
+        auto* producer = dynamic_cast<DcpProducer*>(connection.get());
         if (producer) {
             producer->notifySeqnoAvailable(vbid, bySeqno);
         }
@@ -425,9 +435,12 @@ void DcpConnMap::seqnoAckVBPassiveStream(Vbid vbid) {
     // only Producers).
     // @todo-durability: not clear yet if for Consumers we can simplify by
     //     keeping a 1-to-1 VB-to-Consumer mapping
-    for (auto& connection : vbConns[vbid.get()]) {
-        auto* consumer = dynamic_cast<DcpConsumer*>(&connection.get());
-        // Check that this connection is actually a consumer
+    for (auto& weakPtr : vbConns[vbid.get()]) {
+        auto connection = weakPtr.lock();
+        if (!connection) {
+            continue;
+        }
+        auto* consumer = dynamic_cast<DcpConsumer*>(connection.get());
         if (consumer) {
             // Note: Sync Repl enabled at Consumer only if Producer supports it.
             //     This is to prevent that 6.5 Consumers send DCP_SEQNO_ACK to

engines/ep/src/dcp/producer.cc

@@ -553,7 +553,7 @@ ENGINE_ERROR_CODE DcpProducer::streamRequest(
     notifyStreamReady(vbucket);
 
     if (add_vb_conn_map) {
-        engine_.getDcpConnMap().addVBConnByVBId(*this, vbucket);
+        engine_.getDcpConnMap().addVBConnByVBId(shared_from_this(), vbucket);
     }
 
     return rv;

engines/ep/tests/mock/mock_dcp_conn_map.h

@@ -54,11 +54,13 @@ class MockDcpConnMap : public DcpConnMap {
     /// return if the named handler exists for the vbid in the vbConns structure
     bool doesConnHandlerExist(Vbid vbid, const std::string& name) const {
         const auto& list = vbConns[vbid.get()];
-        return std::find_if(list.begin(),
-                            list.end(),
-                            [&name](const ConnHandler& c) -> bool {
-                                return c.getName() == name;
-                            }) != list.end();
+        return std::find_if(
+                       list.begin(),
+                       list.end(),
+                       [&name](const std::weak_ptr<ConnHandler>& c) -> bool {
+                           auto p = c.lock();
+                           return p && p->getName() == name;
+                       }) != list.end();
     }
 
 protected: