stfsender: support cancellation of init process

Author: ironMann

src/StfSender/StfSenderDevice.cxx

@@ -153,10 +153,7 @@ void StfSenderDevice::Reset()
   DDDLOG("StfBuilderDevice::Reset()");
 
   I().mDeviceRunning = false;
-  // wait the Info thread, before closing mTfSchedulerRpcClient
-  if (I().mInfoThread.joinable()) {
-    I().mInfoThread.join();
-  }
+  // stop the receiver thread
   if (I().mReceiverThread.joinable()) {
     I().mReceiverThread.join();
   }
@@ -222,11 +219,20 @@ void StfSenderDevice::InitTask()
     if (!standalone()) {
       // contact the scheduler on gRPC
       while (I().mTfSchedulerRpcClient.should_retry_start() && !I().mTfSchedulerRpcClient.start(I().mDiscoveryConfig)) {
-        std::this_thread::sleep_for(150ms);
+
+        // try to reach the scheduler unless we should exit
+        if (NewStatePending()) {
+          IDDLOG("InitTask: The control system requested abort.");
+          AbortInitTask();
+          ChangeState(fair::mq::Transition::ErrorFound);
+          return;
+        }
+
+        std::this_thread::sleep_for(250ms);
       }
 
       // Did we fail to connect to the TfScheduler?
-      if (!I().mTfSchedulerRpcClient.should_retry_start()) {
+      if (!I().mTfSchedulerRpcClient.started()) {
         EDDLOG("InitTask: Failed to connect to TfScheduler. Exiting.");
         ChangeState(fair::mq::Transition::ErrorFound);
         return;
@@ -298,7 +304,7 @@ void StfSenderDevice::ResetTask()
 {
   I().mRunning = false;
 
-    // Stop the pipeline
+  // Stop the pipeline
   I().stopPipeline();
 
   // stop the receiver thread

src/StfSender/StfSenderDevice.h

@@ -76,6 +76,16 @@ class StfSenderDevice : public DataDistDevice
   virtual void InitTask() override final;
   virtual void ResetTask() override final;
 
+  void AbortInitTask() {
+    DDDLOG("Aborting InitTask...");
+    if (mI && I().mDiscoveryConfig) {
+      auto& lStatus = I().mDiscoveryConfig->status();
+      lStatus.mutable_info()->set_process_state(BasicInfo::ABORTED);
+      I().mDiscoveryConfig->write();
+    }
+    ResetTask();
+  }
+
   virtual void PreRun() final;
   virtual void PostRun() final;
   virtual bool ConditionalRun() final;
@@ -88,6 +98,21 @@ class StfSenderDevice : public DataDistDevice
     StfSenderInstance()
     : IFifoPipeline(ePipelineSize) {}
 
+    ~StfSenderInstance() {
+      mRunning = false;
+      mDeviceRunning = false;
+
+      // stop the info thread
+      if (mInfoThread.joinable()) {
+        mInfoThread.join();
+      }
+
+      // stop the receiver thread
+      if (mReceiverThread.joinable()) {
+        mReceiverThread.join();
+      }
+    }
+
     /// Configuration
     std::string mInputChannelName;
     bool mStandalone = false;

src/StfSender/StfSenderOutputUCX.cxx

@@ -148,7 +148,6 @@ void StfSenderOutputUCX::stop()
     for (auto &lWorker : mDataWorkers) {
       ucp_worker_destroy(lWorker.ucp_worker);
     }
-    ucp_cleanup(ucp_context);
   }
   DDDLOG("StfSenderOutputUCX::stop: closed all connections.");
 
@@ -159,6 +158,8 @@ void StfSenderOutputUCX::stop()
       ucx::util::destroy_rkey_for_region(ucp_context, lMapping.ucp_mem, lMapping.ucp_rkey_buf);
     }
   }
+
+  ucp_cleanup(ucp_context);
   DDDLOG("StfSenderOutputUCX::stop: revoked all rkeys.");
 }
 

src/common/rpc/TfSchedulerRpcClient.h

@@ -59,7 +59,7 @@ class TfSchedulerRpcClient {
         return false;
       }
 
-      IDDLOG_RL(1000, "TfScheduler instance configuration not found. Retrying.");
+      IDDLOG_RL(5000, "TfScheduler instance configuration not found. Retrying.");
       return false;
     }
 
@@ -79,15 +79,20 @@ class TfSchedulerRpcClient {
 
     IDDLOG("Connected to TfScheduler RPC endpoint={}", lEndpoint);
 
+    mStarted = true;
+
     return true;
   }
 
   void stop() {
     mTfSchedulerConf.Clear();
     mStub.reset(nullptr);
     mChannel.reset();
+    mStarted = false;
   }
 
+  bool started() const { return mStarted; }
+
   void updateTimeInformation(BasicInfo &pInfo);
 
   // rpc HeartBeat(BasicInfo) returns (google.protobuf.Empty) { }
@@ -137,6 +142,8 @@ class TfSchedulerRpcClient {
 
   // keep looking for the TfScheduler instance
   bool mShouldRetryStart = true;
+  // mark successful start
+  bool mStarted = false;
 };
 
 } /* namespace o2::DataDistribution */