Perf[MQB]: make independent item pools for channels

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>
bloomberg · Oct 26, 2024 · 5dbe10b · 5dbe10b
1 parent bb7d697
commit 5dbe10b
Show file tree

Hide file tree

Showing 13 changed files with 24 additions and 74 deletions.
diff --git a/src/groups/mqb/mqbmock/mqbmock_cluster.cpp b/src/groups/mqb/mqbmock/mqbmock_cluster.cpp
@@ -136,7 +136,6 @@ void Cluster::_initializeNetcluster()
     d_netCluster_mp.load(new (*d_allocator_p)
                              mqbnet::MockCluster(d_clusterDefinition,
                                                  d_bufferFactory_p,
-                                                 &d_itemPool,
                                                  d_allocator_p),
                          d_allocator_p);
 
@@ -221,7 +220,6 @@ Cluster::Cluster(bdlbb::BlobBufferFactory* bufferFactory,
 , d_timeSource(&d_scheduler)
 , d_isStarted(false)
 , d_clusterDefinition(allocator)
-, d_itemPool(mqbnet::Channel::k_ITEM_SIZE, allocator)
 , d_channels(allocator)
 , d_negotiator_mp()
 , d_transportManager(&d_scheduler,

diff --git a/src/groups/mqb/mqbmock/mqbmock_cluster.h b/src/groups/mqb/mqbmock/mqbmock_cluster.h
@@ -178,9 +178,6 @@ class Cluster : public mqbi::Cluster {
     mqbcfg::ClusterDefinition d_clusterDefinition;
     // Cluster definition
 
-    mqbnet::Channel::ItemPool d_itemPool;
-    // Item pool
-
     TestChannelMap d_channels;
     // Test channels
 
@@ -429,9 +426,6 @@ class Cluster : public mqbi::Cluster {
     /// Get a modifiable reference to this object's time source.
     bdlmt::EventSchedulerTestTimeSource& _timeSource();
 
-    /// Get a modifiable reference to this object's item pool.
-    mqbnet::Channel::ItemPool& _itemPool();
-
     /// Get a modifiable reference to this object's cluster data.
     mqbc::ClusterData* _clusterData();
 
@@ -585,11 +579,6 @@ inline bdlmt::EventSchedulerTestTimeSource& Cluster::_timeSource()
     return d_timeSource;
 }
 
-inline mqbnet::Channel::ItemPool& Cluster::_itemPool()
-{
-    return d_itemPool;
-}
-
 inline mqbc::ClusterData* Cluster::_clusterData()
 {
     return d_clusterData_mp.get();

diff --git a/src/groups/mqb/mqbnet/mqbnet_channel.cpp b/src/groups/mqb/mqbnet/mqbnet_channel.cpp
@@ -63,7 +63,6 @@ void Channel::Stats::reset()
 // -------------
 
 Channel::Channel(bdlbb::BlobBufferFactory* blobBufferFactory,
-                 ItemPool*                 itemPool,
                  const bsl::string&        name,
                  bslma::Allocator*         allocator)
 : d_allocators(allocator)
@@ -73,7 +72,9 @@ Channel::Channel(bdlbb::BlobBufferFactory* blobBufferFactory,
 , d_ackBuilder(blobBufferFactory, d_allocator_p)
 , d_confirmBuilder(blobBufferFactory, d_allocator_p)
 , d_rejectBuilder(blobBufferFactory, d_allocator_p)
-, d_itemPool_p(itemPool)
+, d_itemPool(sizeof(Item),
+             bsls::BlockGrowth::BSLS_CONSTANT,
+             d_allocators.get(bsl::string("ChannelPool-") + name))
 , d_buffer(1024, allocator)
 , d_secondaryBuffer(1024, allocator)
 , d_doStop(false)
@@ -108,7 +109,7 @@ Channel::~Channel()
 
 void Channel::deleteItem(void* item, void* cookie)
 {
-    static_cast<Channel*>(cookie)->d_itemPool_p->deleteObject(
+    static_cast<Channel*>(cookie)->d_itemPool.deleteObject(
         static_cast<Item*>(item));
 }
 
@@ -119,7 +120,7 @@ Channel::writePut(const bmqp::PutHeader&                    ph,
                   bool                                      keepWeakPtr)
 {
     bslma::ManagedPtr<Item> item(
-        new (d_itemPool_p->allocate())
+        new (d_itemPool.allocate())
             Item(ph, data, keepWeakPtr, state, d_allocator_p),
         this,
         deleteItem);
@@ -136,7 +137,7 @@ Channel::writePush(const bsl::shared_ptr<bdlbb::Blob>&       payload,
                    const bmqp::Protocol::SubQueueInfosArray& subQueueInfos,
                    const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(queueId,
                                           msgId,
                                           flags,
@@ -160,7 +161,7 @@ Channel::writePush(int                                       queueId,
                    const bmqp::Protocol::SubQueueInfosArray& subQueueInfos,
                    const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(queueId,
                                           msgId,
                                           flags,
@@ -182,7 +183,7 @@ Channel::writeAck(int                                       status,
                   const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
     bslma::ManagedPtr<Item> item(
-        new (d_itemPool_p->allocate())
+        new (d_itemPool.allocate())
             Item(status, correlationId, guid, queueId, state, d_allocator_p),
         this,
         deleteItem);
@@ -195,7 +196,7 @@ Channel::writeConfirm(int                                       queueId,
                       const bmqt::MessageGUID&                  guid,
                       const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(queueId,
                                           subQueueId,
                                           guid,
@@ -213,7 +214,7 @@ Channel::writeReject(int                                       queueId,
                      const bmqt::MessageGUID&                  guid,
                      const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(queueId,
                                           subQueueId,
                                           guid,
@@ -230,7 +231,7 @@ Channel::writeBlob(const bdlbb::Blob&                        data,
                    bmqp::EventType::Enum                     type,
                    const bsl::shared_ptr<bmqu::AtomicState>& state)
 {
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(data, type, state, d_allocator_p),
                                  this,
                                  deleteItem);
@@ -253,7 +254,7 @@ void Channel::resetChannel()
         d_stateCondition.signal();
     }
     // Wake up the writing thread in case it is blocked by 'popFront'
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(d_allocator_p),
                                  this,
                                  deleteItem);
@@ -336,7 +337,7 @@ void Channel::flush()
         return;  // RETURN
     }
 
-    bslma::ManagedPtr<Item> item(new (d_itemPool_p->allocate())
+    bslma::ManagedPtr<Item> item(new (d_itemPool.allocate())
                                      Item(d_allocator_p),
                                  this,
                                  deleteItem);

diff --git a/src/groups/mqb/mqbnet/mqbnet_channel.h b/src/groups/mqb/mqbnet/mqbnet_channel.h
@@ -348,8 +348,6 @@ class Channel {
 
   public:
     // PUBLIC TYPES
-
-    typedef bdlma::ConcurrentPool ItemPool;
     typedef bmqc::MonitoredQueue<
         bdlcc::SingleConsumerQueue<bslma::ManagedPtr<Item> > >
         ItemQueue;
@@ -366,10 +364,6 @@ class Channel {
         e_HWM = 5  // HWM
     };
 
-  public:
-    // CONSTANTS
-    static const int k_ITEM_SIZE = sizeof(Item);  // for ItemPool
-
   private:
     // CONSTANTS
     static const int k_NAGLE_PACKET_SIZE = 1024 * 1024;  // 1MB;
@@ -390,7 +384,7 @@ class Channel {
 
     bmqp::RejectEventBuilder d_rejectBuilder;
 
-    ItemPool* d_itemPool_p;
+    bdlma::ConcurrentPool d_itemPool;
     // Pool of 'Item' objects.
 
     ItemQueue d_buffer;
@@ -513,7 +507,6 @@ class Channel {
 
     /// Create a new object using the specified `allocator`.
     Channel(bdlbb::BlobBufferFactory* blobBufferFactory,
-            ItemPool*                 itemPool,
             const bsl::string&        name,
             bslma::Allocator*         allocator);
 

diff --git a/src/groups/mqb/mqbnet/mqbnet_channel.t.cpp b/src/groups/mqb/mqbnet/mqbnet_channel.t.cpp
@@ -725,9 +725,7 @@ static void test1_write()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)
@@ -793,9 +791,7 @@ static void test2_highWatermark()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)
@@ -885,9 +881,7 @@ static void test3_highWatermarkInWriteCb()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)
@@ -977,9 +971,7 @@ static void test4_controlBlob()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)
@@ -1048,9 +1040,7 @@ static void test5_reconnect()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)
@@ -1129,9 +1119,7 @@ static void test6_weakData()
 // ------------------------------------------------------------------------
 {
     bdlbb::PooledBlobBufferFactory bufferFactory(k_BUFFER_SIZE, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
-    mqbnet::Channel channel(&bufferFactory, &itemPool, "test", s_allocator_p);
+    mqbnet::Channel channel(&bufferFactory, "test", s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannelEx> testChannel(
         new (*s_allocator_p)

diff --git a/src/groups/mqb/mqbnet/mqbnet_clusterimp.cpp b/src/groups/mqb/mqbnet/mqbnet_clusterimp.cpp
@@ -45,12 +45,11 @@ namespace mqbnet {
 ClusterNodeImp::ClusterNodeImp(ClusterImp*                cluster,
                                const mqbcfg::ClusterNode& config,
                                bdlbb::BlobBufferFactory*  blobBufferFactory,
-                               Channel::ItemPool*         itemPool,
                                bslma::Allocator*          allocator)
 : d_cluster_p(cluster)
 , d_config(config, allocator)
 , d_description(allocator)
-, d_channel(blobBufferFactory, itemPool, config.name(), allocator)
+, d_channel(blobBufferFactory, config.name(), allocator)
 , d_identity(allocator)
 , d_isReading(false)
 {
@@ -181,7 +180,6 @@ ClusterImp::ClusterImp(const bsl::string&                      name,
                        const bsl::vector<mqbcfg::ClusterNode>& nodesConfig,
                        int                                     selfNodeId,
                        bdlbb::BlobBufferFactory* blobBufferFactory,
-                       Channel::ItemPool*        itemPool,
                        bslma::Allocator*         allocator)
 : d_allocator_p(allocator)
 , d_name(name, allocator)
@@ -199,7 +197,7 @@ ClusterImp::ClusterImp(const bsl::string&                      name,
     bsl::vector<mqbcfg::ClusterNode>::const_iterator nodeIt;
     for (nodeIt = d_nodesConfig.begin(); nodeIt != d_nodesConfig.end();
          ++nodeIt) {
-        d_nodes.emplace_back(this, *nodeIt, blobBufferFactory, itemPool);
+        d_nodes.emplace_back(this, *nodeIt, blobBufferFactory);
         d_nodesList.emplace_back(&d_nodes.back());
         if (nodeIt->id() == selfNodeId) {
             d_selfNode = d_nodesList.back();

diff --git a/src/groups/mqb/mqbnet/mqbnet_clusterimp.h b/src/groups/mqb/mqbnet/mqbnet_clusterimp.h
@@ -113,7 +113,6 @@ class ClusterNodeImp : public ClusterNode {
     ClusterNodeImp(ClusterImp*                cluster,
                    const mqbcfg::ClusterNode& config,
                    bdlbb::BlobBufferFactory*  blobBufferFactory,
-                   Channel::ItemPool*         itemPool,
                    bslma::Allocator*          allocator);
 
     /// Destructor.
@@ -277,7 +276,6 @@ class ClusterImp : public Cluster {
                const bsl::vector<mqbcfg::ClusterNode>& nodesConfig,
                int                                     selfNodeId,
                bdlbb::BlobBufferFactory*               blobBufferFactory,
-               Channel::ItemPool*                      itemPool,
                bslma::Allocator*                       allocator);
 
     /// Destructor

diff --git a/src/groups/mqb/mqbnet/mqbnet_dummysession.t.cpp b/src/groups/mqb/mqbnet/mqbnet_dummysession.t.cpp
@@ -52,18 +52,14 @@ static void test1_BreathingTest()
 
     mqbcfg::ClusterDefinition      clusterConfig(s_allocator_p);
     bdlbb::PooledBlobBufferFactory bufferFactory(1024, s_allocator_p);
-    mqbnet::Channel::ItemPool      itemPool(mqbnet::Channel::k_ITEM_SIZE,
-                                       s_allocator_p);
     mqbnet::MockCluster            mockCluster(clusterConfig,
                                     &bufferFactory,
-                                    &itemPool,
                                     s_allocator_p);
 
     mqbcfg::ClusterNode     clusterNodeConfig(s_allocator_p);
     mqbnet::MockClusterNode mockClusterNode(&mockCluster,
                                             clusterNodeConfig,
                                             &bufferFactory,
-                                            &itemPool,
                                             s_allocator_p);
 
     bsl::shared_ptr<bmqio::TestChannel> testChannel;

diff --git a/src/groups/mqb/mqbnet/mqbnet_mockcluster.cpp b/src/groups/mqb/mqbnet/mqbnet_mockcluster.cpp
@@ -41,12 +41,11 @@ namespace mqbnet {
 MockClusterNode::MockClusterNode(MockCluster*               cluster,
                                  const mqbcfg::ClusterNode& config,
                                  bdlbb::BlobBufferFactory*  blobBufferFactory,
-                                 Channel::ItemPool*         itemPool,
                                  bslma::Allocator*          allocator)
 : d_cluster_p(cluster)
 , d_config(config, allocator)
 , d_description(allocator)
-, d_channel(blobBufferFactory, itemPool, config.name(), allocator)
+, d_channel(blobBufferFactory, config.name(), allocator)
 , d_identity(allocator)
 , d_isReading(false)
 {
@@ -156,7 +155,6 @@ void MockCluster::notifyObserversOfNodeStateChange(ClusterNode* node,
 
 MockCluster::MockCluster(const mqbcfg::ClusterDefinition& config,
                          bdlbb::BlobBufferFactory*        blobBufferFactory,
-                         Channel::ItemPool*               itemPool,
                          bslma::Allocator*                allocator)
 : d_allocator_p(allocator)
 , d_config(config, allocator)
@@ -171,7 +169,7 @@ MockCluster::MockCluster(const mqbcfg::ClusterDefinition& config,
     bsl::vector<mqbcfg::ClusterNode>::const_iterator nodeIt;
     for (nodeIt = d_config.nodes().begin(); nodeIt != d_config.nodes().end();
          ++nodeIt) {
-        d_nodes.emplace_back(this, *nodeIt, blobBufferFactory, itemPool);
+        d_nodes.emplace_back(this, *nodeIt, blobBufferFactory);
         d_nodesList.emplace_back(&d_nodes.back());
     }
 }

diff --git a/src/groups/mqb/mqbnet/mqbnet_mockcluster.h b/src/groups/mqb/mqbnet/mqbnet_mockcluster.h
@@ -100,7 +100,6 @@ class MockClusterNode : public ClusterNode {
     MockClusterNode(MockCluster*               cluster,
                     const mqbcfg::ClusterNode& config,
                     bdlbb::BlobBufferFactory*  blobBufferFactory,
-                    Channel::ItemPool*         itemPool,
                     bslma::Allocator*          allocator);
 
     /// Destructor.
@@ -251,7 +250,6 @@ class MockCluster : public Cluster {
     /// `allocator`.
     MockCluster(const mqbcfg::ClusterDefinition& config,
                 bdlbb::BlobBufferFactory*        blobBufferFactory,
-                Channel::ItemPool*               itemPool,
                 bslma::Allocator*                allocator);
 
     /// Destructor

diff --git a/src/groups/mqb/mqbnet/mqbnet_transportmanager.cpp b/src/groups/mqb/mqbnet/mqbnet_transportmanager.cpp
@@ -352,7 +352,6 @@ TransportManager::TransportManager(bdlmt::EventScheduler*    scheduler,
 , d_state(e_STOPPED)
 , d_scheduler_p(scheduler)
 , d_blobBufferFactory_p(blobBufferFactory)
-, d_itemPool(Channel::k_ITEM_SIZE, bsls::BlockGrowth::BSLS_CONSTANT, allocator)
 , d_negotiator_mp(negotiator)
 , d_statController_p(statController)
 , d_tcpSessionFactory_mp(0)
@@ -525,7 +524,6 @@ int TransportManager::createCluster(
                                                       nodes,
                                                       myNodeId,
                                                       d_blobBufferFactory_p,
-                                                      &d_itemPool,
                                                       d_allocator_p),
                                        d_allocator_p);
 

diff --git a/src/groups/mqb/mqbnet/mqbnet_transportmanager.h b/src/groups/mqb/mqbnet/mqbnet_transportmanager.h
@@ -195,8 +195,6 @@ class TransportManager {
     // BlobBufferFactory to use by the
     // sessions
 
-    Channel::ItemPool d_itemPool;
-
     bslma::ManagedPtr<Negotiator> d_negotiator_mp;
     // Negotiation to use