Merge pull request #2841 from glevkovich/memory_management_dev1_pr2

Memory management dev1 pr2

bftengine/src/bcstatetransfer/BCStateTran.cpp

@@ -338,6 +338,7 @@ BCStateTran::BCStateTran(const Config &config, IAppState *const stateApi, DataSt
 }
 
 BCStateTran::~BCStateTran() {
+  LOG_TRACE(logger_, "~BCStateTran");
   ConcordAssert(!running_);
   ConcordAssert(cacheOfVirtualBlockForResPages.empty());
   ConcordAssert(pendingItemDataMsgs.empty());
@@ -470,7 +471,7 @@ void BCStateTran::startRunningImpl(IReplicaForStateTransfer *r) {
 
 // timer is cancelled in the calling context, see ReplicaForStateTransfer::stop
 void BCStateTran::stopRunningImpl() {
-  LOG_INFO(logger_, "Stopping");
+  LOG_TRACE(logger_, "stopRunningImpl");
   ConcordAssert(running_);
   ConcordAssertNE(replicaForStateTransfer_, nullptr);
   // This one should always be first!
@@ -484,6 +485,7 @@ void BCStateTran::stopRunningImpl() {
     stReset(g.txn(), true, false, false);
   }
   replicaForStateTransfer_ = nullptr;
+  LOG_TRACE(logger_, "Done stopRunningImpl");
 }
 
 // Create a CheckpointDesc for the given checkpointNumber.

bftengine/src/bcstatetransfer/BCStateTran.hpp

@@ -868,6 +868,7 @@ class BCStateTran : public IStateTransfer {
     static constexpr uint64_t MAX_PENDING_BLOCKS_SIZE = 1000ULL;
 
     Recorders() {
+      LOG_TRACE(ST_SRC_LOG, "Recorders: Thread ID: " << std::this_thread::get_id() << KVLOG(this));
       auto& registrar = concord::diagnostics::RegistrarSingleton::getInstance();
       // common component
       registrar.perf.registerComponent("state_transfer",
@@ -898,6 +899,7 @@ class BCStateTran : public IStateTransfer {
                                         src_next_block_wait_duration});
     }
     ~Recorders() {
+      LOG_TRACE(ST_SRC_LOG, "~Recorders: Thread ID: " << std::this_thread::get_id() << KVLOG(this));
       auto& registrar = concord::diagnostics::RegistrarSingleton::getInstance();
       registrar.perf.unRegisterComponent("state_transfer");
       registrar.perf.unRegisterComponent("state_transfer_dest");

bftengine/src/bftengine/BFTEngine.cpp

@@ -87,12 +87,14 @@ void ReplicaInternal::start() {
 }
 
 void ReplicaInternal::stop() {
+  LOG_TRACE(GL, "ReplicaInternal::stop started");
   unique_lock<std::mutex> lk(debugWaitLock_);
   if (replica_->isRunning()) {
     replica_->stop();
   }
 
   debugWait_.notify_all();
+  LOG_TRACE(GL, "ReplicaInternal::stop done");
 }
 
 void ReplicaInternal::SetAggregator(std::shared_ptr<concordMetrics::Aggregator> a) { replica_->SetAggregator(a); }

bftengine/src/bftengine/ReadOnlyReplica.cpp

@@ -287,9 +287,7 @@ void ReadOnlyReplica::registerStatusHandlers() {
         bj.endNested();
         bj.endJson();
 
-        char *cstr = new char[bj.getJson().length() + 1];
-        std::strcpy(cstr, bj.getJson().c_str());
-        return cstr;
+        return bj.getJson();
       });
   concord::diagnostics::RegistrarSingleton::getInstance().status.registerHandler(h);
 }

bftengine/src/bftengine/ReplicaImp.cpp

@@ -884,7 +884,10 @@ void ReplicaImp::startConsensusProcess(PrePrepareMsg *pp) {
 }
 
 void ReplicaImp::startConsensusProcess(PrePrepareMsg *pp, bool isCreatedEarlier) {
-  if (!isCurrentPrimary()) return;
+  if (!isCurrentPrimary()) {
+    delete pp;
+    return;
+  }
   TimeRecorder scoped_timer(*histograms_.startConsensusProcess);
   if (getReplicaConfig().timeServiceEnabled) {
     pp->setTime(time_service_manager_->getClockTimePoint());
@@ -4349,12 +4352,13 @@ ReplicaImp::ReplicaImp(bool firstTime,
       viewChangeProtocolEnabled{config.viewChangeProtocolEnabled},
       autoPrimaryRotationEnabled{config.autoPrimaryRotationEnabled},
       restarted_{!firstTime},
+      internalThreadPool("ReplicaImp::internalThreadPool"),
       MAIN_THREAD_ID{std::this_thread::get_id()},
+      postExecThread_{"ReplicaImp::postExecThread"},
       replyBuffer{static_cast<char *>(std::malloc(config_.getmaxReplyMessageSize() - sizeof(ClientReplyMsgHeader)))},
       timeOfLastStateSynch{getMonotonicTime()},    // TODO(GG): TBD
       timeOfLastViewEntrance{getMonotonicTime()},  // TODO(GG): TBD
       timeOfLastAgreedView{getMonotonicTime()},    // TODO(GG): TBD
-
       pm_{pm},
       sm_{sm ? sm : std::make_shared<concord::secretsmanager::SecretsManagerPlain>()},
       metric_view_{metrics_.RegisterGauge("view", 0)},
@@ -4539,11 +4543,10 @@ ReplicaImp::ReplicaImp(bool firstTime,
   controller = new ControllerWithSimpleHistory(
       config_.getcVal(), config_.getfVal(), config_.getreplicaId(), getCurrentView(), primaryLastUsedSeqNum);
 
-  if (retransmissionsLogicEnabled)
+  if (retransmissionsLogicEnabled) {
     retransmissionsManager =
-        new RetransmissionsManager(&internalThreadPool, &getIncomingMsgsStorage(), kWorkWindowSize, 0);
-  else
-    retransmissionsManager = nullptr;
+        std::make_unique<RetransmissionsManager>(&internalThreadPool, &getIncomingMsgsStorage(), kWorkWindowSize, 0);
+  }
 
   ticks_gen_ = concord::cron::TicksGenerator::create(internalBFTClient_,
                                                      *clientsManager,
@@ -4609,13 +4612,19 @@ ReplicaImp::~ReplicaImp() {
 }
 
 void ReplicaImp::stop() {
-  if (retransmissionsLogicEnabled) timers_.cancel(retranTimer_);
+  LOG_DEBUG(GL, "ReplicaImp::stop started");
+  if (retransmissionsLogicEnabled) {
+    timers_.cancel(retranTimer_);
+  }
   timers_.cancel(slowPathTimer_);
   timers_.cancel(infoReqTimer_);
   timers_.cancel(statusReportTimer_);
   timers_.cancel(clientRequestsRetransmissionTimer_);
-  if (viewChangeProtocolEnabled) timers_.cancel(viewChangeTimer_);
+  if (viewChangeProtocolEnabled) {
+    timers_.cancel(viewChangeTimer_);
+  }
   ReplicaForStateTransfer::stop();
+  LOG_DEBUG(GL, "ReplicaImp::stop done");
 }
 
 void ReplicaImp::addTimers() {

bftengine/src/bftengine/ReplicaImp.hpp

@@ -27,7 +27,6 @@
 #include "InternalReplicaApi.hpp"
 #include "ClientsManager.hpp"
 #include "CheckpointInfo.hpp"
-#include "SimpleThreadPool.hpp"
 #include "Bitmap.hpp"
 #include "OpenTracing.hpp"
 #include "RequestHandler.h"
@@ -89,7 +88,7 @@ class ReplicaImp : public InternalReplicaApi, public ReplicaForStateTransfer {
   concord::util::SimpleThreadPool internalThreadPool;  // TODO(GG): !!!! rename
 
   // retransmissions manager (can be disabled)
-  RetransmissionsManager* retransmissionsManager = nullptr;
+  unique_ptr<RetransmissionsManager> retransmissionsManager;
 
   // controller
   ControllerBase* controller = nullptr;
@@ -374,7 +373,7 @@ class ReplicaImp : public InternalReplicaApi, public ReplicaForStateTransfer {
 
   // InternalReplicaApi
   bool isCollectingState() const override {
-    LOG_DEBUG(GL, "Thread ID: " << std::this_thread::get_id());
+    LOG_TRACE(GL, "Thread ID: " << std::this_thread::get_id());
     return isCollectingState_;
   }
   void startCollectingState(std::string&& reason = "");

bftengine/src/bftengine/RequestThreadPool.hpp

@@ -32,7 +32,10 @@ class RequestThreadPool final {
   static auto& getThreadPool(uint16_t level) {
     // Currently we need 2 level thread pools.
     static std::array<concord::util::ThreadPool, PoolLevel::MAXLEVEL> threadBag = {
-        ReplicaConfig::instance().threadbagConcurrencyLevel1, ReplicaConfig::instance().threadbagConcurrencyLevel2};
+        concord::util::ThreadPool("RequestThreadPool::threadBag_ConcurrencyLevel1",
+                                  ReplicaConfig::instance().threadbagConcurrencyLevel1),
+        concord::util::ThreadPool("RequestThreadPool::threadBag_ConcurrencyLevel2",
+                                  ReplicaConfig::instance().threadbagConcurrencyLevel2)};
     return threadBag.at(level);
   }
 

bftengine/src/preprocessor/PreProcessor.cpp

@@ -366,6 +366,7 @@ PreProcessor::PreProcessor(shared_ptr<MsgsCommunicator> &msgsCommunicator,
       numOfReplicas_(myReplica.getReplicaConfig().numReplicas + myReplica.getReplicaConfig().numRoReplicas),
       numOfClientProxies_(myReplica.getReplicaConfig().numOfClientProxies),
       clientBatchingEnabled_(myReplica.getReplicaConfig().clientBatchingEnabled),
+      threadPool_("PreProcessor::threadPool"),
       memoryPool_(maxExternalMsgSize_, timers),
       metricsComponent_{concordMetrics::Component("preProcessor", std::make_shared<concordMetrics::Aggregator>())},
       metricsLastDumpTime_(0),
@@ -439,13 +440,20 @@ PreProcessor::PreProcessor(shared_ptr<MsgsCommunicator> &msgsCommunicator,
 }
 
 PreProcessor::~PreProcessor() {
+  LOG_TRACE(logger(), "~PreProcessor start");
   msgLoopDone_ = true;
   msgLoopSignal_.notify_all();
   cancelTimers();
   threadPool_.stop();
-  if (msgLoopThread_.joinable()) msgLoopThread_.join();
-  if (!memoryPoolEnabled_)
-    for (const auto &result : preProcessResultBuffers_) delete[] result->buffer;
+  if (msgLoopThread_.joinable()) {
+    msgLoopThread_.join();
+  }
+  if (!memoryPoolEnabled_) {
+    for (const auto &result : preProcessResultBuffers_) {
+      delete[] result->buffer;
+    }
+  }
+  LOG_TRACE(logger(), "~PreProcessor done");
 }
 
 void PreProcessor::addTimers() {

bftengine/src/preprocessor/tests/preprocessor_test.cpp

@@ -133,7 +133,7 @@ class DummyReplica : public InternalReplicaApi {
  private:
   bool primary_ = true;
   IncomingMsgsStorage* incomingMsgsStorage_ = nullptr;
-  concord::util::SimpleThreadPool pool_;
+  concord::util::SimpleThreadPool pool_{""};
   bftEngine::impl::ReplicasInfo replicasInfo_;
   set<ReplicaId> replicaIds_;
 };

bftengine/tests/testRequestThreadPool/testRequestThreadPool.cpp

@@ -85,7 +85,7 @@ TEST(testRequestThreadPool, sameTP) {
   std::vector<std::future<void>> tasks;
 
   try {
-    auto pool = concord::util::ThreadPool{};
+    auto pool = concord::util::ThreadPool{""};
     for (uint32_t i = 0; i < 10000; i++) {
       tasks.push_back(pool.async([&thread_pool_addr]() {
         std::vector<size_t> thread_pool_addr_tp;

client/client_pool/src/concord_client_pool.cpp

@@ -256,7 +256,8 @@ std::unique_ptr<ConcordClientPool> ConcordClientPool::create(config_pool::Concor
 ConcordClientPool::ConcordClientPool(config_pool::ConcordClientPoolConfig &config,
                                      std::shared_ptr<concordMetrics::Aggregator> aggregator,
                                      bool delay_behavior)
-    : metricsComponent_{concordMetrics::Component("ClientPool", std::make_shared<concordMetrics::Aggregator>())},
+    : jobs_thread_pool_{"ConcordClientPool::jobs_thread_pool"},
+      metricsComponent_{concordMetrics::Component("ClientPool", std::make_shared<concordMetrics::Aggregator>())},
       ClientPoolMetrics_{metricsComponent_.RegisterCounter("requests_counter"),
                          metricsComponent_.RegisterCounter("executed_requests_counter"),
                          metricsComponent_.RegisterCounter("rejected_counter"),
@@ -286,7 +287,8 @@ ConcordClientPool::ConcordClientPool(config_pool::ConcordClientPoolConfig &confi
 
 ConcordClientPool::ConcordClientPool(config_pool::ConcordClientPoolConfig &config,
                                      std::shared_ptr<concordMetrics::Aggregator> aggregator)
-    : metricsComponent_{concordMetrics::Component("ClientPool", std::make_shared<concordMetrics::Aggregator>())},
+    : jobs_thread_pool_{"ConcordClientPool::jobs_thread_pool"},
+      metricsComponent_{concordMetrics::Component("ClientPool", std::make_shared<concordMetrics::Aggregator>())},
       ClientPoolMetrics_{metricsComponent_.RegisterCounter("requests_counter"),
                          metricsComponent_.RegisterCounter("executed_requests_counter"),
                          metricsComponent_.RegisterCounter("rejected_counter"),

client/thin-replica-client/include/client/thin-replica-client/grpc_connection.hpp

@@ -86,10 +86,10 @@ class GrpcConnection {
       : logger_(logging::getLogger("concord.client.thin_replica.trscon")),
         address_(address),
         client_id_(client_id),
+        grpc_connection_pool_{std::make_unique<concord::util::ThreadPool>("GrpcConnection::grpc_connection_pool", 1)},
         data_timeout_(std::chrono::seconds(data_operation_timeout_seconds)),
         hash_timeout_(std::chrono::seconds(hash_operation_timeout_seconds)),
         snapshot_timeout_(std::chrono::seconds(snapshot_operation_timeout_seconds)) {
-    grpc_connection_pool_ = std::make_unique<concord::util::ThreadPool>(1);
     ConcordAssertNE(grpc_connection_pool_, nullptr);
     LOG_INFO(logger_,
              KVLOG(data_operation_timeout_seconds, hash_operation_timeout_seconds, snapshot_operation_timeout_seconds));

client/thin-replica-client/include/client/thin-replica-client/replica_state_snapshot_client.hpp

@@ -58,7 +58,7 @@ class ReplicaStateSnapshotClient {
   ReplicaStateSnapshotClient(std::unique_ptr<ReplicaStateSnapshotClientConfig> config)
       : logger_(logging::getLogger("concord.client.replica_stream_snapshot")),
         config_(std::move(config)),
-        threadpool_(config_->concurrency_level),
+        threadpool_("ReplicaStateSnapshotClient::threadpool", config_->concurrency_level),
         count_of_concurrent_request_{0} {}
   void readSnapshotStream(const SnapshotRequest& request,
                           std::shared_ptr<concord::client::concordclient::SnapshotQueue> remote_queue);

client/thin-replica-client/test/replica_stream_snapshot_client_test.cpp

@@ -163,7 +163,7 @@ void getGrpcConnections(bool full_fake, vector<shared_ptr<GrpcConnection>>& grpc
 TEST(replica_stream_snapshot_client_test, test_destructor_always_successful) {
   vector<shared_ptr<GrpcConnection>> grpc_connections;
   getGrpcConnections(true, grpc_connections, 7);
-  ThreadPool thread_pool{10};
+  ThreadPool thread_pool{"", 10};
   auto read_snapshot = [&grpc_connections]() {
     auto rss_config = std::make_unique<ReplicaStateSnapshotClientConfig>(grpc_connections, 8);
     auto rss = std::make_unique<ReplicaStateSnapshotClient>(std::move(rss_config));
@@ -208,7 +208,7 @@ TEST(replica_stream_snapshot_client_test, test_real_action) {
     // responsible for shutting down the server for this call to ever return.
     (servers[i])->Wait();
   };
-  ThreadPool server_thread_pool{7};
+  ThreadPool server_thread_pool{"", 7};
   vector<std::future<void>> server_results;
   for (size_t i = 0; i < 7; ++i) {
     server_results.push_back(server_thread_pool.async(run_server, i));
@@ -222,7 +222,7 @@ TEST(replica_stream_snapshot_client_test, test_real_action) {
 
   vector<shared_ptr<GrpcConnection>> grpc_connections;
   getGrpcConnections(false, grpc_connections, 7);
-  ThreadPool thread_pool{10};
+  ThreadPool thread_pool{"", 10};
   auto read_snapshot = [&grpc_connections](size_t len) {
     auto rss_config = std::make_unique<ReplicaStateSnapshotClientConfig>(grpc_connections, 8);
     auto rss = std::make_unique<ReplicaStateSnapshotClient>(std::move(rss_config));
@@ -232,7 +232,7 @@ TEST(replica_stream_snapshot_client_test, test_real_action) {
     ::client::replica_state_snapshot_client::SnapshotRequest rss_request;
     rss_request.snapshot_id = len;
     rss->readSnapshotStream(rss_request, remote_queue);
-    ThreadPool read_thread_pool{1};
+    ThreadPool read_thread_pool{"", 1};
     read_thread_pool.async(
         [&remote_queue](size_t l) {
           size_t num_received = 0;

communication/src/TlsConnectionManager.cpp

@@ -76,6 +76,7 @@ void ConnectionManager::stop() {
     LOG_DEBUG(logger_, "Closing connection from: " << config_.selfId_ << ", to: " << id);
     syncCloseConnection(conn);
   }
+  LOG_TRACE(logger_, "Done stopping connection manager for " << config_.selfId_);
 }
 
 void ConnectionManager::setReceiver(NodeNum, IReceiver* receiver) { receiver_ = receiver; }

diagnostics/include/diagnostics_server.h

@@ -158,6 +158,7 @@ class Server {
   }
 
   void stop() {
+    LOG_INFO(logger, std::boolalpha << KVLOG(shutdown_));
     if (!shutdown_) {
       LOG_INFO(logger, "Shutting down diagnostics server main thread.");
       shutdown_.store(true);

kvbc/benchmark/state_snapshot_benchmarks/hash_state_benchmark.cpp

@@ -184,7 +184,7 @@ int run(int argc, char* argv[]) {
     point_lookup_batch_size++;
   }
 
-  auto thread_pool = ThreadPool{static_cast<std::uint32_t>(point_lookup_threads)};
+  auto thread_pool = ThreadPool{"hash_state_benchmark::thread_pool", static_cast<std::uint32_t>(point_lookup_threads)};
 
   std::cout << "Hashing state with a point lookup batch size = " << point_lookup_batch_size
             << ", point lookup threads = " << point_lookup_threads

kvbc/include/categorization/kv_blockchain.h

@@ -254,9 +254,9 @@ class KeyValueBlockchain {
   VersionedRawBlock last_raw_block_;
 
   // currently we are operating with single thread
-  util::ThreadPool thread_pool_{1};
+  util::ThreadPool thread_pool_{"categorization::KeyValueBlockchain::thread_pool", 1};
   // For concurrent deletion of the categories inside a block.
-  util::ThreadPool prunning_thread_pool_{2};
+  util::ThreadPool prunning_thread_pool_{"categorization::KeyValueBlockchain::prunning_thread_pool_,", 2};
 
   // metrics
   std::shared_ptr<concordMetrics::Aggregator> aggregator_;

kvbc/include/v4blockchain/detail/blockchain.h

@@ -120,7 +120,7 @@ class Blockchain {
   std::atomic<BlockId> last_reachable_block_id_{INVALID_BLOCK_ID};
   std::atomic<BlockId> genesis_block_id_{INVALID_BLOCK_ID};
   std::shared_ptr<concord::storage::rocksdb::NativeClient> native_client_;
-  util::ThreadPool thread_pool_{1};
+  util::ThreadPool thread_pool_{"v4blockchain::detail::Blockchain::thread_pool", 1};
   std::optional<std::future<concord::crypto::BlockDigest>> future_digest_;
   bool need_compaction_{false};
   std::mutex compaction_mutex_;

kvbc/include/v4blockchain/v4_blockchain.h

@@ -276,8 +276,8 @@ class KeyValueBlockchain {
   const ::rocksdb::Snapshot *snap_shot_{nullptr};
   std::map<kvbc::BlockId, const ::rocksdb::Snapshot *> chkpnt_snap_shots_;
   // const ::rocksdb::Snapshot *chkpoint_snap_shot_{nullptr};
-  util::ThreadPool thread_pool_{1};
-  util::ThreadPool compaction_thread_pool_{1};
+  util::ThreadPool thread_pool_{"v4blockchain::KeyValueBlockchain::thread_pool", 1};
+  util::ThreadPool compaction_thread_pool_{"v4blockchain::KeyValueBlockchain::compaction_thread_pool_", 1};
 
   // Metrics
   std::shared_ptr<concordMetrics::Aggregator> aggregator_;

kvbc/src/Replica.cpp

@@ -476,7 +476,8 @@ Replica::Replica(ICommunication *comm,
       aggregator_(aggregator),
       pm_{pm},
       secretsManager_{secretsManager},
-      blocks_io_workers_pool((replicaConfig.numWorkerThreadsForBlockIO > 0) ? replicaConfig.numWorkerThreadsForBlockIO
+      blocks_io_workers_pool("Replica::blocks_io_workers_pool",
+                             (replicaConfig.numWorkerThreadsForBlockIO > 0) ? replicaConfig.numWorkerThreadsForBlockIO
                                                                             : std::thread::hardware_concurrency()),
       AdaptivePruningManager_{
           concord::performance::IntervalMappingResourceManager::createIntervalMappingResourceManager(

kvbc/src/migrations/block_merkle_latest_ver_cf_migration.cpp

@@ -111,7 +111,7 @@ void BlockMerkleLatestVerCfMigration::clearExistingLatestVerCf() {
 
 void BlockMerkleLatestVerCfMigration::iterateAndMigrate() {
   auto blockchain = Blockchain{db_};
-  ThreadPool tp(iteration_batch_size_);
+  ThreadPool tp("BlockMerkleLatestVerCfMigration::iterateAndMigrate::tp", iteration_batch_size_);
   for (auto block_id = INITIAL_GENESIS_BLOCK_ID; block_id <= blockchain.getLastReachableBlockId();
        block_id += iteration_batch_size_) {
     std::vector<std::future<void>> tasks;

kvbc/test/v4blockchain/v4_blockchain_test.cpp

@@ -111,7 +111,7 @@ class v4_kvbc : public Test {
                                      v4blockchain::KeyValueBlockchain* blockchain,
                                      std::vector<std::future<void>>& tasks) {
     auto last_block = blockchain->getLastReachableBlockId();
-    concord::util::ThreadPool tp;
+    concord::util::ThreadPool tp{""};
 
     // Keys are:
     // <category_name>_key_<blockId>_<key_id>

kvbc/tools/migrations/v4migration_tool/src/migration_bookeeper.cpp

@@ -71,7 +71,8 @@ void BookKeeper::initialize_migration() {
     }
   }
   const auto point_lookup_threads = config_["point-lookup-threads"].as<std::uint64_t>();
-  batch_thread_pool_ = std::make_unique<ThreadPool>(static_cast<std::uint32_t>(point_lookup_threads));
+  batch_thread_pool_ = std::make_unique<ThreadPool>("migration_bookeeper::batch_thread_pool",
+                                                    static_cast<std::uint32_t>(point_lookup_threads));
 }
 
 void BookKeeper::migrate(const std::shared_ptr<concord::kvbc::adapter::ReplicaBlockchain>& from,

storage/include/s3/client.hpp

@@ -392,7 +392,7 @@ class Client : public concord::storage::IDBClient {
   logging::Logger logger_ = logging::getLogger("concord.storage.s3");
   uint16_t initialDelay_ = 100;
   Metrics metrics_;
-  util::ThreadPool thread_pool_{std::thread::hardware_concurrency()};
+  util::ThreadPool thread_pool_{"GetObjectResponseData::thread_pool", std::thread::hardware_concurrency()};
 };
 
 }  // namespace concord::storage::s3