Refactor[mqbs::FileStore]: Encapsulate write-head cursor accessors

src/groups/mqb/mqbblp/mqbblp_recoverymanager.cpp

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 #include <bsls_assert.h>
 #include <mqbblp_recoverymanager.h>
 #include <mqbs_filestoreprintutil.h>
 
@@ -2011,7 +2011,7 @@
 
     bmqp_ctrlmsg::PartitionSequenceNumber selfSequenceNum;
     selfSequenceNum.primaryLeaseId() = fs->writeHeadLeaseId();
-    selfSequenceNum.sequenceNumber() = fs->sequenceNumber();
+    selfSequenceNum.sequenceNumber() = fs->writeHeadSeqNum();
 
     const FileTransferInfo& fti = primarySyncCtx->fileTransferInfo();
 
@@ -4373,7 +4373,7 @@
 
     bmqp_ctrlmsg::PartitionSequenceNumber tmp;
     tmp.primaryLeaseId() = fs->writeHeadLeaseId();
-    tmp.sequenceNumber() = fs->sequenceNumber();
+    tmp.sequenceNumber() = fs->writeHeadSeqNum();
     primarySyncCtx.setSelfPartitionSequenceNum(tmp);
 
     if (!fs->syncPoints().empty()) {
@@ -4469,7 +4469,7 @@
 
     response.partitionId()    = req.partitionId();
     response.primaryLeaseId() = fs->writeHeadLeaseId();
-    response.sequenceNum()    = fs->sequenceNumber();
+    response.sequenceNum()    = fs->writeHeadSeqNum();
     if (!fs->syncPoints().empty()) {
         response.lastSyncPointOffsetPair() = fs->syncPoints().back();
     }
@@ -4578,7 +4578,7 @@
 
     bmqp_ctrlmsg::PartitionSequenceNumber selfPSN;
     selfPSN.primaryLeaseId() = fs->writeHeadLeaseId();
-    selfPSN.sequenceNumber() = fs->sequenceNumber();
+    selfPSN.sequenceNumber() = fs->writeHeadSeqNum();
 
     if (requesterUptoPSN <= requesterPSN) {
         BALL_LOG_WARN << d_clusterData_p->identity().description()

src/groups/mqb/mqbblp/mqbblp_storagemanager.cpp

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 #include <ball_log.h>
 #include <mqbblp_storagemanager.h>
 
 #include <mqbscm_version.h>
@@ -364,7 +364,7 @@
                                      : "**none**")
                     << ", "
                     << mqbs::printPSN(fs->writeHeadLeaseId(),
-                                      fs->sequenceNumber())
+                                      fs->writeHeadSeqNum())
                     << ")";
             }
         }

src/groups/mqb/mqbc/mqbc_recoverymanager.cpp

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 #include <ball_log.h>
 #include <bsls_assert.h>
 #include <mqbc_partitionfsm.h>
 #include <mqbc_recoverymanager.h>
@@ -232,7 +232,7 @@
         BSLS_ASSERT_SAFE(receiveDataCtx.d_currPSN.primaryLeaseId() ==
                          fs.writeHeadLeaseId());
         BSLS_ASSERT_SAFE(receiveDataCtx.d_currPSN.sequenceNumber() ==
-                         fs.sequenceNumber());
+                         fs.writeHeadSeqNum());
     }
     else {
         BSLS_ASSERT_SAFE(recoveryCtx.d_mappedJournalFd.isValid() &&
@@ -646,12 +646,12 @@
         BSLS_ASSERT_SAFE(receiveDataCtx.d_currPSN.primaryLeaseId() ==
                          fs->writeHeadLeaseId());
         BSLS_ASSERT_SAFE(receiveDataCtx.d_currPSN.sequenceNumber() ==
-                         fs->sequenceNumber());
+                         fs->writeHeadSeqNum());
 
         fs->processStorageEvent(blob, true /* isPartitionSyncEvent */, source);
 
         receiveDataCtx.d_currPSN.primaryLeaseId() = fs->writeHeadLeaseId();
-        receiveDataCtx.d_currPSN.sequenceNumber() = fs->sequenceNumber();
+        receiveDataCtx.d_currPSN.sequenceNumber() = fs->writeHeadSeqNum();
 
         if (receiveDataCtx.d_currPSN == receiveDataCtx.d_endPSN) {
             receiveDataCtx.d_expectChunks = false;

src/groups/mqb/mqbc/mqbc_storagemanager.cpp

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 #include <ball_log.h>
 #include <ball_logthrottle.h>
 #include <bsls_assert.h>
 #include <mqbc_storagemanager.h>
@@ -1925,7 +1925,7 @@
     BSLS_ASSERT_SAFE(fs);
     if (fs->isOpen()) {
         nodePSNCtx.d_PSN.primaryLeaseId() = fs->writeHeadLeaseId();
-        nodePSNCtx.d_PSN.sequenceNumber() = fs->sequenceNumber();
+        nodePSNCtx.d_PSN.sequenceNumber() = fs->writeHeadSeqNum();
     }
     else {
         const int rc = d_recoveryManager_mp->recoverPSN(&nodePSNCtx.d_PSN,

src/groups/mqb/mqbc/mqbc_storagemanager.t.cpp

@@ -2924,7 +2924,7 @@ static void test18_primaryHealingWatchdogRetry()
     const int k_PRIMARY_LEASE_ID =
         storageManager.fileStore(k_PARTITION_ID).writeHeadLeaseId();
     const int k_PRIMARY_SEQ_NUM =
-        storageManager.fileStore(k_PARTITION_ID).sequenceNumber();
+        storageManager.fileStore(k_PARTITION_ID).writeHeadSeqNum();
 
     static const int             k_REQUEST_ID = 1;
     bmqp_ctrlmsg::ControlMessage message;
@@ -3555,7 +3555,7 @@ static void test23_replicaHealingReceivesReplicaDataRqstDropInvalidPid()
 
     // 5. Send a storage event (PUT) and verify it is buffered, not processed
     const bsls::Types::Uint64 seqNumBefore =
-        storageManager.fileStore(k_PARTITION_ID).sequenceNumber();
+        storageManager.fileStore(k_PARTITION_ID).writeHeadSeqNum();
 
     bmqp::StorageEventBuilder seb(mqbs::FileStoreProtocol::k_VERSION,
                                   bmqp::EventType::e_STORAGE,
@@ -3593,8 +3593,9 @@ static void test23_replicaHealingReceivesReplicaDataRqstDropInvalidPid()
 
     // Sequence number has not advanced; this proves that we did not process
     // the PUT.
-    BMQTST_ASSERT_EQ(storageManager.fileStore(k_PARTITION_ID).sequenceNumber(),
-                     seqNumBefore);
+    BMQTST_ASSERT_EQ(
+        storageManager.fileStore(k_PARTITION_ID).writeHeadSeqNum(),
+        seqNumBefore);
 
     BMQTST_ASSERT_EQ(storageManager.partitionHealthState(k_PARTITION_ID),
                      mqbc::PartitionFSM::State::e_REPLICA_HEALING);

src/groups/mqb/mqbc/mqbc_storageutil.cpp

@@ -2134,8 +2134,8 @@ void StorageUtil::recoveredQueuesCb(
                 queueKey);
             BSLS_ASSERT_SAFE(infoMapCit != queueKeyInfoMap->end());
 
-            const mqbs::DataStoreRecordKey current(handle.sequenceNum(),
-                                                   handle.primaryLeaseId());
+            const mqbs::DataStoreRecordKey current(handle.primaryLeaseId(),
+                                                   handle.sequenceNum());
 
             const unsigned int numGhosts = infoMapCit->second.advanceAndCount(
                 current);
@@ -3735,7 +3735,7 @@ void StorageUtil::forceIssueAdvisoryAndSyncPt(mqbc::ClusterData*   clusterData,
                       << fs->config().partitionId()
                       << "]: successfully issued a forced SyncPt: "
                       << mqbs::printPSN(fs->writeHeadLeaseId(),
-                                        fs->sequenceNumber())
+                                        fs->writeHeadSeqNum())
                       << ".";
     }
     else {
@@ -3744,7 +3744,7 @@ void StorageUtil::forceIssueAdvisoryAndSyncPt(mqbc::ClusterData*   clusterData,
                        << "]: failed to force-issue SyncPt, rc: " << rc
                        << ", current PSN: "
                        << mqbs::printPSN(fs->writeHeadLeaseId(),
-                                         fs->sequenceNumber());
+                                         fs->writeHeadSeqNum());
     }
 }
 

src/groups/mqb/mqbs/mqbs_datastore.cpp

@@ -176,8 +176,8 @@ bsl::ostream& DataStoreRecordKey::print(bsl::ostream& stream,
 
     bslim::Printer printer(&stream, level, spacesPerLevel);
     printer.start();
-    printer.printAttribute("sequenceNum", d_sequenceNum);
     printer.printAttribute("primaryLeaseId", d_primaryLeaseId);
+    printer.printAttribute("sequenceNum", d_sequenceNum);
     printer.end();
 
     return stream;

src/groups/mqb/mqbs/mqbs_datastore.h

@@ -192,8 +192,8 @@ struct DataStoreRecordKey {
     // CREATORS
     DataStoreRecordKey();
 
-    DataStoreRecordKey(const bsls::Types::Uint64 sequenceNum,
-                       unsigned int              primaryLeaseId);
+    DataStoreRecordKey(unsigned int        primaryLeaseId,
+                       bsls::Types::Uint64 sequenceNum);
 
     // ACCESSORS
 
@@ -837,9 +837,8 @@ inline DataStoreRecordKey::DataStoreRecordKey()
     // NOTHING
 }
 
-inline DataStoreRecordKey::DataStoreRecordKey(
-    const bsls::Types::Uint64 sequenceNum,
-    unsigned int              primaryLeaseId)
+inline DataStoreRecordKey::DataStoreRecordKey(unsigned int primaryLeaseId,
+                                              bsls::Types::Uint64 sequenceNum)
 : d_sequenceNum(sequenceNum)
 , d_primaryLeaseId(primaryLeaseId)
 {
@@ -851,8 +850,8 @@ template <class HASH_ALGORITHM>
 void hashAppend(HASH_ALGORITHM& hashAlgo, const mqbs::DataStoreRecordKey& key)
 {
     using bslh::hashAppend;  // for ADL
-    hashAppend(hashAlgo, key.d_sequenceNum);
     hashAppend(hashAlgo, key.d_primaryLeaseId);
+    hashAppend(hashAlgo, key.d_sequenceNum);
 }
 
 // --------------------------------
@@ -864,8 +863,8 @@ template <class TYPE>
 inline DataStoreRecordKeyHashAlgo::result_type
 DataStoreRecordKeyHashAlgo::operator()(const TYPE& type) const
 {
-    return type.d_sequenceNum +
-           (static_cast<bsls::Types::Uint64>(type.d_primaryLeaseId) << 32);
+    return (static_cast<bsls::Types::Uint64>(type.d_primaryLeaseId) << 32) +
+           type.d_sequenceNum;
 }
 
 // -----------------------------

src/groups/mqb/mqbs/mqbs_datastore.t.cpp

@@ -50,7 +50,7 @@
                                                 index % k_KEYS_PER_LEASE) +
                                             1;
 
-    return mqbs::DataStoreRecordKey(sequenceNum, primaryLeaseId);
+    return mqbs::DataStoreRecordKey(primaryLeaseId, sequenceNum);
 }
 
 static void test1_breathingTest()
@@ -78,7 +78,7 @@
         BMQTST_ASSERT_EQ(keyDefault.d_primaryLeaseId, 0U);
 
         // Valued constructor
-        mqbs::DataStoreRecordKey keyValued(k_SEQUENCE_NUM, k_PRIMARY_LEASE_ID);
+        mqbs::DataStoreRecordKey keyValued(k_PRIMARY_LEASE_ID, k_SEQUENCE_NUM);
         BMQTST_ASSERT_EQ(keyValued.d_sequenceNum, k_SEQUENCE_NUM);
         BMQTST_ASSERT_EQ(keyValued.d_primaryLeaseId, k_PRIMARY_LEASE_ID);
     }
@@ -324,7 +324,7 @@
 
     bsls::Types::Int64 begin = bsls::TimeUtil::getTimer();
     for (size_t i = 0; i < k_NUM_ITERATIONS; ++i) {
-        hasher(mqbs::DataStoreRecordKey(i, 7));
+        hasher(mqbs::DataStoreRecordKey(7, i));
     }
     bsls::Types::Int64 end = bsls::TimeUtil::getTimer();
 
@@ -357,7 +357,7 @@
 
     bsls::Types::Int64 begin = bsls::TimeUtil::getTimer();
     for (size_t i = 0; i < k_NUM_ITERATIONS; ++i) {
-        hasher(mqbs::DataStoreRecordKey(i, 7));
+        hasher(mqbs::DataStoreRecordKey(7, i));
     }
     bsls::Types::Int64 end = bsls::TimeUtil::getTimer();
 
@@ -390,15 +390,15 @@
         k_NUM_ELEMS,
         bmqtst::TestHelperUtil::allocator());
     // Warmup
     for (size_t i = 1; i <= 1000; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
 
     ht.clear();
 
     bsls::Types::Int64 begin = bsls::TimeUtil::getTimer();
     for (size_t i = 1; i <= k_NUM_ELEMS; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
     bsls::Types::Int64 end = bsls::TimeUtil::getTimer();
 
@@ -434,15 +434,15 @@
         ht(k_NUM_ELEMS, bmqtst::TestHelperUtil::allocator());
 
     // Warmup
     for (size_t i = 1; i <= 1000; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
 
     ht.clear();
 
     bsls::Types::Int64 begin = bsls::TimeUtil::getTimer();
     for (size_t i = 1; i <= k_NUM_ELEMS; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
     bsls::Types::Int64 end = bsls::TimeUtil::getTimer();
 
@@ -479,7 +479,7 @@
     // same as: bslh::Hash<> hasher;
     for (auto _ : state) {
         for (int i = 0; i < state.range(); ++i) {
-            hasher(mqbs::DataStoreRecordKey(i, 7));
+            hasher(mqbs::DataStoreRecordKey(7, i));
         }
     }
 }
@@ -503,7 +503,7 @@
 
     for (auto _ : state) {
         for (int i = 0; i < state.range(0); ++i) {
-            hasher(mqbs::DataStoreRecordKey(i, 7));
+            hasher(mqbs::DataStoreRecordKey(7, i));
         }
     }
 }
@@ -529,14 +529,14 @@
         bmqtst::TestHelperUtil::allocator());
     // Warmup
     for (size_t i = 1; i <= 1000; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
 
     ht.clear();
 
     for (auto _ : state) {
         for (int i = 1; i <= state.range(0); ++i) {
-            ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+            ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
         }
     }
 }
@@ -564,13 +564,13 @@
 
     // Warmup
     for (size_t i = 1; i <= 1000; ++i) {
-        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+        ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
     }
 
     ht.clear();
     for (auto _ : state) {
         for (int i = 1; i <= state.range(0); ++i) {
-            ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(i, 7), i));
+            ht.insert(bsl::make_pair(mqbs::DataStoreRecordKey(7, i), i));
         }
     }
 }

src/groups/mqb/mqbs/mqbs_filebackedstorage.t.cpp

@@ -275,7 +275,7 @@ class MockDataStore : public mqbs::DataStore {
             reinterpret_cast<mqbs::DataStoreConfig::RecordIterator*>(handle);
 
         InsertRc insertRc = d_records.insert(bsl::make_pair(
-            mqbs::DataStoreRecordKey(sequenceNum, primaryLeaseId),
+            mqbs::DataStoreRecordKey(primaryLeaseId, sequenceNum),
             mqbs::DataStoreRecord(recType, recOffset)));
 
         // Needed for FileBackedStorage::gcExpiredMessages
@@ -316,7 +316,7 @@ class MockDataStore : public mqbs::DataStore {
             reinterpret_cast<mqbs::DataStoreConfig::RecordIterator*>(handle);
 
         InsertRc insertRc = d_records.insert(bsl::make_pair(
-            mqbs::DataStoreRecordKey(sequenceNum, primaryLeaseId),
+            mqbs::DataStoreRecordKey(primaryLeaseId, sequenceNum),
             mqbs::DataStoreRecord(recType, recOffset)));
 
         *iter = insertRc.first;
@@ -860,7 +860,7 @@ BMQTST_TEST_F(Test, supportedOperations)
 
     {
         const bsls::Types::Uint64    k_SEQUENCE_NUM = 1024;
-        mqbs::DataStoreRecordKey     key(k_SEQUENCE_NUM, k_PRIMARY_LEASE_ID);
+        mqbs::DataStoreRecordKey     key(k_PRIMARY_LEASE_ID, k_SEQUENCE_NUM);
         const mqbs::RecordType::Enum k_RECORD_TYPE =
             mqbs::RecordType::e_MESSAGE;
         mqbs::DataStoreRecord       record(k_RECORD_TYPE, k_RECORD_OFFSET);
@@ -873,7 +873,7 @@ BMQTST_TEST_F(Test, supportedOperations)
 
     {
         const bsls::Types::Uint64    k_SEQUENCE_NUM = 1025;
-        mqbs::DataStoreRecordKey     key(k_SEQUENCE_NUM, k_PRIMARY_LEASE_ID);
+        mqbs::DataStoreRecordKey     key(k_PRIMARY_LEASE_ID, k_SEQUENCE_NUM);
         const mqbs::RecordType::Enum k_RECORD_TYPE =
             mqbs::RecordType::e_CONFIRM;
         mqbs::DataStoreRecord       record(k_RECORD_TYPE, k_RECORD_OFFSET);
@@ -889,7 +889,7 @@ BMQTST_TEST_F(Test, supportedOperations)
 
     {
         const bsls::Types::Uint64    k_SEQUENCE_NUM = 1026;
-        mqbs::DataStoreRecordKey     key(k_SEQUENCE_NUM, k_PRIMARY_LEASE_ID);
+        mqbs::DataStoreRecordKey     key(k_PRIMARY_LEASE_ID, k_SEQUENCE_NUM);
         const mqbs::RecordType::Enum k_RECORD_TYPE =
             mqbs::RecordType::e_DELETION;
         mqbs::DataStoreRecord       record(k_RECORD_TYPE, k_RECORD_OFFSET);
@@ -1865,7 +1865,7 @@ BMQTST_TEST_F(Test, addQueueOpRecordHandle)
     const mqbs::RecordType::Enum k_RECORD_TYPE = mqbs::RecordType::e_QUEUE_OP;
     const bsls::Types::Uint64    k_RECORD_OFFSET = 4096;
 
-    mqbs::DataStoreRecordKey    key(k_SEQUENCE_NUM, k_PRIMARY_LEASE_ID);
+    mqbs::DataStoreRecordKey    key(k_PRIMARY_LEASE_ID, k_SEQUENCE_NUM);
     mqbs::DataStoreRecord       record(k_RECORD_TYPE, k_RECORD_OFFSET);
     mqbs::DataStoreRecordHandle handle;
     d_tester.insertDataStoreRecord(&handle, key, record);