MB-58961: Remove AS::maxScanSeqno

paolococchi · paolococchi · commit 424b5131fe19 · 2023-10-30T10:10:47.000Z
The member is set by DCPBackfill<type> instances for being used in the final AS::completeBackfill calls. Remove the member and just pass that information by arg onto the ::completeBackfill() call. No logic change. Change-Id: If1995392fa4f03bfdf850ac2a9aafd11e390a05a Reviewed-on: https://review.couchbase.org/c/kv_engine/+/199229 Well-Formed: Restriction Checker Reviewed-by: Trond Norbye <trond.norbye@couchbase.com> Tested-by: Paolo Cocchi <paolo.cocchi@couchbase.com>
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -565,19 +565,21 @@ bool ActiveStream::backfillReceived(std::unique_ptr<Item> itm,
     return true;
 }
 
-void ActiveStream::completeBackfill(std::chrono::steady_clock::duration runtime,
+void ActiveStream::completeBackfill(uint64_t maxScanSeqno,
+                                    std::chrono::steady_clock::duration runtime,
                                     size_t diskBytesRead) {
-    // maxSeqno is not needed for InOrder completion
     completeBackfillInner(
-            BackfillType::InOrder, 0 /*maxSeqno*/, runtime, diskBytesRead);
+            BackfillType::InOrder, maxScanSeqno, runtime, diskBytesRead);
 }
 
 void ActiveStream::completeOSOBackfill(
-        uint64_t maxSeqno,
+        uint64_t maxScanSeqno,
         std::chrono::steady_clock::duration runtime,
         size_t diskBytesRead) {
-    completeBackfillInner(
-            BackfillType::OutOfSequenceOrder, maxSeqno, runtime, diskBytesRead);
+    completeBackfillInner(BackfillType::OutOfSequenceOrder,
+                          maxScanSeqno,
+                          runtime,
+                          diskBytesRead);
     firstMarkerSent = true;
 }
 
@@ -1949,7 +1951,7 @@ bool ActiveStream::tryAndScheduleOSOBackfill(DcpProducer& producer,
 
 void ActiveStream::completeBackfillInner(
         BackfillType backfillType,
-        uint64_t maxSeqno,
+        uint64_t maxScanSeqno,
         std::chrono::steady_clock::duration runtime,
         size_t diskBytesRead) {
     {
@@ -1963,21 +1965,20 @@ void ActiveStream::completeBackfillInner(
         }
 
         if (backfillType == BackfillType::InOrder) {
+            const auto vb = engine->getVBucket(vb_);
+            if (!vb) {
+                log(spdlog::level::level_enum::warn,
+                    "{} ActiveStream::completeBackfillInner(): Vbucket "
+                    "does not exist",
+                    logPrefix);
+                return;
+            }
+
             // In-order backfills may require a seqno-advanced message if
             // there is a stream filter present (e.g. only streaming a single
             // collection).
             if (isSeqnoAdvancedEnabled() &&
                 isSeqnoGapAtEndOfSnapshot(maxScanSeqno)) {
-                const auto vb = engine->getVBucket(vb_);
-                if (!vb) {
-                    log(spdlog::level::level_enum::warn,
-                        "{} ActiveStream::completeBackfillInner(): Vbucket "
-                        "does not exist",
-                        logPrefix);
-                }
-
-                // Not VB:  Vbucket doens't exist anymore; We still need to send
-                //          a SeqnoAdvance to bump the peer to end-of-snapshot.
                 // Active:  We must send a SeqnoAdvanced to bump the DCP
                 //          client's seqno to snap-end.
                 // Replica: Vbucket may transition backfill->memory without
@@ -1988,7 +1989,7 @@ void ActiveStream::completeBackfillInner(
                 const auto replicaVucketSeqnoAdvance =
                         maxScanSeqno > lastBackfilledSeqno &&
                         maxScanSeqno == lastSentSnapEndSeqno;
-                if (!vb || vb->getState() != vbucket_state_replica ||
+                if (vb->getState() != vbucket_state_replica ||
                     replicaVucketSeqnoAdvance) {
                     queueSeqnoAdvanced();
                 }
@@ -2007,8 +2008,6 @@ void ActiveStream::completeBackfillInner(
             if (!isCollectionEnabledStream() && maxScanSeqno > lastReadSeqno) {
                 lastReadSeqno.store(maxScanSeqno);
             }
-            // reset last seqno seen by backfill
-            maxScanSeqno = 0;
         }
 
         if (isBackfilling()) {
@@ -2055,19 +2054,19 @@ void ActiveStream::completeBackfillInner(
                     logPrefix);
             } else if (
                     producer->isOutOfOrderSnapshotsEnabledWithSeqnoAdvanced() &&
-                    maxSeqno != lastBackfilledSeqno) {
+                    maxScanSeqno != lastBackfilledSeqno) {
                 pushToReadyQ(std::make_unique<SeqnoAdvanced>(
-                        opaque_, vb_, sid, maxSeqno));
-                lastSentSeqnoAdvance.store(maxSeqno);
+                        opaque_, vb_, sid, maxScanSeqno));
+                lastSentSeqnoAdvance.store(maxScanSeqno);
             }
 
             // Now that the OSO backfill has ended, we can tweak
             // lastReadSeqno so that it reflects the end of the snapshot
             // we've just processed. This ensures any pending backfill which
             // follows continues from maxSeqno and not the max seqno of the
             // collection(s) in the OSO scan, which could be way less.
-            if (maxSeqno > lastReadSeqno) {
-                lastReadSeqno = maxSeqno;
+            if (maxScanSeqno > lastReadSeqno) {
+                lastReadSeqno = maxScanSeqno;
             }
 
             pushToReadyQ(std::make_unique<OSOSnapshot>(
diff --git a/engines/ep/src/dcp/active_stream.h b/engines/ep/src/dcp/active_stream.h
@@ -241,20 +241,24 @@ class ActiveStream : public Stream,
                           backfill_source_t backfill_source);
 
     /**
+     * @param maxScanSeqno the maximum seqno of the snapshot supplying the OSO
+     *        backfill. A SeqnoAdvanced maybe sent if the last backfilled
+     *        item is not the maxSeqno item
      * @param runtime The total runtime the backfill took, measured as active
-     * time executing (i.e. total BackfillManagerTask::run() durations for
-     * this backfill)
+     *        time executing (i.e. total BackfillManagerTask::run() durations
+     *        for this backfill)
      * @param diskBytesRead The total number of bytes read from disk during
      *        this backfill.
      */
-    void completeBackfill(std::chrono::steady_clock::duration runtime,
+    void completeBackfill(uint64_t maxScanSeqno,
+                          std::chrono::steady_clock::duration runtime,
                           size_t diskBytesRead);
 
     /**
      * Queues a single "Out of Seqno Order" marker with the 'end' flag
      * into the ready queue.
      *
-     * @param maxSeqno the maximum seqno of the snapshot supplying the OSO
+     * @param maxScanSeqno the maximum seqno of the snapshot supplying the OSO
      *        backfill. A SeqnoAdvanced maybe sent if the last backfilled
      *        item is not the maxSeqno item
      * @param runtime The total runtime the backfill took, measured as active
@@ -263,7 +267,7 @@ class ActiveStream : public Stream,
      * @param diskBytesRead The total number of bytes read from disk during
      *        this backfill.
      */
-    void completeOSOBackfill(uint64_t maxSeqno,
+    void completeOSOBackfill(uint64_t maxScanSeqno,
                              std::chrono::steady_clock::duration runtime,
                              size_t diskBytesRead);
 
@@ -428,22 +432,6 @@ class ActiveStream : public Stream,
      */
     bool isIgnoringPurgedTombstones() const;
 
-    /**
-     * Used to set the last read seqno from a scan of the data store layer.
-     * (This is for external use only and ensures that
-     * maxScanSeqno is zero prior to being set).
-     * @param seqno last read seqno during a data store layer scan
-     */
-    void setBackfillScanLastRead(uint64_t seqno) {
-        std::unique_lock<std::mutex> lh(streamMutex);
-        /*
-         * maxScanSeqno should only be modified once per
-         * backfill
-         */
-        Expects(maxScanSeqno == 0);
-        maxScanSeqno = seqno;
-    }
-
     /**
      * Method to get the collections filter of the stream
      * @return the filter object
@@ -887,15 +875,6 @@ class ActiveStream : public Stream,
     //! Last snapshot end seqno sent to the DCP client
     std::atomic<uint64_t> lastSentSnapEndSeqno;
 
-    /*
-     * This is the highest seqno seen by KVStore when performing a scan
-     * for the current snapshot we are back filling for. This is regardless of
-     * collection or visibility. This used inform completeBackfill() of the last
-     * read seqno from disk, to help make a decision on if we should enqueue an
-     * SeqnoAdvanced op (see ::completeBackfill() for more info).
-     */
-    uint64_t maxScanSeqno{0};
-
     /* Flag used by checkpointCreatorTask that is set before all items are
        extracted for given checkpoint cursor, and is unset after all retrieved
        items are added to the readyQ */
diff --git a/engines/ep/src/dcp/backfill_by_seqno_disk.cc b/engines/ep/src/dcp/backfill_by_seqno_disk.cc
@@ -193,7 +193,6 @@ backfill_status_t DCPBackfillBySeqnoDisk::scan() {
         if (historyScan) {
             transitionState(backfill_state_scanning_history_snapshot);
         } else {
-            stream->setBackfillScanLastRead(scanCtx->lastReadSeqno);
             transitionState(backfill_state_completing);
         }
         return backfill_success;
@@ -234,7 +233,8 @@ void DCPBackfillBySeqnoDisk::complete(bool cancelled) {
     }
 
     const auto diskBytesRead = scanCtx ? scanCtx->diskBytesRead : 0;
-    stream->completeBackfill(runtime, diskBytesRead);
+    const auto maxScanSeqno = scanCtx ? scanCtx->lastReadSeqno : 0;
+    stream->completeBackfill(maxScanSeqno, runtime, diskBytesRead);
 
     auto severity = cancelled ? spdlog::level::level_enum::info
                               : spdlog::level::level_enum::debug;
diff --git a/engines/ep/src/dcp/backfill_disk.cc b/engines/ep/src/dcp/backfill_disk.cc
@@ -390,10 +390,10 @@ backfill_status_t DCPBackfillDisk::scanHistory() {
     Expects(kvstore);
     switch (kvstore->scanAllVersions(bySeqnoCtx)) {
     case scan_success:
-        stream->setBackfillScanLastRead(bySeqnoCtx.lastReadSeqno);
         // Call complete and transition straight to done (via complete). This
         // avoids the sub-class (by_id or by_seq) complete function being called
-        stream->completeBackfill(runtime, bySeqnoCtx.diskBytesRead);
+        stream->completeBackfill(
+                bySeqnoCtx.lastReadSeqno, runtime, bySeqnoCtx.diskBytesRead);
         transitionState(backfill_state_completing);
         transitionState(backfill_state_done);
         return backfill_success;
diff --git a/engines/ep/src/dcp/backfill_memory.cc b/engines/ep/src/dcp/backfill_memory.cc
@@ -324,8 +324,6 @@ backfill_status_t DCPBackfillMemoryBuffered::scan() {
         ++rangeItr;
     }
 
-    stream->setBackfillScanLastRead(endSeqno);
-
     /* Backfill has ran to completion */
     complete(false);
 
@@ -350,7 +348,7 @@ void DCPBackfillMemoryBuffered::complete(bool cancelled) {
 
     /* [EPHE TODO]: invalidate cursor sooner before it gets deleted */
 
-    stream->completeBackfill(runtime, 0);
+    stream->completeBackfill(endSeqno, runtime, 0);
 
     auto severity = cancelled ? spdlog::level::level_enum::info
                               : spdlog::level::level_enum::debug;
diff --git a/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc b/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc
@@ -2063,7 +2063,7 @@ TEST_P(STParamPersistentBucketTest, test_mb22451) {
     // flag to true
     EXPECT_TRUE(mock_stream->public_getPendingBackfill())
         << "handleSlowStream should set pendingBackfill to True";
-    mock_stream->completeBackfill({}, {});
+    mock_stream->completeBackfill(0, {}, {});
     EXPECT_FALSE(mock_stream->public_isBackfillTaskRunning())
         << "completeBackfill should set isBackfillTaskRunning to False";
     EXPECT_TRUE(mock_stream->isBackfilling())
