[BP] MB-58961: Remove ActiveStream::lastReadSeqnoUnSnapshotted

That is set to the last seqno that a stream has got from checkpoints
but still not pushed into the stream readyQ.

Used for passing that information down to
ActiveStream::snapshot(<items>) and setting AS::lastReadSeqno. That
doesn't need to be a class member as that information is computed
locally when the stream makes up a snapshot from checkpoint.

Also, lastReadSeqnoUnSnapshotted represents a non-atomic information on
the stream state, as that is updated for every item processed under
streamLock in the middle of the execution of
AS::nextCheckpointItemTask().

Change-Id: Iac36349920203e81ba9898137a902c2d1def2048
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/198886
Reviewed-by: Trond Norbye <[email protected]>
Tested-by: Paolo Cocchi <[email protected]>
Well-Formed: Restriction Checker

Author: paolococchi

engines/ep/docs/stats.org

@@ -741,8 +741,6 @@ another colon.  For example, if your client is named, =slave1=, the
 | last_sent_seqno               | The last seqno sent by this stream                    |
 | last_sent_snap_end_seqno      | The last snapshot end seqno sent by active stream     |
 | last_read_seqno               | The last seqno read by this stream from disk or memory|
-| last_read_seqno_unsnapshotted | The last sequence number queued from memory, but is   |
-|                               | yet to be put in a snapshot                           |
 | ready_queue_memory            | Memory occupied by elements in the DCP readyQ         |
 | memory_phase                  | The amount of items sent during the memory phase      |
 | opaque                        | The unique stream identifier                          |

engines/ep/src/dcp/active_stream.cc

@@ -66,7 +66,6 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       includeValue(includeVal),
       includeXattributes(includeXattrs),
       includeDeletedUserXattrs(includeDeletedUserXattrs),
-      lastReadSeqnoUnSnapshotted(st_seqno),
       lastSentSeqno(st_seqno),
       lastSentSeqnoAdvance(0),
       curChkSeqno(st_seqno),
@@ -846,8 +845,6 @@ void ActiveStream::addStats(const AddStatFn& add_stat, const CookieIface* c) {
         addStat("last_sent_snap_end_seqno",
                 lastSentSnapEndSeqno.load(std::memory_order_relaxed));
         addStat("last_read_seqno", lastReadSeqno.load());
-        addStat("last_read_seqno_unsnapshotted",
-                lastReadSeqnoUnSnapshotted.load());
         addStat("ready_queue_memory", getReadyQueueMemory());
         addStat("backfill_buffer_bytes", bufferedBackfill.bytes.load());
         addStat("backfill_buffer_items", bufferedBackfill.items.load());
@@ -1317,6 +1314,7 @@ void ActiveStream::processItems(
          * sent meaning the snapshot was never completed.
          */
         std::optional<uint64_t> highNonVisibleSeqno;
+        uint64_t newLastReadSeqno = 0;
         for (auto& qi : outstandingItemsResult.items) {
             if (qi->getOperation() == queue_op::checkpoint_end) {
                 // At the end of each checkpoint remove its snapshot range, so
@@ -1337,7 +1335,8 @@ void ActiveStream::processItems(
                     snapshot(outstandingItemsResult,
                              mutations,
                              visibleSeqno,
-                             highNonVisibleSeqno);
+                             highNonVisibleSeqno,
+                             newLastReadSeqno);
                     /* clear out all the mutations since they are already put
                        onto the readyQ */
                     mutations.clear();
@@ -1365,7 +1364,8 @@ void ActiveStream::processItems(
             }
 
             if (shouldProcessItem(*qi)) {
-                lastReadSeqnoUnSnapshotted = qi->getBySeqno();
+                newLastReadSeqno = qi->getBySeqno();
+
                 // Check if the item is allowed on the stream, note the filter
                 // updates itself for collection deletion events
                 if (filter.checkAndUpdate(*qi)) {
@@ -1387,7 +1387,8 @@ void ActiveStream::processItems(
             snapshot(outstandingItemsResult,
                      mutations,
                      visibleSeqno,
-                     highNonVisibleSeqno);
+                     highNonVisibleSeqno,
+                     newLastReadSeqno);
         } else if (isSeqnoAdvancedEnabled()) {
             // Note that we cannot enter this case if supportSyncReplication()
             // returns true (see isSeqnoAdvancedEnabled). This means that we
@@ -1476,13 +1477,13 @@ bool ActiveStream::shouldProcessItem(const Item& item) {
 void ActiveStream::snapshot(const OutstandingItemsResult& meta,
                             std::deque<std::unique_ptr<DcpResponse>>& items,
                             uint64_t maxVisibleSeqno,
-                            std::optional<uint64_t> highNonVisibleSeqno) {
+                            std::optional<uint64_t> highNonVisibleSeqno,
+                            uint64_t newLastReadSeqno) {
     if (items.empty()) {
         return;
     }
 
-    /* This assumes that all items in the "items deque" is put onto readyQ */
-    lastReadSeqno.store(lastReadSeqnoUnSnapshotted);
+    lastReadSeqno.store(newLastReadSeqno);
 
     if (isCurrentSnapshotCompleted()) {
         // Get OptionalSeqnos which for the items list types should have values

engines/ep/src/dcp/active_stream.h

@@ -632,8 +632,7 @@ class ActiveStream : public Stream,
         std::atomic<size_t> sent = 0;
     } backfillItems;
 
-    /* The last sequence number queued from disk or memory and is
-       snapshotted and put onto readyQ */
+    // The last sequence number queued into the readyQ from disk or memory
     AtomicMonotonic<uint64_t, ThrowExceptionPolicy> lastReadSeqno;
 
     /* backfill ById or BySeqno updates this member during the scan, then
@@ -694,11 +693,13 @@ class ActiveStream : public Stream,
      * @param highNonVisibleSeqno the snapEnd seqno that includes any non
      * visible mutations i.e. prepares and aborts. This is only used when
      * collections is enabled and sync writes are not supported on the stream.
+     * @param newLastReadSeqno The new lastReadSeqno, see member for details.
      */
     void snapshot(const OutstandingItemsResult& meta,
                   std::deque<std::unique_ptr<DcpResponse>>& items,
                   uint64_t maxVisibleSeqno,
-                  std::optional<uint64_t> highNonVisibleSeqno);
+                  std::optional<uint64_t> highNonVisibleSeqno,
+                  uint64_t newLastReadSeqno);
 
     void endStream(cb::mcbp::DcpStreamEndStatus reason);
 
@@ -839,10 +840,6 @@ class ActiveStream : public Stream,
     //! Number of times a backfill is paused.
     cb::RelaxedAtomic<uint64_t> numBackfillPauses{0};
 
-    /* The last sequence number queued from memory, but is yet to be
-       snapshotted and put onto readyQ */
-    std::atomic<uint64_t> lastReadSeqnoUnSnapshotted;
-
     //! The last sequence number sent to the network layer
     AtomicMonotonic<uint64_t, ThrowExceptionPolicy> lastSentSeqno;