Create DCP consumer / producer without holding the lock

Previously we grabbed a write lock to the underlying map and
checked if we already had the cookie registered (which would
_ALWAYS_ fail as the core won't allow DCP open on a connection
which is already marked as DCP). After that it would check
if we had a connection with the same logical name before we
would create the object and insert it into the map.

We only need the lock while we check for the name and insert
the new stream.

Remove the code to check for an existing consumer/producer
for the cookie and the unit tests for that (it is tested
via DcpTest::CantDcpOpenTwice that the second DCP open
won't reach the underlying engine and is refused from
the packet validators)

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

Author: trondn

engines/ep/src/dcp/dcpconnmap.cc

@@ -75,37 +75,26 @@ DcpConsumer* DcpConnMap::newConsumer(const void* cookie,
     std::string conn_name("eq_dcpq:");
     conn_name.append(name);
 
-    auto handle = connStore->getCookieToConnectionMapHandle();
+    auto consumer = makeConsumer(engine, cookie, conn_name, consumerName);
+    EP_LOG_DEBUG("{} Connection created", consumer->logHeader());
+    auto* rawPtr = consumer.get();
 
-    const auto& connForCookie = handle->findConnHandlerByCookie(cookie);
-    if (connForCookie) {
-        connForCookie->setDisconnect();
-        EP_LOG_INFO(
-                "Failed to create Dcp Consumer because connection "
-                "({}) already exists.",
-                cookie);
-        return nullptr;
-    }
+    // Get a write-handle!
+    auto handle = connStore->getCookieToConnectionMapHandle();
 
-    /*
-     *  If we request a connection of the same name then
-     *  mark the existing connection as "want to disconnect".
-     */
+    //  If we request a connection of the same name then
+    //  mark the existing connection as "want to disconnect".
     const auto& connForName = handle->findConnHandlerByName(conn_name);
     if (connForName) {
         EP_LOG_INFO(
                 "{} Disconnecting existing Dcp Consumer {} as it has the "
-                "same "
-                "name as a new connection {}",
+                "same name as a new connection {}",
                 connForName->logHeader(),
                 connForName->getCookie(),
                 cookie);
         connForName->setDisconnect();
     }
 
-    auto consumer = makeConsumer(engine, cookie, conn_name, consumerName);
-    EP_LOG_DEBUG("{} Connection created", consumer->logHeader());
-    auto* rawPtr = consumer.get();
     handle->addConnByCookie(cookie, std::move(consumer));
     return rawPtr;
 }
@@ -160,16 +149,13 @@ DcpProducer* DcpConnMap::newProducer(const void* cookie,
     std::string conn_name("eq_dcpq:");
     conn_name.append(name);
 
+    auto producer = std::make_shared<DcpProducer>(
+            engine, cookie, conn_name, flags, true /*startTask*/);
+    EP_LOG_DEBUG("{} Connection created", producer->logHeader());
+    auto* result = producer.get();
+
+    // Get a write-handle!
     auto handle = connStore->getCookieToConnectionMapHandle();
-    const auto& connForCookie = handle->findConnHandlerByCookie(cookie);
-    if (connForCookie) {
-        connForCookie->flagDisconnect();
-        EP_LOG_INFO(
-                "Failed to create Dcp Producer because connection "
-                "({}) already exists.",
-                cookie);
-        return nullptr;
-    }
 
     // If we request a connection of the same name then mark the
     // existing connection as "want to disconnect" and pull it out from
@@ -182,8 +168,7 @@ DcpProducer* DcpConnMap::newProducer(const void* cookie,
     if (connForName) {
         EP_LOG_INFO(
                 "{} Disconnecting existing Dcp Producer {} as it has the "
-                "same "
-                "name as a new connection {}",
+                "same name as a new connection {}",
                 connForName->logHeader(),
                 connForName->getCookie(),
                 cookie);
@@ -197,12 +182,7 @@ DcpProducer* DcpConnMap::newProducer(const void* cookie,
         // here will lead to issues described MB-36451.
     }
 
-    auto producer = std::make_shared<DcpProducer>(
-            engine, cookie, conn_name, flags, true /*startTask*/);
-    EP_LOG_DEBUG("{} Connection created", producer->logHeader());
-    auto* result = producer.get();
     handle->addConnByCookie(cookie, producer);
-
     return result;
 }
 

engines/ep/tests/ep_testsuite_dcp.cc

@@ -1749,61 +1749,6 @@ static enum test_result test_dcp_producer_open(EngineIface* h) {
     return SUCCESS;
 }
 
-static enum test_result test_dcp_producer_open_same_cookie(EngineIface* h) {
-    auto* cookie = testHarness->create_cookie(h);
-    const std::string name("unittest");
-    uint32_t opaque = 0;
-    const uint32_t seqno = 0;
-    auto dcp = requireDcpIface(h);
-
-    checkeq(ENGINE_SUCCESS,
-            dcp->open(cookie,
-                      opaque,
-                      seqno,
-                      cb::mcbp::request::DcpOpenPayload::Producer,
-                      name,
-                      R"({"consumer_name":"replica1"})"),
-            "Failed dcp producer open connection.");
-
-    const auto stat_type("eq_dcpq:" + name + ":type");
-    auto type = get_str_stat(h, stat_type.c_str(), "dcp");
-    checkeq(0, type.compare("producer"), "Producer not found");
-    /*
-     * Number of references is 2 (as opposed to 1) because a
-     * mock_connstuct is initialised to having 1 reference
-     * to represent a client being connected to it.
-     */
-    checkeq(2,
-            testHarness->get_number_of_mock_cookie_references(cookie),
-            "Number of cookie references is not two");
-    /*
-     * engine_data needs to be reset so that it passes the check that
-     * a connection does not already exist on the same socket.
-     */
-    testHarness->store_engine_specific(cookie, nullptr);
-
-    checkeq(ENGINE_DISCONNECT,
-            dcp->open(cookie,
-                      opaque++,
-                      seqno,
-                      cb::mcbp::request::DcpOpenPayload::Producer,
-                      name,
-                      R"({"consumer_name":"replica1"})"),
-            "Failed to return ENGINE_DISCONNECT");
-
-    checkeq(2,
-            testHarness->get_number_of_mock_cookie_references(cookie),
-            "Number of cookie references is not two");
-
-    testHarness->destroy_cookie(cookie);
-
-    checkeq(1,
-            testHarness->get_number_of_mock_cookie_references(cookie),
-            "Number of cookie references is not one");
-
-    return SUCCESS;
-}
-
 static enum test_result test_dcp_noop(EngineIface* h) {
     auto* cookie = testHarness->create_cookie(h);
     const std::string name("unittest");
@@ -8246,8 +8191,6 @@ BaseTestCase testsuite_testcases[] = {
                  cleanup),
         TestCase("test open producer", test_dcp_producer_open,
                  test_setup, teardown, nullptr, prepare, cleanup),
-        TestCase("test open producer same cookie", test_dcp_producer_open_same_cookie,
-                 test_setup, teardown, nullptr, prepare, cleanup),
         TestCase("test dcp noop", test_dcp_noop, test_setup, teardown, nullptr,
                  prepare, cleanup),
         TestCase("test dcp noop failure", test_dcp_noop_fail, test_setup,

engines/ep/tests/module_tests/dcp_test.cc

@@ -1340,35 +1340,6 @@ TEST_P(ConnectionTest, test_mb17042_duplicate_name_consumer_connections) {
         << "Dead connections still remain";
 }
 
-TEST_P(ConnectionTest, test_mb17042_duplicate_cookie_producer_connections) {
-    MockDcpConnMap connMap(*engine);
-    connMap.initialize();
-    auto* cookie = create_mock_cookie(engine);
-    // Create a new Dcp producer
-    DcpProducer* producer = connMap.newProducer(cookie,
-                                                "test_producer1",
-                                                /*flags*/ 0);
-
-    // Create a duplicate Dcp producer
-    DcpProducer* duplicateproducer = connMap.newProducer(cookie,
-                                                         "test_producer2",
-                                                         /*flags*/ 0);
-
-    EXPECT_TRUE(producer->doDisconnect()) << "producer doDisconnect == false";
-    EXPECT_EQ(nullptr, duplicateproducer) << "duplicateproducer is not null";
-
-    // Disconnect the producer connection
-    connMap.disconnect(cookie);
-    // Cleanup the deadConnections
-    connMap.manageConnections();
-    // Should be zero deadConnections
-    EXPECT_EQ(0, connMap.getNumberOfDeadConnections())
-        << "Dead connections still remain";
-}
-
-
-
-
 TEST_P(ConnectionTest, test_producer_unknown_ctrl_msg) {
     auto* cookie = create_mock_cookie(engine);
     /* Create a new Dcp producer */
@@ -1386,28 +1357,6 @@ TEST_P(ConnectionTest, test_producer_unknown_ctrl_msg) {
     destroy_mock_cookie(cookie);
 }
 
-TEST_P(ConnectionTest, test_mb17042_duplicate_cookie_consumer_connections) {
-    MockDcpConnMap connMap(*engine);
-    connMap.initialize();
-    auto* cookie = create_mock_cookie(engine);
-    // Create a new Dcp consumer
-    DcpConsumer* consumer = connMap.newConsumer(cookie, "test_consumer1");
-
-    // Create a duplicate Dcp consumer
-    DcpConsumer* duplicateconsumer =
-            connMap.newConsumer(cookie, "test_consumer2");
-    EXPECT_TRUE(consumer->doDisconnect()) << "consumer doDisconnect == false";
-    EXPECT_EQ(nullptr, duplicateconsumer) << "duplicateconsumer is not null";
-
-    // Disconnect the consumer connection
-    connMap.disconnect(cookie);
-    // Cleanup the deadConnections
-    connMap.manageConnections();
-    // Should be zero deadConnections
-    EXPECT_EQ(0, connMap.getNumberOfDeadConnections())
-        << "Dead connections still remain";
-}
-
 TEST_P(ConnectionTest, test_update_of_last_message_time_in_consumer) {
     auto* cookie = create_mock_cookie(engine);
     Vbid vbid(0);
@@ -1774,12 +1723,12 @@ class DcpConnMapTest : public ::testing::Test {
      * MB-36915: With a recent change, we unconditionally acquire an exclusive
      * lock to vbstate in KVBucket::setVBucketState. But, the new lock
      * introduces a potential deadlock by lock-inversion on connLock and
-     * vbstateLock in EPE::dcpOpen if a connection with the same cookie or name
+     * vbstateLock in EPE::dcpOpen if a connection with the same name
      * already exists in conn-map. TSAN easily spots the issue as soon as we
      * have an execution where two threads run in parallel and execute the code
      * responsible for the potential deadlock, which is what this test achieves.
      */
-    void testLockInversionInSetVBucketStateAndNewProducer(ConnExistsBy by);
+    void testLockInversionInSetVBucketStateAndNewProducer();
 
     SynchronousEPEngineUniquePtr engine;
     const Vbid vbid = Vbid(0);
@@ -2227,8 +2176,7 @@ TEST_F(DcpConnMapTest, ConnAggStats) {
     destroy_mock_cookie(statsCookie);
 }
 
-void DcpConnMapTest::testLockInversionInSetVBucketStateAndNewProducer(
-        ConnExistsBy by) {
+void DcpConnMapTest::testLockInversionInSetVBucketStateAndNewProducer() {
     auto& connMap = dynamic_cast<MockDcpConnMap&>(engine->getDcpConnMap());
     auto* cookie = create_mock_cookie(engine.get());
     const std::string connName = "producer";
@@ -2277,28 +2225,15 @@ void DcpConnMapTest::testLockInversionInSetVBucketStateAndNewProducer(
     // Note: ActiveStream::setDead executed only if re-creating the same
     // connection (ie, same cookie or connection name).
     auto* cookie2 = create_mock_cookie(engine.get());
-    switch (by) {
-    case ConnExistsBy::Cookie: {
-        const std::string newConnName = "newName";
-        EXPECT_FALSE(connMap.newProducer(cookie, newConnName, flags));
-        // Note: Connection flagged as disconnected but not yet removed
-        // from conn-map. See comments in ConnMap::newProducer.
-        EXPECT_TRUE(connMap.doesVbConnExist(vbid, "eq_dcpq:" + connName));
-        break;
-    }
-    case ConnExistsBy ::Name: {
-        producer = connMap.newProducer(cookie2, connName, flags);
-        ASSERT_TRUE(producer);
-        // Check that the connection has been re-created with the same name
-        // and exists in vbConns at stream-req
-        {
-            SCOPED_TRACE("");
-            streamRequest(*producer);
-        }
-        EXPECT_TRUE(connMap.doesVbConnExist(vbid, "eq_dcpq:" + connName));
-        break;
-    }
+    producer = connMap.newProducer(cookie2, connName, flags);
+    ASSERT_TRUE(producer);
+    // Check that the connection has been re-created with the same name
+    // and exists in vbConns at stream-req
+    {
+        SCOPED_TRACE("");
+        streamRequest(*producer);
     }
+    EXPECT_TRUE(connMap.doesVbConnExist(vbid, "eq_dcpq:" + connName));
 
     t1.join();
 
@@ -2308,14 +2243,9 @@ void DcpConnMapTest::testLockInversionInSetVBucketStateAndNewProducer(
     destroy_mock_cookie(cookie2);
 }
 
-TEST_F(DcpConnMapTest,
-       AvoidLockInversionInSetVBucketStateAndNewProducerExistingCookie) {
-    testLockInversionInSetVBucketStateAndNewProducer(ConnExistsBy::Cookie);
-}
-
 TEST_F(DcpConnMapTest,
        AvoidLockInversionInSetVBucketStateAndNewProducerExistingName) {
-    testLockInversionInSetVBucketStateAndNewProducer(ConnExistsBy::Name);
+    testLockInversionInSetVBucketStateAndNewProducer();
 }
 
 class NotifyTest : public DCPTest {