Feat[MQB]: Monolithic Virtual Storage #334

dorjesinpo · 2024-06-24T15:45:36Z

Replacing VirtualStorage collections of GUIDs, one for each app, with one collection per queue. The new collection allocates consecutive memory (vector) if needed to keep apps states for each GUID. Each VirtualStorage representing an App, gets assigned an ordinal to use as an index to access the state.
This allows us to:

get rid of insert operations for each App upon receiving each PUT in Primary, or PUSH in Replica or Proxy.
get rid of remove operations for each App upon message deletion
align the delivery (PUSHing); QueueEngine can access all Apps states. This lays the ground for Apps batching. Reliable Broadcast will rely on App batching (as in the case of synthetic Apps) to address scaling concerns.
get rid of maintaining Storage Iterators for each App (calling beforeMessageRemoved per App)
get rid of most Apps iterations. There are still some in evaluateAutoSubscriptions, initialization, and purging.

Replacing the extra VirtualStorageCatalog in RelayQueueEngine with new type of storage PushStream. It is short-lived, keeping GUIDs only until they get PUSHed.

CONFIRM logic is now inverted. CONFIRM is not a delete operation now but a change of the state.

678098

Some surface level comments

678098 · 2024-06-27T12:15:58Z

src/groups/mqb/mqba/mqba_application.cpp

+ resources.d_blobSpPool_p = &d_blobSpPool;
+ resources.d_bufferFactory_p = &d_bufferFactory;
+ resources.d_scheduler_p = d_scheduler_p;
+ resources.d_pushElementsPool_p = &d_pushElementsPool;


What about passing these fields to the constructor instead? My concern is about adding a new field in resources, with a field access like here it's easy to forget a new field somewhere. With a constructor extended with a new field, the compiler will point to all places where a new field is missing

Yeah, that was the old way and the constructor signature kept growing. I thought, it would help to encapsulate everything in Resources so when we need something new from mqba::Application far away from it, we don't have to change signatures across the entire chain of constructors.
The concern about missing field is addressed by the fact that if a component needs the field, the compilation will fail.
The concern about missing field value can be addressed by an assert

678098 · 2024-06-27T12:18:41Z

src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp

+ bmqp::SubQueueInfo(bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID));
+ }
+ else {
+ // Assume, RelayQueueEngien will use upstreramSubQueueIds as the


typo

Suggested change

// Assume, RelayQueueEngien will use upstreramSubQueueIds as the

// Assume, RelayQueueEngine will use upstreramSubQueueIds as the

678098 · 2024-06-27T12:19:19Z

src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp

+ // d_rdaInfo,
+ // bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID,
+ // mqbu::StorageKey::k_NULL_KEY);
+ // }


This code is not needed anymore, right?

src/groups/mqb/mqbblp/mqbblp_queueenginetester.h

678098 · 2024-06-27T12:27:54Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

@@ -397,6 +404,14 @@ void QueueEngineUtil::logRejectMessage(
 << "deleted." << MWCTSK_ALARMLOG_END;
 }

+bsls::Types::Int64 QueueEngineUtil::getMessageQueueTime(


In my previous PR, I turned this getMessageQueueTime to a local function, because it's not used outside the QueueEngineUtil cpp file. If it's still not used outside, let's keep it hidden from QueueEngineUtil interface.

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

678098 · 2024-06-27T12:44:04Z

src/groups/mqb/mqbblp/mqbblp_routers.t.cpp

 allocator)
 , d_iterator(d_storage.getIterator(mqbu::StorageKey()))
 , d_bufferFactory(32, allocator)
- , d_queue_sp(new (*allocator) mqbmock::Queue(0, allocator), allocator)
+ , d_queue_sp(new(*allocator) mqbmock::Queue(0, allocator), allocator)


This might be a bug in the local clang-format

Suggested change

, d_queue_sp(new(*allocator) mqbmock::Queue(0, allocator), allocator)

, d_queue_sp(new (*allocator) mqbmock::Queue(0, allocator), allocator)

The clang-format does without the space but the C++ Formatting Check does not pass so I had to manually insert the space.

678098 · 2024-06-27T12:45:20Z

src/groups/mqb/mqbblp/mqbblp_routers.t.cpp

 TEST_PROLOG(mwctst::TestHelper::e_DEFAULT);

+ // bslma::DefaultAllocatorGuard _defAllocGuard(&ma);


Changes in this file are not needed, right?

pniedzielski

Overall comments:

The logic for RelayQueueEngine becomes much clearer with this refactor. Even more so than in making way for reliable broadcast, I find this to be the most important change in this PR.
Maybe as the cost of that, I find the code for PushStream quite opaque. The code feels strongly coupled to the routing logic, whereas really it's a generic data structure with a few indices. I'm not sure if you hadn't explained the design in person to me that I would have been able to reconstruct it from this code. I don't need it to be pulled out into an mwcc-style component, but making the shape of the data structure clearer, in docs and in code, would go a long way towards making the routing logic more transparent.
I have many comments below about ClusterResources. Not because it's an important change, but because it's a change that (necessarily) touches so many places. It's a good change, rather than adding more and more parameters to all these constructors. However, I'm very concerned that each of the fields is nullptr in different places, and when I see an instance of ClusterResources, I have no good way of knowing what should and should not be nullptr. Ideally, none of them could be nullptr, and in the few places we keep them null we bite the bullet and construct something for them to point to. If some of them truly are optional, I'd like that to be documented. If all of them are optional, I think that needs to be reconsidered.
In Doxygen comments (triple slashes), backticks are used, not single ticks. I started noting those down, but the changes are extensive enough that I missed many of them.

I haven't taken a very deep look at the InMemoryStorage + FileBackedStorage changes.

pniedzielski · 2024-07-17T21:06:11Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+
+ PushStream(bdlma::ConcurrentPool* pushElementsPool);
+
+ iterator add(const bmqt::MessageGUID& guid);


This overload of add has a different return type than the one below. Although both member functions add something, what they actually do is quite different. They should probably have different names.

Good idea. Renaming it to findOrAppendMessage

pniedzielski · 2024-07-17T21:15:17Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+ /// Return the number of remaining Elements in the corresponding GUID.
+ unsigned int remove(Element* element);
+
+ /// Remove all PushStream Elements corresponding to the specified


The return values here are not documented, but you use them in mqbblp_relayqueueengine.cpp(538) (line number after applying the proposed patch). I assumed this was an error code when first looking at the interface, and didn't connect that it was the same as remove above. In retrospect, that's not so obscure, but better to document here too.

pniedzielski · 2024-07-17T21:21:43Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+
+ struct App {
+ Elements d_elements;
+ bsl::shared_ptr<RelayQueueEngine_AppState> d_app;


This is the coupling that means this data structure can't be its own component.

What we could try to do is to forward declare RelayQueueEngine_AppState and move PushStream, RelayQueueEngine_PushStorageIterator, and RelayQueueEngine_VirtualPushStorageIterator into mqbblp_pushstream.
And rename the latter two to PushStreamIterator and VirtualPushStreamIterator

I like this strategy. It should tame this file quite a bit.

pniedzielski · 2024-07-17T21:30:37Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+ // GUID or for each App. An 'Element' holding PUSH context for a GUID and
+ // an App is in two lists - for the GUID for the App. Removal can be
+ // random for the GUID list and always sequential for the App list.
+ // The storage still supports 'mqbi::StorageIterator' interface but the


This sentence is very concerning. What is the reason for supporting this interface? This isn't a Storage, and we really should not take an existing interface, change its semantics, and reuse it.

Adding comment

// The 'mqbi::StorageIterator' interface support is important for QueueEngines // which access data or delivery exclusively using 'mqbi::StorageIterator' // interface for both data in "real" storage or 'PushStream'.

pniedzielski · 2024-07-17T22:32:18Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+ apps.erase(d_iteratorApp);
+}
+
+inline mqbi::AppMessage* PushStream::Element::appState()


Should this and appView be const overloads of one another? If someone wants a const view of the AppMessage, they can store it in a pointer-to-const, or better yet, take a const view of the Element first.

The difference between appState and appView is more than a const. The latter will not allocate any state and the former will

pniedzielski · 2024-07-19T15:36:41Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

 };

+struct PushStream {


This documentation comment is quite terse, and seems to put explaining why this data structure is shaped the way it is before what properties the data structure has. I could eventually parse that information out, but a sentence at the start of "This data structure stores contexts of PUSHs, allowing for efficient sequential access by GUID and by App, and random-access removal for GUID." All the information here is good, but I could only make sense of it since you've explained the design before.

Drawing out the shape of the data structure also helped me here. I'm not sure if ASCII art is truly worth it, but it's complex to see what this data structure is just from the code (even though it's not terribly complex at a high level).

pniedzielski · 2024-07-19T15:39:46Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+
+ bdlma::ConcurrentPool* d_pushElementsPool_p;
+
+ PushStream(bdlma::ConcurrentPool* pushElementsPool);


Precondition on pushElementsPool is v. important.

pniedzielski · 2024-07-19T15:40:59Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+// class RelayQueueEngine_PushStorageIterator
+// ==========================================
+
+class RelayQueueEngine_PushStorageIterator : public mqbi::StorageIterator {


As mentioned above, should this really be a StorageIterator? Or rather, it's not a StorageIterator; it has different semantics. Is it one for convenience or is there a deeper need for it to be one?

pniedzielski · 2024-07-19T15:56:46Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+ unsigned int removeAll();
+
+ /// Create new Element associated with the specified 'info',
+ // 'upstreamSubQueueId', and 'iterator'.


Triple slash this line

pniedzielski · 2024-07-19T15:57:21Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h

+ // MANIPULATORS
+
+ /// If the specified 'isOutOfOrder' is 'true', insert the specified
+ // 'msgGUID' into the corresponding App redelivery list. Otherwise, insert


Triple slash this line.

pniedzielski · 2024-07-19T15:58:06Z

(I think perhaps having multiple commits could have made this review go faster.)

pniedzielski · 2024-07-19T18:09:23Z

src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp

 // CREATORS
 VirtualStorageCatalog::VirtualStorageCatalog(mqbi::Storage* storage,
 bslma::Allocator* allocator)
 : d_storage_p(storage)
 , d_virtualStorages(allocator)
+, d_avaialbleOrdinals(allocator)


Typo: available

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

678098 · 2024-07-20T05:30:16Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+ const bool isBroadcast = d_queue_p->isDeliverAll();
+
+ if (isBroadcast) {


After your changes, isBroadcast const used only once in this method. Therefore this is possible:

Suggested change

const bool isBroadcast = d_queue_p->isDeliverAll();

if (isBroadcast) {

if (d_queue_p->isDeliverAll()) {

More general question, why do we check if it's a broadcast in two different ways:
a. const bool isBroadcast = d_queue_p->isDeliverAll();
b. QueueEngineUtil::isBroadcastMode(d_queue_p) -> equal to return queue->isDeliverAll() && queue->isAtMostOnce();
Don't we need to also check d_queue_p->isAtMostOnce() here?

Why don't we introduce bool isBroadcast() const for mqbi::Queue *d_queue_p? Doing so will simplify QueueEngineUtil interface.

Up until reliable broadcast, isDeliverAll() and isAtMostOnce() are synonyms. Reliable broadcast can be isDeliverAll() && !isAtMostOnce(). Let's return to this, when we introduce this queue mode

678098 · 2024-07-24T17:53:04Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+ const mqbi::AppMessage& appView = d_currentMessage->appMessageView(
+ ordinal);
+ Routers::Result result = Routers::e_SUCCESS;
+
+ if (appView.isNew()) {


What do you think of a small refactor by early return here, so we don't create result until it's needed?

Suggested change

const mqbi::AppMessage& appView = d_currentMessage->appMessageView(

ordinal);

Routers::Result result = Routers::e_SUCCESS;

if (appView.isNew()) {

const mqbi::AppMessage& appView = d_currentMessage->appMessageView(

ordinal);

if (!appView.isNew()) {

d_doRepeat = true;

return true; // RETURN

}

const Routers::Result result = app.selectConsumer(...

678098 · 2024-07-24T18:02:13Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+ d_timeDelta = getMessageQueueTime(d_currentMessage->attributes());
+ }
+ return d_timeDelta.value();
+}


From the signature, it looks like an accessor, but instead it caches the field and returns it. Do we really need this method?
Why don't we reset d_timeDelta on reset()?

Do we really need this method?

We need the cache to avoid extra work.

Yes, we need to reset it in reset

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

678098 · 2024-07-24T19:05:10Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+ d_appOrdinal = ordinal;
+ d_isAuthorized = true;
+
+ return true;


Suggested change

return true;

return true; // RETURN

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

kaikulimu

Some comments on mqbblp::RootQueueEngine

kaikulimu · 2024-07-24T18:43:04Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

- size_t numMessages =
- app->deliverMessages(&delay, key, *d_queueState_p->storage(), appId);
+ size_t numMessages = app->deliverMessages(&delay,
+ d_realStorageIter_mp.get(),


Improve alignment

d_realStorageIter_mp.get() corresponds to mqbi::StorageIterator* reader in QueueEngineUtil_AppState::deliverMessages(). Could you explain the meaning of reader?

reader is the interface to read date out of whatever storage we use.
Adding

/// Use the specified `reader` to read data for delivery.

kaikulimu · 2024-07-24T19:07:04Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.h

@@ -140,6 +140,12 @@ class RootQueueEngine BSLS_KEYWORD_FINAL : public mqbi::QueueEngine {
 // Throttler for when reject messages
 // are dumped into temp files.

+ bslma::ManagedPtr<mqbi::StorageIterator> d_storageIter_mp;


Could you please add more explanation to the difference between d_storageIter_mp and d_realStorageIter_mp? This is confusing to any reader who does not already have deep understanding of the component.

Added

// Storage iterator to the logical stream of messages. // Queue Engine iterates this one sequentially.

and

// Storage iterator to access storage state. // Queue Engine uses this one to access random message (as in the case of // redelivery).

kaikulimu · 2024-07-24T19:53:41Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+ // The message is gone because of either GC or purge
+ // In either case, start at the beginning


Suggested change

// The message is gone because of either GC or purge

// In either case, start at the beginning

// The message is gone because of either GC or purge.

// In either case, start at the beginning.

kaikulimu · 2024-07-24T20:00:55Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.h

- mqbi::Storage& storage,
- const bsl::string& appId);
+ size_t deliverMessages(bsls::TimeInterval* delay,
+ mqbi::StorageIterator* reader,


Explain the meaning of reader

/// Use the specified `reader` to read data for delivery.

kaikulimu · 2024-07-24T20:21:32Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+ else {
+ start = d_storageIter_mp.get();
+ }
+
 bsls::TimeInterval delay;


Can be const bsls::TimeInterval delay; and const size_t numMessages

the delay is an out parameter for deliverMessages
the numMessages is const, indeed

kaikulimu · 2024-07-24T21:32:24Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp


 app->setUpstreamSubQueueId(upstreamSubQueueId);
+ // Do not copy resumePoint. New RootQueueEngine redelivers everything
+ // unconfirmed; its ierator is at the beginning.


Suggested change

// unconfirmed; its ierator is at the beginning.

// unconfirmed; its iterator is at the beginning.

kaikulimu · 2024-07-26T19:20:43Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+ start = storageIter_mp.get();
+ }
+ else {
+ start = d_storageIter_mp.get();


Suggested change

start = d_storageIter_mp.get();

// TODO comment

start = d_storageIter_mp.get();

Where does d_storageIter_mp point to at this point of time? Can you put a small comment explaining?

Added

// 'start' points at either the resume point (if found) or the first // unconfirmed message of the 'app' (if not found).

and

// 'start' points at the next message in the logical stream (common // for all apps).

kaikulimu · 2024-07-26T19:27:38Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

 }
 }
+ if (!context.isEmpty()) {
+ // Report 'queue time' metric for the entire queue


Was this 'queue time' metric reported at another place in the existing code? Did you move the logic here?

Yes, it was reported in the QueueEngineUtil_AppState/QueueEngineUtil_AppsDeliveryContext upon checking if this is Primary (RootQueueEngine).
Moved it to. the RootQueueEngine

kaikulimu · 2024-07-26T19:39:49Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

- for (Apps::iterator iter = d_apps.begin(); iter != d_apps.end(); ++iter) {
- iter->value()->beforeMessageRemoved(msgGUID, true);
+
+ // PRECONDITIONS


No need to write // PRECONDITIONS twice

kaikulimu · 2024-07-26T19:49:29Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

@@ -1820,8 +1878,10 @@ void RootQueueEngine::loadInternals(mqbcmd::QueueEngine* out) const
 consumerState.appId() = iter->key1();

 if (d_queueState_p->storage()->hasVirtualStorage(iter->key1())) {
- consumerState.isAtEndOfStorage().makeValue(
- iter->value()->d_storageIter_mp->atEnd());
+ bool isAtEndOfStorage = iter->value()->isAtEndOfStorage() &&


Suggested change

bool isAtEndOfStorage = iter->value()->isAtEndOfStorage() &&

const bool isAtEndOfStorage = iter->value()->isAtEndOfStorage() &&

pniedzielski · 2024-08-15T15:19:57Z

src/groups/mqb/mqba/mqba_application.cpp

- resources.d_bufferFactory_p = &d_bufferFactory;
- resources.d_scheduler_p = d_scheduler_p;
- resources.d_pushElementsPool_p = &d_pushElementsPool;
+ mqbi::ClusterResources resources(


👍🏻

Recording for posterity: one of the things we thought about and could implement in the future if needed is to make the pools lazy-initialized, so they're only created when first needed. We're not doing this now, but it could be worthwhile.

pniedzielski · 2024-08-15T15:20:59Z

src/groups/mqb/mqbi/mqbi_storage.h

@@ -669,17 +670,17 @@ inline AppMessage::AppMessage(const bmqp::RdaInfo& rdaInfo)
 // NOTHING
 }

-inline void AppMessage::onPush()
+inline void AppMessage::setPushState()


To me, at least, this change makes the code a little clearer.

pniedzielski · 2024-08-15T15:21:57Z

src/groups/mqb/mqbi/mqbi_cluster.h

+inline ClusterResources::ClusterResources(
+ bdlmt::EventScheduler* scheduler,
+ bdlbb::BlobBufferFactory* bufferFactory,
+ BlobSpPool* blobSpPool)
+: d_scheduler_p(scheduler)
+, d_bufferFactory_p(bufferFactory)
+, d_blobSpPool_p(blobSpPool)
+, d_pushElementsPool()
+{
+ BSLS_ASSERT_SAFE(d_scheduler_p);
+ BSLS_ASSERT_SAFE(d_bufferFactory_p);
+ BSLS_ASSERT_SAFE(d_blobSpPool_p);
+}
+
+inline ClusterResources::ClusterResources(
+ bdlmt::EventScheduler* scheduler,
+ bdlbb::BlobBufferFactory* bufferFactory,
+ BlobSpPool* blobSpPool,
+ bdlma::ConcurrentPool* pushElementsPool)
+: d_scheduler_p(scheduler)
+, d_bufferFactory_p(bufferFactory)
+, d_blobSpPool_p(blobSpPool)
+, d_pushElementsPool(pushElementsPool)
+{
+ BSLS_ASSERT_SAFE(d_scheduler_p);
+ BSLS_ASSERT_SAFE(d_bufferFactory_p);
+ BSLS_ASSERT_SAFE(d_blobSpPool_p);
+ BSLS_ASSERT_SAFE(d_pushElementsPool);
+}


I like this solution.

pniedzielski · 2024-08-15T15:23:01Z

src/groups/mqb/mqbblp/mqbblp_pushstream.h

@@ -0,0 +1,697 @@
+// Copyright 2024 Bloomberg Finance L.P.


kaikulimu

lgtm

678098

Notes about PUSH STREAM

678098 · 2024-08-13T01:26:06Z

src/groups/mqb/mqba/mqba_application.cpp

@@ -152,6 +153,8 @@ Application::Application(bdlmt::EventScheduler* scheduler,
 bdlf::PlaceHolders::_2), // allocator
 k_BLOB_POOL_GROWTH_STRATEGY,
 d_allocators.get("BlobSpPool"))
+, d_pushElementsPool(sizeof(mqbblp::PushStream::Element),
+ d_allocators.get("PushElementsPool"))


Note that the default implementation of this concurrent pool uses geometric growth limited by 32. Does it worth it to set a larger const growth?

https://github.com/bloomberg/bde/blob/6ec108c973912eea7f009df971c553e7b8fa5ed5/groups/bdl/bdlma/bdlma_concurrentpool.cpp#L45

We do not expect large volume here as it is highly volatile. Once message is pushed, it gets removed

src/groups/mqb/mqbblp/mqbblp_clusterproxy.h

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

src/groups/mqb/mqbblp/mqbblp_pushstream.h

678098 · 2024-08-16T03:45:00Z

src/groups/mqb/mqbblp/mqbblp_pushstream.h

+ Apps::iterator itApp = d_apps.find(upstreamSubQueueId);
+
+ unsigned int numMessages = 0;
+ if (itApp != d_apps.end()) {
+ numMessages = removeApp(itApp);
+ }
+
+ return numMessages;


Suggested change

Apps::iterator itApp = d_apps.find(upstreamSubQueueId);

unsigned int numMessages = 0;

if (itApp != d_apps.end()) {

numMessages = removeApp(itApp);

}

return numMessages;

Apps::iterator itApp = d_apps.find(upstreamSubQueueId);

if (itApp == d_apps.end()) { // predict unlikely?

return 0; // RETURN

}

return removeApp(itApp);

src/groups/mqb/mqbblp/mqbblp_pushstream.h

678098 · 2024-08-16T03:52:49Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+ << "' could not redeliver GUID: '" << *it
+ << "' (not in the storage)";
+ }
+ else if (!reader->appMessageView(ordinal()).isPending()) {


Should this condition also be UNLIKELY?

hmm. I think, this would be a confirmed or gc'ed message. BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY would not hurt

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

src/groups/mqb/mqbblp/mqbblp_remotequeue.t.cpp

bmq-oss-ci

Build 186 of commit 42995d7 has completed with FAILURE

678098 · 2024-08-29T13:46:00Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

+{
+ if (!d_pushElementsPool_sp) {
+ d_pushElementsPool_sp.load(
+ new (*allocator) bdlma::ConcurrentPool(sizeof(Element), allocator),


Without bslma::Default::allocator(allocator) guard this code will crash if we provide allocator == 0.
We might either add a precondition or use bslma::Default::allocator(allocator) before allocating memory.

678098 · 2024-08-29T13:50:07Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

+ // loaded in 'appData', 'options' or 'attributes' routines.
+ d_appData_sp.reset();
+ d_options_sp.reset();
+ d_attributes.reset();


In a normal workflow, we do 3 resets just before we set these fields again in loadMessageAndAttributes. I think making a lazy reset might be good for performance.
For example, reset only d_appData_sp (which we check anyway in loadMessageAndAttributes), and reassign all 3 fields where needed.
We might do it in a separate PR with performance measurements

We explicitly call loadMessageAndAttributes when accessing d_attributes and d_options_sp, so we could leave them. On the other hand, an instance of PushStreamIterator would be in an inconsistent state. Is the benefit worth it?

This inconsistency only means that some fields are not reseted between calls, but we do have the d_appData_sp pointer which indicates that these fields are not reseted. So we can guarantee safe usage via class API. The non-initialized fields will only be visible if someone looks at the binary representation of this class with debugger.

678098 · 2024-08-29T14:01:25Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

+ // doKeepGuid because of the d_iterator
+
+ if (d_iterator->second.numElements() == 0) {
+ // d_currentElement->eraseFromStream(d_owner_p->d_stream);


Is this code needed?

678098 · 2024-08-29T14:23:13Z

src/groups/mqb/mqbblp/mqbblp_pushstream.t.cpp

+#include <mqbcfg_messages.h>
+#include <mqbs_inmemorystorage.h>
+
+#include <mwcu_memoutstream.h>


Suggested change

#include <mqbcfg_messages.h>

#include <mqbs_inmemorystorage.h>

#include <mwcu_memoutstream.h>

// MQB

#include <mqbcfg_messages.h>

#include <mqbs_inmemorystorage.h>

// MWC

#include <mwcu_memoutstream.h>

678098 · 2024-08-29T14:36:41Z

src/groups/mqb/mqbblp/mqbblp_pushstream.t.cpp

+ case 1: test1_basic(); break;
+ case 2: test2_iterations(); break;


Typically we have a descending order, not sure if UTs runner relies on this

Suggested change

case 1: test1_basic(); break;

case 2: test2_iterations(); break;

case 2: test2_iterations(); break;

case 1: test1_basic(); break;

hmm. I thought otherwise. For example, mqbblp_remotequeue.t

bmq-oss-ci

Build 229 of commit 1604b8c has completed with FAILURE

bmq-oss-ci

Build 248 of commit 31fc7b9 has completed with FAILURE

bmq-oss-ci

Build 261 of commit 31d7712 has completed with FAILURE

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

bmq-oss-ci

Build 268 of commit 8f4320e has completed with FAILURE

dorjesinpo added the enhancement New feature or request label Jun 24, 2024

dorjesinpo self-assigned this Jun 24, 2024

dorjesinpo requested a review from a team as a code owner June 24, 2024 15:45

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch 7 times, most recently from 9273dab to db1187f Compare June 26, 2024 21:52

678098 reviewed Jun 27, 2024

View reviewed changes

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch from db1187f to 5a63af8 Compare June 27, 2024 17:00

pniedzielski requested changes Jul 19, 2024

View reviewed changes

678098 self-requested a review July 19, 2024 17:12

pniedzielski reviewed Jul 19, 2024

View reviewed changes

kaikulimu self-requested a review July 22, 2024 22:15

678098 reviewed Jul 25, 2024

View reviewed changes

kaikulimu requested changes Jul 26, 2024

View reviewed changes

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch 2 times, most recently from 8c1652c to 1606a42 Compare August 12, 2024 16:42

pniedzielski previously approved these changes Aug 15, 2024

View reviewed changes

kaikulimu previously approved these changes Aug 15, 2024

View reviewed changes

678098 reviewed Aug 16, 2024

View reviewed changes

dorjesinpo dismissed stale reviews from kaikulimu and pniedzielski via ea7463d August 21, 2024 17:31

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch 4 times, most recently from 71c0b35 to 777a3a3 Compare August 21, 2024 18:25

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch from 777a3a3 to 42995d7 Compare August 21, 2024 21:28

dorjesinpo assigned 678098 and unassigned dorjesinpo Aug 21, 2024

bmq-oss-ci bot reviewed Aug 27, 2024

View reviewed changes

678098 changed the title ~~Monolithic Virtual Storage~~ Feat[MQB]: Monolithic Virtual Storage Aug 29, 2024

678098 reviewed Aug 29, 2024

View reviewed changes

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch 2 times, most recently from c2a1396 to 1604b8c Compare September 3, 2024 15:24

bmq-oss-ci bot reviewed Sep 3, 2024

View reviewed changes

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch from 1604b8c to 31fc7b9 Compare September 10, 2024 13:58

bmq-oss-ci bot reviewed Sep 10, 2024

View reviewed changes

678098 previously approved these changes Sep 11, 2024

View reviewed changes

dorjesinpo dismissed 678098’s stale review via 244398f September 18, 2024 19:45

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch 2 times, most recently from 244398f to 31d7712 Compare September 18, 2024 19:47

bmq-oss-ci bot reviewed Sep 18, 2024

View reviewed changes

dorjesinpo added 9 commits September 19, 2024 11:17

WIP

68ff8f6

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

Monolithic Virtual Storage

ffac11c

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

formatting

fd3f8fe

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

addressing reviews

cef1837

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

merge

85d4a09

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

addressing review

61705d6

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

adding UT

1c84a1e

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

merge

cac0d8f

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

addressing review

8f4320e

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

dorjesinpo force-pushed the dev/refactor-virtusl-storage branch from 31d7712 to 8f4320e Compare September 19, 2024 15:17

678098 approved these changes Sep 19, 2024

View reviewed changes

bmq-oss-ci bot reviewed Sep 19, 2024

View reviewed changes

	// Assume, RelayQueueEngien will use upstreramSubQueueIds as the
	// Assume, RelayQueueEngine will use upstreramSubQueueIds as the

	, d_queue_sp(new(*allocator) mqbmock::Queue(0, allocator), allocator)
	, d_queue_sp(new (*allocator) mqbmock::Queue(0, allocator), allocator)

		TEST_PROLOG(mwctst::TestHelper::e_DEFAULT);

		// bslma::DefaultAllocatorGuard _defAllocGuard(&ma);


		PushStream(bdlma::ConcurrentPool* pushElementsPool);

		iterator add(const bmqt::MessageGUID& guid);


		bdlma::ConcurrentPool* d_pushElementsPool_p;

		PushStream(bdlma::ConcurrentPool* pushElementsPool);

		const bool isBroadcast = d_queue_p->isDeliverAll();

		if (isBroadcast) {

		// The message is gone because of either GC or purge
		// In either case, start at the beginning

	// unconfirmed; its ierator is at the beginning.
	// unconfirmed; its iterator is at the beginning.

	start = d_storageIter_mp.get();
	// TODO comment
	start = d_storageIter_mp.get();

	bool isAtEndOfStorage = iter->value()->isAtEndOfStorage() &&
	const bool isAtEndOfStorage = iter->value()->isAtEndOfStorage() &&

		case 1: test1_basic(); break;
		case 2: test2_iterations(); break;

Feat[MQB]: Monolithic Virtual Storage #334

Are you sure you want to change the base?

Feat[MQB]: Monolithic Virtual Storage #334

Conversation

dorjesinpo commented Jun 24, 2024 • edited Loading

678098 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

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

pniedzielski 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pniedzielski commented Jul 19, 2024

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kaikulimu left a comment • edited Loading

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

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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kaikulimu left a comment

Choose a reason for hiding this comment

678098 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

Choose a reason for hiding this comment

bmq-oss-ci bot 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

678098 Sep 11, 2024 • edited Loading

dorjesinpo commented Jun 24, 2024 •

edited

Loading

kaikulimu left a comment •

edited

Loading

678098 Sep 11, 2024 •

edited

Loading