Skip to content

Commit b2c29cb

Browse files
alexander-e1offalexander-e1off
authored
Feat[mqbc, mqbs] FSM mode: Enhance rollover to dynamically change partition max file sizes (#1060)
Signed-off-by: Aleksandr Ivanov <aivanov71@bloomberg.net>
1 parent 8c0aac2 commit b2c29cb

20 files changed

Lines changed: 1269 additions & 288 deletions

‎src/applications/bmqstoragetool/m_bmqstoragetool_journalfile.cpp‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,7 @@ JournalFile::makeJournalOpRecord(unsigned int primaryLeaseId,
113113
bsls::Types::Uint64 sequenceNumber)
114114
{
115115
OffsetPtr<JournalOpRecord> rec(d_block, d_currPos);
116-
new (rec.get()) JournalOpRecord(JournalOpType::e_SYNCPOINT,
117-
SyncPointType::e_REGULAR,
116+
new (rec.get()) JournalOpRecord(SyncPointType::e_REGULAR,
118117
1234567, // seqNum
119118
25, // leaderTerm
120119
121, // leaderNodeId

‎src/applications/bmqstoragetool/m_bmqstoragetool_printer.cpp‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -224,9 +224,11 @@ void printJournalFileMeta(bsl::ostream& ostream,
224224
}
225225
p << epochValue;
226226

227-
p << syncPt.sequenceNum() << syncPt.primaryNodeId()
228-
<< syncPt.primaryLeaseId() << syncPt.dataFileOffsetDwords()
229-
<< syncPt.qlistFileOffsetWords();
227+
const mqbs::JournalOpRecord::SyncPointData& spd =
228+
syncPt.syncPointData();
229+
p << spd.sequenceNum() << spd.primaryNodeId()
230+
<< spd.primaryLeaseId() << spd.dataFileOffsetDwords()
231+
<< spd.qlistFileOffsetWords();
230232
}
231233
}
232234
printer << s.str();

‎src/applications/bmqstoragetool/m_bmqstoragetool_recordprinter.h‎

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -270,16 +270,27 @@ void RecordDetailsPrinter<PRINTER_TYPE>::printRecord(
270270
const RecordDetails<mqbs::JournalOpRecord>& rec)
271271
{
272272
d_fields.push_back("JournalOpType");
273-
d_fields.push_back("SyncPointType");
274-
d_fields.push_back("SyncPtPrimaryLeaseId");
275-
d_fields.push_back("SyncPtSequenceNumber");
276-
d_fields.push_back("PrimaryNodeId");
277-
d_fields.push_back("DataFileOffsetDwords");
278-
279-
*d_printer_mp << rec.d_record.type() << rec.d_record.syncPointType()
280-
<< rec.d_record.primaryLeaseId()
281-
<< rec.d_record.sequenceNum() << rec.d_record.primaryNodeId()
282-
<< rec.d_record.dataFileOffsetDwords();
273+
if (mqbs::JournalOpType::e_SYNCPOINT == rec.d_record.type()) {
274+
d_fields.push_back("SyncPointType");
275+
d_fields.push_back("SyncPtPrimaryLeaseId");
276+
d_fields.push_back("SyncPtSequenceNumber");
277+
d_fields.push_back("PrimaryNodeId");
278+
d_fields.push_back("DataFileOffsetDwords");
279+
const mqbs::JournalOpRecord::SyncPointData& spd =
280+
rec.d_record.syncPointData();
281+
*d_printer_mp << rec.d_record.type() << rec.d_record.syncPointType()
282+
<< spd.primaryLeaseId() << spd.sequenceNum()
283+
<< spd.primaryNodeId() << spd.dataFileOffsetDwords();
284+
}
285+
else if (mqbs::JournalOpType::e_RESIZE_STORAGE == rec.d_record.type()) {
286+
d_fields.push_back("MaxJournalFileSize");
287+
d_fields.push_back("MaxDataFileSize");
288+
d_fields.push_back("MaxQlistFileSize");
289+
const mqbs::JournalOpRecord::ResizeStorageData& rsd =
290+
rec.d_record.resizeStorageData();
291+
*d_printer_mp << rsd.maxJournalFileSize() << rsd.maxDataFileSize()
292+
<< rsd.maxQlistFileSize();
293+
}
283294
}
284295

285296
} // close namespace RecordPrinter

‎src/applications/bmqtool/m_bmqtool_storageinspector.cpp‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -538,10 +538,11 @@ void StorageInspector::processCommand(
538538
}
539539
printer << epochValue;
540540

541-
printer << syncPt.sequenceNum() << syncPt.primaryNodeId()
542-
<< syncPt.primaryLeaseId()
543-
<< syncPt.dataFileOffsetDwords()
544-
<< syncPt.qlistFileOffsetWords();
541+
const mqbs::JournalOpRecord::SyncPointData& spd =
542+
syncPt.syncPointData();
543+
printer << spd.sequenceNum() << spd.primaryNodeId()
544+
<< spd.primaryLeaseId() << spd.dataFileOffsetDwords()
545+
<< spd.qlistFileOffsetWords();
545546
}
546547
}
547548
}

‎src/groups/mqb/mqbblp/mqbblp_recoverymanager.cpp‎

Lines changed: 47 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -2154,7 +2154,8 @@ int RecoveryManager::syncPeerPartition(PrimarySyncContext* primarySyncCtx,
21542154
syncPtRec->syncPointType());
21552155

21562156
bsls::Types::Uint64 qlistMapOffset =
2157-
static_cast<bsls::Types::Uint64>(syncPtRec->qlistFileOffsetWords()) *
2157+
static_cast<bsls::Types::Uint64>(
2158+
syncPtRec->syncPointData().qlistFileOffsetWords()) *
21582159
bmqp::Protocol::k_WORD_SIZE;
21592160

21602161
if (fti.qlistFd().fileSize() < qlistMapOffset) {
@@ -2172,7 +2173,8 @@ int RecoveryManager::syncPeerPartition(PrimarySyncContext* primarySyncCtx,
21722173
}
21732174

21742175
bsls::Types::Uint64 dataMapOffset =
2175-
static_cast<bsls::Types::Uint64>(syncPtRec->dataFileOffsetDwords()) *
2176+
static_cast<bsls::Types::Uint64>(
2177+
syncPtRec->syncPointData().dataFileOffsetDwords()) *
21762178
bmqp::Protocol::k_DWORD_SIZE;
21772179

21782180
if (fti.dataFd().fileSize() < dataMapOffset) {
@@ -2242,10 +2244,12 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
22422244
}
22432245

22442246
// So we have encountered a JournalOp record in the stream. Currently,
2245-
// there is 1 type of JournalOp record: SYNCPOINT (see
2246-
// mqbs::JournalOpType). We know that SYNCPOINT can be of 2 sub-types:
2247-
// REGULAR and ROLLOVER (see mqbs::SyncPointType). We don't care about
2248-
// sub-type, it can be either.
2247+
// there are 2 types of JournalOp record: SYNCPOINT and RESIZE_STORAGE
2248+
// (see mqbs::JournalOpType). RESIZE_STORAGE is supported only in FSM
2249+
// mode. (should be never encountered here). We know that SYNCPOINT
2250+
// can be of 2 sub-types: REGULAR and ROLLOVER (see
2251+
// mqbs::SyncPointType). We don't care about sub-type, it can be
2252+
// either.
22492253

22502254
// Load the position of journal record.
22512255
bmqu::BlobPosition syncPointPosition;
@@ -2300,7 +2304,9 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
23002304
}
23012305

23022306
if (mqbs::JournalOpType::e_SYNCPOINT != journalOpRec->type()) {
2303-
// This should not occur. Per BlazingMQ replication algo, only a
2307+
// This should not occur. RESIZE_STORAGE record type is supported
2308+
// only in FSM mode and should not be encountered here.
2309+
// Per BlazingMQ replication algo, only a
23042310
// JournalOp record of type SYNCPOINT should be encountered.
23052311

23062312
BMQTSK_ALARMLOG_ALARM("RECOVERY")
@@ -2340,7 +2346,7 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
23402346
// same reason).
23412347

23422348
if (syncPointRecHeader->primaryLeaseId() <
2343-
journalOpRec->primaryLeaseId()) {
2349+
journalOpRec->syncPointData().primaryLeaseId()) {
23442350
// This indicates bug in BlazingMQ replication logic.
23452351

23462352
BALL_LOG_ERROR << d_clusterData_p->identity().description()
@@ -2351,17 +2357,18 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
23512357
<< syncPointRecHeader->primaryLeaseId() << ", "
23522358
<< syncPointRecHeader->sequenceNumber()
23532359
<< "). Sequence number in sync point("
2354-
<< journalOpRec->primaryLeaseId() << ", "
2355-
<< journalOpRec->sequenceNum()
2360+
<< journalOpRec->syncPointData().primaryLeaseId()
2361+
<< ", "
2362+
<< journalOpRec->syncPointData().sequenceNum()
23562363
<< "). Source: " << source->nodeDescription()
23572364
<< ". Ignoring this message.";
23582365
continue; // CONTINUE
23592366
}
23602367

23612368
if (syncPointRecHeader->primaryLeaseId() ==
2362-
journalOpRec->primaryLeaseId()) {
2369+
journalOpRec->syncPointData().primaryLeaseId()) {
23632370
if (syncPointRecHeader->sequenceNumber() !=
2364-
journalOpRec->sequenceNum()) {
2371+
journalOpRec->syncPointData().sequenceNum()) {
23652372
// If leaseId's match, sequence numbers must match too. Again
23662373
// look at the above comment or see
23672374
// 'FileStore::writeJournalRecord'.
@@ -2374,8 +2381,8 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
23742381
<< syncPointRecHeader->primaryLeaseId() << ", "
23752382
<< syncPointRecHeader->sequenceNumber()
23762383
<< "). Sequence number in sync point ("
2377-
<< journalOpRec->primaryLeaseId() << ", "
2378-
<< journalOpRec->sequenceNum()
2384+
<< journalOpRec->syncPointData().primaryLeaseId() << ", "
2385+
<< journalOpRec->syncPointData().sequenceNum()
23792386
<< "). Source: " << source->nodeDescription()
23802387
<< ". Ignoring this message." << BMQTSK_ALARMLOG_END;
23812388
continue; // CONTINUE
@@ -2404,8 +2411,8 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
24042411
<< syncPointRecHeader->primaryLeaseId() << ", "
24052412
<< syncPointRecHeader->sequenceNumber()
24062413
<< "). Sequence number in sync point ("
2407-
<< journalOpRec->primaryLeaseId() << ", "
2408-
<< journalOpRec->sequenceNum()
2414+
<< journalOpRec->syncPointData().primaryLeaseId() << ", "
2415+
<< journalOpRec->syncPointData().sequenceNum()
24092416
<< "). Source: " << source->nodeDescription();
24102417

24112418
bmqp_ctrlmsg::SyncPoint dummySyncPt;
@@ -2423,12 +2430,13 @@ bool RecoveryManager::hasSyncPoint(bmqp_ctrlmsg::SyncPoint* syncPoint,
24232430
// payload, not the RecordHeader (same reasoning as above -- new
24242431
// primary could be issuing a SyncPt on behalf of previous primary).
24252432

2426-
syncPoint->primaryLeaseId() = journalOpRec->primaryLeaseId();
2427-
syncPoint->sequenceNum() = journalOpRec->sequenceNum();
2433+
syncPoint->primaryLeaseId() =
2434+
journalOpRec->syncPointData().primaryLeaseId();
2435+
syncPoint->sequenceNum() = journalOpRec->syncPointData().sequenceNum();
24282436
syncPoint->dataFileOffsetDwords() =
2429-
journalOpRec->dataFileOffsetDwords();
2437+
journalOpRec->syncPointData().dataFileOffsetDwords();
24302438
syncPoint->qlistFileOffsetWords() =
2431-
journalOpRec->qlistFileOffsetWords();
2439+
journalOpRec->syncPointData().qlistFileOffsetWords();
24322440
*syncPointHeaderPosition = iter.headerPosition();
24332441
*journalOffset = static_cast<bsls::Types::Uint64>(
24342442
header.journalOffsetWords()) *
@@ -3192,22 +3200,24 @@ void RecoveryManager::startRecovery(
31923200
return; // RETURN
31933201
}
31943202

3195-
if (0 == journalOpRec.primaryLeaseId() ||
3196-
0 == journalOpRec.sequenceNum()) {
3203+
if (0 == journalOpRec.syncPointData().primaryLeaseId() ||
3204+
0 == journalOpRec.syncPointData().sequenceNum()) {
31973205
BMQTSK_ALARMLOG_ALARM("RECOVERY")
31983206
<< d_clusterData_p->identity().description() << ": For Partition ["
31993207
<< partitionId << "], "
32003208
<< "last sync point has invalid primaryLeaseId: "
3201-
<< journalOpRec.primaryLeaseId()
3202-
<< " or sequenceNum: " << journalOpRec.sequenceNum()
3209+
<< journalOpRec.syncPointData().primaryLeaseId()
3210+
<< " or sequenceNum: "
3211+
<< journalOpRec.syncPointData().sequenceNum()
32033212
<< ". Ignoring this sync point. Recovery will proceed as if this "
32043213
<< "node had no local recoverable files for this partition."
32053214
<< BMQTSK_ALARMLOG_END;
32063215
return; // RETURN
32073216
}
32083217

32093218
bsls::Types::Uint64 dataFileOffset =
3210-
static_cast<bsls::Types::Uint64>(journalOpRec.dataFileOffsetDwords()) *
3219+
static_cast<bsls::Types::Uint64>(
3220+
journalOpRec.syncPointData().dataFileOffsetDwords()) *
32113221
bmqp::Protocol::k_DWORD_SIZE;
32123222

32133223
if (bdls::FilesystemUtil::getFileSize(recoveryCtx.fileSet().dataFile()) <
@@ -3225,7 +3235,8 @@ void RecoveryManager::startRecovery(
32253235
}
32263236

32273237
bsls::Types::Uint64 qlistFileOffset =
3228-
static_cast<bsls::Types::Uint64>(journalOpRec.qlistFileOffsetWords()) *
3238+
static_cast<bsls::Types::Uint64>(
3239+
journalOpRec.syncPointData().qlistFileOffsetWords()) *
32293240
bmqp::Protocol::k_WORD_SIZE;
32303241

32313242
if (bdls::FilesystemUtil::getFileSize(recoveryCtx.fileSet().qlistFile()) <
@@ -3244,10 +3255,12 @@ void RecoveryManager::startRecovery(
32443255

32453256
// Retrieved old sync point is valid.
32463257

3247-
syncPoint.primaryLeaseId() = journalOpRec.primaryLeaseId();
3248-
syncPoint.sequenceNum() = journalOpRec.sequenceNum();
3249-
syncPoint.dataFileOffsetDwords() = journalOpRec.dataFileOffsetDwords();
3250-
syncPoint.qlistFileOffsetWords() = journalOpRec.qlistFileOffsetWords();
3258+
syncPoint.primaryLeaseId() = journalOpRec.syncPointData().primaryLeaseId();
3259+
syncPoint.sequenceNum() = journalOpRec.syncPointData().sequenceNum();
3260+
syncPoint.dataFileOffsetDwords() =
3261+
journalOpRec.syncPointData().dataFileOffsetDwords();
3262+
syncPoint.qlistFileOffsetWords() =
3263+
journalOpRec.syncPointData().qlistFileOffsetWords();
32513264

32523265
recoveryCtx.setOldSyncPoint(syncPoint);
32533266
recoveryCtx.setOldSyncPointOffset(lastSyncPointOffset);
@@ -4805,7 +4818,8 @@ void RecoveryManager::processPartitionSyncDataRequest(
48054818
syncPtRec->syncPointType());
48064819

48074820
bsls::Types::Uint64 qlistMapOffset =
4808-
static_cast<bsls::Types::Uint64>(syncPtRec->qlistFileOffsetWords()) *
4821+
static_cast<bsls::Types::Uint64>(
4822+
syncPtRec->syncPointData().qlistFileOffsetWords()) *
48094823
bmqp::Protocol::k_WORD_SIZE;
48104824
if (static_cast<bsls::Types::Uint64>(fileSet.qlistFileSize()) <
48114825
qlistMapOffset) {
@@ -4830,7 +4844,8 @@ void RecoveryManager::processPartitionSyncDataRequest(
48304844
}
48314845

48324846
bsls::Types::Uint64 dataMapOffset =
4833-
static_cast<bsls::Types::Uint64>(syncPtRec->dataFileOffsetDwords()) *
4847+
static_cast<bsls::Types::Uint64>(
4848+
syncPtRec->syncPointData().dataFileOffsetDwords()) *
48344849
bmqp::Protocol::k_DWORD_SIZE;
48354850
if (static_cast<bsls::Types::Uint64>(fileSet.dataFileSize()) <
48364851
dataMapOffset) {

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -673,6 +673,11 @@ class DataStore : public mqbi::DispatcherClient {
673673
virtual int writeSyncPointRecord(const bmqp_ctrlmsg::SyncPoint& syncPoint,
674674
SyncPointType::Enum type) = 0;
675675

676+
/// Write a RESIZE_STORAGE record to the data store with the specified
677+
/// `maxFileSizes`.
678+
virtual int writeResizeStorageRecord(
679+
const bmqp_ctrlmsg::PartitionMaxFileSizes& maxFileSizes) = 0;
680+
676681
/// Remove the record identified by the specified `handle`. Return zero
677682
/// on success, non-zero value if `handle` is invalid. Behavior is
678683
/// undefined unless `handle` represents a record in the data store.

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

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -376,6 +376,12 @@ class MockDataStore : public mqbs::DataStore {
376376
return 0;
377377
}
378378

379+
int writeResizeStorageRecord(const bmqp_ctrlmsg::PartitionMaxFileSizes&)
380+
BSLS_KEYWORD_OVERRIDE
381+
{
382+
return 0;
383+
}
384+
379385
int removeRecord(const mqbs::DataStoreRecordHandle&) BSLS_KEYWORD_OVERRIDE
380386
{
381387
return 0;
@@ -1757,7 +1763,7 @@ BMQTST_TEST(garbageCollect)
17571763
mqbi::StorageResult::e_SUCCESS);
17581764

17591765
const bdlt::Datetime currentTimeUtc;
1760-
bsls::Types::Uint64 secondsFromEpoch = 5;
1766+
bsls::Types::Uint64 secondsFromEpoch = 5;
17611767

17621768
mqbs::ReplicatedStorage& storage = tester.storage();
17631769

0 commit comments

Comments
 (0)