Perf[MQB]: separate BlobSpPools and BufferFactories per FileStore thread

678098 · 678098 · commit 1621b518931e · 2025-02-28T15:21:13.000Z
Signed-off-by: Evgeny Malygin &lt;emalygin@bloomberg.net&gt;
diff --git a/src/groups/mqb/mqbc/mqbc_storagemanager.t.cpp b/src/groups/mqb/mqbc/mqbc_storagemanager.t.cpp
@@ -867,6 +867,11 @@ struct TestHelper {
         const mqbcfg::PartitionConfig& partitionCfg =
             d_cluster_mp->_clusterDefinition().partitionConfig();
 
+        bmqp::BlobPoolUtil::BlobSpPoolSp blobSpPool_sp =
+            bmqp::BlobPoolUtil::createBlobPool(
+                &d_cluster_mp->_clusterData()->bufferFactory(),
+                bmqtst::TestHelperUtil::allocator());
+
         mqbs::DataStoreConfig dsCfg;
         dsCfg.setScheduler(&d_cluster_mp->_scheduler())
             .setBufferFactory(&d_cluster_mp->_clusterData()->bufferFactory())
@@ -894,7 +899,7 @@ struct TestHelper {
                            d_cluster_mp->dispatcher(),
                            &d_cluster_mp->netCluster(),
                            &d_cluster_mp->_clusterData()->stats(),
-                           &d_cluster_mp->_clusterData()->blobSpPool(),
+                           blobSpPool_sp,
                            &d_cluster_mp->_clusterData()->stateSpPool(),
                            &threadPool,
                            d_cluster_mp->isCSLModeEnabled(),
diff --git a/src/groups/mqb/mqbc/mqbc_storageutil.cpp b/src/groups/mqb/mqbc/mqbc_storageutil.cpp
@@ -42,6 +42,7 @@
 #include <bdlb_print.h>
 #include <bdlb_scopeexit.h>
 #include <bdlb_stringrefutil.h>
+#include <bdlbb_pooledblobbufferfactory.h>
 #include <bdlf_bind.h>
 #include <bdlf_memfn.h>
 #include <bdlma_localsequentialallocator.h>
@@ -1225,8 +1226,37 @@ int StorageUtil::assignPartitionDispatcherThreads(
         mqbi::DispatcherClientType::e_QUEUE);
     BSLS_ASSERT_SAFE(numProcessors > 0);
 
+    bslma::Allocator*              baseAllocator = allocators->baseAllocator();
+    bsl::vector<bslma::Allocator*> queueThreadAllocators(numProcessors,
+                                                         baseAllocator);
+    bsl::vector<bsl::shared_ptr<bdlbb::BlobBufferFactory> > bufferFactories(
+        numProcessors,
+        baseAllocator);
+    bsl::vector<bmqp::BlobPoolUtil::BlobSpPoolSp> blobSpPools(numProcessors,
+                                                              baseAllocator);
+
+    // Allocater per-thread resources
+    for (int i = 0; i < numProcessors; ++i) {
+        bmqu::MemOutStream allocatorName(baseAllocator);
+        allocatorName << "QueueDispatcherThread" << i;
+
+        bslma::Allocator* queueThreadAllocator = allocators->get(
+            allocatorName.str());
+        bsl::shared_ptr<bdlbb::PooledBlobBufferFactory> bufferFactory =
+            bsl::allocate_shared<bdlbb::PooledBlobBufferFactory>(
+                queueThreadAllocator,
+                4096,
+                bsls::BlockGrowth::BSLS_CONSTANT);
+
+        queueThreadAllocators.at(i) = queueThreadAllocator;
+        bufferFactories.at(i)       = bufferFactory;
+        blobSpPools.at(
+            i) = bmqp::BlobPoolUtil::createBlobPool(bufferFactory.get(),
+                                                    queueThreadAllocator);
+    }
+
     for (int i = 0; i < config.numPartitions(); ++i) {
-        int                   processorId = i % numProcessors;
+        const int             processorId = i % numProcessors;
         mqbs::DataStoreConfig dsCfg;
         dsCfg.setScheduler(&clusterData->scheduler())
             .setBufferFactory(&clusterData->bufferFactory())
@@ -1250,27 +1280,19 @@ int StorageUtil::assignPartitionDispatcherThreads(
             dsCfg.setQueueDeletionCb(queueDeletionCb.value());
         }
 
-        // Get named allocator from associated bmqma::CountingAllocatorStore
-        bslma::Allocator* fileStoreAllocator = allocators->get(
-            bsl::string("Partition") + bsl::to_string(i));
-
-        bsl::shared_ptr<mqbs::FileStore> fsSp(
-            new (*fileStoreAllocator)
-                mqbs::FileStore(dsCfg,
-                                processorId,
-                                dispatcher,
-                                clusterData->membership().netCluster(),
-                                &clusterData->stats(),
-                                blobSpPool,
-                                &clusterData->stateSpPool(),
-                                threadPool,
-                                cluster.isCSLModeEnabled(),
-                                cluster.isFSMWorkflow(),
-                                replicationFactor,
-                                fileStoreAllocator),
-            fileStoreAllocator);
-
-        (*fileStores)[i] = fsSp;
+        (*fileStores)[i] = bsl::allocate_shared<mqbs::FileStore>(
+            queueThreadAllocators.at(processorId),
+            dsCfg,
+            processorId,
+            dispatcher,
+            clusterData->membership().netCluster(),
+            &clusterData->stats(),
+            blobSpPools.at(processorId),
+            &clusterData->stateSpPool(),
+            threadPool,
+            cluster.isCSLModeEnabled(),
+            cluster.isFSMWorkflow(),
+            replicationFactor);
     }
 
     return rc_SUCCESS;
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.cpp b/src/groups/mqb/mqbs/mqbs_filestore.cpp
@@ -3947,7 +3947,8 @@ void FileStore::processReceiptEvent(unsigned int         primaryLeaseId,
     if (itNode == d_nodes.end()) {
         // no prior history about this node
         d_nodes.insert(
-            bsl::make_pair(nodeId, NodeContext(d_blobSpPool_p, recordKey)));
+            bsl::make_pair(nodeId,
+                           NodeContext(d_blobSpPool_sp.get(), recordKey)));
         from = d_unreceipted.begin();
     }
     else if (itNode->second.d_key < recordKey) {
@@ -5207,7 +5208,7 @@ void FileStore::aliasMessage(bsl::shared_ptr<bdlbb::Blob>* appData,
 
         bdlbb::BlobBuffer optionsBlobBuffer(optionsBufferSp, optionsSize);
 
-        *options = d_blobSpPool_p->getObject();
+        *options = d_blobSpPool_sp->getObject();
         (*options)->appendDataBuffer(optionsBlobBuffer);
     }
 
@@ -5218,7 +5219,7 @@ void FileStore::aliasMessage(bsl::shared_ptr<bdlbb::Blob>* appData,
     bdlbb::BlobBuffer appDataBlobBuffer(appDataBufferSp,
                                         record.d_appDataUnpaddedLen);
 
-    *appData = d_blobSpPool_p->getObject();
+    *appData = d_blobSpPool_sp->getObject();
     (*appData)->appendDataBuffer(appDataBlobBuffer);
 }
 
@@ -5238,7 +5239,7 @@ FileStore::FileStore(const DataStoreConfig&  config,
                      mqbi::Dispatcher*       dispatcher,
                      mqbnet::Cluster*        cluster,
                      mqbstat::ClusterStats*  clusterStats,
-                     BlobSpPool*             blobSpPool,
+                     BlobSpPoolSp            blobSpPool_sp,
                      StateSpPool*            statePool,
                      bdlmt::FixedThreadPool* miscWorkThreadPool,
                      bool                    isCSLModeEnabled,
@@ -5252,7 +5253,7 @@ FileStore::FileStore(const DataStoreConfig&  config,
 , d_partitionDescription(allocator)
 , d_dispatcherClientData()
 , d_clusterStats_p(clusterStats)
-, d_blobSpPool_p(blobSpPool)
+, d_blobSpPool_sp(blobSpPool_sp)
 , d_statePool_p(statePool)
 , d_aliasedBufferDeleterSpPool(1024, d_allocators.get("AliasedBufferDeleters"))
 , d_isOpen(false)
@@ -5282,7 +5283,7 @@ FileStore::FileStore(const DataStoreConfig&  config,
 , d_nagglePacketCount(k_NAGLE_PACKET_COUNT)
 , d_storageEventBuilder(FileStoreProtocol::k_VERSION,
                         bmqp::EventType::e_STORAGE,
-                        d_blobSpPool_p,
+                        d_blobSpPool_sp.get(),
                         allocator)
 {
     // PRECONDITIONS
@@ -6578,9 +6579,9 @@ FileStore::generateReceipt(NodeContext*         nodeContext,
         if (itNode == d_nodes.end()) {
             // no prior history about this node
             itNode = d_nodes
-                         .insert(
-                             bsl::make_pair(nodeId,
-                                            NodeContext(d_blobSpPool_p, key)))
+                         .insert(bsl::make_pair(
+                             nodeId,
+                             NodeContext(d_blobSpPool_sp.get(), key)))
                          .first;
         }
         nodeContext = &itNode->second;
@@ -6610,7 +6611,7 @@ FileStore::generateReceipt(NodeContext*         nodeContext,
         // The pointer `nodeContext->d_blob_sp` might be in a write queue, so
         // it's not safe to modify or replace the blob under this pointer.
         // Instead, we get another shared pointer to another blob.
-        nodeContext->d_blob_sp = d_blobSpPool_p->getObject();
+        nodeContext->d_blob_sp = d_blobSpPool_sp->getObject();
         bmqp::ProtocolUtil::buildReceipt(nodeContext->d_blob_sp.get(),
                                          d_config.partitionId(),
                                          primaryLeaseId,
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.h b/src/groups/mqb/mqbs/mqbs_filestore.h
@@ -141,8 +141,8 @@ class FileStore BSLS_KEYWORD_FINAL : public DataStore {
 
   public:
     // TYPES
-
     typedef bmqp::BlobPoolUtil::BlobSpPool BlobSpPool;
+    typedef bmqp::BlobPoolUtil::BlobSpPoolSp BlobSpPoolSp;
 
     /// Pool of shared pointers to AtomicStates
     typedef bdlcc::SharedObjectPool<
@@ -290,7 +290,7 @@ class FileStore BSLS_KEYWORD_FINAL : public DataStore {
     // used to report partition level
     // metrics.
 
-    mutable BlobSpPool* d_blobSpPool_p;
+    mutable BlobSpPoolSp d_blobSpPool_sp;
     // Pool of shared pointers to blobs to
     // use.
 
@@ -697,7 +697,7 @@ class FileStore BSLS_KEYWORD_FINAL : public DataStore {
               mqbi::Dispatcher*       dispatcher,
               mqbnet::Cluster*        cluster,
               mqbstat::ClusterStats*  clusterStats,
-              BlobSpPool*             blobSpPool,
+              BlobSpPoolSp            blobSpPool,
               StateSpPool*            statePool,
               bdlmt::FixedThreadPool* miscWorkThreadPool,
               bool                    isCSLModeEnabled,
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.t.cpp b/src/groups/mqb/mqbs/mqbs_filestore.t.cpp
@@ -255,7 +255,7 @@ struct Tester {
                                          &d_dispatcher,
                                          d_cluster_mp.get(),
                                          &d_clusterStats,
-                                         d_blobSpPool_sp.get(),
+                                         d_blobSpPool_sp,
                                          &d_statePool,
                                          &d_miscWorkThreadPool,
                                          false,  // isCSLModeEnabled
