MB-59189: Do not access the cursor without a queueLock

Logging added under MB-58961 in neo accesses the cursor's position
without a checkpoint manager lock, which is unsafe, as the checkpoint
manager might have to dedup the item this cursor is currently on at any
time.

Return the Item the cursor gets registered on in CursorRegResult, so the
calling code can access it without going via a cursor deref to print the
info log.

Change-Id: Ie819af7fd12743d18579f17652eb00cf47e55c9e
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/200830
Reviewed-by: Paolo Cocchi <[email protected]>
Tested-by: Vesko Karaganev <[email protected]>
Well-Formed: Restriction Checker
Reviewed-by: Jim Walker <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: veselink1

engines/ep/src/checkpoint_manager.cc

@@ -328,10 +328,12 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(
         // Note: The legacy logic (below) for cursor-registering is based on
         // Checkpoint::getMinimumCursorSeqno()/getHighSeqno() that are
         // meaningless if all mutations have been expelled.
+        auto pos = (*ckptIt)->begin();
         newCursor = std::make_shared<CheckpointCursor>(
-                name, ckptIt, (*ckptIt)->begin(), droppable, 0);
+                name, ckptIt, pos, droppable, 0);
         result.nextSeqno = lastBySeqno + 1;
         result.cursor.setCursor(newCursor);
+        result.position = *pos;
 
         // Trigger backfill only if there's a gap between lastProcessedSeqno and
         // nextSeqnoAvailableFromCheckpoint
@@ -366,10 +368,12 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(
             if (lastProcessedSeqno < st) {
                 // Requested sequence number is before the start of this
                 // checkpoint, position cursor at the checkpoint begin.
+                auto pos = (*ckptIt)->begin();
                 newCursor = std::make_shared<CheckpointCursor>(
-                        name, ckptIt, (*ckptIt)->begin(), droppable, 0);
+                        name, ckptIt, pos, droppable, 0);
                 result.nextSeqno = st;
                 result.cursor.setCursor(newCursor);
+                result.position = *pos;
                 // Set tryBackfill:true only if lastProcessedSeqno to st is non
                 // contiguous. Note this is likely the fix for MB-53616/MB-58302
                 // which were noted in master only, and likely backported to the
@@ -424,6 +428,7 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(
                 newCursor = std::make_shared<CheckpointCursor>(
                         name, ckptIt, pos, droppable, distance);
                 result.cursor.setCursor(newCursor);
+                result.position = *pos;
                 break;
             }
         }

engines/ep/src/cursor.h

@@ -11,6 +11,9 @@
 
 #pragma once
 
+#include "ep_types.h"
+#include <platform/atomic.h>
+
 #include <folly/SharedMutex.h>
 #include <memory>
 #include <mutex>
@@ -94,4 +97,6 @@ struct CursorRegResult {
     uint64_t nextSeqno;
     // The registered cursor
     Cursor cursor;
+    // The item the cursor was registered on.
+    queued_item position;
 };

engines/ep/src/dcp/active_stream.cc

@@ -261,8 +261,7 @@ void ActiveStream::registerCursor(CheckpointManager& chkptmgr,
         curChkSeqno = result.nextSeqno;
         cursor = result.cursor;
 
-        const auto lockedCursor = result.cursor.lock();
-        Expects(lockedCursor);
+        Expects(result.position);
         log(spdlog::level::level_enum::info,
             "{} ActiveStream::registerCursor name \"{}\", "
             "lastProcessedSeqno:{}, pendingBackfill:{}, "
@@ -272,8 +271,8 @@ void ActiveStream::registerCursor(CheckpointManager& chkptmgr,
             lastProcessedSeqno,
             pendingBackfill,
             result.tryBackfill,
-            ::to_string((*lockedCursor->getPos())->getOperation()),
-            (*lockedCursor->getPos())->getBySeqno(),
+            ::to_string(result.position->getOperation()),
+            result.position->getBySeqno(),
             result.nextSeqno);
     } catch (std::exception& error) {
         log(spdlog::level::level_enum::warn,
@@ -1813,8 +1812,7 @@ void ActiveStream::scheduleBackfill_UNLOCKED(DcpProducer& producer,
             return;
         }
 
-        const auto lockedCursor = registerResult.cursor.lock();
-        Expects(lockedCursor);
+        Expects(registerResult.position);
         log(spdlog::level::level_enum::info,
             "{} ActiveStream::scheduleBackfill_UNLOCKED register cursor with "
             "name \"{}\" lastProcessedSeqno:{}, result{{tryBackfill:{}, op:{}, "
@@ -1823,8 +1821,8 @@ void ActiveStream::scheduleBackfill_UNLOCKED(DcpProducer& producer,
             name_,
             lastReadSeqno.load(),
             registerResult.tryBackfill,
-            ::to_string((*lockedCursor->getPos())->getOperation()),
-            (*lockedCursor->getPos())->getBySeqno(),
+            ::to_string(registerResult.position->getOperation()),
+            registerResult.position->getBySeqno(),
             registerResult.nextSeqno);
 
         curChkSeqno = registerResult.nextSeqno;