[BP] MB-36802: Don't return estimate=0 for dcp-takeover stats before backfill

[Backport of MB-35594.]

The 'dcp-takeover' stats are used by ns_server to estimate how many
mutations are remaining on a DCP stream. However, the estimate value
is not updated until the backfill task has run once (and scanned the
disk file). As such, if 'dcp-takeover' stats are requested before that
first backfil task has run, then they can incorrectly report '0'
backfill items.

To address this, change backfillRemaining to be of type
boost::optional, initialized to an empty optional.

Only when the backfill scan has completed (when the number of items
remaining is determined) is the optional populated.

Then, when stats are requested use a new status value
"calculating-item-count" if the optional is empty (i.e. before scan).

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

Author: owend74

engines/ep/src/dcp/backfill_disk.cc

@@ -256,7 +256,7 @@ backfill_status_t DCPBackfillDisk::create() {
         stream->setDead(status);
         transitionState(backfill_state_done);
     } else {
-        stream->incrBackfillRemaining(scanCtx->documentCount);
+        stream->setBackfillRemaining(scanCtx->documentCount);
         stream->markDiskSnapshot(startSeqno, scanCtx->maxSeqno);
         transitionState(backfill_state_scanning);
     }

engines/ep/src/dcp/backfill_memory.cc

@@ -90,7 +90,7 @@ backfill_status_t DCPBackfillMemory::run() {
     }
 
     /* Put items onto readyQ of the DCP stream */
-    stream->incrBackfillRemaining(items.size());
+    stream->setBackfillRemaining(items.size());
 
     /* Mark disk snapshot */
     stream->markDiskSnapshot(startSeqno, adjustedEndSeqno);
@@ -225,11 +225,11 @@ backfill_status_t DCPBackfillMemoryBuffered::create() {
        remaining count */
     while (rangeItr.curr() != rangeItr.end()) {
         if (static_cast<uint64_t>((*rangeItr).getBySeqno()) >= startSeqno) {
-            /* Incr backfill remaining
+            /* Set backfill remaining
                [EPHE TODO]: This will be inaccurate if do not backfill till end
                             of the iterator
              */
-            stream->incrBackfillRemaining(rangeItr.count());
+            stream->setBackfillRemaining(rangeItr.count());
 
             /* Determine the endSeqno of the current snapshot.
                We want to send till requested endSeqno, but if that cannot

engines/ep/src/dcp/stream.cc

@@ -256,7 +256,7 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       isBackfillTaskRunning(false),
       pendingBackfill(false),
       lastReadSeqno(st_seqno),
-      backfillRemaining(0),
+      backfillRemaining(),
       lastReadSeqnoUnSnapshotted(st_seqno),
       lastSentSeqno(st_seqno),
       curChkSeqno(st_seqno),
@@ -635,6 +635,16 @@ void ActiveStream::setVBucketStateAckRecieved() {
     notifyStreamReady();
 }
 
+void ActiveStream::setBackfillRemaining(size_t value) {
+    std::lock_guard<std::mutex> guard(streamMutex);
+    backfillRemaining = value;
+}
+
+void ActiveStream::clearBackfillRemaining() {
+    std::lock_guard<std::mutex> guard(streamMutex);
+    backfillRemaining.reset();
+}
+
 std::unique_ptr<DcpResponse> ActiveStream::backfillPhase(
         std::lock_guard<std::mutex>& lh) {
     auto resp = nextQueuedItem();
@@ -662,15 +672,15 @@ std::unique_ptr<DcpResponse> ActiveStream::backfillPhase(
         // Only DcpResponse objects representing items from "disk" have a size
         // so only update backfillRemaining when non-zero
         if (resp->getApproximateSize()) {
-            if (backfillRemaining.load(std::memory_order_relaxed) > 0) {
-                backfillRemaining.fetch_sub(1, std::memory_order_relaxed);
+            Expects(backfillRemaining.is_initialized());
+            if (*backfillRemaining > 0) {
+                (*backfillRemaining)--;
             }
         }
     }
 
     if (!isBackfillTaskRunning && readyQ.empty()) {
         // Given readyQ.empty() is True resp will be NULL
-        backfillRemaining.store(0, std::memory_order_relaxed);
         // The previous backfill has completed.  Check to see if another
         // backfill needs to be scheduled.
         if (pendingBackfill) {
@@ -875,15 +885,24 @@ void ActiveStream::addTakeoverStats(ADD_STAT add_stat, const void *cookie,
         return;
     }
 
-    size_t total = backfillRemaining.load(std::memory_order_relaxed);
+    size_t total = 0;
+    const char* status = nullptr;
     if (isBackfilling()) {
-        add_casted_stat("status", "backfilling", add_stat, cookie);
+        if (backfillRemaining) {
+            status = "backfilling";
+            total += *backfillRemaining;
+        } else {
+            status = "calculating-item-count";
+        }
     } else {
-        add_casted_stat("status", "in-memory", add_stat, cookie);
+        status = "in-memory";
+    }
+    add_casted_stat("status", status, add_stat, cookie);
+
+    if (backfillRemaining) {
+        add_casted_stat(
+                "backfillRemaining", *backfillRemaining, add_stat, cookie);
     }
-    add_casted_stat("backfillRemaining",
-                    backfillRemaining.load(std::memory_order_relaxed),
-                    add_stat, cookie);
 
     size_t vb_items = vb.getNumItems();
     size_t chk_items = 0;
@@ -1507,6 +1526,9 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
         producer->scheduleBackfillManager(
                 *vbucket, shared_from_this(), backfillStart, backfillEnd);
         isBackfillTaskRunning.store(true);
+        /// Number of backfill items is unknown until the Backfill task
+        /// completes the scan phase - reset backfillRemaining counter.
+        backfillRemaining.reset();
     } else {
         if (reschedule) {
             // Infrequent code path, see comment below.

engines/ep/src/dcp/stream.h

@@ -250,9 +250,12 @@ class ActiveStream : public Stream,
 
     void setVBucketStateAckRecieved();
 
-    void incrBackfillRemaining(size_t by) {
-        backfillRemaining.fetch_add(by, std::memory_order_relaxed);
-    }
+    /// Set the number of backfill items remaining to the given value.
+    void setBackfillRemaining(size_t value);
+
+    /// Clears the number of backfill items remaining, setting to an empty
+    /// (unknown) value.
+    void clearBackfillRemaining();
 
     void markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno);
 
@@ -397,12 +400,13 @@ class ActiveStream : public Stream,
        snapshotted and put onto readyQ */
     AtomicMonotonic<uint64_t, ThrowExceptionPolicy> lastReadSeqno;
 
-    /* backfillRemaining is a stat recording the amount of
-     * items remaining to be read from disk.  It is an atomic
-     * because otherwise the function incrBackfillRemaining
-     * must acquire the streamMutex lock.
+    /* backfillRemaining is a stat recording the amount of items remaining to
+     * be read from disk.
+     * Before the number of items to be backfilled has been determined (disk
+     * scanned) it is empty.
+     * Guarded by streamMutex.
      */
-    std::atomic<size_t> backfillRemaining;
+    boost::optional<size_t> backfillRemaining;
 
     std::unique_ptr<DcpResponse> backfillPhase(std::lock_guard<std::mutex>& lh);
 

engines/ep/tests/ep_test_apis.cc

@@ -1503,24 +1503,6 @@ void wait_for_stat_to_be_gte(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
     }
 }
 
-void wait_for_stat_to_be_lte(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
-                             const char *stat, int final,
-                             const char* stat_key,
-                             const time_t max_wait_time_in_secs) {
-    useconds_t sleepTime = 128;
-    WaitTimeAccumulator<int> accumulator("to be less than or equal to", stat,
-                                         stat_key, final,
-                                         max_wait_time_in_secs);
-    for (;;) {
-        auto current = get_int_stat(h, h1, stat, stat_key);
-        if (current <= final) {
-            break;
-        }
-        accumulator.incrementAndAbortIfLimitReached(current, sleepTime);
-        decayingSleep(&sleepTime);
-    }
-}
-
 void wait_for_expired_items_to_be(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
                                   int final,
                                   const time_t max_wait_time_in_secs) {

engines/ep/tests/ep_test_apis.h

@@ -435,10 +435,15 @@ void wait_for_stat_to_be_gte(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
                              const char *stat, int final,
                              const char* stat_key = NULL,
                              const time_t max_wait_time_in_secs = 60);
-void wait_for_stat_to_be_lte(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
-                             const char *stat, int final,
+
+template <typename T>
+void wait_for_stat_to_be_lte(ENGINE_HANDLE* h,
+                             ENGINE_HANDLE_V1* h1,
+                             const char* stat,
+                             T final,
                              const char* stat_key = NULL,
                              const time_t max_wait_time_in_secs = 60);
+
 void wait_for_expired_items_to_be(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
                                   int final,
                                   const time_t max_wait_time_in_secs = 60);
@@ -641,6 +646,29 @@ void wait_for_stat_to_be(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
     }
 }
 
+template <typename T>
+void wait_for_stat_to_be_lte(ENGINE_HANDLE* h,
+                             ENGINE_HANDLE_V1* h1,
+                             const char* stat,
+                             T final,
+                             const char* stat_key,
+                             const time_t max_wait_time_in_secs) {
+    useconds_t sleepTime = 128;
+    WaitTimeAccumulator<T> accumulator("to be less than or equal to",
+                                       stat,
+                                       stat_key,
+                                       final,
+                                       max_wait_time_in_secs);
+    for (;;) {
+        auto current = get_stat<T>(h, h1, stat, stat_key);
+        if (current <= final) {
+            break;
+        }
+        accumulator.incrementAndAbortIfLimitReached(current, sleepTime);
+        decayingSleep(&sleepTime);
+    }
+}
+
 /**
  * Function that does an exponential wait for a 'val' to reach 'expected'
  *

engines/ep/tests/ep_testsuite_dcp.cc

@@ -687,8 +687,8 @@ ENGINE_ERROR_CODE TestDcpConsumer::openStreams() {
             std::stringstream stats_takeover;
             stats_takeover << "dcp-vbtakeover " << ctx.vbucket
             << " " << name.c_str();
-            wait_for_stat_to_be_lte(h, h1, "estimate", static_cast<int>(est),
-                                    stats_takeover.str().c_str());
+            wait_for_stat_to_be_lte(
+                    h, h1, "estimate", est, stats_takeover.str().c_str());
         }
 
         if (ctx.flags & DCP_ADD_STREAM_FLAG_DISKONLY) {

engines/ep/tests/mock/mock_stream.h

@@ -121,7 +121,7 @@ class MockActiveStream : public ActiveStream {
         return lastReadSeqno;
     }
 
-    int getNumBackfillItemsRemaining() const {
+    boost::optional<size_t> getNumBackfillItemsRemaining() const {
         return backfillRemaining;
     }