Perf[MQB]: fix linear iteration over all queues on any replication #525

678098 · 2024-11-19T12:50:17Z

Notify only weak consistency queues on replicated batch, and only ones that were affected by storage event builder flush.
Remove obsolete d_hasNewMessages flag from mqbblp::LocalQueue.
Cancel notifications for queues that changed their consistency from weak to strong on domain reconfigure.
Bugfix: set FileBackedStorage::d_hasReceipts flag on domain reconfigure when consistency changes.

Signed-off-by: Evgeny Malygin <[email protected]>

678098 · 2024-11-25T16:11:39Z

src/groups/mqb/mqbs/mqbs_filebackedstorage.h

@@ -203,7 +203,7 @@ class FileBackedStorage BSLS_KEYWORD_FINAL : public ReplicatedStorage {
    bmqp::SchemaLearner::Context d_schemaLearnerContext;
    // Context for replicated data.

-    const bool d_hasReceipts;
+    bool d_hasReceipts;


We set this flag once and didn't update it before. It is also used here for early return:

blazingmq/src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp

Lines 216 to 221 in c0f8832

bool FileBackedStorage::hasReceipt(const bmqt::MessageGUID& msgGUID) const

{

if (d_hasReceipts) {

// Weak consistency

return true; // RETURN

}

678098 · 2024-11-25T16:13:06Z

src/groups/mqb/mqbi/mqbi_storage.h

@@ -408,6 +408,10 @@ class Storage {
                          const bsls::Types::Int64 messageTtl,
                          const int                maxDeliveryAttempts) = 0;

+    /// Set the consistency level associated to this storage to the specified
+    /// `value`.
+    virtual bool setConsistency(const mqbconfm::Consistency& value) = 0;


I decided to introduce a separate setter instead of extending configure, because sometimes we call configure without setting new consistency.

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci

Build 385 of commit 4fc5276 has completed with FAILURE

dorjesinpo

Looks good, we just need to revisit cancelReplicationNotification logic

dorjesinpo · 2024-11-26T15:19:30Z

src/groups/mqb/mqbs/mqbs_filestore.cpp

+                 d_replicationNotifications.begin();
+             it != d_replicationNotifications.end();
+             it++) {
+            StoragesMap::iterator storageIt = d_storages.find(*it);


There is some possibility to avoid this second lookup but then we need to rely on storage being empty when we remove it. Maybe, we can just add a comment now.

Added a comment, let's revisit it later

dorjesinpo · 2024-11-26T15:21:53Z

src/groups/mqb/mqbs/mqbs_filestore.cpp

-            ReplicatedStorage* rs = it->second;
-            if (rs->queue()) {
-                rs->queue()->onReplicatedBatch();
+        // good time to flush weak consistency queues.


This overlaps with the code in dispatcherFlush . Maybe, we can refactor a bit

I removed code duplication and also splitted dispatcherFlush -> flushStorage, flushQueues

dorjesinpo · 2024-11-26T15:38:39Z

src/groups/mqb/mqbs/mqbs_filestore.h

+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// No effect if `queueKey` is not found in this storage or if there are no
+    /// current registered notifications for this queue.
+    void cancelReplicationNotification(const mqbu::StorageKey& queueKey);


When we apply new config, the old messages may keep their logic. For example, DataStoreRecord::d_hasReceipt is always true, if at the time of message creation, the consistency was weak and false otherwise, until we get the receipt. And generating receipts should be controlled by header flags (not the configuration)

In this light, is there a possibility to simplify and get by without cancelReplicationNotification? The notification of "in-transit" messages will treat then as "weak" which is correct. As I understand, those messages will never get receipts.

The statement about old messages keeping their logic has one flaw when reconfiguring from strong to weak. That is because of the early return in

bool FileBackedStorage::hasReceipt(const bmqt::MessageGUID& msgGUID) const { if (d_hasReceipts) { // Weak consistency return true; // RETURN }

We have options here

Remove the early return

Accept the fact that "strong to weak" change applies to existing messages.

anything else?

I thing either is fine.

This early return is helpful because it prevents us from doing this lookup a few lines below:

RecordHandleMap::const_iterator it = d_handles.find(msgGUID);

I think it's nice for performance that we don't access OrderedHashMapWithHistory here. Option (2) is fine by me. We do have a thread race anyway between reconfigure and incoming messages.

I also removed cancelReplicationNotification to keep things simple

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci

Build 391 of commit f7be85e has completed with FAILURE

dorjesinpo

This is getting better and better.
One more concern about flushQueue API.

dorjesinpo · 2024-11-26T22:33:32Z

src/groups/mqb/mqbi/mqbi_storage.h

+    /// undefined unless this cluster node is the primary for this partition.
+    virtual void flushStorage() = 0;
+
+    /// Flush all associated weak consistency queues.  Behaviour is


But we flush only if d_storageEventBuilder is empty. See another comment about flushQueues

Updated the comment

dorjesinpo · 2024-11-26T22:45:02Z

src/groups/mqb/mqbblp/mqbblp_localqueue.cpp

@@ -538,7 +533,7 @@ void LocalQueue::postMessage(const bmqp::PutHeader&              putHeader,

    // If 'FileStore::d_storageEventBuilder' is flushed, flush all relevant
    // queues (call 'afterNewMessage' to deliver accumulated data)
-    d_state_p->storage()->dispatcherFlush(false, true);
+    d_state_p->storage()->flushQueues();


This (already existing) logic is a bit unclear. Please, check the following reasoning:

LocalQueue::postMessage calls put which adds to the builder in the FileStore.

FileStore tracks the builder size and does FileStore::flushIfNeeded which used to call notifyQueuesOnReplicatedBatch

Then, why do we need this call dispatcherFlush(false, true)/flushQueues() here?

If we don't, then there is no need to have flushQueues as an interface method. It is entirely private business of FileStore (and something like notifyQueuesOnReplicatedBatch seems like a better name)

That is correct, this call is unnecessary. Removed & simplified

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci

Build 392 of commit 7a4de87 has completed with FAILURE

dorjesinpo

thank you!

chrisbeard · 2024-12-06T16:09:29Z

src/groups/mqb/mqbs/mqbs_datastore.h

-    /// all associated queues.  Behavior is undefined unless this node is
-    /// the primary for this partition.
-    virtual void dispatcherFlush(bool storage, bool queues) = 0;
+    /// all associated queues.


@678098 should this comment from 685-687 be removed? Looks like flushStorage() docs don't need this bit anymore

Will clean this up

678098 requested a review from a team as a code owner November 19, 2024 12:50

678098 requested review from dorjesinpo and chrisbeard and removed request for a team November 19, 2024 12:50

Register queues for notifications

0e2382d

Signed-off-by: Evgeny Malygin <[email protected]>

678098 force-pushed the t2577_fix_queue_linear_iteration branch from e4b086b to 0e2382d Compare November 22, 2024 14:55

678098 added 6 commits November 22, 2024 15:29

Remove d_hasNewMessages flag

3a2b28a

Signed-off-by: Evgeny Malygin <[email protected]>

Storage key lookup

73c5912

Signed-off-by: Evgeny Malygin <[email protected]>

Better docs

764c0a7

Signed-off-by: Evgeny Malygin <[email protected]>

Better docs

b31d995

Signed-off-by: Evgeny Malygin <[email protected]>

Fix: handle replication notifications on domain reconfigure

b9f5335

Signed-off-by: Evgeny Malygin <[email protected]>

Fix: correctly set d_hasReceipts for FileBackedStorage on configure

c0f8832

Signed-off-by: Evgeny Malygin <[email protected]>

678098 commented Nov 25, 2024

View reviewed changes

678098 assigned dorjesinpo Nov 25, 2024

Fix: setConsistency returns void

4fc5276

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci bot reviewed Nov 25, 2024

View reviewed changes

dorjesinpo reviewed Nov 26, 2024

View reviewed changes

dorjesinpo assigned 678098 and unassigned dorjesinpo Nov 26, 2024

678098 added 3 commits November 26, 2024 18:26

Refactor[MQB]: split dispatcherFlush into flushStorage, flushQueues

8c1790e

Signed-off-by: Evgeny Malygin <[email protected]>

Docs[MQB]: add a comment

0124dd3

Signed-off-by: Evgeny Malygin <[email protected]>

Remove cancelReplicationNotification

1949b2b

Signed-off-by: Evgeny Malygin <[email protected]>

678098 requested a review from dorjesinpo November 26, 2024 18:49

678098 assigned dorjesinpo and unassigned 678098 Nov 26, 2024

Docs: fix docs

f7be85e

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci bot reviewed Nov 26, 2024

View reviewed changes

dorjesinpo reviewed Nov 26, 2024

View reviewed changes

dorjesinpo assigned 678098 Nov 26, 2024

dorjesinpo removed their assignment Nov 26, 2024

Remove flushQueues

7a4de87

Signed-off-by: Evgeny Malygin <[email protected]>

bmq-oss-ci bot reviewed Nov 28, 2024

View reviewed changes

678098 assigned dorjesinpo and unassigned 678098 Nov 29, 2024

dorjesinpo approved these changes Dec 3, 2024

View reviewed changes

dorjesinpo merged commit efe2770 into bloomberg:main Dec 3, 2024
40 checks passed

678098 deleted the t2577_fix_queue_linear_iteration branch December 3, 2024 21:46

chrisbeard reviewed Dec 6, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Perf[MQB]: fix linear iteration over all queues on any replication #525

Perf[MQB]: fix linear iteration over all queues on any replication #525

678098 commented Nov 19, 2024 •

edited

Loading

678098 Nov 25, 2024

678098 Nov 25, 2024

bmq-oss-ci bot left a comment

dorjesinpo left a comment

dorjesinpo Nov 26, 2024

678098 Nov 26, 2024 •

edited

Loading

dorjesinpo Nov 26, 2024

678098 Nov 26, 2024 •

edited

Loading

dorjesinpo Nov 26, 2024

678098 Nov 26, 2024 •

edited

Loading

bmq-oss-ci bot left a comment

dorjesinpo left a comment

dorjesinpo Nov 26, 2024

678098 Nov 28, 2024

dorjesinpo Nov 26, 2024

678098 Nov 28, 2024

bmq-oss-ci bot left a comment

dorjesinpo left a comment

chrisbeard Dec 6, 2024

678098 Dec 6, 2024

	bool FileBackedStorage::hasReceipt(const bmqt::MessageGUID& msgGUID) const
	{
	if (d_hasReceipts) {
	// Weak consistency
	return true; // RETURN
	}

Perf[MQB]: fix linear iteration over all queues on any replication #525

Perf[MQB]: fix linear iteration over all queues on any replication #525

Conversation

678098 commented Nov 19, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

dorjesinpo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

678098 Nov 26, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

678098 Nov 26, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

678098 Nov 26, 2024 • edited Loading

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

dorjesinpo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

dorjesinpo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

678098 commented Nov 19, 2024 •

edited

Loading

678098 Nov 26, 2024 •

edited

Loading

678098 Nov 26, 2024 •

edited

Loading

678098 Nov 26, 2024 •

edited

Loading