Skip to content

Commit b181724

Browse files
678098678098
committed
Fix[MQB]: add state transtition UNKNOWN -> RST_PRIMARY -> UNKNOWN
Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>
1 parent 69145fe commit b181724

8 files changed

Lines changed: 176 additions & 35 deletions

‎src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -285,7 +285,8 @@ void ClusterStateManager::do_updatePrimaryInPFSMs(
285285

286286
// PRECONDITIONS
287287
BSLS_ASSERT_SAFE(d_cluster_p->inDispatcherThread());
288-
BSLS_ASSERT_SAFE(d_clusterFSM.isSelfHealed());
288+
BSLS_ASSERT_SAFE(d_clusterFSM.isSelfHealed() ||
289+
event.first == ClusterStateTableEvent::e_RST_PRIMARY);
289290

290291
const ClusterStateTableEvent::Enum eventType = event.first;
291292
const bsl::vector<int>& modifiedPartitions =

‎src/groups/mqb/mqbc/mqbc_clusterstatetable.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -461,6 +461,7 @@ class ClusterStateTable
461461
logUnexpectedCSLCommit_and_abort,
462462
UNKNOWN);
463463
CST_CFG(UNKNOWN, STOP_NODE, stopPFSMs, STOPPED);
464+
CST_CFG(UNKNOWN, RST_PRIMARY, updatePrimaryInPFSMs, UNKNOWN);
464465
CST_CFG(FOL_HEALING,
465466
SLCT_LDR,
466467
stopWatchDog_cancelRequests_reapplyEvent,

‎src/groups/mqb/mqbc/mqbc_clusterutil.cpp‎

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -654,15 +654,28 @@ void ClusterUtil::onPartitionPrimaryAssignment(
654654

655655
ns->addPartitionRaw(partitionId);
656656

657-
// Notify the storage about (potentially same) mapping. This must be
658-
// done before calling
659-
// 'ClusterQueueHelper::afterPartitionPrimaryAssignment' (via
660-
// d_afterPartitionPrimaryAssignmentCb), because ClusterQueueHelper
661-
// assumes that storage is aware of the mapping.
662-
storageManager->setPrimaryForPartition(partitionId, primary, leaseId);
657+
if (!clusterData->clusterConfig()
658+
.clusterAttributes()
659+
.isFSMWorkflow()) {
660+
// In legacy mode, set primary directly. In FSM workflow,
661+
// setting is done through the Partition FSM's 'do_setPrimary'
662+
// action upon processing 'e_DETECT_SELF_REPLICA' or
663+
// 'e_DETECT_SELF_PRIMARY', keeping d_partitionInfoVec in sync
664+
// with the PFSM state.
665+
storageManager->setPrimaryForPartition(partitionId,
666+
primary,
667+
leaseId);
668+
}
663669
}
664670
else {
665-
storageManager->clearPrimaryForPartition(partitionId, oldPrimary);
671+
if (!clusterData->clusterConfig()
672+
.clusterAttributes()
673+
.isFSMWorkflow()) {
674+
// In legacy mode, clear primary directly. In FSM workflow,
675+
// clearing is done through the Partition FSM's
676+
// 'do_cleanupMetadata' action upon processing 'e_RST_UNKNOWN'.
677+
storageManager->clearPrimaryForPartition(partitionId, oldPrimary);
678+
}
666679
}
667680
}
668681

‎src/groups/mqb/mqbc/mqbc_partitionfsm.h‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,9 @@ class PartitionFSMEventData {
6969
bmqp_ctrlmsg::PartitionSequenceNumber>
7070
PartitionSeqNumDataRange;
7171

72+
/// The placeholder value for invalid or unspecified leaseId.
73+
static const unsigned int k_INVALID_LEASE_ID = 0;
74+
7275
private:
7376
// DATA
7477

@@ -131,8 +134,8 @@ class PartitionFSMEventData {
131134
int requestId,
132135
int partitionId,
133136
int incrementCount,
134-
mqbnet::ClusterNode* primary = 0,
135-
unsigned int primaryLeaseId = 0,
137+
mqbnet::ClusterNode* primary = 0,
138+
unsigned int primaryLeaseId = k_INVALID_LEASE_ID,
136139
const bmqp_ctrlmsg::PartitionSequenceNumber& seqNum =
137140
bmqp_ctrlmsg::PartitionSequenceNumber(),
138141
const bmqp_ctrlmsg::PartitionSequenceNumber&
@@ -327,7 +330,7 @@ inline PartitionFSMEventData::PartitionFSMEventData()
327330
, d_partitionId(mqbi::Storage::k_INVALID_PARTITION_ID)
328331
, d_incrementCount(1)
329332
, d_primary_p(0)
330-
, d_primaryLeaseId(0) // Invalid placeholder LeaseId
333+
, d_primaryLeaseId(k_INVALID_LEASE_ID)
331334
, d_partitionSequenceNumber()
332335
, d_firstSyncPointAfterRolloverSequenceNumber()
333336
, d_partitionMaxFileSizes()
@@ -409,7 +412,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
409412
, d_partitionId(partitionId)
410413
, d_incrementCount(incrementCount)
411414
, d_primary_p(0)
412-
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
415+
, d_primaryLeaseId(k_INVALID_LEASE_ID)
413416
, d_partitionSequenceNumber(seqNum)
414417
, d_firstSyncPointAfterRolloverSequenceNumber(
415418
firstSyncPointAfterRollloverSeqNum)
@@ -432,7 +435,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
432435
, d_partitionId(partitionId)
433436
, d_incrementCount(incrementCount)
434437
, d_primary_p(0)
435-
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
438+
, d_primaryLeaseId(k_INVALID_LEASE_ID)
436439
, d_partitionSequenceNumber()
437440
, d_firstSyncPointAfterRolloverSequenceNumber()
438441
, d_partitionMaxFileSizes()
@@ -453,7 +456,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
453456
, d_partitionId(partitionId)
454457
, d_incrementCount(incrementCount)
455458
, d_primary_p(0)
456-
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
459+
, d_primaryLeaseId(k_INVALID_LEASE_ID)
457460
, d_partitionSequenceNumber()
458461
, d_firstSyncPointAfterRolloverSequenceNumber()
459462
, d_partitionMaxFileSizes()

‎src/groups/mqb/mqbc/mqbc_partitionstatetable.h‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -332,6 +332,8 @@ class PartitionStateTableActions {
332332

333333
virtual void do_processLiveData(const ARGS& args) = 0;
334334

335+
virtual void do_setPrimary(const ARGS& args) = 0;
336+
335337
virtual void do_cleanupMetadata(const ARGS& args) = 0;
336338

337339
virtual void do_startSendDataChunks(const ARGS& args) = 0;
@@ -380,10 +382,11 @@ class PartitionStateTableActions {
380382
const ARGS& args);
381383

382384
void
383-
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
385+
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
384386
const ARGS& args);
385387

386-
void do_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
388+
void
389+
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
387390
const ARGS& args);
388391

389392
void do_storeReplicaSeq_primaryStateResponse_checkQuorumMaxFileSizesAndSeq(
@@ -520,12 +523,12 @@ class PartitionStateTable
520523
PST_CFG(
521524
UNKNOWN,
522525
DETECT_SELF_PRIMARY,
523-
startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq,
526+
setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq,
524527
PRIMARY_HEALING_STG1);
525528
PST_CFG(
526529
UNKNOWN,
527530
DETECT_SELF_REPLICA,
528-
startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest,
531+
setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest,
529532
REPLICA_WAITING);
530533
PST_CFG(UNKNOWN,
531534
PRIMARY_STATE_RQST,
@@ -859,9 +862,10 @@ void PartitionStateTableActions<ARGS>::
859862

860863
template <typename ARGS>
861864
void PartitionStateTableActions<ARGS>::
862-
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
865+
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
863866
const ARGS& args)
864867
{
868+
do_setPrimary(args);
865869
do_startWatchDog(args);
866870
do_openRecoveryFileSet(args);
867871
do_storeSelfSeq(args);
@@ -871,9 +875,10 @@ void PartitionStateTableActions<ARGS>::
871875

872876
template <typename ARGS>
873877
void PartitionStateTableActions<ARGS>::
874-
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
878+
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
875879
const ARGS& args)
876880
{
881+
do_setPrimary(args);
877882
do_startWatchDog(args);
878883
do_openRecoveryFileSet(args);
879884
do_storeSelfSeq(args);

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

Lines changed: 122 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -336,11 +336,13 @@ void StorageManager::dispatchEventToPartition(PartitionFSM::Event::Enum event,
336336
// PRECONDITIONS
337337
BSLS_ASSERT_SAFE(eventDataVec.size() >= 1);
338338

339-
// NOTE: it is assumed that all elements in 'eventDataVec' have the same
340-
// 'partitionId'.
341339
const int partitionId = eventDataVec[0].partitionId();
342340
BSLS_ASSERT_SAFE(0 <= partitionId &&
343341
partitionId < static_cast<int>(d_fileStores.size()));
342+
// Verify other events with 0-indexed event's partitionId
343+
for (size_t i = 1; i < eventDataVec.size(); ++i) {
344+
BSLS_ASSERT_SAFE(partitionId == eventDataVec[i].partitionId());
345+
}
344346

345347
if (d_cluster_p->isStopping()) {
346348
BALL_LOG_WARN << d_clusterData_p->identity().description()
@@ -353,17 +355,88 @@ void StorageManager::dispatchEventToPartition(PartitionFSM::Event::Enum event,
353355
mqbs::FileStore* fs = d_fileStores[partitionId].get();
354356
BSLS_ASSERT_SAFE(fs);
355357
if (fs->inDispatcherThread()) {
356-
d_partitionFSMVec[partitionId]->enqueueEvent(
357-
PartitionFSM::EventWithData(event, eventDataVec));
358+
executeEventInPartitionThread(event, eventDataVec);
358359
}
359360
else {
360361
fs->execute(bdlf::BindUtil::bind(
361-
&PartitionFSM::enqueueEvent,
362-
d_partitionFSMVec[partitionId].get(),
363-
PartitionFSM::EventWithData(event, eventDataVec)));
362+
&StorageManager::executeEventInPartitionThread,
363+
this,
364+
event,
365+
eventDataVec));
364366
}
365367
}
366368

369+
void StorageManager::executeEventInPartitionThread(
370+
PartitionFSM::Event::Enum event,
371+
const EventData& eventDataVec)
372+
{
373+
// Thread: QUEUE dispatcher
374+
375+
// PRECONDITIONS
376+
BSLS_ASSERT_SAFE(eventDataVec.size() >= 1);
377+
378+
const int partitionId = eventDataVec[0].partitionId();
379+
BSLS_ASSERT_SAFE(0 <= partitionId &&
380+
partitionId < static_cast<int>(d_fileStores.size()));
381+
BSLS_ASSERT_SAFE(d_fileStores[partitionId]->inDispatcherThread());
382+
383+
// Verify events
384+
if (eventDataVec.size() == 1) {
385+
const PartitionFSMEventData& evt = eventDataVec[0];
386+
387+
// Do not perform extra checks if primaryLeaseId is unspecified.
388+
if (PartitionFSMEventData::k_INVALID_LEASE_ID !=
389+
evt.primaryLeaseId()) {
390+
const PartitionInfo& pinfo = d_partitionInfoVec[partitionId];
391+
392+
const bool isLeaseIdOutdated = evt.primaryLeaseId() <
393+
pinfo.primaryLeaseId();
394+
const bool isPrimaryMismatch = evt.primaryLeaseId() ==
395+
pinfo.primaryLeaseId() &&
396+
evt.primary() && pinfo.primary() &&
397+
evt.primary()->nodeId() !=
398+
pinfo.primary()->nodeId();
399+
400+
if (isLeaseIdOutdated || isPrimaryMismatch) {
401+
BALL_LOG_WARN
402+
<< d_clusterData_p->identity().description()
403+
<< " Partition [" << partitionId
404+
<< "]: dropping stale event: source "
405+
<< evt.source()->nodeDescription()
406+
<< ", event primaryLeaseId [" << evt.primaryLeaseId()
407+
<< "], current primaryLeaseId [" << pinfo.primaryLeaseId()
408+
<< "], event primary ["
409+
<< (evt.primary() ? evt.primary()->nodeDescription()
410+
: "null")
411+
<< "], current primary ["
412+
<< (pinfo.primary() ? pinfo.primary()->nodeDescription()
413+
: "null")
414+
<< "]";
415+
416+
if (0 <= evt.requestId()) {
417+
bmqp_ctrlmsg::ControlMessage controlMsg;
418+
controlMsg.rId() = evt.requestId();
419+
bmqp_ctrlmsg::Status& response =
420+
controlMsg.choice().makeStatus();
421+
422+
response.category() =
423+
bmqp_ctrlmsg::StatusCategory::E_REFUSED;
424+
response.code() = mqbi::ClusterErrorCode::e_UNKNOWN;
425+
response.message() = "Primary mismatch";
426+
427+
d_clusterData_p->messageTransmitter().sendMessageSafe(
428+
controlMsg,
429+
evt.source());
430+
}
431+
return;
432+
}
433+
}
434+
}
435+
436+
d_partitionFSMVec[partitionId]->enqueueEvent(
437+
PartitionFSM::EventWithData(event, eventDataVec));
438+
}
439+
367440
void StorageManager::setPrimaryStatusForPartitionDispatched(
368441
int partitionId,
369442
bmqp_ctrlmsg::PrimaryStatus::Value value)
@@ -1580,10 +1653,13 @@ void StorageManager::do_replicaStateResponse(const EventWithData& event)
15801653
BSLS_ASSERT_SAFE(eventDataVec.size() == 1);
15811654

15821655
const PartitionFSMEventData& eventData = eventDataVec[0];
1583-
int partitionId = eventData.partitionId();
1656+
const int partitionId = eventData.partitionId();
15841657

15851658
BSLS_ASSERT_SAFE(0 <= partitionId &&
15861659
partitionId < static_cast<int>(d_fileStores.size()));
1660+
BSLS_ASSERT_SAFE(eventData.source());
1661+
BSLS_ASSERT_SAFE(eventData.source()->nodeId() ==
1662+
d_partitionInfoVec[partitionId].primary()->nodeId());
15871663

15881664
bmqp_ctrlmsg::ControlMessage controlMsg;
15891665
controlMsg.rId() = eventData.requestId();
@@ -1609,10 +1685,6 @@ void StorageManager::do_replicaStateResponse(const EventWithData& event)
16091685
response.partitionMaxFileSizes() = getSelfPartitionMaxFileSizes(
16101686
partitionId);
16111687

1612-
BSLS_ASSERT_SAFE(eventData.source());
1613-
BSLS_ASSERT_SAFE(eventData.source()->nodeId() ==
1614-
d_partitionInfoVec[partitionId].primary()->nodeId());
1615-
16161688
fileStore(partitionId).sendMessage(controlMsg, eventData.source());
16171689

16181690
BALL_LOG_INFO << d_clusterData_p->identity().description()
@@ -2795,6 +2867,36 @@ void StorageManager::do_processLiveData(const EventWithData& event)
27952867
source);
27962868
}
27972869

2870+
void StorageManager::do_setPrimary(const EventWithData& event)
2871+
{
2872+
// executed by the *QUEUE DISPATCHER* thread associated with the
2873+
// paritionId contained in 'event'
2874+
2875+
const EventData& eventDataVec = event.second;
2876+
BSLS_ASSERT_SAFE(eventDataVec.size() == 1);
2877+
2878+
const PartitionFSMEventData& eventData = eventDataVec[0];
2879+
const int partitionId = eventData.partitionId();
2880+
mqbnet::ClusterNode* primary = eventData.primary();
2881+
const unsigned int leaseId = eventData.primaryLeaseId();
2882+
2883+
// PRECONDITIONS
2884+
BSLS_ASSERT_SAFE(0 <= partitionId &&
2885+
partitionId < static_cast<int>(d_fileStores.size()));
2886+
BSLS_ASSERT_SAFE(d_fileStores[partitionId]->inDispatcherThread());
2887+
BSLS_ASSERT_SAFE(primary);
2888+
2889+
PartitionInfo& pinfo = d_partitionInfoVec[partitionId];
2890+
if (pinfo.primary() && (pinfo.primary()->nodeId() == primary->nodeId())) {
2891+
pinfo.setPrimaryLeaseId(leaseId);
2892+
return; // RETURN
2893+
}
2894+
2895+
pinfo.setPrimary(primary);
2896+
pinfo.setPrimaryLeaseId(leaseId);
2897+
pinfo.setPrimaryStatus(bmqp_ctrlmsg::PrimaryStatus::E_PASSIVE);
2898+
}
2899+
27982900
void StorageManager::do_cleanupMetadata(const EventWithData& event)
27992901
{
28002902
// executed by the *QUEUE DISPATCHER* thread associated with the
@@ -2815,6 +2917,12 @@ void StorageManager::do_cleanupMetadata(const EventWithData& event)
28152917
d_partitionFSMVec[partitionId]->state() ==
28162918
PartitionFSM::State::e_STOPPED);
28172919

2920+
StorageUtil::clearPrimaryForPartition(
2921+
d_fileStores[partitionId].get(),
2922+
&d_partitionInfoVec[partitionId],
2923+
d_clusterData_p->identity().description(),
2924+
partitionId);
2925+
28182926
d_nodeToContextMapVec[partitionId].clear();
28192927
d_numReplicaDataResponsesReceivedVec[partitionId] = 0;
28202928
d_recoveryManager_mp->resetReceiveDataCtx(partitionId);
@@ -4905,6 +5013,8 @@ void StorageManager::processReplicaStateRequest(
49055013
message.rId().isNull() ? -1 : message.rId().value(),
49065014
partitionId,
49075015
1,
5016+
source,
5017+
replicaStateRequest.latestSequenceNumber().primaryLeaseId(),
49085018
replicaStateRequest.latestSequenceNumber(),
49095019
replicaStateRequest.firstSyncPointAfterRolloverSequenceNumber(),
49105020
replicaStateRequest.partitionMaxFileSizes());

‎src/groups/mqb/mqbc/mqbc_storagemanager.h‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -458,12 +458,18 @@ class StorageManager BSLS_KEYWORD_FINAL
458458
void onPartitionRecovery(int partitionId);
459459

460460
/// Dispatch the event to *QUEUE DISPATCHER* thread associated with
461-
/// the partitionId as per the specified `eventDataVec` with the
461+
/// the `partitionId` as per the specified `eventDataVec` with the
462462
/// specified `event`. If we are already in *QUEUE DISPATCHER* thread,
463463
/// then execute the event in place.
464464
void dispatchEventToPartition(PartitionFSM::Event::Enum event,
465465
const EventData& eventDataVec);
466466

467+
/// Verify and execute the event in *QUEUE DISPATCHER* thread associated
468+
/// with the `partitionId` as per the specified `eventDataVec` with the
469+
/// specified `event`.
470+
void executeEventInPartitionThread(PartitionFSM::Event::Enum event,
471+
const EventData& eventDataVec);
472+
467473
/// Set the primary status of the specified `partitionId` to the specified
468474
/// `value`.
469475
///
@@ -668,6 +674,8 @@ class StorageManager BSLS_KEYWORD_FINAL
668674

669675
void do_processLiveData(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;
670676

677+
void do_setPrimary(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;
678+
671679
void do_cleanupMetadata(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;
672680

673681
void

0 commit comments

Comments
 (0)