Merge pull request #215 from sy-c/master

v2.9.2

Author: sy-c

doc/releaseNotes.md

@@ -436,3 +436,6 @@ This file describes the main feature changes for each readout.exe released versi
 - DB improvements: occ.getRole() workaround for source id, logging.
 - Added logging of some EOR stats (TF number, etc) as for bookkeeping.
 - o2-readout-monitor: indexing by source for correct rates computation.
+
+## v2.9.2 - 31/03/2022
+- Cosmetics improvements related to threads cleanup order on exit.

src/ReadoutEquipment.cxx

@@ -826,3 +826,7 @@ int ReadoutEquipment::processRdh(DataBlockContainerReference& block)
   return 0;
 }
 
+void ReadoutEquipment::abortThread() {
+  // ensure thread is stopped
+  readoutThread = nullptr;
+}

src/ReadoutEquipment.h

@@ -58,6 +58,10 @@ class ReadoutEquipment
   // get current memory pool usage (available and total)
   int getMemoryUsage(size_t& numberOfPagesAvailable, size_t& numberOfPagesInPool);
 
+  // this should be called first in derived classes destructor
+  // to ensure the associated thread is stopped before releasing resources
+  void abortThread();
+
  private:
   std::unique_ptr<Thread> readoutThread;
   static Thread::CallbackResult threadCallback(void* arg);

src/ReadoutStats.cxx

@@ -38,13 +38,12 @@ ReadoutStats::ReadoutStats() {
 }
 
 ReadoutStats::~ReadoutStats() {
-  stopPublish();
-
   if (publishThread != nullptr) {
     shutdownThread = 1;
     publishThread->join();
     publishThread = nullptr;
   }
+  zmqCleanup();
 }
 
 void ReadoutStats::reset()
@@ -220,17 +219,18 @@ void ReadoutStats::threadLoop() {
 void ReadoutStats::publishNow() {
   #ifdef WITH_ZMQ
   if (zmqEnabled) {
-    fflush(stdout);
     uint64_t newUpdate = counters.notify.load();
     ReadoutStatsCounters snapshot;
     memcpy((void *)&snapshot, (void *)&gReadoutStats.counters, sizeof(snapshot));
     snapshot.timestamp = ((std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch())).count())/1000000.0;
     publishMutex.lock();
-    if ((newUpdate != lastUpdate) || (snapshot.timestamp.load()-lastPublishTimestamp > publishInterval - 0.1)) {
-      zmq_send(zmqHandle, &snapshot, sizeof(snapshot), ZMQ_DONTWAIT);
-      lastUpdate = newUpdate;
-      lastPublishTimestamp = snapshot.timestamp;
-    }   
+    if (zmqHandle != nullptr) {
+      if ((newUpdate != lastUpdate) || (snapshot.timestamp.load()-lastPublishTimestamp > publishInterval - 0.1)) {
+	zmq_send(zmqHandle, &snapshot, sizeof(snapshot), ZMQ_DONTWAIT);
+	lastUpdate = newUpdate;
+	lastPublishTimestamp = snapshot.timestamp;
+      }
+    }
     publishMutex.unlock();
   }
  #endif

src/ReadoutVersion.h

@@ -9,5 +9,5 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-#define READOUT_VERSION "2.9.1"
+#define READOUT_VERSION "2.9.2"
 

src/mainReadout.cxx

@@ -1326,7 +1326,11 @@ Readout::~Readout() {
   dataConsumers.clear();
   agg = nullptr;
   agg_output = nullptr;
-  readoutDevices.clear();
+  // ensure readout equipment threads stopped before releasing resources
+  for (const auto &d : readoutDevices) {
+    d->abortThread();
+  }
+  readoutDevices.clear(); // after aggregator, because they own the data blocks
 }
 
 #ifdef WITH_OCC
@@ -1764,6 +1768,7 @@ int main(int argc, char* argv[])
 
   gReadoutStats.counters.state = stringToUint64("> exit");
   gReadoutStats.counters.notify++;
+  gReadoutStats.stopPublish();
 
   theReadout = nullptr;