Run -> CondRun; Seems to be better

sawenzel · sawenzel · commit 8d2118e549bf · 2021-06-15T20:45:15.000+02:00
CondRun seems to cope better with signal
handling. At least it seems to be stable now
on both Linux and Mac (the Run()-based implementation
still had problems on Mac).
diff --git a/run/O2PrimaryServerDevice.h b/run/O2PrimaryServerDevice.h
@@ -39,6 +39,7 @@
 #include <atomic>
 #include "PrimaryServerState.h"
 #include "SimPublishChannelHelper.h"
+#include <chrono>
 
 namespace o2
 {
@@ -319,60 +320,60 @@ class O2PrimaryServerDevice final : public FairMQDevice
     return true;
   }
 
-  void Run() override
+  bool ConditionalRun() override
   {
-    while (mState != O2PrimaryServerState::Stopped) {
-      // we might come here in IDLE mode
-      if (mState == O2PrimaryServerState::Idle) {
-        if (mWaitingControlInput.load() == 0) {
-          if (mControlThread.joinable()) {
-            mControlThread.join();
-          }
-          mControlThread = std::thread(&O2PrimaryServerDevice::waitForControlInput, this);
+    // we might come here in IDLE mode
+    if (mState == O2PrimaryServerState::Idle) {
+      if (mWaitingControlInput.load() == 0) {
+        if (mControlThread.joinable()) {
+          mControlThread.join();
         }
+        mControlThread = std::thread(&O2PrimaryServerDevice::waitForControlInput, this);
       }
+    }
 
-      auto& channel = fChannels.at("primary-get").at(0);
-      PrimaryChunkRequest requestpayload;
-      std::unique_ptr<FairMQMessage> request(channel.NewSimpleMessage(requestpayload));
-      auto bytes = channel.Receive(request);
-      if (bytes < 0) {
-        LOG(ERROR) << "Some error/interrupt occurred on socket during receive";
-        // This may likely be due to an external signal which disrupts the polling
-        LOG(INFO) << "Get current state : " << GetCurrentStateName();
-        LOG(INFO) << "New state pending : " << NewStatePending();
-        if (NewStatePending()) { // new state is typically pending if (term) signal was received
-          WaitForNextState();
-          // ask ourselves for termination of this loop
-          stateTransition(O2PrimaryServerState::Stopped, "CONDRUN");
-        }
-        continue;
+    auto& channel = fChannels.at("primary-get").at(0);
+    PrimaryChunkRequest requestpayload;
+    std::unique_ptr<FairMQMessage> request(channel.NewSimpleMessage(requestpayload));
+    auto bytes = channel.Receive(request);
+    if (bytes < 0) {
+      LOG(ERROR) << "Some error/interrupt occurred on socket during receive";
+      if (NewStatePending()) { // new state is typically pending if (term) signal was received
+        WaitForNextState();
+        // ask ourselves for termination of this loop
+        stateTransition(O2PrimaryServerState::Stopped, "CONDRUN");
       }
+      return false;
+    }
 
-      TStopwatch timer;
-      timer.Start();
-      auto& r = *((PrimaryChunkRequest*)(request->GetData()));
-      LOG(INFO) << "PARTICLE REQUEST IN STATE " << PrimStateToString[(int)mState.load()] << " from " << r.workerid << ":" << r.requestid;
-
-      auto prestate = mState.load();
-      auto more = HandleRequest(request, 0, channel);
-      if (!more) {
-        if (mAsService) {
-          if (prestate == O2PrimaryServerState::ReadyToServe || prestate == O2PrimaryServerState::WaitingEvent) {
-            stateTransition(O2PrimaryServerState::Idle, "CONDRUN");
-          }
-        } else {
-          stateTransition(O2PrimaryServerState::Stopped, "CONDRUN");
+    TStopwatch timer;
+    timer.Start();
+    auto& r = *((PrimaryChunkRequest*)(request->GetData()));
+    LOG(INFO) << "PARTICLE REQUEST IN STATE " << PrimStateToString[(int)mState.load()] << " from " << r.workerid << ":" << r.requestid;
+
+    auto prestate = mState.load();
+    auto more = HandleRequest(request, 0, channel);
+    if (!more) {
+      if (mAsService) {
+        if (prestate == O2PrimaryServerState::ReadyToServe || prestate == O2PrimaryServerState::WaitingEvent) {
+          stateTransition(O2PrimaryServerState::Idle, "CONDRUN");
         }
+      } else {
+        stateTransition(O2PrimaryServerState::Stopped, "CONDRUN");
       }
-      timer.Stop();
-      auto time = timer.CpuTime();
-      LOG(INFO) << "COND-RUN TOOK " << time << " s";
     }
-    // wait for info thread
+    timer.Stop();
+    auto time = timer.CpuTime();
+    LOG(INFO) << "COND-RUN TOOK " << time << " s";
+    return mState != O2PrimaryServerState::Stopped;
+  }
+
+  void PostRun() override
+  {
     while (!mInfoThreadStopped) {
       LOG(INFO) << "Waiting info thread";
-      sleep(1);
+      using namespace std::chrono_literals;
+      std::this_thread::sleep_for(100ms);
     }
   }
 
