Skip to content

Fix[mqb]: FileStore must flush storage before control messages #1202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dorjesinpo merged 3 commits into from
Mar 20, 2026
Merged

Fix[mqb]: FileStore must flush storage before control messages #1202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f7f7ef1
move controlmessagetransmitter to mqbnet
dorjesinpo Mar 18, 2026
aab7115
FileStore must flush storage before control messages
dorjesinpo Mar 18, 2026
8babee8
Addressing review
dorjesinpo Mar 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/groups/mqb/mqbblp/mqbblp_cluster.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,9 @@ void Cluster::startDispatched(bsl::ostream* errorDescription, int* rc)
bmqp_ctrlmsg::NodeStatusAdvisory& advisory =
clusterMsg.choice().makeNodeStatusAdvisory();
advisory.status() = bmqp_ctrlmsg::NodeStatus::E_STARTING;
d_clusterData.messageTransmitter().broadcastMessage(controlMsg, true);
d_clusterData.messageTransmitter().broadcastMessage(
controlMsg,
d_clusterData.transportManager());

// Start a StatMonitorSnapshotRecorder to track system stats during
// recovery
Expand Down Expand Up @@ -327,6 +329,10 @@ void Cluster::stopDispatched()
// guaranteed that this object will be kept alive.

// Teardown all ClusterNodeSession
BALL_LOG_INFO << description() << " teardown "
<< d_clusterData.membership().clusterNodeSessionMap().size()
<< " ClusterNodeSession";

for (ClusterNodeSessionMapIter it =
d_clusterData.membership().clusterNodeSessionMap().begin();
it != d_clusterData.membership().clusterNodeSessionMap().end();
Expand Down Expand Up @@ -679,7 +685,9 @@ void Cluster::continueShutdownDispatched(

d_clusterData.membership().setSelfNodeStatus(
bmqp_ctrlmsg::NodeStatus::E_UNAVAILABLE);
d_clusterData.messageTransmitter().broadcastMessage(controlMsg, true);
d_clusterData.messageTransmitter().broadcastMessage(
controlMsg,
d_clusterData.transportManager());

// Make sure all partitions done sending last sync points and advisories.
// Synchronize with all Queue Dispatcher threads
Expand Down
5 changes: 3 additions & 2 deletions src/groups/mqb/mqbblp/mqbblp_clusterorchestrator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -797,8 +797,9 @@ void ClusterOrchestrator::transitionToAvailable()
bmqp_ctrlmsg::NodeStatusAdvisory& advisory =
clusterMsg.choice().makeNodeStatusAdvisory();
advisory.status() = bmqp_ctrlmsg::NodeStatus::E_AVAILABLE;
d_clusterData_p->messageTransmitter().broadcastMessage(controlMsg,
true);
d_clusterData_p->messageTransmitter().broadcastMessage(
controlMsg,
d_clusterData_p->transportManager());

d_wasAvailableAdvisorySent = true;
}
Expand Down
2 changes: 1 addition & 1 deletion src/groups/mqb/mqbblp/mqbblp_clusterproxy.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ void ClusterProxy::startDispatched()
// and all session to build initial state and then schedule a refresh to
// find active node if there is none after processing all pending events.

d_activeNodeManager.initialize(&d_clusterData.transportManager());
d_activeNodeManager.initialize(d_clusterData.transportManager());

// Ready to read.
d_clusterData.membership().netCluster()->enableRead();
Expand Down
3 changes: 1 addition & 2 deletions src/groups/mqb/mqbc/mqbc_clusterdata.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,8 +119,7 @@ ClusterData::ClusterData(
allocator))
, d_cluster_p(cluster)
, d_messageTransmitter(resources.blobSpPool(),
cluster,
transportManager,
membership().netCluster(),
allocator)
, d_requestManager(bmqp::EventType::e_CONTROL,
resources.blobSpPool(),
Expand Down
14 changes: 7 additions & 7 deletions src/groups/mqb/mqbc/mqbc_clusterdata.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,14 +26,14 @@

// MQB
#include <mqbc_clustermembership.h>
#include <mqbc_controlmessagetransmitter.h>
#include <mqbc_electorinfo.h>
#include <mqbcfg_clusterquorummanager.h>
#include <mqbcfg_messages.h>
#include <mqbi_cluster.h>
#include <mqbi_dispatcher.h>
#include <mqbi_domain.h>
#include <mqbnet_cluster.h>
#include <mqbnet_controlmessagetransmitter.h>
#include <mqbnet_elector.h>
#include <mqbnet_multirequestmanager.h>
#include <mqbnet_transportmanager.h>
Expand Down Expand Up @@ -185,7 +185,7 @@ class ClusterData {
mqbi::Cluster* d_cluster_p;

/// Control message transmitter to use.
ControlMessageTransmitter d_messageTransmitter;
mqbnet::ControlMessageTransmitter d_messageTransmitter;

/// Request manager to use.
RequestManagerType d_requestManager;
Expand Down Expand Up @@ -270,7 +270,7 @@ class ClusterData {
mqbcfg::ClusterQuorumManager& quorumManager();

/// Get a modifiable reference to this object's messageTransmitter.
ControlMessageTransmitter& messageTransmitter();
mqbnet::ControlMessageTransmitter& messageTransmitter();

/// Get a modifiable reference to this object's requestManager.
RequestManagerType& requestManager();
Expand All @@ -282,7 +282,7 @@ class ClusterData {
mqbi::DomainFactory* domainFactory();

/// Get a modifiable reference to this object's transportManager.
mqbnet::TransportManager& transportManager();
mqbnet::TransportManager* transportManager();

/// Get a modifiable reference to this object's cluster stats.
mqbstat::ClusterStats& stats();
Expand Down Expand Up @@ -403,7 +403,7 @@ inline mqbcfg::ClusterQuorumManager& ClusterData::quorumManager()
return d_quorumManager;
}

inline ControlMessageTransmitter& ClusterData::messageTransmitter()
inline mqbnet::ControlMessageTransmitter& ClusterData::messageTransmitter()
{
return d_messageTransmitter;
}
Expand All @@ -423,9 +423,9 @@ inline mqbi::DomainFactory* ClusterData::domainFactory()
return d_domainFactory_p;
}

inline mqbnet::TransportManager& ClusterData::transportManager()
inline mqbnet::TransportManager* ClusterData::transportManager()
{
return *d_transportManager_p;
return d_transportManager_p;
}

inline mqbstat::ClusterStats& ClusterData::stats()
Expand Down
27 changes: 9 additions & 18 deletions src/groups/mqb/mqbc/mqbc_storagemanager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1613,8 +1613,7 @@ void StorageManager::do_replicaStateResponse(const EventWithData& event)
BSLS_ASSERT_SAFE(eventData.source()->nodeId() ==
d_partitionInfoVec[partitionId].primary()->nodeId());

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< ": sent response " << controlMsg
Expand Down Expand Up @@ -1673,8 +1672,7 @@ void StorageManager::do_failureReplicaStateResponse(const EventWithData& event)
<< eventData.source()->nodeDescription() << ".";
}

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());
}

void StorageManager::do_logFailureReplicaStateResponse(
Expand Down Expand Up @@ -1915,8 +1913,7 @@ void StorageManager::do_primaryStateResponse(const EventWithData& event)
response.partitionMaxFileSizes() = getSelfPartitionMaxFileSizes(
partitionId);

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< ": sent response " << controlMsg
Expand Down Expand Up @@ -1951,8 +1948,7 @@ void StorageManager::do_failurePrimaryStateResponse(const EventWithData& event)
response.code() = mqbi::ClusterErrorCode::e_NOT_PRIMARY;
response.message() = "Not a primary";

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId
Expand Down Expand Up @@ -2181,8 +2177,7 @@ void StorageManager::do_replicaDataResponsePush(const EventWithData& event)
partitionId);
}

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
destNode);
fileStore(partitionId).sendMessage(controlMsg, destNode);

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId << "]: " << "Sent response "
Expand Down Expand Up @@ -2394,8 +2389,7 @@ void StorageManager::do_replicaDataResponseDrop(const EventWithData& event)
eventData.partitionSeqNumDataRange().first;
response.endSequenceNumber() = eventData.partitionSeqNumDataRange().second;

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId << "]: " << "Sent response "
Expand Down Expand Up @@ -2506,8 +2500,7 @@ void StorageManager::do_replicaDataResponsePull(const EventWithData& event)
eventData.partitionSeqNumDataRange().first;
response.endSequenceNumber() = eventData.partitionSeqNumDataRange().second;

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
destNode);
fileStore(partitionId).sendMessage(controlMsg, destNode);

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId << "]: Sent response "
Expand Down Expand Up @@ -2546,8 +2539,7 @@ void StorageManager::do_failureReplicaDataResponsePull(
status.code() = mqbi::ClusterErrorCode::e_STORAGE_FAILURE;
status.message() = "Failed to send data chunks";

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
destNode);
fileStore(partitionId).sendMessage(controlMsg, destNode);

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId
Expand Down Expand Up @@ -3919,8 +3911,7 @@ void StorageManager::do_replicaDataResponseResize(const EventWithData& event)
response.partitionMaxFileSizes() = getSelfPartitionMaxFileSizes(
partitionId);

d_clusterData_p->messageTransmitter().sendMessageSafe(controlMsg,
eventData.source());
fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
<< " Partition [" << partitionId << "]: " << "Sent response "
Expand Down
19 changes: 9 additions & 10 deletions src/groups/mqb/mqbc/mqbc_storageutil.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1412,15 +1412,15 @@ StorageUtil::findMinReqDiskSpace(const mqbcfg::PartitionConfig& config)
return minimumRequiredDiskSpace;
}

void StorageUtil::transitionToActivePrimary(PartitionInfo* partitionInfo,
mqbc::ClusterData* clusterData,
int partitionId)
void StorageUtil::transitionToActivePrimary(PartitionInfo* partitionInfo,
mqbs::FileStore* fs,
int partitionId)

@678098 678098 Mar 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
int partitionId)
)

I am curious why don't we make partitionId a field in FileStore, we can use it here (extract from fs).
Current API allows to work with FileStore with provided incorrect partitionId
Also there are different members in FileStore that require partitionId arg

@678098 678098 Mar 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you agree with this, I can do it in another PR

{
// executed by *QUEUE_DISPATCHER* thread associated with 'partitionId'

// PRECONDITIONS
BSLS_ASSERT_SAFE(partitionInfo);
BSLS_ASSERT_SAFE(clusterData);
BSLS_ASSERT_SAFE(fs);

partitionInfo->setPrimaryStatus(bmqp_ctrlmsg::PrimaryStatus::E_ACTIVE);

Expand All @@ -1434,7 +1434,7 @@ void StorageUtil::transitionToActivePrimary(PartitionInfo* partitionInfo,
primaryAdv.primaryLeaseId() = partitionInfo->primaryLeaseId();
primaryAdv.status() = bmqp_ctrlmsg::PrimaryStatus::E_ACTIVE;

clusterData->messageTransmitter().broadcastMessageSafe(controlMsg);
fs->broadcastMessage(controlMsg);
}

void StorageUtil::onPartitionPrimarySync(
Expand Down Expand Up @@ -1505,7 +1505,7 @@ void StorageUtil::onPartitionPrimarySync(

// Broadcast self as active primary of this partition. This must be done
// before invoking 'FileStore::setActivePrimary'.
transitionToActivePrimary(pinfo, clusterData, partitionId);
transitionToActivePrimary(pinfo, fs, partitionId);

partitionPrimaryStatusCb(partitionId, status, pinfo->primaryLeaseId());

Expand Down Expand Up @@ -3412,7 +3412,7 @@ void StorageUtil::processShutdownEventDispatched(ClusterData* clusterData,
primaryAdv.primaryLeaseId() = pinfo->primaryLeaseId();
primaryAdv.status() = bmqp_ctrlmsg::PrimaryStatus::E_PASSIVE;

clusterData->messageTransmitter().broadcastMessageSafe(controlMsg);
fs->broadcastMessage(controlMsg);
}

// Notify partition that self node is shutting down (this occurs
Expand Down Expand Up @@ -3862,11 +3862,10 @@ void StorageUtil::forceIssueAdvisoryAndSyncPt(mqbc::ClusterData* clusterData,
primaryAdv.status() = bmqp_ctrlmsg::PrimaryStatus::E_ACTIVE;

if (destination) {
clusterData->messageTransmitter().sendMessageSafe(controlMsg,
destination);
fs->sendMessage(controlMsg, destination);
}
else {
clusterData->messageTransmitter().broadcastMessageSafe(controlMsg);
fs->broadcastMessage(controlMsg);
}
const int rc = fs->issueSyncPoint();
bmqp_ctrlmsg::PartitionSequenceNumber psn;
Expand Down
9 changes: 4 additions & 5 deletions src/groups/mqb/mqbc/mqbc_storageutil.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -540,14 +540,13 @@ struct StorageUtil {

/// Transition self to active primary of the specified `partitionId` and
/// load this info into the specified `partitionInfo`. Then, broadcast
/// a primary status advisory to peers using the specified
/// `clusterData`.
/// a primary status advisory to peers via the specified `fs`.
///
/// THREAD: Executed by the queue dispatcher thread associated with
/// 'partitionId'.
static void transitionToActivePrimary(PartitionInfo* partitionInfo,
mqbc::ClusterData* clusterData,
int partitionId);
static void transitionToActivePrimary(PartitionInfo* partitionInfo,
mqbs::FileStore* fs,
int partitionId);

/// Callback executed after primary sync for the specified 'partitionId'
/// is complete with the specified 'status'. Use the specified 'fs',
Expand Down
1 change: 0 additions & 1 deletion src/groups/mqb/mqbc/package/mqbc.mem
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ mqbc_clusterstateledgerutil
mqbc_clusterstatemanager
mqbc_clusterstatetable
mqbc_clusterutil
mqbc_controlmessagetransmitter
mqbc_electorinfo
mqbc_incoreclusterstateledger
mqbc_incoreclusterstateledgeriterator
Expand Down
Loading
Loading