Proper termination of detached thread

sawenzel · sawenzel · commit 94a8dafa67c1 · 2021-06-01T08:29:59.000+02:00
Wait for proper termination of detached thread,
and thereby prevent ZMQ socket problem.
diff --git a/run/O2PrimaryServerDevice.h b/run/O2PrimaryServerDevice.h
@@ -153,8 +153,8 @@ class O2PrimaryServerDevice final : public FairMQDevice
       }
       std::unique_ptr<FairMQMessage> request(channel.NewSimpleMessage(-1));
       while (mState != O2PrimaryServerState::Stopped) {
-        LOG(INFO) << "READY TO RECEIVE INFO REQUEST";
-        if (channel.Receive(request) > 0) {
+        int timeout = 100; // 100ms --> so as not to block and allow for proper termination of this thread
+        if (channel.Receive(request, timeout) > 0) {
           LOG(INFO) << "INFO REQUEST RECEIVED";
           if (*(int*)(request->GetData()) == (int)O2PrimaryServerInfoRequest::Status) {
             LOG(INFO) << "Received status request";
@@ -172,6 +172,7 @@ class O2PrimaryServerDevice final : public FairMQDevice
           }
         }
       }
+      mInfoThreadStopped = true;
     };
     threads.push_back(std::thread(lambda));
     threads.back().detach();
@@ -360,7 +361,11 @@ class O2PrimaryServerDevice final : public FairMQDevice
       timer.Stop();
       auto time = timer.CpuTime();
       LOG(INFO) << "COND-RUN TOOK " << time << " s";
-      // return mState.load() != O2PrimaryServerState::Stopped; // will be taken down by external driver
+    }
+    // wait for info thread
+    while (!mInfoThreadStopped) {
+      LOG(INFO) << "Waiting info thread";
+      sleep(1);
     }
   }
 
@@ -553,6 +558,7 @@ class O2PrimaryServerDevice final : public FairMQDevice
 
   std::atomic<O2PrimaryServerState> mState{O2PrimaryServerState::Initializing};
   std::atomic<int> mWaitingControlInput{0};
+  std::atomic<bool> mInfoThreadStopped{false};
 
   bool mAsService = false;
 };

Original file line number	Diff line number	Diff line change
`@@ -153,8 +153,8 @@ class O2PrimaryServerDevice final : public FairMQDevice`
`153`	`153`	`}`
`154`	`154`	`std::unique_ptr<FairMQMessage> request(channel.NewSimpleMessage(-1));`
`155`	`155`	`while (mState != O2PrimaryServerState::Stopped) {`
`156`		`- LOG(INFO) << "READY TO RECEIVE INFO REQUEST";`
`157`		`- if (channel.Receive(request) > 0) {`
	`156`	`+ int timeout = 100; // 100ms --> so as not to block and allow for proper termination of this thread`
	`157`	`+ if (channel.Receive(request, timeout) > 0) {`
`158`	`158`	`LOG(INFO) << "INFO REQUEST RECEIVED";`
`159`	`159`	`if ((int)(request->GetData()) == (int)O2PrimaryServerInfoRequest::Status) {`
`160`	`160`	`LOG(INFO) << "Received status request";`
`@@ -172,6 +172,7 @@ class O2PrimaryServerDevice final : public FairMQDevice`
`172`	`172`	`}`
`173`	`173`	`}`
`174`	`174`	`}`
	`175`	`+ mInfoThreadStopped = true;`
`175`	`176`	`};`
`176`	`177`	`threads.push_back(std::thread(lambda));`
`177`	`178`	`threads.back().detach();`
`@@ -360,7 +361,11 @@ class O2PrimaryServerDevice final : public FairMQDevice`
`360`	`361`	`timer.Stop();`
`361`	`362`	`auto time = timer.CpuTime();`
`362`	`363`	`LOG(INFO) << "COND-RUN TOOK " << time << " s";`
`363`		`- // return mState.load() != O2PrimaryServerState::Stopped; // will be taken down by external driver`
	`364`	`+ }`
	`365`	`+ // wait for info thread`
	`366`	`+ while (!mInfoThreadStopped) {`
	`367`	`+ LOG(INFO) << "Waiting info thread";`
	`368`	`+ sleep(1);`
`364`	`369`	`}`
`365`	`370`	`}`
`366`	`371`
`@@ -553,6 +558,7 @@ class O2PrimaryServerDevice final : public FairMQDevice`
`553`	`558`
`554`	`559`	`std::atomic<O2PrimaryServerState> mState{O2PrimaryServerState::Initializing};`
`555`	`560`	`std::atomic<int> mWaitingControlInput{0};`
	`561`	`+ std::atomic<bool> mInfoThreadStopped{false};`
`556`	`562`
`557`	`563`	`bool mAsService = false;`
`558`	`564`	`};`