MB-34910: Track HPS/HCS as member variables of the ADM::State

Instead of deriving the HPS and HCS when they are read, keep
a tracked/cached counter in the ADM::State which is simply
returned when needed.

This is also required to simplify the warmup of the ADM as
we can just assign the pre-warmup value.

Note: highPreparedSeqno is tracked as weak monotonic counter
becase updateHighPreparedSeqno can run and make no change,
rather than detect when no-change occured we just tolerate
assignment of the current value.

Change-Id: I914637548fdfecd419d6492561a4a3319fe88b26
Reviewed-on: http://review.couchbase.org/111863
Reviewed-by: Ben Huddleston <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: jimwwalker

engines/ep/src/durability/active_durability_monitor.cc

@@ -189,6 +189,8 @@ struct ActiveDurabilityMonitor::State {
      */
     void performQueuedAckForChain(const ReplicationChain& chain);
 
+    void updateHighCompletedSeqno();
+
 private:
     /**
      * Advance the current Position (iterator and seqno).
@@ -228,6 +230,12 @@ struct ActiveDurabilityMonitor::State {
     // Stores the last aborted seqno.
     Monotonic<int64_t> lastAbortedSeqno = 0;
 
+    // Stores the highPreparedSeqno
+    WeaklyMonotonic<int64_t> highPreparedSeqno = 0;
+
+    // Stores the highCompletedSeqno
+    Monotonic<int64_t> highCompletedSeqno = 0;
+
     // Cumulative count of accepted (tracked) SyncWrites.
     size_t totalAccepted = 0;
     // Cumulative count of Committed SyncWrites.
@@ -413,12 +421,11 @@ int64_t ActiveDurabilityMonitor::getHighPreparedSeqno() const {
     if (!s->firstChain) {
         return 0;
     }
-    return s->getNodeWriteSeqno(s->getActive());
+    return s->highPreparedSeqno;
 }
 
 int64_t ActiveDurabilityMonitor::getHighCompletedSeqno() const {
-    const auto s = state.rlock();
-    return std::max(s->lastCommittedSeqno, s->lastAbortedSeqno);
+    return state.rlock()->highCompletedSeqno;
 }
 
 bool ActiveDurabilityMonitor::isDurabilityPossible() const {
@@ -932,6 +939,7 @@ void ActiveDurabilityMonitor::commit(const SyncWrite& sw) {
     {
         auto s = state.wlock();
         s->lastCommittedSeqno = sw.getBySeqno();
+        s->updateHighCompletedSeqno();
         s->totalCommitted++;
         // Note:
         // - Level Majority locally-satisfied first at Active by-logic
@@ -965,6 +973,7 @@ void ActiveDurabilityMonitor::abort(const SyncWrite& sw) {
     }
     auto s = state.wlock();
     s->lastAbortedSeqno = sw.getBySeqno();
+    s->updateHighCompletedSeqno();
     s->totalAborted++;
 }
 
@@ -1046,6 +1055,8 @@ void ActiveDurabilityMonitor::toOStream(std::ostream& os) const {
     const auto s = state.rlock();
     os << "ActiveDurabilityMonitor[" << this
        << "] #trackedWrites:" << s->trackedWrites.size()
+       << " highPreparedSeqno:" << s->highPreparedSeqno
+       << " highCompletedSeqno:" << s->highCompletedSeqno
        << " lastTrackedSeqno:" << s->lastTrackedSeqno
        << " lastCommittedSeqno:" << s->lastCommittedSeqno
        << " lastAbortedSeqno:" << s->lastAbortedSeqno << "\n";
@@ -1326,6 +1337,7 @@ void ActiveDurabilityMonitor::State::updateHighPreparedSeqno(
     while ((next = getNodeNext(active)) != trackedWrites.end() &&
            static_cast<uint64_t>(next->getBySeqno()) <=
                    adm.vb.getPersistenceSeqno()) {
+        highPreparedSeqno = next->getBySeqno();
         advanceNodePosition(active);
         removeForCommitIfSatisfied();
     }
@@ -1347,6 +1359,7 @@ void ActiveDurabilityMonitor::State::updateHighPreparedSeqno(
             break;
         }
 
+        highPreparedSeqno = next->getBySeqno();
         advanceNodePosition(active);
         removeForCommitIfSatisfied();
     }
@@ -1355,6 +1368,10 @@ void ActiveDurabilityMonitor::State::updateHighPreparedSeqno(
     //     Position::lastAckSeqno for the local (Active) tracking.
 }
 
+void ActiveDurabilityMonitor::State::updateHighCompletedSeqno() {
+    highCompletedSeqno = std::max(lastCommittedSeqno, lastAbortedSeqno);
+}
+
 void ActiveDurabilityMonitor::checkForCommit() {
     // Identify all SyncWrites which are now committed, transferring them into
     // the completedQueue (under the correct locks).