Merge branch 'master' into nkumar4/clientVerifiability-POC

Author: nkumar04

CMakeLists.txt

@@ -44,6 +44,7 @@ option(USE_GRPC "Enable GRPC and Protobuf" ON)
 option(USE_OPENSSL "Enable use of OpenSSL" ON)
 option(BUILD_THIRDPARTY "Wheter to build third party librarie or use preinstalled ones" OFF)
 option(CODECOVERAGE "Enable Code Coverage Metrics in Clang" OFF)
+option(ENABLE_RESTART_RECOVERY_TESTS "Enable tests for restart recovery" OFF)
 
 if(USE_OPENSSL AND NOT BUILD_THIRDPARTY)
     set(OPENSSL_ROOT_DIR /usr/local/ssl) # not to confuse with system ssl libs

Makefile

@@ -56,6 +56,7 @@ CONCORD_BFT_CMAKE_ASAN?=FALSE
 CONCORD_BFT_CMAKE_TSAN?=FALSE
 CONCORD_BFT_CMAKE_CODECOVERAGE?=FALSE
 CONCORD_BFT_CMAKE_USE_FAKE_CLOCK_IN_TIME_SERVICE?=FALSE
+ENABLE_RESTART_RECOVERY_TESTS?=FALSE
 ifeq (${CONCORD_BFT_CMAKE_ASAN},TRUE)
 	CONCORD_BFT_CMAKE_CXX_FLAGS_RELEASE='-O0 -g'
 else ifeq (${CONCORD_BFT_CMAKE_TSAN},TRUE)
@@ -88,7 +89,8 @@ CONCORD_BFT_CMAKE_FLAGS?= \
 			-DLEAKCHECK=${CONCORD_BFT_CMAKE_ASAN} \
 			-DTHREADCHECK=${CONCORD_BFT_CMAKE_TSAN} \
 			-DCODECOVERAGE=${CONCORD_BFT_CMAKE_CODECOVERAGE} \
-			-DTXN_SIGNING_ENABLED=${CONCORD_BFT_CMAKE_TRANSACTION_SIGNING_ENABLED}
+			-DTXN_SIGNING_ENABLED=${CONCORD_BFT_CMAKE_TRANSACTION_SIGNING_ENABLED} \
+			-DENABLE_RESTART_RECOVERY_TESTS=${ENABLE_RESTART_RECOVERY_TESTS}
 
 
 # The consistency parameter makes sense only at MacOS.

bftengine/include/bftengine/DbCheckpointManager.hpp

@@ -136,6 +136,10 @@ class DbCheckpointManager {
   }
   inline auto getLastDbCheckpointSeqNum() const { return lastCheckpointSeqNum_; }
   std::string getDiskUsageInfo();
+  // return a map of pair<checkpoint_id, size_on_disk>
+  // checkpoint_id = 0 indicates rocksdb size
+  // only used for apollo test
+  std::map<uint64_t, uint64_t> getDbSize();
 
  private:
   logging::Logger getLogger() {

bftengine/src/bcstatetransfer/BCStateTran.cpp

@@ -1481,8 +1481,12 @@ bool BCStateTran::onMessage(const AskForCheckpointSummariesMsg *m, uint32_t msgL
     if (!psd_->hasCheckpointDesc(i)) continue;
 
     DataStore::CheckpointDesc cpDesc = psd_->getCheckpointDesc(i);
-    std::unique_ptr<CheckpointSummaryMsg> msg =
-        std::unique_ptr<CheckpointSummaryMsg>(CheckpointSummaryMsg::create(cpDesc.rvbData.size()));
+    auto deleter = [](CheckpointSummaryMsg *msg) {
+      char *bytes = reinterpret_cast<char *>(msg);
+      delete[] bytes;
+    };
+    auto msg = std::unique_ptr<CheckpointSummaryMsg, decltype(deleter)>(
+        CheckpointSummaryMsg::create(cpDesc.rvbData.size()), deleter);
 
     msg->checkpointNum = i;
     msg->maxBlockId = cpDesc.maxBlockId;

bftengine/src/bftengine/DbCheckpointManager.cpp

@@ -148,7 +148,7 @@ uint64_t DbCheckpointManager::directorySize(const _fs::path& directory, const bo
   uint64_t size{0};
   try {
     if (_fs::exists(directory)) {
-      for (const auto& entry : _fs::recursive_directory_iterator(directory)) {
+      for (const auto& entry : _fs::directory_iterator(directory)) {
         if (_fs::is_regular_file(entry) && !_fs::is_symlink(entry)) {
           if (_fs::hard_link_count(entry) > 1 && excludeHardLinks) continue;
           size += _fs::file_size(entry);
@@ -348,7 +348,8 @@ void DbCheckpointManager::updateMetrics() {
   if (const auto it = dbCheckptMetadata_.dbCheckPoints_.crbegin(); it != dbCheckptMetadata_.dbCheckPoints_.crend()) {
     _fs::path path(checkpointDir);
     _fs::path chkptIdPath = path / std::to_string(it->first);
-    auto lastDbCheckpointSize = directorySize(chkptIdPath, false, true);
+    // db checkpoint directory does not have any sub-directory
+    auto lastDbCheckpointSize = directorySize(chkptIdPath, false, false);
     lastDbCheckpointSizeInMb_.Get().Set(lastDbCheckpointSize / (1024 * 1024));
     metrics_.UpdateAggregator();
     LOG_INFO(getLogger(), "rocksdb check point id:" << it->first << ", size: " << HumanReadable{lastDbCheckpointSize});
@@ -398,4 +399,20 @@ std::string DbCheckpointManager::getDiskUsageInfo() {
   }
   return ss.str();
 }
+std::map<uint64_t, uint64_t> DbCheckpointManager::getDbSize() {
+  auto find_db_size = [&](const std::string& path) -> uint64_t {
+    _fs::path dbPath(path);
+    return directorySize(dbPath, false, false);
+  };
+  auto dbSizeMap = std::map<uint64_t, uint64_t>{};
+  dbSizeMap[0] = find_db_size(dbClient_->getPath());
+  {
+    std::scoped_lock lock(lock_);
+    for (const auto& [db_chkpt_id, _] : dbCheckptMetadata_.dbCheckPoints_) {
+      (void)_;
+      dbSizeMap[db_chkpt_id] = find_db_size(dbClient_->getPathForCheckpoint(db_chkpt_id));
+    }
+  }
+  return dbSizeMap;
+}
 }  // namespace bftEngine::impl

bftengine/src/bftengine/KeyExchangeManager.cpp

@@ -145,10 +145,12 @@ void KeyExchangeManager::loadPublicKeys() {
 void KeyExchangeManager::exchangeTlsKeys(const std::string& type, const SeqNum& bft_sn) {
   auto keys = concord::util::crypto::Crypto::instance().generateECDSAKeyPair(
       concord::util::crypto::KeyFormat::PemFormat, concord::util::crypto::CurveType::secp384r1);
-  std::string root_path =
-      bftEngine::ReplicaConfig::instance().certificatesRootPath + "/" + std::to_string(repID_) + "/" + type;
+  bool use_unified_certs = bftEngine::ReplicaConfig::instance().useUnifiedCertificates;
+  const std::string base_path =
+      bftEngine::ReplicaConfig::instance().certificatesRootPath + "/" + std::to_string(repID_);
+  std::string root_path = (use_unified_certs) ? base_path : base_path + "/" + type;
 
-  std::string cert_path = root_path + "/" + type + ".cert";
+  std::string cert_path = (use_unified_certs) ? root_path + "/node.cert" : root_path + "/" + type + ".cert";
   std::string prev_key_pem = concord::util::crypto::Crypto::instance()
                                  .RsaHexToPem(std::make_pair(SigManager::instance()->getSelfPrivKey(), ""))
                                  .first;

bftengine/src/bftengine/ReplicaImp.cpp

@@ -2448,6 +2448,7 @@ void ReplicaImp::startExecution(SeqNum seqNumber,
   if (isCurrentPrimary()) {
     metric_consensus_duration_.finishMeasurement(seqNumber);
     metric_post_exe_duration_.addStartTimeStamp(seqNumber);
+    metric_consensus_end_to_core_exe_duration_.addStartTimeStamp(seqNumber);
   }
 
   consensus_times_.end(seqNumber);
@@ -4341,9 +4342,13 @@ ReplicaImp::ReplicaImp(bool firstTime,
       metric_total_preexec_requests_executed_{metrics_.RegisterCounter("totalPreExecRequestsExecuted")},
       metric_received_restart_ready_{metrics_.RegisterCounter("receivedRestartReadyMsg", 0)},
       metric_received_restart_proof_{metrics_.RegisterCounter("receivedRestartProofMsg", 0)},
-      metric_consensus_duration_{metrics_, "consensusDuration", 1000, true},
-      metric_post_exe_duration_{metrics_, "postExeDuration", 1000, true},
-      metric_primary_batching_duration_{metrics_, "primaryBatchingDuration", 10000, true},
+      metric_consensus_duration_{metrics_, "consensusDuration", 1000, 100, true},
+      metric_post_exe_duration_{metrics_, "postExeDuration", 1000, 100, true},
+      metric_core_exe_func_duration_{metrics_, "postExeCoreFuncDuration", 1000, 100, true},
+      metric_consensus_end_to_core_exe_duration_{metrics_, "consensusEndToExeStartDuration", 1000, 100, true},
+      metric_post_exe_thread_idle_time_{metrics_, "PostExeThreadIdleDuration", 1000, 100, true},
+      metric_post_exe_thread_active_time_{metrics_, "PostExeThreadActiveDuration", 1000, 100, true},
+      metric_primary_batching_duration_{metrics_, "primaryBatchingDuration", 10000, 1000, true},
       consensus_times_(histograms_.consensus),
       checkpoint_times_(histograms_.checkpointFromCreationToStable),
       time_in_active_view_(histograms_.timeInActiveView),
@@ -4817,6 +4822,11 @@ void ReplicaImp::startPrePrepareMsgExecution(PrePrepareMsg *ppMsg,
     // send internal message that will call to finishExecutePrePrepareMsg
     ConcordAssert(activeExecutions_ == 0);
     activeExecutions_ = 1;
+    if (isCurrentPrimary()) {
+      metric_post_exe_thread_active_time_.addStartTimeStamp(0);
+      metric_post_exe_thread_idle_time_.finishMeasurement(0);
+    }
+
     InternalMessage im = FinishPrePrepareExecutionInternalMsg{ppMsg, nullptr};  // TODO(GG): check....
     getIncomingMsgsStorage().pushInternalMsg(std::move(im));
   }
@@ -4908,6 +4918,10 @@ void ReplicaImp::executeAllPrePreparedRequests(bool allowParallelExecution,
 
   ConcordAssert(activeExecutions_ == 0);
   activeExecutions_ = 1;
+  if (isCurrentPrimary()) {
+    metric_post_exe_thread_active_time_.addStartTimeStamp(0);
+    metric_post_exe_thread_idle_time_.finishMeasurement(0);
+  }
   if (shouldRunRequestsInParallel) {
     PostExecJob *j = new PostExecJob(ppMsg, requestSet, time, *this);
     postExecThread_.add(j);
@@ -5032,7 +5046,14 @@ void ReplicaImp::executeRequests(PrePrepareMsg *ppMsg, Bitmap &requestSet, Times
       span.setTag("rid", config_.getreplicaId());
       span.setTag("cid", ppMsg->getCid());
       span.setTag("seq_num", ppMsg->seqNumber());
+      if (isCurrentPrimary()) {
+        metric_consensus_end_to_core_exe_duration_.finishMeasurement(ppMsg->seqNumber());
+        metric_core_exe_func_duration_.addStartTimeStamp(ppMsg->seqNumber());
+      }
       bftRequestsHandler_->execute(*pAccumulatedRequests, time, ppMsg->getCid(), span);
+      if (isCurrentPrimary()) {
+        metric_core_exe_func_duration_.finishMeasurement(ppMsg->seqNumber());
+      }
     }
   } else {
     LOG_INFO(
@@ -5067,6 +5088,10 @@ void ReplicaImp::executeRequests(PrePrepareMsg *ppMsg, Bitmap &requestSet, Times
 void ReplicaImp::finishExecutePrePrepareMsg(PrePrepareMsg *ppMsg,
                                             IRequestsHandler::ExecutionRequestsQueue *pAccumulatedRequests) {
   activeExecutions_ = 0;
+  if (isCurrentPrimary()) {
+    metric_post_exe_thread_idle_time_.addStartTimeStamp(0);
+    metric_post_exe_thread_active_time_.finishMeasurement(0);
+  }
 
   if (pAccumulatedRequests != nullptr) {
     sendResponses(ppMsg, *pAccumulatedRequests);

bftengine/src/bftengine/ReplicaImp.hpp

@@ -314,6 +314,10 @@ class ReplicaImp : public InternalReplicaApi, public ReplicaForStateTransfer {
   CounterHandle metric_received_restart_proof_;
   PerfMetric<uint64_t> metric_consensus_duration_;
   PerfMetric<uint64_t> metric_post_exe_duration_;
+  PerfMetric<uint64_t> metric_core_exe_func_duration_;
+  PerfMetric<uint64_t> metric_consensus_end_to_core_exe_duration_;
+  PerfMetric<uint64_t> metric_post_exe_thread_idle_time_;
+  PerfMetric<uint64_t> metric_post_exe_thread_active_time_;
   PerfMetric<std::string> metric_primary_batching_duration_;
   //*****************************************************
   RollingAvgAndVar consensus_time_;

bftengine/src/preprocessor/PreProcessor.cpp

@@ -363,7 +363,7 @@ PreProcessor::PreProcessor(shared_ptr<MsgsCommunicator> &msgsCommunicator,
                            metricsComponent_.RegisterAtomicGauge("preProcessingTimeAvg", 0),
                            metricsComponent_.RegisterAtomicGauge("launchAsyncPreProcessJobTimeAvg", 0),
                            metricsComponent_.RegisterAtomicGauge("PreProcInFlyRequestsNum", 0)},
-      metric_pre_exe_duration_{metricsComponent_, "metric_pre_exe_duration_", 10000, true},
+      metric_pre_exe_duration_{metricsComponent_, "metric_pre_exe_duration_", 10000, 1000, true},
       totalPreProcessingTime_(true),
       launchAsyncJobTimeAvg_(true),
       preExecReqStatusCheckPeriodMilli_(myReplica_.getReplicaConfig().preExecReqStatusCheckTimerMillisec),

client/bftclient/include/bftclient/bft_client.h

@@ -35,7 +35,9 @@ typedef std::shared_ptr<bft::communication::ICommunication> SharedCommPtr;
 
 class Client {
  public:
-  Client(SharedCommPtr comm, const ClientConfig& config);
+  Client(SharedCommPtr comm,
+         const ClientConfig& config,
+         std::shared_ptr<concordMetrics::Aggregator> aggregator = nullptr);
 
   void setAggregator(const std::shared_ptr<concordMetrics::Aggregator>& aggregator) {
     metrics_.setAggregator(aggregator);