Fix[BMQ,MQB]: default allocations (#830)

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>

Author: 678098

src/groups/bmq/bmqimp/bmqimp_application.cpp

@@ -381,8 +381,8 @@ bmqt::GenericResult::Enum Application::startChannel()
     bsls::TimeInterval attemptInterval;
     attemptInterval.setTotalMilliseconds(k_RECONNECT_INTERVAL_MS);
 
-    bmqio::Status         status;
-    bmqio::ConnectOptions options;
+    bmqio::Status         status(&d_allocator);
+    bmqio::ConnectOptions options(&d_allocator);
     options.setEndpoint(out.str())
         .setNumAttempts(k_RECONNECT_COUNT)
         .setAttemptInterval(attemptInterval)
@@ -392,12 +392,13 @@ bmqt::GenericResult::Enum Application::startChannel()
         &status,
         &d_connectHandle_mp,
         options,
-        bdlf::BindUtil::bind(&Application::channelStateCallback,
-                             this,
-                             options.endpoint(),
-                             bdlf::PlaceHolders::_1,    // event
-                             bdlf::PlaceHolders::_2,    // status
-                             bdlf::PlaceHolders::_3));  // channel
+        bdlf::BindUtil::bindS(&d_allocator,
+                              &Application::channelStateCallback,
+                              this,
+                              options.endpoint(),
+                              bdlf::PlaceHolders::_1,    // event
+                              bdlf::PlaceHolders::_2,    // status
+                              bdlf::PlaceHolders::_3));  // channel
     if (!status) {
         BALL_LOG_ERROR << id() << "Failed to connect to broker at '"
                        << d_sessionOptions.brokerUri()
@@ -453,7 +454,7 @@ void Application::printStats(bool isFinal)
     //         executed by the *SCHEDULER* thread
     // (and by the *MAIN* thread (in destructor))
 
-    bmqu::MemOutStream os;
+    bmqu::MemOutStream os(&d_allocator);
 
     os << "#### stats [delta = last "
        << d_sessionOptions.statsDumpInterval().seconds() << " seconds] ####\n";
@@ -482,8 +483,8 @@ void Application::printStats(bool isFinal)
     os << "::::: TCP Channels >>";
     if (isFinal) {
         // For the final stats, no need to print the 'delta' columns
-        bmqst::Table                  table;
-        bmqst::BasicTableInfoProvider tip;
+        bmqst::Table                  table(&d_allocator);
+        bmqst::BasicTableInfoProvider tip(&d_allocator);
         bmqio::StatChannelFactoryUtil::initializeStatsTable(
             &table,
             &tip,
@@ -577,14 +578,15 @@ Application::Application(
       bmqp::BlobPoolUtil::createBlobPool(&d_blobBufferFactory,
                                          d_allocators.get("BlobSpPool")))
 , d_scheduler(bsls::SystemClockType::e_MONOTONIC, &d_allocator)
-, d_channelFactory(ntcCreateInterfaceConfig(sessionOptions, allocator),
+, d_channelFactory(ntcCreateInterfaceConfig(sessionOptions, &d_allocator),
                    &d_blobBufferFactory,
-                   allocator)
+                   &d_allocator)
 , d_resolvingChannelFactory(
       bmqio::ResolvingChannelFactoryConfig(
           &d_channelFactory,
           bmqex::ExecutionPolicyUtil::oneWay().alwaysBlocking().useExecutor(
-              bmqex::SystemExecutor())),
+              bmqex::SystemExecutor()),
+          allocator),
       allocator)
 , d_reconnectingChannelFactory(
       bmqio::ReconnectingChannelFactoryConfig(&d_resolvingChannelFactory,

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

@@ -4533,7 +4533,7 @@ void BrokerSession::cancel(const bsl::shared_ptr<Queue>&       queue,
     // Cancel pending configureQueue
     if (queue->pendingConfigureId() != Queue::k_INVALID_CONFIGURE_ID) {
         // Prepare canceled response.
-        bmqp_ctrlmsg::ControlMessage controlMessage;
+        bmqp_ctrlmsg::ControlMessage controlMessage(d_allocator_p);
         controlMessage.rId().makeValue(queue->pendingConfigureId());
 
         bmqp::ControlMessageUtil::makeStatus(&controlMessage,
@@ -4569,7 +4569,7 @@ void BrokerSession::cancel(int                                 groupId,
     BSLS_ASSERT_SAFE(d_fsmThreadChecker.inSameThread());
 
     // Cancel all pending requests with CANCELED reason.
-    bmqp_ctrlmsg::ControlMessage controlMessage;
+    bmqp_ctrlmsg::ControlMessage controlMessage(d_allocator_p);
     bmqp::ControlMessageUtil::makeStatus(&controlMessage, status, -1, reason);
     if (groupId >= 0) {
         d_requestManager.cancelAllRequests(controlMessage, groupId);
@@ -4695,7 +4695,8 @@ void BrokerSession::cancelPendingMessages(
     bmqp::AckEventBuilder  ackBuilder(d_blobSpPool_p, d_allocator_p);
     bsl::shared_ptr<Event> ackEvent = createEvent();
 
-    MessageCorrelationIdContainer::KeyIdsCb callback = bdlf::BindUtil::bind(
+    MessageCorrelationIdContainer::KeyIdsCb callback = bdlf::BindUtil::bindS(
+        d_allocator_p,
         &BrokerSession::cancelPendingMessageImp,
         this,
         &ackBuilder,
@@ -5924,12 +5925,13 @@ int BrokerSession::startAsync()
     // Add to the FSM event queue
     bsl::shared_ptr<Event> queueEvent = createEvent();
     queueEvent->configureAsRequestEvent(
-        bdlf::BindUtil::bind(&BrokerSession::doStart,
-                             this,
-                             &fsmAcceptedSemaphore,
-                             &startStatus,
-                             bdlf::PlaceHolders::_1,  // eventImpl
-                             createDTSpan("bmq.session.start")));
+        bdlf::BindUtil::bindS(d_allocator_p,
+                              &BrokerSession::doStart,
+                              this,
+                              &fsmAcceptedSemaphore,
+                              &startStatus,
+                              bdlf::PlaceHolders::_1,  // eventImpl
+                              createDTSpan("bmq.session.start")));
 
     if (enqueueFsmEvent(queueEvent) != bmqt::GenericResult::e_SUCCESS) {
         return bmqt::GenericResult::e_REFUSED;  // RETURN

src/groups/bmq/bmqimp/bmqimp_eventqueue.cpp

@@ -579,7 +579,7 @@ EventQueue::timedPopFront(const bsls::TimeInterval& timeout,
         BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
 
         bmqt::SessionEventType::Enum type;
-        bsl::string                  errorDescription;
+        bslstl::StringRef            errorDescription;
         if (BSLS_PERFORMANCEHINT_PREDICT_LIKELY(rc == -1)) {
             // We timed out.. create a timeout event
             type             = bmqt::SessionEventType::e_TIMEOUT;

src/groups/bmq/bmqst/bmqst_tableschema.cpp

@@ -167,11 +167,12 @@ TableSchemaColumn& TableSchema::addColumn(const bslstl::StringRef& name,
 
 void TableSchema::addDefaultIdColumn(const bslstl::StringRef& name)
 {
-    Column::ValueFn columnFn = bdlf::BindUtil::bind(defaultIdColumn,
-                                                    bdlf::PlaceHolders::_1,
-                                                    bdlf::PlaceHolders::_2,
-                                                    bdlf::PlaceHolders::_3,
-                                                    bdlf::PlaceHolders::_4);
+    Column::ValueFn columnFn = bdlf::BindUtil::bindS(d_allocator_p,
+                                                     defaultIdColumn,
+                                                     bdlf::PlaceHolders::_1,
+                                                     bdlf::PlaceHolders::_2,
+                                                     bdlf::PlaceHolders::_3,
+                                                     bdlf::PlaceHolders::_4);
 
     addColumn(name, columnFn);
 }

src/groups/mqb/mqbs/mqbs_inmemorystorage.cpp

@@ -75,6 +75,7 @@ InMemoryStorage::InMemoryStorage(const bmqt::Uri&        uri,
 , d_virtualStorageCatalog(
       this,
       allocatorStore ? allocatorStore->get("VirtualHandles") : d_allocator_p)
+, d_queueOpRecordHandles(d_allocator_p)
 , d_ttlSeconds(config.messageTtl())
 , d_isEmpty(1)
 , d_currentlyAutoConfirming()