Generating unique Subscription ids for each request #135

dorjesinpo · 2023-10-25T14:39:41Z

Issue number of the reported bug or feature request: #133

2 Subscription Ids is not enough; we need 3.

In the order of increasing uniqueness:

CorrelationId - user provided.
SubscriptionHandle - uniquely identifies a Subscription in QueueOptions. But if reused for different App in the same queue, creates the issue
New unique id generated for each Subscription for each Configure request.

The existing logic involved storing request context associated with subscriptions (like, CorrelationId) until response arrives which makes the context effective. The logic is slightly refactored.

Ideally, the context belongs to BrokerSession::RequestManagerType::RequestSp but that calls for a bigger refactoring of the SDK.

quarter-note · 2023-10-25T18:12:49Z

To @678098, assuming that this is ready for review.

678098

Some comments

678098 · 2023-10-25T18:15:15Z

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

@@ -5239,7 +5245,13 @@ BrokerSession::createConfigureQueueContext(const bsl::shared_ptr<Queue>& queue,

            bmqp_ctrlmsg::Subscription subscription(d_allocator_p);

-            subscription.sId() = cit->first.id();
+            static bsls::AtomicUint s_nextId(0);


This static local variable initialization is not thread safe in general. The probability to observe a bug is small though.

Here are some details and advices:

https://bloomberg.github.io/bde-resources/doxygen/bde_api_prod/group__bslmt__once.html#_details

And here is how we did a safe initialization for another component:

blazingmq/src/groups/bmq/bmqt/bmqt_correlationid.cpp

Lines 33 to 42 in 08a6d16

CorrelationId CorrelationId::autoValue()

{

static bsls::AtomicInt* g_id_p = 0; // A unique id for each AutoValue

BSLMT_ONCE_DO

{

g_id_p = new bsls::AtomicInt(0);

// Heap allocate it to prevent 'exit-time-destructor needed' compiler

// warning. Causes valgrind-reported memory leak.

}

678098 · 2023-10-25T18:26:17Z

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

-            subscription.sId() = cit->first.id();
+            static bsls::AtomicUint s_nextId(0);
+
+            unsigned int internalSubscriptionId = s_nextId.add(1);


Suggested change

unsigned int internalSubscriptionId = s_nextId.add(1);

const unsigned int internalSubscriptionId = s_nextId.add(1);

678098 · 2023-10-25T18:32:27Z

src/groups/bmq/bmqimp/bmqimp_event.h

@@ -500,7 +500,7 @@ class Event {
    /// event's type() is MESSAGEEVENT, 'messageEventMode()' is READ and the
    /// underlying raw event is of type ACK, PUT or PUSH.
    void addCorrelationId(const bmqt::CorrelationId& correlationId,
-                          unsigned int               subscriptionId =
+                          unsigned int               subscriptionHandleId =
                              bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID);


The default value of bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID is a good default to initialize handle id in this case, but these integers have different meanings now, so might be good to introduce another constant k_DEFAULT_SUBSCRIPTION_HANDLE_ID.

678098 · 2023-10-25T18:35:03Z

src/groups/bmq/bmqimp/bmqimp_queuemanager.h

@@ -363,8 +352,9 @@ class QueueManager {
    /// `correlationId`, and return a shared pointer to the Queue object (if
    /// found), or an empty shared pointer (if not found).
    QueueSp lookupQueueBySubscriptionId(bmqt::CorrelationId* correlationId,
-                                        int                  queueId,
-                                        unsigned int subscriptionId) const;
+                                        unsigned int* subscriptionHandle,


We have subscriptionHandle in some places like here, and subscriptionHandleId in other places.

678098 · 2023-10-25T18:53:08Z

src/groups/bmq/bmqimp/bmqimp_queue.h

+
+    SubscriptionHandle result(cit->second);
+
+    d_registeredInternalSubscriptionIds.erase(cit);


Is it correct that Queue is accessed from one thread only? So modifications of d_registeredInternalSubscriptionIds container do not break anything?

Yes. It is the FSM thread. So is createConfigureQueueContext! So, there is no need for atomic or Once (or static).
Will commit new changes.

Signed-off-by: Vitaly Dzhitenov <[email protected]>

678098

LGTM!

678098 · 2023-11-05T00:46:33Z

src/groups/bmq/bmqt/bmqt_subscription.cpp

@@ -43,7 +43,7 @@ const int Subscription::k_DEFAULT_CONSUMER_PRIORITY        = 0;

 unsigned int SubscriptionHandle::nextId()
 {
-    static bsls::AtomicUint s_id = 0;
+    static bsls::AtomicUint s_id = k_INVALID_HANDLE_ID;


#141

I created an issue about this line. It's out of the scope of this PR and could be fixed in another PR.

dorjesinpo added the bug Something isn't working label Oct 25, 2023

dorjesinpo requested review from quarter-note and 678098 October 25, 2023 14:39

dorjesinpo requested a review from a team as a code owner October 25, 2023 14:39

dorjesinpo force-pushed the issue-133 branch from e6eec92 to f4c9784 Compare October 25, 2023 15:58

quarter-note assigned 678098 Oct 25, 2023

678098 reviewed Oct 25, 2023

View reviewed changes

678098 assigned dorjesinpo and unassigned 678098 Oct 25, 2023

dorjesinpo added 2 commits October 26, 2023 10:05

Generating unique Subscription ids for each request

c2649c0

Signed-off-by: Vitaly Dzhitenov <[email protected]>

clang-format

0c2681e

Signed-off-by: Vitaly Dzhitenov <[email protected]>

dorjesinpo force-pushed the issue-133 branch from f4c9784 to 0c2681e Compare October 26, 2023 15:08

Use BrokerSession::d_nextInternalSubscriptionId

b6e41ea

Signed-off-by: Vitaly Dzhitenov <[email protected]>

dorjesinpo force-pushed the issue-133 branch from d0c12fe to b6e41ea Compare October 26, 2023 15:41

dorjesinpo assigned 678098 and unassigned dorjesinpo Oct 26, 2023

678098 approved these changes Oct 27, 2023

View reviewed changes

678098 assigned dorjesinpo and unassigned 678098 Oct 27, 2023

678098 reviewed Nov 5, 2023

View reviewed changes

quarter-note merged commit cf0d323 into bloomberg:main Nov 6, 2023
6 checks passed

678098 mentioned this pull request Nov 7, 2023

[Do not merge]Sample: same sId assert #132

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Generating unique Subscription ids for each request #135

Generating unique Subscription ids for each request #135

dorjesinpo commented Oct 25, 2023

quarter-note commented Oct 25, 2023

678098 left a comment

678098 Oct 25, 2023

678098 Oct 25, 2023

678098 Oct 25, 2023 •

edited

Loading

678098 Oct 25, 2023

678098 Oct 25, 2023 •

edited

Loading

dorjesinpo Oct 26, 2023 •

edited

Loading

678098 left a comment

678098 Nov 5, 2023

	CorrelationId CorrelationId::autoValue()
	{
	static bsls::AtomicInt* g_id_p = 0; // A unique id for each AutoValue

	BSLMT_ONCE_DO
	{
	g_id_p = new bsls::AtomicInt(0);
	// Heap allocate it to prevent 'exit-time-destructor needed' compiler
	// warning. Causes valgrind-reported memory leak.
	}

	unsigned int internalSubscriptionId = s_nextId.add(1);
	const unsigned int internalSubscriptionId = s_nextId.add(1);


		SubscriptionHandle result(cit->second);

		d_registeredInternalSubscriptionIds.erase(cit);

Generating unique Subscription ids for each request #135

Generating unique Subscription ids for each request #135

Conversation

dorjesinpo commented Oct 25, 2023

quarter-note commented Oct 25, 2023

678098 left a comment

Choose a reason for hiding this comment

678098 Oct 25, 2023

Choose a reason for hiding this comment

678098 Oct 25, 2023

Choose a reason for hiding this comment

678098 Oct 25, 2023 • edited Loading

Choose a reason for hiding this comment

678098 Oct 25, 2023

Choose a reason for hiding this comment

678098 Oct 25, 2023 • edited Loading

Choose a reason for hiding this comment

dorjesinpo Oct 26, 2023 • edited Loading

Choose a reason for hiding this comment

678098 left a comment

Choose a reason for hiding this comment

678098 Nov 5, 2023

Choose a reason for hiding this comment

678098 Oct 25, 2023 •

edited

Loading

678098 Oct 25, 2023 •

edited

Loading

dorjesinpo Oct 26, 2023 •

edited

Loading