Skip to content

Commit 6d4a104

Browse files
Perf[bmqc]: stop MQTP queues in parallel (bloomberg#1557)
Split the serial stop loop in `MultiQueueThreadPool::stop()` into two phases so all queues drain concurrently rather than sequentially. The first loop signals every queue (pushBack NULL, disablePushBack), the second waits on each queue's `d_finished` semaphore. This is safe because each queue's shutdown is self-contained: its own semaphore, ref count, and backing SingleConsumerQueue with no cross-queue state. The ref count nets to zero per queue (+1 enqueue monitor, -1 release initial reference), so finish order is irrelevant. Signed-off-by: Srinath Satuluri <satulurisrinath@gmail.com>
1 parent 70f0a9b commit 6d4a104

1 file changed

Lines changed: 6 additions & 0 deletions

File tree

src/groups/bmq/bmqc/bmqc_multiqueuethreadpool.h

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -865,6 +865,7 @@ inline void MultiQueueThreadPool<TYPE>::stop()
865865
&d_warningMonitorEventHandle);
866866
}
867867

868+
// Signal all queues to stop.
868869
for (size_t i = 0; i < d_queues.size(); ++i) {
869870
QueueInfo& info = *d_queues[i];
870871

@@ -882,6 +883,11 @@ inline void MultiQueueThreadPool<TYPE>::stop()
882883
// It is possible that something is enqueued to the queue between the
883884
// last monitor event and `disablePushBack()` call, this is expected.
884885
info.d_queue_p->disablePushBack();
886+
}
887+
888+
// Wait for all queues to finish, then drain and delete.
889+
for (size_t i = 0; i < d_queues.size(); ++i) {
890+
QueueInfo& info = *d_queues[i];
885891

886892
const bsls::TimeInterval timeout =
887893
bsls::SystemTime::nowMonotonicClock().addSeconds(

0 commit comments

Comments
 (0)