tfscheduler: implement tf scheduling throttling based on setting provided from aliecs

Author: ironMann

src/TfScheduler/TfSchedulerInstance.cxx

@@ -43,6 +43,7 @@ TfSchedulerInstanceHandler::TfSchedulerInstanceHandler(DataDistDevice& pDev,
   lStatus.mutable_info()->set_process_id(pProcessId);
   lStatus.mutable_info()->set_ip_address(Config::getNetworkIfAddressOption(*pDev.GetConfig()));
   lStatus.mutable_partition()->set_partition_id(mPartitionInfo.mPartitionId);
+  lStatus.set_allocated_partition_params(&mPartitionInfo.mParameters);
 
   lStatus.set_stf_sender_count(mPartitionInfo.mStfSenderIdList.size());
   for (const auto &lStfSenderId : mPartitionInfo.mStfSenderIdList) {

src/TfScheduler/TfSchedulerStfInfo.cxx

@@ -23,6 +23,7 @@
 #include <set>
 #include <tuple>
 #include <algorithm>
+#include <random>
 
 namespace o2::DataDistribution
 {
@@ -456,6 +457,10 @@ void TfSchedulerStfInfo::DropThread()
 
 void TfSchedulerStfInfo::addStfInfo(const StfSenderStfInfo &pStfInfo, SchedulerStfInfoResponse &pResponse)
 {
+  using namespace std::chrono_literals;
+  static thread_local std::default_random_engine lGen;
+  static thread_local std::uniform_real_distribution<double> lUniformDist(0, 100.0);
+
   const std::uint64_t lRunNumber = pStfInfo.partition().run_number();
   const auto lStfId = pStfInfo.stf_id();
 
@@ -494,6 +499,32 @@ void TfSchedulerStfInfo::addStfInfo(const StfSenderStfInfo &pStfInfo, SchedulerS
           SchedulerStfInfoResponse::DROP_SCHED_DISCARDED);
         return;
       }
+
+      { // check if this tf is rejected because of throttling
+        static std::uint64_t lThrottlingRejectedSize = 0;
+
+        if (lStfId > mLastThrottledStfId) {
+
+          // decide if discarding on throttling
+          bool lTfDropped = (lUniformDist(lGen) <= mPercentageToBuild) ? false : true;
+          if (lTfDropped) {
+            mDroppedThrottlingStfs.SetEvent(lStfId);
+            mDroppedStfs.SetEvent(lStfId);
+            mNotScheduledTfsCount++;
+            // this will only accurately represent the rate
+            DDMON_RATE("tfscheduler", "tf.rejected.tf", lThrottlingRejectedSize);
+            lThrottlingRejectedSize = 0;
+          }
+          mLastThrottledStfId = lStfId;
+        }
+
+        if (mDroppedThrottlingStfs.GetEvent(lStfId)) {
+          mTfSizeTotalRejected += pStfInfo.stf_size();
+          lThrottlingRejectedSize += pStfInfo.stf_size();
+          pResponse.set_status(SchedulerStfInfoResponse::DROP_STFS_THROTTLING);
+          return;
+        }
+      }
     }
 
     // Drop not running

src/TfScheduler/TfSchedulerStfInfo.h

@@ -28,6 +28,8 @@
 
 #include <Utilities.h>
 
+#include <boost/algorithm/string/trim.hpp>
+
 #include <vector>
 #include <map>
 #include <thread>
@@ -87,13 +89,37 @@ class TfSchedulerStfInfo
   : mDiscoveryConfig(pDiscoveryConfig),
     mConnManager(pConnManager),
     mTfBuilderInfo(pTfBuilderInfo),
-    mDroppedStfs(24ULL * 3600 * 88), // 1h of running ~ 1MiB size
-    mBuiltTfs(24ULL * 3600 * 88)
+    mDroppedStfs(48ULL * 3600 * 88), // 1h of running ~ 1MiB size
+    mDroppedThrottlingStfs(48ULL * 3600 * 88),
+    mBuiltTfs(48ULL * 3600 * 88)
   { }
 
   ~TfSchedulerStfInfo() { }
 
   void start() {
+    // get parameters
+
+    // Build TfPercentage from AliECS
+    if (mDiscoveryConfig->status().partition_params().param_values().count("BuildTfPercentage") > 0) {
+      double lTfPercent = 100.0;
+      auto lBuildPercentString = mDiscoveryConfig->status().partition_params().param_values().at("BuildTfPercentage");
+      boost::algorithm::trim(lBuildPercentString);
+
+      try {
+        if (lBuildPercentString.empty()) {
+          throw boost::bad_lexical_cast();
+        }
+        lTfPercent = boost::lexical_cast<double>(lBuildPercentString);
+
+      } catch(boost::bad_lexical_cast const &e) {
+        EDDLOG("Error while parsing AliECS parameter 'BuildTfPercentage'. str_value={} what={}", lBuildPercentString, e.what());
+        lTfPercent = 100.0;
+      }
+
+      mPercentageToBuild = std::clamp(lTfPercent, 0.0, 100.0);
+      IDDLOG("TfScheduler parameters: BuildTfPercentage={:.4}", mPercentageToBuild);
+    }
+
     mStfInfoMap.clear();
 
     mRunning = true;
@@ -169,6 +195,7 @@ class TfSchedulerStfInfo
 
   /// Discovery configuration
   std::shared_ptr<ConsulTfScheduler> mDiscoveryConfig;
+  double mPercentageToBuild = 100.0;
 
   /// RPC clients to StfSenders and TfBuilders
   TfSchedulerConnManager &mConnManager;
@@ -202,14 +229,17 @@ class TfSchedulerStfInfo
     std::map<std::string, std::uint64_t> mMaxStfIdPerStfSender;
     // total sizes
     std::uint64_t mTfSizeTotalScheduled = 0;
-    std::uint64_t mTfSizeTotalRejected = 0;
+    std::atomic_uint64_t mTfSizeTotalRejected = 0;
 
     std::uint64_t mStaleTfCount = 0;
     std::uint64_t mScheduledTfs = 0;
 
     EventRecorder mDroppedStfs;
+    EventRecorder mDroppedThrottlingStfs;
+    std::uint64_t mLastThrottledStfId = 0;
     EventRecorder mBuiltTfs;
 
+
     void reset() {
       // NOTE: only call when holding mGlobalStfInfoLock
       mLastStfId = 0;
@@ -227,6 +257,8 @@ class TfSchedulerStfInfo
       mTfSizeTotalRejected = 0;
 
       mDroppedStfs.reset();
+      mDroppedThrottlingStfs.reset();
+      mLastThrottledStfId = 0;
       mBuiltTfs.reset();
 
       if (!mStfInfoMap.empty()) {

src/common/discovery/ConfigConsul.h

@@ -19,6 +19,7 @@
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-parameter"
+#include <google/protobuf/util/json_util.h>
 #include "discovery.pb.h"
 #pragma GCC diagnostic pop
 
@@ -464,6 +465,7 @@ class ConsulConfig : public Config {
     static const std::string sPartitionIdSubKey = "/partition-id"s;
     static const std::string sRequestCreateTimeKey = "/request-create-time"s;
     static const std::string sStfSenderListSubKey = "/stf-sender-id-list"s;
+    static const std::string sPartParamsKey = "/parameters"s;
 
     static_assert(std::is_same_v<T, TfSchedulerConfigStatus>, "Only TfScheduler can call this method.");
 
@@ -475,6 +477,7 @@ class ConsulConfig : public Config {
       const auto lReqPartitionIdKey = lPartReqKey + sPartitionIdSubKey;
       const auto lReqCreateTimeKey = lPartReqKey + sRequestCreateTimeKey;
       const auto lReqStfSenderListKey = lPartReqKey + sStfSenderListSubKey;
+      const auto lReqParametersKey = lPartReqKey + sPartParamsKey;
 
       std::scoped_lock lLock(mConsulLock);
 
@@ -514,14 +517,40 @@ class ConsulConfig : public Config {
         }
 
         // list of all FLPs
-        // ../<part-id>/stf-sender-id-list: <str date-time>
+        // ../<part-id>/stf-sender-id-list: <str>
         auto lFlpIdList = std::find_if(std::begin(lReqItems), std::end(lReqItems),
           [&] (KeyValue const& p) { return p.key == lReqStfSenderListKey; });
         if (lFlpIdList == std::end(lReqItems)) {
           EDDLOG("Invalid new partition request. Missing key: {}", lReqStfSenderListKey);
           break;
         }
 
+        // [optional] parameters from AliECS
+        // ../<part-id>/parameters: <protobuf PartitionParameters as json>
+        auto lPartParams = std::find_if(std::begin(lReqItems), std::end(lReqItems),
+          [&] (KeyValue const& p) { return p.key == lReqParametersKey; });
+        if (lPartParams != std::end(lReqItems)) {
+
+          // just validate the parameters can be decoded
+          PartitionParameters lPartParamsProto;
+
+          // Parse the json_string
+          google::protobuf::util::JsonParseOptions lJsonOptions;
+          lJsonOptions.ignore_unknown_fields = true;
+          lJsonOptions.case_insensitive_enum_parsing = true;
+
+          if (JsonStringToMessage(lPartParams->value, &lPartParamsProto, lJsonOptions).ok()) {
+            pNewPartitionRequest.mParameters = std::move(lPartParamsProto);
+
+            for (const auto lParamKV : pNewPartitionRequest.mParameters.param_values()) {
+              IDDLOG("Partition request parameters: {:30} : {}", lParamKV.first, lParamKV.second);
+            }
+
+          } else {
+            EDDLOG("Cannot parse partition parameter json object. json_str={}", lPartParams->value);
+          }
+        }
+
         // validate the request fields
         // partition name, check if already exist
         const std::string lPartitionId = boost::trim_copy(lPartitionIdIt->value);

src/common/discovery/ConfigParameters.h

@@ -14,6 +14,12 @@
 #ifndef ALICEO2_DATADIST_CONFIG_PARAMETERS_H_
 #define ALICEO2_DATADIST_CONFIG_PARAMETERS_H_
 
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#include "discovery.pb.h"
+#pragma GCC diagnostic pop
+
+
 #include <string>
 #include <vector>
 #include <cassert>
@@ -45,6 +51,8 @@ struct PartitionRequest {
   std::string   mPartitionId;
   std::string   mReqCreatedTime;
   std::vector<std::string> mStfSenderIdList;
+
+  PartitionParameters mParameters;
 };
 
 

src/common/discovery/discovery.proto

@@ -52,15 +52,21 @@ message PartitionInfo {
   uint64 run_number     = 3;
 }
 
+message PartitionParameters {
+  map<string, string> param_values = 1;
+}
+
 message TfSchedulerConfigStatus {
-            BasicInfo       info                = 1;
-            PartitionInfo   partition           = 2;
+            BasicInfo           info                = 1;
+            PartitionInfo       partition           = 2;
+
+            string              rpc_endpoint        = 3;
 
-            string          rpc_endpoint        = 3;
+            uint32              stf_sender_count    = 4;
+  repeated  string              stf_sender_id_list  = 5;
+            PartitionState      partition_state     = 6;
 
-            uint32          stf_sender_count    = 4;
-  repeated  string          stf_sender_id_list  = 5;
-            PartitionState  partition_state     = 6;
+            PartitionParameters partition_params    = 10; // Parameters from AliECS
 }
 
 message StfBuilderConfigStatus {
@@ -256,6 +262,7 @@ message SchedulerStfInfoResponse {
     DROP_SCHED_DISCARDED      = 4; // scheduler already processed that tf
     DROP_STFS_INCOMPLETE      = 5; // all StfSenders are not reachable
     DROP_STFS_BUFFER_FULL     = 6; // buffers on StfSender are full
+    DROP_STFS_THROTTLING      = 7; // Not scheduling because of throttling settings
   }
 
   StfInfoStatus  status = 1;