Skip to content
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

Wait ACKs for all messages posted by bmqtool #231

Merged
merged 2 commits into from
Apr 8, 2024
Merged

Wait ACKs for all messages posted by bmqtool #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
be49213
Wait ACKs for all messages posted by bmqtool
syuzvinsky Apr 5, 2024
041457d
Rename variable
syuzvinsky Apr 5, 2024
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
2 changes: 1 addition & 1 deletion src/applications/bmqtool/bmqtoolcmd.xsd
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@
<element name='latencyReport' type='string' default=""/>
<element name='dumpMsg' type='boolean' default="false"/>
<element name='confirmMsg' type='boolean' default="false"/>
<element name='eventSize' type='int' default="1"/>
<element name='eventSize' type='long' default="1"/>
<element name='msgSize' type='int' default="1024"/>
<element name='postRate' type='int' default="1"/>
<element name='eventsCount' type='string' default="0"/>
Expand Down
23 changes: 15 additions & 8 deletions src/applications/bmqtool/m_bmqtool_application.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -768,6 +768,12 @@ void Application::onMessageEvent(const bmqa::MessageEvent& event)

// Write to log file
d_fileLogger.writeAckMessage(message);

if (d_numExpectedAcks != 0 &&
d_numExpectedAcks == ++d_numAcknowledged) {
BALL_LOG_INFO << "All posted messages have been acknowledged";
d_shutdownSemaphore_p->post();
dorjesinpo marked this conversation as resolved.
Show resolved Hide resolved
}
}
else {
// Message is a push message
Expand Down Expand Up @@ -931,7 +937,7 @@ void Application::producerThread()
}

eventBuilder.reset();
for (int msgId = 0; msgId < d_parameters_p->eventSize();
for (bsl::uint64_t msgId = 0; msgId < d_parameters_p->eventSize();
++msgId, ++msgSeqId) {
bmqa::Message& msg = eventBuilder.startMessage();
int length = 0;
Expand Down Expand Up @@ -1046,12 +1052,7 @@ void Application::producerThread()
}
}

// Finished posting messages in auto mode?
// If shutDownGrace is set, signal to the main thread to exit.
if (d_parameters_p->mode() == ParametersMode::e_AUTO &&
d_parameters_p->shutdownGrace() != 0) {
// We do not need to sleep the grace period, since it is done
// by the main thread, in the stop() function.
if (!bmqt::QueueFlagsUtil::isAck(d_parameters_p->queueFlags())) {
dorjesinpo marked this conversation as resolved.
Show resolved Hide resolved
Copy link
Collaborator

Choose a reason for hiding this comment

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

Could you verify the use of shutdownGrace now? Do we still need it?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

I will double check this with Jean-Louis and update the response after that.

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

Discussed this with Julien Bibollet.
The conclusion was that shutdownGrace was designed, and is still needed in consuming mode: some integration tests imply running several instances of the consumer, and there is some nondeterminism in how many messages each particular instance should receive, only the total number of messages is defined.
So, the test code will be waiting for a publisher to complete, and after that it will be waiting until each client stops receiving messages and terminate.
In the mean time, using shutdownGrace in a publisher thread looks rather like a hack, and counting ACKs is a better way to make sure that all messages have reached the cluster.

d_shutdownSemaphore_p->post();
}
}
Expand Down Expand Up @@ -1131,6 +1132,8 @@ Application::Application(Parameters* parameters,
, d_latencies(allocator)
, d_autoReadInProgress(false)
, d_autoReadActivity(false)
, d_numExpectedAcks(0)
, d_numAcknowledged(0)
{
// NOTHING
}
Expand Down Expand Up @@ -1194,8 +1197,12 @@ int Application::run()
d_shutdownSemaphore_p->post();
}
else {
// Start the thread
if (bmqt::QueueFlagsUtil::isWriter(d_parameters_p->queueFlags())) {
d_numExpectedAcks = d_parameters_p->eventsCount() *
d_parameters_p->eventSize();
Copy link
Collaborator

Choose a reason for hiding this comment

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

I think it's worth to change these params types to xs:long in xsd, so the product here is less prone to overflow with "seemingly not too big" param values.

Copy link
Collaborator

@678098 678098 Apr 2, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

At the same time, I am not sure if we should add more constraints for param values, like:

BSLS_ASSERT(eventsCount() >= 0);
BSLS_ASSERT(eventsCount() <= 2^31);
BSLS_ASSERT(eventSize() >= 1);
BSLS_ASSERT(eventSize() <= 2^31);

Another way, these constraints might be introduced in xsd if possible.

d_numAcknowledged = 0;

// Start the thread
rc = bslmt::ThreadUtil::create(
&d_runningThread,
bdlf::MemFnUtil::memFn(&Application::producerThread, this));
Expand Down
13 changes: 13 additions & 0 deletions src/applications/bmqtool/m_bmqtool_application.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,19 @@ class Application : public bmqa::SessionEventHandler {
// message was seen during the current
// grace period.

bsl::uint64_t d_numExpectedAcks;
// Auto-produce mode only. The total number of messages
// the tool will send. After posting is finished
// the tool will be waiting for this number of ACK
// messages, after which the shutdown semaphore will
// be posted.

bsl::uint64_t d_numAcknowledged;
// Auto-produce mode only. The number of acknowledged
// messages. When the value of this field becomes equal
// to d_numExpectedAcks, the shutdown semaphore will be
// posted.

// PRIVATE MANIPULATORS
// (virtual: bmqa::SessionEventHandler)

Expand Down
Loading
Loading