MB-24817: During takeover, hold stream lock until vb is set to dead

During DCP takeover, when we decide to set the vbucket to dead state
on the producer side, we also transition the stream state from
STREAM_TAKEOVER_WAIT to STREAM_TAKEOVER_SEND state to send out any
remaining items that were received before the vbucket was set to
dead state.

The stream lock must be held until the vbucket is set to the dead
state, so that we do not prematurely finish sending out the last
items when the vbucket is not dead yet (that is if the vbucket
is not dead, it could get some more items).

This commit addresses the issue and also handles the ordering of
locks involved at stream level, ep store level and vbucket level
in the scenario.

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

Author: manudhundi

src/dcp/stream.cc

@@ -422,42 +422,68 @@ void ActiveStream::snapshotMarkerAckReceived() {
 }
 
 void ActiveStream::setVBucketStateAckRecieved() {
-    LockHolder lh(streamMutex);
-    if (state_ == STREAM_TAKEOVER_WAIT) {
-        if (takeoverState == vbucket_state_pending) {
-            producer->getLogger().log(EXTENSION_LOG_INFO,
-                "(vb %" PRIu16 ") Receive ack for set vbucket state to pending "
-                "message", vb_);
-
-            takeoverState = vbucket_state_active;
-            transitionState(STREAM_TAKEOVER_SEND);
-            lh.unlock();
+    RCPtr<VBucket> vbucket = engine->getVBucket(vb_);
+    if (!vbucket) {
+        producer->getLogger().log(EXTENSION_LOG_WARNING,
+                                  "(vb %" PRIu16
+                                  ") not present during ack for set vbucket "
+                                  "during takeover",
+                                  vb_);
+        return;
+    }
 
-            engine->getEpStore()->setVBucketState(vb_, vbucket_state_dead,
-                                                  false, false);
-            RCPtr<VBucket> vbucket = engine->getVBucket(vb_);
-            producer->getLogger().log(EXTENSION_LOG_NOTICE,
-                "(vb %" PRIu16 ") Vbucket marked as dead, last sent seqno: %"
-                PRIu64 ", high seqno: %" PRIu64,
-                vb_, lastSentSeqno.load(), vbucket->getHighSeqno());
+    {
+        /* Order in which the below 3 locks are acquired is important to avoid
+           any potential lock inversion problems */
+        LockHolder epVbSetLh(engine->getEpStore()->getVbSetMutexLock());
+        WriterLockHolder vbStateLh(vbucket->getStateLock());
+        LockHolder lh(streamMutex);
+        if (state_ == STREAM_TAKEOVER_WAIT) {
+            if (takeoverState == vbucket_state_pending) {
+                producer->getLogger().log(EXTENSION_LOG_INFO,
+                        "(vb %" PRIu16 ") Receive ack for set vbucket state to "
+                        "pending message",
+                        vb_);
+
+                takeoverState = vbucket_state_active;
+                transitionState(STREAM_TAKEOVER_SEND);
+
+                engine->getEpStore()->setVBucketState_UNLOCKED(
+                                                        vb_,
+                                                        vbucket_state_dead,
+                                                        false /* transfer */,
+                                                        false /* notify_dcp */,
+                                                        epVbSetLh,
+                                                        &vbStateLh);
+
+                producer->getLogger().log(EXTENSION_LOG_NOTICE,
+                        "(vb %" PRIu16 ") Vbucket marked as dead, last sent "
+                        "seqno: %" PRIu64 ", high seqno: %" PRIu64,
+                        vb_,
+                        lastSentSeqno.load(),
+                        vbucket->getHighSeqno());
+            } else {
+                producer->getLogger().log(EXTENSION_LOG_NOTICE,
+                        "(vb %" PRIu16 ") Receive ack for set vbucket state to "
+                        "active message",
+                        vb_);
+                endStream(END_STREAM_OK);
+            }
         } else {
-            producer->getLogger().log(EXTENSION_LOG_INFO,
-                "(vb %" PRIu16 ") Receive ack for set vbucket state to active "
-                "message", vb_);
-            endStream(END_STREAM_OK);
-            lh.unlock();
-        }
-
-        bool inverse = false;
-        if (itemsReady.compare_exchange_strong(inverse, true)) {
-            producer->notifyStreamReady(vb_);
+            producer->getLogger().log(EXTENSION_LOG_WARNING,
+                    "(vb %" PRIu16 ") Unexpected ack for set vbucket op on "
+                    "stream '%s' state '%s'",
+                    vb_,
+                    name_.c_str(),
+                    stateName(state_));
+            return;
         }
-    } else {
-        producer->getLogger().log(EXTENSION_LOG_WARNING,
-            "(vb %" PRIu16 ") Unexpected ack for set vbucket op on stream '%s' "
-            "state '%s'", vb_, name_.c_str(), stateName(state_));
     }
 
+    bool inverse = false;
+    if (itemsReady.compare_exchange_strong(inverse, true)) {
+        producer->notifyStreamReady(vb_);
+    }
 }
 
 DcpResponse* ActiveStream::backfillPhase() {

src/ep.cc

@@ -1115,11 +1115,13 @@ ENGINE_ERROR_CODE EventuallyPersistentStore::setVBucketState(uint16_t vbid,
     return setVBucketState_UNLOCKED(vbid, to, transfer, notify_dcp, lh);
 }
 
-ENGINE_ERROR_CODE EventuallyPersistentStore::setVBucketState_UNLOCKED(uint16_t vbid,
-                                                           vbucket_state_t to,
-                                                           bool transfer,
-                                                           bool notify_dcp,
-                                                           LockHolder& vbset) {
+ENGINE_ERROR_CODE EventuallyPersistentStore::setVBucketState_UNLOCKED(
+                                            uint16_t vbid,
+                                            vbucket_state_t to,
+                                            bool transfer,
+                                            bool notify_dcp,
+                                            LockHolder& vbset,
+                                            WriterLockHolder* vbStateLock) {
     RCPtr<VBucket> vb = vbMap.getBucket(vbid);
     if (vb && to == vb->getState()) {
         return ENGINE_SUCCESS;
@@ -1128,7 +1130,11 @@ ENGINE_ERROR_CODE EventuallyPersistentStore::setVBucketState_UNLOCKED(uint16_t v
     if (vb) {
         vbucket_state_t oldstate = vb->getState();
 
-        vb->setState(to);
+        if (vbStateLock) {
+            vb->setState_UNLOCKED(to, *vbStateLock);
+        } else {
+            vb->setState(to);
+        }
 
         if (oldstate != to && notify_dcp) {
             bool closeInboundStreams = false;

src/ep.h

@@ -515,11 +515,53 @@ class EventuallyPersistentStore {
         return vbMap.getPersistenceSeqno(vb);
     }
 
-    /* transfer should be set to true *only* if this vbucket is becoming master
-     * as the result of the previous master cleanly handing off control. */
+    /**
+     * Sets the vbucket or creates a vbucket with the desired state
+     *
+     * @param vbid vbucket id
+     * @param state desired state of the vbucket
+     * @param transfer indicates that the vbucket is transferred to the active
+     *                 post a failover and/or rebalance
+     * @param notify_dcp indicates whether we must consider closing DCP streams
+     *                    associated with the vbucket
+     *
+     * return status of the operation
+     */
     ENGINE_ERROR_CODE setVBucketState(uint16_t vbid, vbucket_state_t state,
                                       bool transfer, bool notify_dcp = true);
 
+    /**
+     * Sets the vbucket or creates a vbucket with the desired state
+     *
+     * @param vbid vbucket id
+     * @param state desired state of the vbucket
+     * @param transfer indicates that the vbucket is transferred to the active
+     *                 post a failover and/or rebalance
+     * @param notify_dcp indicates whether we must consider closing DCP streams
+     *                    associated with the vbucket
+     * @param vbset LockHolder acquiring the 'vbsetMutex' lock in the
+     *              EventuallyPersistentStore class
+     * @param vbStateLock ptr to WriterLockHolder of 'stateLock' in the vbucket
+     *                    class. if passed as null, the function acquires the
+     *                    vbucket 'stateLock'
+     *
+     * return status of the operation
+     */
+    ENGINE_ERROR_CODE setVBucketState_UNLOCKED(
+                                    uint16_t vbid,
+                                    vbucket_state_t state,
+                                    bool transfer,
+                                    bool notify_dcp,
+                                    LockHolder& vbset,
+                                    WriterLockHolder* vbStateLock = nullptr);
+
+    /**
+     * Returns the 'vbsetMutex'
+     */
+    Mutex& getVbSetMutexLock() {
+        return vbsetMutex;
+    }
+
     /**
      * Physically deletes a VBucket from disk. This function should only
      * be called on a VBucket that has already been logically deleted.
@@ -1022,10 +1064,6 @@ class EventuallyPersistentStore {
 
     bool resetVBucket_UNLOCKED(uint16_t vbid, LockHolder& vbset);
 
-    ENGINE_ERROR_CODE setVBucketState_UNLOCKED(uint16_t vbid, vbucket_state_t state,
-                                               bool transfer, bool notify_dcp,
-                                               LockHolder& vbset);
-
     friend class Warmup;
     friend class Flusher;
     friend class BGFetchCallback;

src/vbucket.cc

@@ -185,22 +185,26 @@ void VBucket::fireAllOps(EventuallyPersistentEngine &engine) {
 }
 
 void VBucket::setState(vbucket_state_t to) {
-    vbucket_state_t oldstate;
-    {
-        WriterLockHolder wlh(stateLock);
-        oldstate = state;
-
-        if (to == vbucket_state_active &&
-            checkpointManager.getOpenCheckpointId() < 2) {
-            checkpointManager.setOpenCheckpointId(2);
-        }
+    WriterLockHolder wlh(stateLock);
+    setState_UNLOCKED(to, wlh);
+}
 
-        LOG(EXTENSION_LOG_NOTICE,
-            "VBucket::setState: transitioning vbucket:%" PRIu16 " from:%s to:%s",
-            id, VBucket::toString(oldstate), VBucket::toString(to));
+void VBucket::setState_UNLOCKED(vbucket_state_t to,
+                                WriterLockHolder &vbStateLock) {
+    vbucket_state_t oldstate = state;
 
-        state = to;
+    if (to == vbucket_state_active &&
+        checkpointManager.getOpenCheckpointId() < 2) {
+        checkpointManager.setOpenCheckpointId(2);
     }
+
+    LOG(EXTENSION_LOG_NOTICE,
+        "VBucket::setState: transitioning vbucket:%" PRIu16 " from:%s to:%s",
+        id,
+        VBucket::toString(oldstate),
+        VBucket::toString(to));
+    
+    state = to;
 }
 
 vbucket_state VBucket::getVBucketState() const {

src/vbucket.h

@@ -278,7 +278,23 @@ class VBucket : public RCValue {
 
     id_type getId() const { return id; }
     vbucket_state_t getState(void) const { return state.load(); }
+
+    /**
+     * Sets the vbucket state to a desired state
+     *
+     * @param to desired vbucket state
+     */
     void setState(vbucket_state_t to);
+
+    /**
+     * Sets the vbucket state to a desired state with the 'stateLock' already
+     * acquired
+     *
+     * @param to desired vbucket state
+     * @param vbStateLock write lock holder on 'stateLock'
+     */
+    void setState_UNLOCKED(vbucket_state_t to, WriterLockHolder& vbStateLock);
+
     RWLock& getStateLock() {return stateLock;}
 
     vbucket_state_t getInitialState(void) { return initialState; }