Skip to content

Commit 60d1ba1

Browse files
dorjesinpodorjesinpo
committed
Fixing bad merge
Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>
1 parent acafdc6 commit 60d1ba1

5 files changed

Lines changed: 48 additions & 54 deletions

File tree

‎src/groups/mqb/mqbraft/mqbraft_partitionraft.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1770,7 +1770,7 @@ unsigned int PartitionRaft::getMessageLenRaw(
17701770
return d_fileStore_sp->getMessageLenRaw(handle);
17711771
}
17721772

1773-
unsigned int PartitionRaft::primaryLeaseId() const
1773+
unsigned int PartitionRaft::writeHeadLeaseId() const
17741774
{
17751775
return static_cast<unsigned int>(d_raftNode_mp->currentTerm());
17761776
}

‎src/groups/mqb/mqbraft/mqbraft_partitionraft.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -497,8 +497,9 @@ class PartitionRaft : public mqbs::RecordStore {
497497
unsigned int getMessageLenRaw(
498498
const mqbs::DataStoreRecordHandle& handle) const BSLS_KEYWORD_OVERRIDE;
499499

500-
/// Return the current primary leaseId for this partition.
501-
unsigned int primaryLeaseId() const BSLS_KEYWORD_OVERRIDE;
500+
/// Return the write-head leaseId for this partition: the lease id of the
501+
/// next record this store writes or applies.
502+
virtual unsigned int writeHeadLeaseId() const BSLS_KEYWORD_OVERRIDE;
502503

503504
/// Return `true` if there was Replication Receipt for the specified
504505
/// `handle`.

‎src/groups/mqb/mqbs/mqbs_datastore.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -854,8 +854,9 @@ class RecordStore {
854854
virtual unsigned int
855855
getMessageLenRaw(const DataStoreRecordHandle& handle) const = 0;
856856

857-
/// Return the current primary leaseId for this partition.
858-
virtual unsigned int primaryLeaseId() const = 0;
857+
/// Return the write-head leaseId for this partition: the lease id of the
858+
/// next record this store writes or applies.
859+
virtual unsigned int writeHeadLeaseId() const = 0;
859860

860861
/// Return `true` if there was Replication Receipt for the specified
861862
/// `handle`.

‎src/groups/mqb/mqbs/mqbs_filestore.cpp‎

Lines changed: 36 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1002,7 +1002,6 @@ int FileStore::openInRecoveryMode(
10021002

10031003
BALL_LOG_INFO << partitionDesc() << "Retrieved PSN: "
10041004
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
1005-
<< ".";
10061005

10071006
// Create file set.
10081007
FileSetSp fileSetSp;
@@ -1108,7 +1107,6 @@ int FileStore::openInRecoveryMode(
11081107
if (0 == writeHeadSeqNum()) {
11091108
BALL_LOG_ERROR << partitionDesc() << "Invalid PSN: "
11101109
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum())
1111-
currentSequenceNumber())
11121110
<< " while primary leaseId is " << "valid ("
11131111
<< d_writeHeadLeaseId << ").";
11141112
return rc_INVALID_SYNC_PT; // RETURN
@@ -1121,7 +1119,7 @@ int FileStore::openInRecoveryMode(
11211119
<< "Internal list of SyncPts is empty, while "
11221120
<< "retrieved PSN is valid: "
11231121
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
1124-
<< ".";
1122+
11251123
return rc_INVALID_SYNC_PT; // RETURN
11261124
}
11271125

@@ -1132,7 +1130,7 @@ int FileStore::openInRecoveryMode(
11321130
<< partitionDesc() << "Invalid leaseId in the last "
11331131
<< "in-memory SyncPt: " << lastSp << ". Retrieved PSN: "
11341132
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
1135-
<< ".";
1133+
11361134
return rc_INVALID_SYNC_PT; // RETURN
11371135
}
11381136

@@ -1142,7 +1140,7 @@ int FileStore::openInRecoveryMode(
11421140
<< partitionDesc() << "Invalid PSN in the last "
11431141
<< "in-memory SyncPt: " << lastSp << ". Retrieved PSN: "
11441142
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
1145-
<< ".";
1143+
11461144
return rc_INVALID_SYNC_PT; // RETURN
11471145
}
11481146
}
@@ -1239,7 +1237,6 @@ int FileStore::openInRecoveryMode(
12391237
BALL_LOG_INFO
12401238
<< partitionDesc() << "Appending a sync point with "
12411239
<< "PSN: " << printPSN(d_writeHeadLeaseId, writeHeadSeqNum() + 1)
1242-
<< printPSN(d_primaryLeaseId, currentSequenceNumber() + 1)
12431240
<< " since journal does not end with a sync point for this "
12441241
<< "partition belonging to a 1-node cluster.";
12451242

@@ -1300,7 +1297,6 @@ int FileStore::openInRecoveryMode(
13001297
<< printPSN(primaryLeaseIdCurr, sequenceNumcurr)
13011298
<< " is higher than storage-retrieved " << "PSN: "
13021299
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum())
1303-
currentSequenceNumber())
13041300
<< ". This is possible if self "
13051301
<< "replica has received primary status advisory "
13061302
<< "before opening FileStore. Thus, we keep current"
@@ -1748,14 +1744,14 @@ int FileStore::recoverMessages(QueueKeyInfoMap* queueKeyInfoMap,
17481744

17491745
// `+1` so that checks in first iteration in the second pass work
17501746
// correctly.
1751-
bsls::Types::Uint64 sequenceNum = currentSequenceNumber() + 1;
1747+
bsls::Types::Uint64 sequenceNum = currentSeqNum + 1;
17521748

17531749
// Have to clear `d_highestSeqNums` for FSM mode that allows recovery to
17541750
// run multiple times (rebuild `d_highestSeqNums` from scratch).
17551751
// In legacy mode, FileStore is recovered at most once.
17561752
d_highestSeqNums.clear();
17571753
if (primaryLeaseId > 0) {
1758-
d_highestSeqNums[primaryLeaseId] = currentSequenceNumber();
1754+
d_highestSeqNums[primaryLeaseId] = currentSeqNum;
17591755
}
17601756

17611757
bsls::Types::Uint64 snapshotSeqNum = 0;
@@ -3309,7 +3305,7 @@ int FileStore::rolloverImpl(bsls::Types::Uint64 timestamp)
33093305
QueueKeyCounterMap queueKeyCounterMap;
33103306
writeRolledOverRecords(newActiveFileSetSp.get(),
33113307
&queueKeyCounterMap,
3312-
currentSequenceNumber());
3308+
writeHeadSeqNum());
33133309

33143310
// Print summary of rolled over queues.
33153311
logRolloverQueueSummary(queueKeyCounterMap);
@@ -4823,8 +4819,8 @@ int FileStore::issueSyncPointInternal(SyncPointType::Enum type,
48234819
// (e.g. when issuing a sync point on behalf of a previous primary).
48244820
int rc = writeSyncPointRecord(syncPoint,
48254821
type,
4826-
d_primaryLeaseId,
4827-
currentSeqNumRef() + 1);
4822+
d_writeHeadLeaseId,
4823+
writeHeadSeqNum() + 1);
48284824
if (0 != rc) {
48294825
BMQTSK_ALARMLOG_ALARM("FILE_IO")
48304826
<< partitionDesc() << "Failed to write sync point: " << syncPoint
@@ -5092,8 +5088,8 @@ int FileStore::writeMessageRecord(const bmqp::StorageHeader& header,
50925088

50935089
BSLS_ASSERT_SAFE(0 < record.d_appDataUnpaddedLen);
50945090

5095-
DataStoreRecordKey key(recHeader.sequenceNumber(),
5096-
recHeader.primaryLeaseId());
5091+
DataStoreRecordKey key(recHeader.primaryLeaseId(),
5092+
recHeader.sequenceNumber());
50975093
DataStoreRecordHandle handle;
50985094
insertDataStoreRecord(&handle, key, record);
50995095

@@ -5227,8 +5223,8 @@ int FileStore::writeQueueCreationRecord(
52275223
DataStoreRecord record(RecordType::e_QUEUE_OP,
52285224
recordOffset,
52295225
queueRecLength);
5230-
DataStoreRecordKey key(recHeader.sequenceNumber(),
5231-
recHeader.primaryLeaseId());
5226+
DataStoreRecordKey key(recHeader.primaryLeaseId(),
5227+
recHeader.sequenceNumber());
52325228

52335229
insertDataStoreRecord(&handle, key, record);
52345230

@@ -5428,8 +5424,8 @@ int FileStore::writeJournalRecord(const bmqp::StorageHeader& header,
54285424
BSLS_ASSERT_SAFE(queueKey);
54295425
BSLS_ASSERT_SAFE(rstorage);
54305426

5431-
DataStoreRecordKey key(recHeader.sequenceNumber(),
5432-
recHeader.primaryLeaseId());
5427+
DataStoreRecordKey key(recHeader.primaryLeaseId(),
5428+
recHeader.sequenceNumber());
54335429
DataStoreRecord record(recordType, recordOffset);
54345430
DataStoreRecordHandle handle;
54355431
insertDataStoreRecord(&handle, key, record);
@@ -5580,7 +5576,6 @@ int FileStore::writeJournalRecord(const bmqp::StorageHeader& header,
55805576
<< "down. No further storage events will be processed by "
55815577
<< "self. Current PSN: "
55825578
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
5583-
<< ".";
55845579
}
55855580
}
55865581
}
@@ -5612,8 +5607,8 @@ int FileStore::writeJournalRecord(const bmqp::StorageHeader& header,
56125607
<< "], queueKey [" << *queueKey << "], appKey ["
56135608
<< *appKey << "].";
56145609

5615-
DataStoreRecordKey key(recHeader.sequenceNumber(),
5616-
recHeader.primaryLeaseId());
5610+
DataStoreRecordKey key(recHeader.primaryLeaseId(),
5611+
recHeader.sequenceNumber());
56175612
DataStoreRecord record(recordType, recordOffset);
56185613
DataStoreRecordHandle handle;
56195614
insertDataStoreRecord(&handle, key, record);
@@ -5747,7 +5742,7 @@ void FileStore::replicateRecord(bmqp::StorageMessageType::Enum type,
57475742
<< FileStoreProtocol::k_JOURNAL_RECORD_SIZE
57485743
<< ", at JOURNAL offset: " << journalOffset << ", rc: " << buildRc
57495744
<< ". PSN was: " << printPSN(d_writeHeadLeaseId, writeHeadSeqNum())
5750-
<< printPSN(d_primaryLeaseId, currentSequenceNumber())
5745+
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum())
57515746
<< ". Current storage "
57525747
<< "event builder size: " << d_storageEventBuilder.eventSize()
57535748
<< ", message count: " << d_storageEventBuilder.messageCount()
@@ -6246,7 +6241,7 @@ int FileStore::openForRaft(bsl::deque<RecoveryRecordInfo>* recoveryIndex)
62466241
fs->d_journal.d_outstandingBytes,
62476242
fs->d_data.d_filePosition,
62486243
fs->d_journal.d_filePosition,
6249-
currentSequenceNumber());
6244+
writeHeadSeqNum());
62506245

62516246
BALL_LOG_INFO << partitionDesc() << "Raft recovery index: "
62526247
<< recoveryIndex->size() << " records.";
@@ -6485,8 +6480,8 @@ int FileStore::writeMessageRecord(mqbi::StorageMessageAttributes* attributes,
64856480

64866481
// Delegate the actual mmap formatting to formatMessageRecord.
64876482
PendingWrite pw(attributes, guid, appData, options, queueKey);
6488-
pw.d_primaryLeaseId = d_primaryLeaseId;
6489-
pw.d_sequenceNumber = ++currentSeqNumRef();
6483+
pw.d_primaryLeaseId = d_writeHeadLeaseId;
6484+
pw.d_sequenceNumber = incrementWriteHeadSeqNum();
64906485

64916486
int formatRc = formatMessageRecord(&pw);
64926487
if (0 != formatRc) {
@@ -6498,7 +6493,7 @@ int FileStore::writeMessageRecord(mqbi::StorageMessageAttributes* attributes,
64986493
// Legacy replication and receipt tracking.
64996494
int flags = 0;
65006495
if (!attributes->hasReceipt()) {
6501-
DataStoreRecordKey key(pw.d_sequenceNumber, pw.d_primaryLeaseId);
6496+
DataStoreRecordKey key(pw.d_primaryLeaseId, pw.d_sequenceNumber);
65026497
RecordIterator it = handleTorRecordIterator(*handle);
65036498
BSLS_ASSERT_SAFE(it != d_records.end());
65046499
d_unreceipted.insert(
@@ -6745,8 +6740,8 @@ int FileStore::writeQueueCreationRecord(DataStoreRecordHandle* handle,
67456740
OffsetPtr<QueueOpRecord> queueOpRec(journal.block(), journalPos);
67466741
new (queueOpRec.get()) QueueOpRecord();
67476742
queueOpRec->header()
6748-
.setPrimaryLeaseId(d_primaryLeaseId)
6749-
.setSequenceNumber(++currentSeqNumRef())
6743+
.setPrimaryLeaseId(d_writeHeadLeaseId)
6744+
.setSequenceNumber(incrementWriteHeadSeqNum())
67506745
.setTimestamp(timestamp);
67516746
queueOpRec->setQueueKey(queueKey).setType(
67526747
isNewQueue ? QueueOpType::e_CREATION : QueueOpType::e_ADDITION);
@@ -6767,7 +6762,7 @@ int FileStore::writeQueueCreationRecord(DataStoreRecordHandle* handle,
67676762
qlistOffset,
67686763
qlistRecTotalLength);
67696764

6770-
DataStoreRecordKey key(currentSequenceNumber(), d_primaryLeaseId);
6765+
DataStoreRecordKey key(d_writeHeadLeaseId, writeHeadSeqNum());
67716766
DataStoreRecord record(RecordType::e_QUEUE_OP,
67726767
recordOffset,
67736768
qlistRecTotalLength);
@@ -6926,7 +6921,7 @@ int FileStore::formatQueueCreationRecord(PendingWrite* pw)
69266921
queueOpRec->setMagic(RecordHeader::k_MAGIC);
69276922
journalPos += FileStoreProtocol::k_JOURNAL_RECORD_SIZE;
69286923

6929-
DataStoreRecordKey key(pw->d_sequenceNumber, pw->d_primaryLeaseId);
6924+
DataStoreRecordKey key(pw->d_primaryLeaseId, pw->d_sequenceNumber);
69306925
DataStoreRecord record(RecordType::e_QUEUE_OP,
69316926
recordOffset,
69326927
qlistRecTotalLength);
@@ -7188,7 +7183,7 @@ int FileStore::writeSyncPointRecord(const bmqp_ctrlmsg::SyncPoint& syncPoint,
71887183
RecordHeader::k_MAGIC);
71897184
journalOpRec->header()
71907185
.setPrimaryLeaseId(primaryLeaseId)
7191-
.setSequenceNumber(writeHeadSeqNum())
7186+
.setSequenceNumber(sequenceNumber)
71927187
.setTimestamp(
71937188
bdlt::EpochUtil::convertToTimeT64(bdlt::CurrentTime::utc()));
71947189
journalPos += FileStoreProtocol::k_JOURNAL_RECORD_SIZE;
@@ -7198,8 +7193,7 @@ int FileStore::writeSyncPointRecord(const bmqp_ctrlmsg::SyncPoint& syncPoint,
71987193

71997194
BALL_LOG_INFO << partitionDesc() << "Wrote SyncPointRecord to journal"
72007195
<< " [type: " << type << ", syncPoint: " << syncPoint
7201-
<< ", PSN: "
7202-
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum())
7196+
<< ", PSN: " << printPSN(primaryLeaseId, sequenceNumber)
72037197
<< ", journal offset: " << recordOffset << "]";
72047198

72057199
return rc_SUCCESS;
@@ -7302,7 +7296,7 @@ int FileStore::formatConfirmRecord(PendingWrite* pw)
73027296
confRec->setMagic(RecordHeader::k_MAGIC);
73037297
journalPos += FileStoreProtocol::k_JOURNAL_RECORD_SIZE;
73047298

7305-
DataStoreRecordKey key(pw->d_sequenceNumber, pw->d_primaryLeaseId);
7299+
DataStoreRecordKey key(pw->d_primaryLeaseId, pw->d_sequenceNumber);
73067300
DataStoreRecord record(RecordType::e_CONFIRM, recordOffset);
73077301
bindOrUpdateRecord(pw, key, record);
73087302

@@ -7418,7 +7412,7 @@ int FileStore::formatQueuePurgeRecord(PendingWrite* pw)
74187412
qRec->setMagic(RecordHeader::k_MAGIC);
74197413
journalPos += FileStoreProtocol::k_JOURNAL_RECORD_SIZE;
74207414

7421-
DataStoreRecordKey key(pw->d_sequenceNumber, pw->d_primaryLeaseId);
7415+
DataStoreRecordKey key(pw->d_primaryLeaseId, pw->d_sequenceNumber);
74227416
DataStoreRecord record(RecordType::e_QUEUE_OP, recordOffset);
74237417
bindOrUpdateRecord(pw, key, record);
74247418

@@ -7475,7 +7469,7 @@ int FileStore::formatQueueDeletionRecord(PendingWrite* pw)
74757469
qRec->setMagic(RecordHeader::k_MAGIC);
74767470
journalPos += FileStoreProtocol::k_JOURNAL_RECORD_SIZE;
74777471

7478-
DataStoreRecordKey key(pw->d_sequenceNumber, pw->d_primaryLeaseId);
7472+
DataStoreRecordKey key(pw->d_primaryLeaseId, pw->d_sequenceNumber);
74797473
DataStoreRecord record(RecordType::e_QUEUE_OP, recordOffset);
74807474
bindOrUpdateRecord(pw, key, record);
74817475

@@ -7517,7 +7511,7 @@ void FileStore::reservePendingRecord(PendingWrite* pw,
75177511
// (still zero) offset before the write drains -- tripping the offset-0
75187512
// assertion in 'loadMessageAttributesRaw'. The receipt is (re)opened when
75197513
// the drained record commits ('onRecordCommittedPrimary').
7520-
DataStoreRecordKey key(sequenceNumber, primaryLeaseId);
7514+
DataStoreRecordKey key(primaryLeaseId, sequenceNumber);
75217515
DataStoreRecord record(recordType);
75227516

75237517
record.d_messagePropertiesInfo = pw->d_attributes.messagePropertiesInfo();
@@ -7705,8 +7699,8 @@ int FileStore::writeFormattedRecord(const bdlbb::Blob& data,
77057699

77067700
// Phase 3: type-specific d_records + outstanding bytes.
77077701
if (needsRecord) {
7708-
DataStoreRecordKey key(recHeader->sequenceNumber(),
7709-
recHeader->primaryLeaseId());
7702+
DataStoreRecordKey key(recHeader->primaryLeaseId(),
7703+
recHeader->sequenceNumber());
77107704
DataStoreRecord dsRecord(recHeader->type(), info->d_journalOffset);
77117705
// dsRecord.d_hasReceipt defaults to true;
77127706

@@ -7845,7 +7839,7 @@ int FileStore::formatMessageRecord(PendingWrite* pw)
78457839
.setMagic(RecordHeader::k_MAGIC);
78467840
journalPos += k_JREC_SIZE;
78477841

7848-
DataStoreRecordKey key(pw->d_sequenceNumber, pw->d_primaryLeaseId);
7842+
DataStoreRecordKey key(pw->d_primaryLeaseId, pw->d_sequenceNumber);
78497843
DataStoreRecord record(RecordType::e_MESSAGE, journalOffset);
78507844
record.d_messageOffset = dataOffset;
78517845
record.d_appDataUnpaddedLen = pw->d_appData->length();
@@ -8109,7 +8103,7 @@ int FileStore::lookupRecord(bsls::Types::Uint64* journalOffset,
81098103
// Legacy-only: never invoked on a Raft partition.
81108104
BSLS_ASSERT(!isRaft());
81118105

8112-
DataStoreRecordKey key(sequenceNumber, primaryLeaseId);
8106+
DataStoreRecordKey key(primaryLeaseId, sequenceNumber);
81138107
RecordConstIterator it = d_records.find(key);
81148108
if (it == d_records.end()) {
81158109
return -1;
@@ -8675,7 +8669,6 @@ void FileStore::processStorageEvent(const bsl::shared_ptr<bdlbb::Blob>& blob,
86758669
<< " while processing a "
86768670
<< "partition-sync storage message. " << "New PSN: "
86778671
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ")";
8678-
<< ")";
86798672
}
86808673
}
86818674
else {
@@ -8822,7 +8815,6 @@ int FileStore::processRecoveryEvent(const bsl::shared_ptr<bdlbb::Blob>& blob)
88228815
<< " with missing sequence number. "
88238816
<< "Expected PSN: "
88248817
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum() + 1)
8825-
currentSequenceNumber() + 1)
88268818
<< ", received PSN: "
88278819
<< printPSN(recHeader->primaryLeaseId(),
88288820
recHeader->sequenceNumber())
@@ -9102,7 +9094,6 @@ void FileStore::setActivePrimary(mqbnet::ClusterNode* primaryNode,
91029094
BALL_LOG_INFO << partitionDesc() << "Primary node is now "
91039095
<< primaryNode->nodeDescription() << " with PSN: "
91049096
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
9105-
<< ".";
91069097

91079098
if (primaryNode->nodeId() != d_config.nodeId()) {
91089099
d_isPrimary = false;
@@ -9333,7 +9324,7 @@ void FileStore::clearPrimary()
93339324
BALL_LOG_INFO << partitionDesc() << "Clearing current primary: "
93349325
<< d_primaryNode_p->nodeDescription() << ". Current PSN: "
93359326
<< printPSN(d_writeHeadLeaseId, writeHeadSeqNum()) << ".";
9336-
<< ".";
9327+
93379328
d_primaryNode_p = 0;
93389329

93399330
// If self has a valid leaseId and zero sequence number (ie, previous

‎src/groups/mqb/mqbs/mqbs_filestore.t.cpp‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -279,6 +279,7 @@ class Tester {
279279
1, // numPartitions
280280
d_clusterStatsRootContext_sp.get(),
281281
d_allocator_p);
282+
mqbs::StorageMonitor* storageMonitor = 0;
282283
d_fs_mp = bslma::ManagedPtrUtil::allocateManaged<mqbs::FileStore>(
283284
d_allocator_p,
284285
d_dsCfg,
@@ -289,10 +290,10 @@ class Tester {
289290
d_blobSpPool_sp.get(),
290291
&d_statePool,
291292
&d_miscWorkThreadPool,
292-
true, // isFSMWorkflow
293-
true, // doesFSMwriteQLIST
294-
1); // replicationFactor
295-
0, // storageMonitor
293+
true, // isFSMWorkflow
294+
true, // doesFSMwriteQLIST
295+
1, // replicationFactor
296+
storageMonitor); // storageMonitor
296297

297298
// To pass `inDispatcherThread` checks:
298299
d_fs_mp->setThreadId(bslmt::ThreadUtil::selfId());

0 commit comments

Comments
 (0)