Remove CheckpointManager arg from Checkpoint::queueDirty()

Change-Id: Iee5283e66b0d830a1d548db1c4af2beab28b41c9
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/156744
Tested-by: Build Bot <[email protected]>
Reviewed-by: Richard de Mellow <[email protected]>

Author: paolococchi

engines/ep/src/checkpoint.cc

@@ -156,6 +156,7 @@ std::ostream& operator<<(std::ostream& os, const CheckpointCursor& c) {
 }
 
 Checkpoint::Checkpoint(
+        CheckpointManager& manager,
         EPStats& st,
         uint64_t id,
         uint64_t snapStart,
@@ -165,7 +166,8 @@ Checkpoint::Checkpoint(
         Vbid vbid,
         CheckpointType checkpointType,
         const std::function<void(int64_t)>& memOverheadChangedCallback)
-    : stats(st),
+    : manager(manager),
+      stats(st),
       checkpointId(id),
       snapStartSeqno(snapStart),
       snapEndSeqno(snapEnd),
@@ -206,8 +208,7 @@ Checkpoint::~Checkpoint() {
     memOverheadChangedCallback(-getMemoryOverhead());
 }
 
-QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi,
-                                        CheckpointManager* checkpointManager) {
+QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi) {
     if (getState() != CHECKPOINT_OPEN) {
         throw std::logic_error(
                 "Checkpoint::queueDirty: checkpointState "
@@ -289,7 +290,7 @@ QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi,
                 // this Checkpoint and see if the existing item for this key is
                 // to the "left" of the cursor (i.e. has already been
                 // processed).
-                for (auto& cursor : checkpointManager->cursors) {
+                for (auto& cursor : manager.cursors) {
                     if ((*(cursor.second->currentCheckpoint)).get() != this) {
                         // Cursor is in another checkpoint, doesn't need
                         // updating here
@@ -346,20 +347,16 @@ QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi,
                     // a single item (we de-dupe below). Track this in an
                     // AggregatedFlushStats in CheckpointManager so that we can
                     // undo these stat updates if the flush fails.
-                    auto backupPCursor = checkpointManager->cursors.find(
+                    auto backupPCursor = manager.cursors.find(
                             CheckpointManager::backupPCursorName);
 
-                    if (backupPCursor == checkpointManager->cursors.end()) {
+                    if (backupPCursor == manager.cursors.end()) {
                         decrCursorIfSameKey();
 
                         // We're not mid-flush, don't need to adjust any stats
                         continue;
                     }
 
-                    // We'd normally use "mutation_id" here which is basically
-                    // the seqno but the backupPCursor may be in a different
-                    // checkpoint and we'd fail a bunch of sanity checks trying
-                    // to read it.
                     auto backupPCursorSeqno =
                             (*(*backupPCursor->second).currentPos)
                                     ->getBySeqno();
@@ -368,8 +365,8 @@ QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi,
                         // the stats we'll use the new item and the flush
                         // will pick up the new item too so we have to match
                         // the original (oldItem) increment with a decrement
-                        checkpointManager->persistenceFailureStatOvercounts
-                                .accountItem(*oldItem);
+                        manager.persistenceFailureStatOvercounts.accountItem(
+                                *oldItem);
                     }
 
                     decrCursorIfSameKey();
@@ -457,7 +454,7 @@ QueueDirtyResult Checkpoint::queueDirty(const queued_item& qi,
     if (qi->getOperation() == queue_op::checkpoint_start ||
         qi->getOperation() == queue_op::checkpoint_end ||
         qi->getOperation() == queue_op::set_vbucket_state) {
-        checkpointManager->notifyFlusher();
+        manager.notifyFlusher();
     }
 
     return rv;

engines/ep/src/checkpoint.h

@@ -398,7 +398,8 @@ class Checkpoint {
     friend class CheckpointManagerTestIntrospector;
 
 public:
-    Checkpoint(EPStats& st,
+    Checkpoint(CheckpointManager& manager,
+               EPStats& st,
                uint64_t id,
                uint64_t snapStart,
                uint64_t snapEnd,
@@ -486,12 +487,9 @@ class Checkpoint {
     /**
      * Queue an item to be written to persistent layer.
      * @param item the item to be persisted
-     * @param checkpointManager the checkpoint manager to which this checkpoint belongs
-     * @param bySeqno the by sequence number assigned to this mutation
      * @return a result indicating the status of the operation.
      */
-    QueueDirtyResult queueDirty(const queued_item& qi,
-                                CheckpointManager* checkpointManager);
+    QueueDirtyResult queueDirty(const queued_item& qi);
 
     /**
      * @return true if the item can be de-duplicated, false otherwise
@@ -677,6 +675,9 @@ class Checkpoint {
      */
     CheckpointIndexKeyType makeIndexKey(const queued_item& item) const;
 
+    // Reference to the CheckpointManager that owns this Checkpoint
+    CheckpointManager& manager;
+
     EPStats& stats;
     uint64_t                       checkpointId;
     uint64_t                       snapStartSeqno;

engines/ep/src/checkpoint_manager.cc

@@ -135,7 +135,7 @@ void CheckpointManager::addNewCheckpoint_UNLOCKED(
             oldOpenCkpt.getHighSeqno());
     queued_item qi = createCheckpointItem(
             oldOpenCkpt.getId(), vbucketId, queue_op::checkpoint_end);
-    oldOpenCkpt.queueDirty(qi, this);
+    oldOpenCkpt.queueDirty(qi);
     ++numItems;
     oldOpenCkpt.setState(CHECKPOINT_CLOSED);
 
@@ -203,7 +203,8 @@ void CheckpointManager::addOpenCheckpoint(
             checkpointList.back()->getState() ==
                     checkpoint_state::CHECKPOINT_CLOSED);
 
-    auto ckpt = std::make_unique<Checkpoint>(stats,
+    auto ckpt = std::make_unique<Checkpoint>(*this,
+                                             stats,
                                              id,
                                              snapStart,
                                              snapEnd,
@@ -217,12 +218,12 @@ void CheckpointManager::addOpenCheckpoint(
     // at creation. So, the cursor will point at the first actual non-meta item
     // after the first cursor-increment.
     queued_item qi = createCheckpointItem(0, Vbid(0xffff), queue_op::empty);
-    ckpt->queueDirty(qi, this);
+    ckpt->queueDirty(qi);
     // Note: We don't include the empty-item in 'numItems'
 
     // This item represents the start of the new checkpoint
     qi = createCheckpointItem(id, vbucketId, queue_op::checkpoint_start);
-    ckpt->queueDirty(qi, this);
+    ckpt->queueDirty(qi);
     ++numItems;
 
     checkpointList.push_back(std::move(ckpt));
@@ -836,7 +837,7 @@ bool CheckpointManager::queueDirty(
         }
     }
 
-    QueueDirtyResult result = openCkpt->queueDirty(qi, this);
+    QueueDirtyResult result = openCkpt->queueDirty(qi);
 
     if (result.status == QueueDirtyStatus::FailureDuplicateItem) {
         // Could not queue into the current checkpoint as it already has a
@@ -857,7 +858,7 @@ bool CheckpointManager::queueDirty(
         // Note this uses the lastBySeqno for snapStart / End.
         checkOpenCheckpoint_UNLOCKED(lh, /*force*/ true, false);
         openCkpt = &getOpenCheckpoint_UNLOCKED(lh);
-        result = openCkpt->queueDirty(qi, this);
+        result = openCkpt->queueDirty(qi);
         if (result.status != QueueDirtyStatus::SuccessNewItem) {
             throw std::logic_error("CheckpointManager::queueDirty(vb:" +
                                    vbucketId.to_string() +
@@ -940,7 +941,7 @@ void CheckpointManager::queueSetVBState(VBucket& vb) {
     vbstate.toItem(*item);
 
     auto& openCkpt = getOpenCheckpoint_UNLOCKED(lh);
-    const auto result = openCkpt.queueDirty(item, this);
+    const auto result = openCkpt.queueDirty(item);
 
     if (result.status == QueueDirtyStatus::SuccessNewItem) {
         ++numItems;