MB-34780: Clear the pending sync writes during bucket shutdown

During bucket deletion all of the pending sync writes needs
to be terminated for bucket deletion to complete. As part
of bucket deletion we tear down all of the DCP streams so
there is no way for the ack/nack's from the replicas to
arrive and unblock the cookie and continue bucket deletion.

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

Author: trondn

daemon/memcached.cc

@@ -1967,6 +1967,7 @@ void DestroyBucketThread::destroy() {
     LOG_INFO("{} Delete bucket [{}]. Notifying engine", connection_id, name);
 
     all_buckets[idx].getEngine()->initiate_shutdown();
+    all_buckets[idx].getEngine()->cancel_all_operations_in_ewb_state();
 
     LOG_INFO("{} Delete bucket [{}]. Engine ready for shutdown",
              connection_id,
@@ -2030,6 +2031,7 @@ void DestroyBucketThread::destroy() {
                         connection.signalIfIdle();
                     }
                 });
+                bucket.getEngine()->cancel_all_operations_in_ewb_state();
                 guard.lock();
                 continue;
             }

engines/ep/src/durability/active_durability_monitor.cc

@@ -972,10 +972,11 @@ std::vector<const void*>
 ActiveDurabilityMonitor::getCookiesForInFlightSyncWrites() {
     auto s = state.wlock();
     auto vec = std::vector<const void*>();
-    for (auto write : s->trackedWrites) {
+    for (auto& write : s->trackedWrites) {
         auto* cookie = write.getCookie();
         if (cookie) {
             vec.push_back(cookie);
+            write.clearCookie();
         }
     }
     return vec;

engines/ep/src/durability/active_durability_monitor.h

@@ -125,9 +125,11 @@ class ActiveDurabilityMonitor : public DurabilityMonitor {
     void processTimeout(std::chrono::steady_clock::time_point asOf);
 
     /**
-     * Get the non-null cookies for all in-flight SyncWrites.
+     * Get (and clear) the non-null cookies for all in-flight SyncWrites.
      * (Null cookies - for example originating from SyncWrites loaded during
-     *  warmup - are not returned).
+     * warmup - are not returned). The reason for clearing the cookies is
+     * to avoid a double notification on the cookie (which is illegal),
+     * so the caller <u>must</u> notify these cookies.
      */
     std::vector<const void*> getCookiesForInFlightSyncWrites();
 

engines/ep/src/durability/durability_monitor_impl.cc

@@ -73,6 +73,10 @@ const void* DurabilityMonitor::SyncWrite::getCookie() const {
     return cookie;
 }
 
+void DurabilityMonitor::SyncWrite::clearCookie() {
+    cookie = nullptr;
+}
+
 std::chrono::steady_clock::time_point
 DurabilityMonitor::SyncWrite::getStartTime() const {
     return startTime;

engines/ep/src/durability/durability_monitor_impl.h

@@ -72,6 +72,8 @@ class DurabilityMonitor::SyncWrite {
 
     const void* getCookie() const;
 
+    void clearCookie();
+
     std::chrono::steady_clock::time_point getStartTime() const;
 
     /**

engines/ep/src/ep_engine.cc

@@ -5821,6 +5821,11 @@ void EventuallyPersistentEngine::initiate_shutdown() {
     dcpConnMap_->shutdownAllConnections();
 }
 
+void EventuallyPersistentEngine::cancel_all_operations_in_ewb_state() {
+    auto eng = acquireEngine(this);
+    kvBucket->releaseRegisteredSyncWrites();
+}
+
 cb::mcbp::Status EventuallyPersistentEngine::stopFlusher(const char** msg,
                                                          size_t* msg_size) {
     (void)msg_size;

engines/ep/src/ep_engine.h

@@ -564,6 +564,7 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
 
     void handleDisconnect(const void *cookie);
     void initiate_shutdown() override;
+    void cancel_all_operations_in_ewb_state() override;
 
     cb::mcbp::Status stopFlusher(const char** msg, size_t* msg_size);
 

engines/ep/src/kv_bucket.cc

@@ -228,10 +228,10 @@ class RespondAmbiguousNotification : public GlobalTask {
 public:
     RespondAmbiguousNotification(EventuallyPersistentEngine& e,
                                  VBucketPtr& vb,
-                                 std::vector<const void*> cookies)
+                                 std::vector<const void*>&& cookies_)
         : GlobalTask(&e, TaskId::RespondAmbiguousNotification, 0, false),
           weakVb(vb),
-          cookies(cookies),
+          cookies(std::move(cookies_)),
           description("Notify clients of Sync Write Ambiguous " +
                       vb->getId().to_string()) {
         for (const auto* cookie : cookies) {
@@ -856,6 +856,29 @@ ENGINE_ERROR_CODE KVBucket::setVBucketState(Vbid vbid,
             vbid, to, meta, transfer, true /*notifyDcp*/, lh);
 }
 
+void KVBucket::releaseRegisteredSyncWrites() {
+    for (size_t vbid = 0; vbid < vbMap.size; ++vbid) {
+        VBucketPtr vb = vbMap.getBucket(Vbid{gsl::narrow<uint16_t>(vbid)});
+        if (!vb) {
+            continue;
+        }
+        folly::SharedMutex::ReadHolder rlh(vb->getStateLock());
+        if (vb->getState() != vbucket_state_active) {
+            continue;
+        }
+
+        auto cookies = vb->getCookiesForInFlightSyncWrites();
+        if (!cookies.empty()) {
+            EP_LOG_INFO("Cancel {} blocked durability requests in {}",
+                        cookies.size(),
+                        vb);
+            ExTask notifyTask = std::make_shared<RespondAmbiguousNotification>(
+                    engine, vb, std::move(cookies));
+            ExecutorPool::get()->schedule(notifyTask);
+        }
+    }
+}
+
 ENGINE_ERROR_CODE KVBucket::setVBucketState_UNLOCKED(
         Vbid vbid,
         vbucket_state_t to,
@@ -887,7 +910,7 @@ ENGINE_ERROR_CODE KVBucket::setVBucketState_UNLOCKED(
             // Get a list of cookies that we should respond to
             auto connectionsToRespondTo = vb->getCookiesForInFlightSyncWrites();
             ExTask notifyTask = std::make_shared<RespondAmbiguousNotification>(
-                    engine, vb, connectionsToRespondTo);
+                    engine, vb, std::move(connectionsToRespondTo));
             ExecutorPool::get()->schedule(notifyTask);
         }
 

engines/ep/src/kv_bucket.h

@@ -283,6 +283,18 @@ class KVBucket : public KVBucketIface {
         }
     }
 
+    /**
+     * Release all cookies blocked for sync write
+     *
+     * This method is called during bucket shutdown to make sure that
+     * all of the cookies waiting for a durable write is released so that
+     * we can continue bucket deletion. As part of bucket deletion one of
+     * the first things we do is to tear down the DCP streams so that
+     * the durable writes will never be notified and would be stuck
+     * waiting for a timeout if we don't explicitly release them.
+     */
+    void releaseRegisteredSyncWrites();
+
     /**
      * Sets the vbucket or creates a vbucket with the desired state
      *

engines/ep/src/vbucket.h

@@ -524,7 +524,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
     void fireAllOps(EventuallyPersistentEngine &engine);
 
     /**
-     * Get the cookies for all in-flight SyncWrites from the ADM
+     * Get (and clear) the cookies for all in-flight SyncWrites from the ADM
      */
     std::vector<const void*> getCookiesForInFlightSyncWrites();