grpc: monitor channel status

Author: ironMann

src/StfSender/StfSenderOutput.cxx

@@ -220,7 +220,12 @@ void StfSenderOutput::StfSchedulerThread()
       continue;
     }
 
-    DDLOGF_RL(1000, DataDistSeverity::debug, "StfSchedulerThread: scheduling stf_id={}", lStfId);
+    if (!mDevice.TfSchedRpcCli().is_ready()) {
+      EDDLOG_RL(1000, "StfSchedulerThread: TfScheduler gRPC connection is not ready stf_id={}", lStfId);
+      continue;
+    }
+
+    DDDLOG_RL(5000, "StfSchedulerThread: scheduling stf_id={}", lStfId);
 
     // move the stf into triage map (before notifying the scheduler to avoid races)
     {
@@ -248,11 +253,11 @@ void StfSenderOutput::StfSchedulerThread()
 
       mDevice.TfSchedRpcCli().StfSenderStfUpdate(lStfInfo, lSchedResponse);
 
-      DDLOGF_RL(3000, DataDistSeverity::debug, "Sent STF announce, stf_id={} stf_size={}", lStfId, lStfInfo.stf_size());
+      DDDLOG_RL(5000, "Sent STF announce, stf_id={} stf_size={}", lStfId, lStfInfo.stf_size());
 
       // check if the scheduler rejected the data
       if (lSchedResponse.status() != SchedulerStfInfoResponse::OK) {
-        DDLOGF_RL(500, DataDistSeverity::info, "TfScheduler rejected the Stf announce. stf_id={} reason={}",
+        IDDLOG_RL(1000, "TfScheduler rejected the Stf announce. stf_id={} reason={}",
           lStfId, SchedulerStfInfoResponse_StfInfoStatus_Name(lSchedResponse.status()));
 
         // remove from the scheduling map

src/TfScheduler/TfSchedulerConnManager.cxx

@@ -30,8 +30,15 @@ using namespace std::chrono_literals;
 
 std::size_t TfSchedulerConnManager::checkStfSenders()
 {
-  // TODO: monitor and reconnect?
-  return mStfSenderRpcClients.size();
+  std::size_t lReadyCnt = 0;
+  for (const auto &lId : mPartitionInfo.mStfSenderIdList) {
+    // this will attempt reconnection on existing connections
+    if (checkStfSenderRpcConn(lId)) {
+      lReadyCnt++;
+    }
+  }
+
+  return lReadyCnt;
 }
 
 void TfSchedulerConnManager::connectTfBuilder(const TfBuilderConfigStatus &pTfBuilderStatus, TfBuilderConnectionResponse &pResponse /*out*/)
@@ -243,20 +250,26 @@ void TfSchedulerConnManager::StfSenderMonitoringThread()
   std::vector<std::uint64_t> lDroppedStfs;
 
   while (mRunning) {
+    std::chrono::milliseconds lSleep = 1000ms;
+
     // make sure all StfSenders are alive
     const std::uint32_t lNumStfSenders = checkStfSenders();
     if (lNumStfSenders < mPartitionInfo.mStfSenderIdList.size()) {
-      IDDLOG("Waiting for StfSenders. connected={} total={}", lNumStfSenders, mPartitionInfo.mStfSenderIdList.size() );
-      std::this_thread::sleep_for(1000ms);
-      continue;
+
+      mStfSenderState = STF_SENDER_STATE_INCOMPLETE;
+
+      WDDLOG_RL(1000, "Waiting for StfSenders. ready={} total={}", lNumStfSenders, mPartitionInfo.mStfSenderIdList.size());
+      lSleep = 250ms;
     }
 
+    mStfSenderState = STF_SENDER_STATE_OK;
+
     // wait for drop futures
     {
       {
         std::scoped_lock lLock(mStfDropFuturesLock);
         for (auto lFutureIt = mStfDropFutures.begin(); lFutureIt != mStfDropFutures.end(); lFutureIt++) {
-          if (std::future_status::ready == lFutureIt->wait_for(std::chrono::seconds(0))) {
+          if (std::future_status::ready == lFutureIt->wait_for(0s)) {
             assert (lFutureIt->valid());
             lDroppedStfs.push_back(lFutureIt->get());
             lFutureIt = mStfDropFutures.erase(lFutureIt);
@@ -267,13 +280,12 @@ void TfSchedulerConnManager::StfSenderMonitoringThread()
       sort(lDroppedStfs.begin(), lDroppedStfs.end());
       for (auto &lDroppedId : lDroppedStfs) {
         lDroppedTotal++;
-        DDLOGF_RL(1000, DataDistSeverity::info, "Dropped SubTimeFrame (cannot schedule). stf_id={} total={}",
-          lDroppedId, lDroppedTotal);
+        IDDLOG_RL(1000, "Dropped SubTimeFrame (cannot schedule). stf_id={} total={}", lDroppedId, lDroppedTotal);
       }
       lDroppedStfs.clear();
     }
 
-    std::this_thread::sleep_for(1000ms);
+    std::this_thread::sleep_for(lSleep);
   }
 
   DDDLOG("Exiting StfSender RPC Monitoring thread.");

src/TfScheduler/TfSchedulerConnManager.h

@@ -38,6 +38,12 @@ namespace o2
 namespace DataDistribution
 {
 
+enum StfSenderState {
+  STF_SENDER_STATE_OK = 1,
+  STF_SENDER_STATE_INITIALIZING,
+  STF_SENDER_STATE_INCOMPLETE
+};
+
 class TfSchedulerConnManager
 {
  public:
@@ -116,15 +122,20 @@ class TfSchedulerConnManager
     return mTfBuilderRpcClients.get(pId);
   }
 
-  void checkStfSenderRpcConn(const std::string &lStfSenderId)
+  bool checkStfSenderRpcConn(const std::string &lStfSenderId)
   {
-    mStfSenderRpcClients.checkStfSenderRpcConn(lStfSenderId);
+    return mStfSenderRpcClients.checkStfSenderRpcConn(lStfSenderId);
   }
 
+  StfSenderState getStfSenderState() const { return mStfSenderState; }
+
 private:
   /// Partition information
   PartitionRequest mPartitionInfo;
 
+  /// StfSender state
+  std::atomic<StfSenderState> mStfSenderState = STF_SENDER_STATE_INITIALIZING;
+
   /// Discovery configuration
   std::shared_ptr<ConsulTfSchedulerInstance> mDiscoveryConfig;
 

src/TfScheduler/TfSchedulerStfInfo.cxx

@@ -119,8 +119,7 @@ void TfSchedulerStfInfo::SchedulingThread()
           } else {
             // TfBuilder was removed in the meantime, e.g. by housekeeping thread because of stale info
             // We drop the current TF as this is not a likely situation
-            WDDLOG(
-              "Selected TfBuilder is not currently reachable. TF will be dropped. tfb_id={:s} tf_id={:d}",
+            WDDLOG("Selected TfBuilder is not currently reachable. TF will be dropped. tfb_id={} tf_id={}",
               lTfBuilderId, lTfId);
 
             mConnManager.dropAllStfsAsync(lTfId);
@@ -204,11 +203,23 @@ void TfSchedulerStfInfo::SchedulingThread()
 
 void TfSchedulerStfInfo::addStfInfo(const StfSenderStfInfo &pStfInfo, SchedulerStfInfoResponse &pResponse)
 {
+  static std::uint64_t sLastDropNotRunning = -1;
+  static std::uint64_t sLastDropIncomplete = -1;
+
   const auto lNumStfSenders = mDiscoveryConfig->status().stf_sender_count();
   const auto lStfId = pStfInfo.stf_id();
 
-  if (!mRunning) {
+  // Drop not running
+  if ((sLastDropNotRunning == lStfId) || !mRunning) {
     pResponse.set_status(SchedulerStfInfoResponse::DROP_NOT_RUNNING);
+    sLastDropNotRunning = lStfId;
+    return;
+  }
+
+  // DROP When not complete
+  if ((sLastDropIncomplete == lStfId) || (mConnManager.getStfSenderState() == StfSenderState::STF_SENDER_STATE_INCOMPLETE)) {
+    pResponse.set_status(SchedulerStfInfoResponse::DROP_STFS_INCOMPLETE);
+    sLastDropIncomplete = lStfId;
     return;
   }
 

src/common/base/DataDistLogger.h

@@ -282,7 +282,6 @@ class DataDistLogger {
 #define WDDLOG(...) DDLOGF(DataDistSeverity::warn, __VA_ARGS__)
 #define EDDLOG(...) DDLOGF(DataDistSeverity::error, __VA_ARGS__)
 
-
 // Log with fmt using ratelimiting (per thread)
 #define DDLOGF_RL(intervalMs, severity, ...)                                                                          \
 do {                                                                                                                  \
@@ -299,6 +298,12 @@ do {
   }                                                                                                                   \
 } while(0)
 
+
+#define DDDLOG_RL(intervalMs, ...) DDLOGF_RL(intervalMs, DataDistSeverity::debug, __VA_ARGS__)
+#define IDDLOG_RL(intervalMs, ...) DDLOGF_RL(intervalMs, DataDistSeverity::info, __VA_ARGS__)
+#define WDDLOG_RL(intervalMs, ...) DDLOGF_RL(intervalMs, DataDistSeverity::warn, __VA_ARGS__)
+#define EDDLOG_RL(intervalMs, ...) DDLOGF_RL(intervalMs, DataDistSeverity::error, __VA_ARGS__)
+
 // Log with fmt using ratelimiting (global)
 #define DDLOGF_GRL(intervalMs, severity, ...)                                                                          \
 do {                                                                                                                  \
@@ -315,6 +320,12 @@ do {
   }                                                                                                                   \
 } while(0)
 
+
+#define DDDLOG_GRL(intervalMs, ...) DDLOGF_GRL(intervalMs, DataDistSeverity::debug, __VA_ARGS__)
+#define IDDLOG_GRL(intervalMs, ...) DDLOGF_GRL(intervalMs, DataDistSeverity::info, __VA_ARGS__)
+#define WDDLOG_GRL(intervalMs, ...) DDLOGF_GRL(intervalMs, DataDistSeverity::warn, __VA_ARGS__)
+#define EDDLOG_GRL(intervalMs, ...) DDLOGF_GRL(intervalMs, DataDistSeverity::error, __VA_ARGS__)
+
 namespace impl {
 struct DataDistLoggerCtx {
 

src/common/base/Utilities.h

@@ -99,6 +99,33 @@ class RunningSamples
     return mCount == 0 ? T(0) : (Sum() / T(mCount));
   }
 
+  T MeanStep() const
+  {
+    T lStepSum = 0;
+    std::size_t lStepsCounted = 0;
+
+    if (mCount < 2) {
+      return T(0);
+    }
+
+    for (auto it = begin()+1; it < end(); it++) {
+      if (*it <= *(it-1)) {
+        continue;
+      }
+
+      lStepSum += (*it - *(it-1));
+      lStepsCounted++;
+    }
+
+    return (lStepsCounted > 0) ? ((T)lStepSum / (T)lStepsCounted) : T(0);
+  }
+
+  T MeanStepFreq() const
+  {
+    const T lMeanStep = MeanStep();
+    return (lMeanStep != T(0)) ? (T(1) / lMeanStep) : T(0);
+  }
+
   std::pair<T, T> MinMax() const
   {
     if (mCount == 0)

src/common/discovery/CMakeLists.txt

@@ -56,7 +56,7 @@ set_target_properties(discovery PROPERTIES
 target_include_directories(discovery
   PUBLIC
     ${CMAKE_CURRENT_BINARY_DIR} # protobuf puts generated c++ files there
-    ${CMAKE_CURRENT_SOURCE_DIR}/../include/discovery
+    ${CMAKE_CURRENT_SOURCE_DIR}
 )
 
 target_link_libraries(discovery

src/common/include/discovery/Config.h src/common/discovery/Config.hsrc/common/include/discovery/Config.h renamed to src/common/discovery/Config.h

@@ -24,6 +24,13 @@
 #include <string>
 #include <map>
 #include <cassert>
+#include <future>
+
+#include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+#include <boost/lexical_cast.hpp>
+#include <boost/asio/ip/host_name.hpp>
 
 namespace o2
 {
@@ -52,13 +59,13 @@ struct ProcessType {
     mType = -1;
   }
 
-  constexpr operator int() const { return mType; }
+  // constexpr operator int() const { return mType; }
 
   constexpr bool operator==(const ProcessType &b) const { return mType == b.mType; }
   constexpr bool operator!=(const ProcessType &b) const { return mType != b.mType; }
   constexpr bool operator<(const ProcessType &b) const { return mType < b.mType; }
 
-  operator std::string() const {
+  std::string to_string() const {
 
     if (*this == ProcessType::StfBuilder)
       return "StfBuilder";
@@ -165,20 +172,72 @@ class Config {
   }
 
   static
-  std::string getIdOption(const FairMQProgOptions& pFMQProgOpt)
+  std::string getIdOption(const ProcessType pProcType, const FairMQProgOptions& pFMQProgOpt)
   {
-    auto lId = pFMQProgOpt.GetValue<std::string>(OptionKeyDiscoveryId);
-    if (lId.empty()) {
-      DDLOG(fair::Severity::ERROR) << "Process must have unique ID for DataDistribution discovery.";
-      throw std::invalid_argument("Process ID for DataDiscovery must be provided.");
+    // check cmdline first
+    {
+      std::string lId = pFMQProgOpt.GetValue<std::string>(OptionKeyDiscoveryId);
+      if (!lId.empty()) {
+        IDDLOG("Parameter <{}> provided on command line. value={}", OptionKeyDiscoveryId, lId);
+        return lId;
+      }
+
+      if (pProcType == ProcessType::StfSender) {
+        const auto lErrorMsg = fmt::format("Parameter <{}> for StfSender must be provided on command line.",
+          OptionKeyDiscoveryId);
+        EDDLOG( lErrorMsg);
+        throw std::invalid_argument(lErrorMsg);
+      }
+    }
+
+    {
+      // get an unique ID
+      std::string lUniquePart = boost::lexical_cast<std::string>(boost::uuids::random_generator()());
+      std::string lUniqueId = pProcType.to_string() + "-" + boost::asio::ip::host_name() + "-" + lUniquePart;
+
+      IDDLOG("Parameter {} not provided, using random value={}", Config::OptionKeyDiscoveryId, lUniqueId);
+      return lUniqueId;
     }
-    return lId;
   }
 
   static
   std::string getPartitionOption(const FairMQProgOptions& pFMQProgOpt)
   {
-    return pFMQProgOpt.GetValue<std::string>(OptionKeyDiscoveryPartition);
+    // check cmdline first
+    {
+      std::string lPartId = pFMQProgOpt.GetValue<std::string>(OptionKeyDiscoveryPartition);
+      if (!lPartId.empty()) {
+        IDDLOG("Parameter <{}> provided on command line. value={}",
+          OptionKeyDiscoveryPartition, lPartId);
+        return lPartId;
+      }
+    }
+
+    // setup for ODC
+    std::promise<std::string> mPartIdPromise;
+    std::future<std::string> mPartIdFuture = mPartIdPromise.get_future();
+
+    pFMQProgOpt.Subscribe<std::string>("o2.datadist", [&](const std::string& pKey, std::string pValue) {
+      IDDLOG("Config::Subscribe received key-value pair <{}>=<{}>", pKey, pValue);
+
+      // partition key for tf builder
+      if (pKey == OptionKeyDiscoveryPartition) {
+        mPartIdPromise.set_value(pValue);
+      } else {
+        EDDLOG("Config::Subscribe unrecognized key value pushed {}={}", pKey, pValue);
+      }
+    });
+
+    // wait for an ODC value
+    mPartIdFuture.wait();
+    const auto lDiscId = mPartIdFuture.get();
+    if (lDiscId.length() == 0) {
+      EDDLOG("Parameter {} provided from ODC. zero length", Config::OptionKeyDiscoveryPartition);
+      throw std::invalid_argument(fmt::format("Invalid ODC parameter <{}>: zero length", Config::OptionKeyDiscoveryPartition));
+    }
+
+    IDDLOG("Parameter {} provided from ODC. value={}", Config::OptionKeyDiscoveryPartition, lDiscId);
+    return lDiscId;
   }
 
   Config() = delete;