sched: debug partition terminate phase

Author: ironMann

src/TfScheduler/TfSchedulerConnManager.h

@@ -33,9 +33,7 @@
 #include <list>
 #include <future>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 enum StfSenderState {
@@ -167,7 +165,7 @@ class TfSchedulerConnManager
 
   /// StfSender RPC-client channels
   std::recursive_mutex mStfSenderClientsLock;
-  StfSenderRpcClientCollection<ConsulTfSchedulerInstance> mStfSenderRpcClients;
+    StfSenderRpcClientCollection<ConsulTfSchedulerInstance> mStfSenderRpcClients;
   /// TfBuilder RPC-client channels
   TfBuilderRpcClientCollection<ConsulTfSchedulerInstance> mTfBuilderRpcClients;
 
@@ -176,7 +174,7 @@ class TfSchedulerConnManager
     std::condition_variable_any mStfDropFuturesCV;
     std::list<std::future<std::uint64_t>> mStfDropFutures;
 };
-}
+
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_TF_SCHEDULER_CONNMANAGER_H_ */

src/TfScheduler/TfSchedulerInstanceRpc.cxx

@@ -18,9 +18,7 @@
 #include <condition_variable>
 #include <stdexcept>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using namespace std::chrono_literals;
@@ -158,42 +156,42 @@ ::grpc::Status TfSchedulerInstanceRpcImpl::HeartBeat(::grpc::ServerContext* /*co
 ::grpc::Status TfSchedulerInstanceRpcImpl::GetPartitionState(::grpc::ServerContext* /*context*/,
   const ::o2::DataDistribution::PartitionInfo* /*request*/, ::o2::DataDistribution::PartitionResponse* response)
 {
-  DDDLOG("gRPC server: GetPartitionState");
-
+  // Terminating?
   if (!accepting_updates()) {
     response->set_partition_state(mPartitionState);
+    DDDLOG("gRPC server: GetPartitionState() state={}", PartitionState_Name(mPartitionState));
     return Status::OK;
   }
 
-  // Terminating?
   switch (mConnManager.getStfSenderState()) {
     case StfSenderState::STF_SENDER_STATE_OK:
     {
-      response->set_partition_state(o2::DataDistribution::PartitionState::PARTITION_CONFIGURED);
+      response->set_partition_state(PartitionState::PARTITION_CONFIGURED);
       response->set_info_message("Partition is fully configured.");
       break;
     }
     case StfSenderState::STF_SENDER_STATE_INITIALIZING:
     {
-      response->set_partition_state(o2::DataDistribution::PartitionState::PARTITION_CONFIGURING);
+      response->set_partition_state(PartitionState::PARTITION_CONFIGURING);
       const auto lMsg = fmt::format("Partition is being configured. Found {} out of {} StfSenders.",
         mConnManager.getStfSenderCount(), mConnManager.getStfSenderSet().size());
       response->set_info_message(lMsg);
       break;
     }
     case StfSenderState::STF_SENDER_STATE_INCOMPLETE:
     {
-      response->set_partition_state(o2::DataDistribution::PartitionState::PARTITION_ERROR);
+      response->set_partition_state(PartitionState::PARTITION_ERROR);
       response->set_info_message("Not all StfSenders are reachable.");
       break;
     }
     default:
     {
-      response->set_partition_state(o2::DataDistribution::PartitionState::PARTITION_ERROR);
+      response->set_partition_state(PartitionState::PARTITION_ERROR);
       response->set_info_message("Unknown partition state.");
     }
   }
 
+  DDDLOG("gRPC server: GetPartitionState() state={}", PartitionState_Name(response->partition_state()));
   return Status::OK;
 }
 
@@ -207,7 +205,7 @@ ::grpc::Status TfSchedulerInstanceRpcImpl::TerminatePartition(::grpc::ServerCont
     updatePartitionState(PartitionState::PARTITION_TERMINATING);
     response->set_info_message("Terminate started.");
   } else {
-    const auto lMsg = fmt::format("Terminate is already requested. partition_id={}", request->partition_id());
+    const auto lMsg = fmt::format("Terminate was already requested. partition_id={}", request->partition_id());
     response->set_info_message(lMsg);
     WDDLOG(lMsg);
   }
@@ -298,5 +296,4 @@ ::grpc::Status TfSchedulerInstanceRpcImpl::StfSenderStfUpdate(::grpc::ServerCont
 }
 
 
-}
 } /* o2::DataDistribution */

src/TfScheduler/TfSchedulerStfInfo.cxx

@@ -22,9 +22,7 @@
 #include <tuple>
 #include <algorithm>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using namespace std::chrono_literals;
@@ -327,7 +325,7 @@ void TfSchedulerStfInfo::HighWatermarkThread()
     }
   }
 
-  DDDLOG("Exiting StfInfo Scheduling thread.");
+  DDDLOG("Exiting HighWatermarkThread thread.");
 }
 
 
@@ -476,5 +474,4 @@ void TfSchedulerStfInfo::addStfInfo(const StfSenderStfInfo &pStfInfo, SchedulerS
   }
 }
 
-}
 } /* o2::DataDistribution */

src/TfScheduler/TfSchedulerStfInfo.h

@@ -33,9 +33,7 @@
 #include <thread>
 #include <chrono>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using namespace std::chrono_literals;
@@ -85,6 +83,7 @@ class TfSchedulerStfInfo
   }
 
   void stop() {
+    DDDLOG("TfSchedulerStfInfo::stop()");
     mRunning = false;
     mDropQueue.stop();
     mCompleteStfsInfoQueue.stop();
@@ -120,7 +119,7 @@ class TfSchedulerStfInfo
 
 private:
   /// Discard timeout for incomplete TFs
-  static constexpr auto sStfDiscardTimeout = 10s;
+  static constexpr auto sStfDiscardTimeout = 5s;
 
   std::atomic_bool mRunning = false;
 
@@ -174,7 +173,7 @@ class TfSchedulerStfInfo
   /// stale cleanup thread
   std::thread mStaleStfThread;
 };
-}
+
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_TF_SCHEDULER_STF_INFO_H_ */

src/common/discovery/StfSenderRpcClient.h

@@ -27,9 +27,7 @@
 #include <map>
 #include <thread>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using grpc::Server;
@@ -68,7 +66,8 @@ class StfSenderRpcClient {
     PartitionInfo lPartInfo; // TODO: specify and check partition ID
     PartitionResponse lRet;
 
-    return mStub->TerminatePartition(&lContext, lPartInfo, &lRet).ok();
+    mStub->TerminatePartition(&lContext, lPartInfo, &lRet);
+    return true; // could have been stopped by the ECS
   }
 
   bool is_ready() const;
@@ -230,8 +229,6 @@ class StfSenderRpcClientCollection {
   std::map<std::string, std::unique_ptr<StfSenderRpcClient>> mClients;
 };
 
-}
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_DATADIST_StfSender_RPC_CLIENT_H_ */
-

src/common/discovery/TfBuilderRpcClient.h

@@ -24,9 +24,7 @@
 #include <map>
 #include <thread>
 
-namespace o2
-{
-namespace DataDistribution
+namespace o2::DataDistribution
 {
 
 using grpc::Server;
@@ -108,13 +106,10 @@ class TfBuilderRpcClientCtx {
     PartitionResponse lResponse;
 
     auto lStatus = mStub->TerminatePartition(&lContext, lPartitionInfo, &lResponse);
-    if (!lStatus.ok()) {
-      return false;
-    }
 
-    if (lStatus.ok() && lResponse.partition_state() != PartitionState::PARTITION_TERMINATING) {
-      EDDLOG("TerminatePartition: TfBuilder. state={} message={}", lResponse.partition_state(), lStatus.error_message());
-    }
+    // this is best effort only. ECS could have already stopped them
+    DDDLOG("TerminatePartition: TfBuilder. tfb_id={} state={} message={}",
+      mTfBuilderConf.info().process_id(), lResponse.partition_state(), lStatus.error_message());
     return true;
   }
 
@@ -279,7 +274,6 @@ class TfBuilderRpcClientCollection {
 };
 
 
-}
 } /* namespace o2::DataDistribution */
 
 #endif /* ALICEO2_DATADIST_TFBUILDER_RPC_CLIENT_H_ */