Merge branch 'AliceO2Group:dev' into fd3_digits

Author: pkurash

Common/SimConfig/src/SimConfig.cxx

@@ -200,7 +200,11 @@ bool SimConfig::determineActiveModulesList(const std::string& version, std::vect
       return false;
     }
     modules = map[version];
-    LOGP(info, "Running with official detector version '{}'", version);
+    static std::string last_version{}; // prevent multiple printouts of same message
+    if (last_version != version) {
+      LOGP(info, "Running with official detector version '{}'", version);
+      last_version = version;
+    }
   }
   // check if specified modules are in list
   if (inputargs.size() != 1 || inputargs[0] != "all") {

Detectors/TPC/base/include/TPCBase/CalArray.h

@@ -26,6 +26,11 @@
 #include <boost/format.hpp>
 #endif
 
+#ifdef NDEBUG
+#undef NDEBUG
+#include <cassert>
+#endif
+
 namespace o2
 {
 namespace tpc
@@ -93,7 +98,11 @@ class CalArray
   int getPadSubsetNumber() const { return mPadSubsetNumber; }
 
   void setValue(const size_t channel, const T& value) { mData[channel] = value; }
-  const T getValue(const size_t channel) const { return mData[channel]; }
+  const T getValue(const size_t channel) const
+  {
+    assert(channel < mData.size());
+    return mData[channel];
+  }
 
   void setValue(const size_t row, const size_t pad, const T& value);
   const T getValue(const size_t row, const size_t pad) const;

Detectors/TPC/base/include/TPCBase/CalDet.h

@@ -30,6 +30,12 @@
 #include "Rtypes.h"
 #endif
 
+#ifndef NDEBUG
+#undef NDEBUG
+// always enable assert
+#include <cassert>
+#endif
+
 namespace o2
 {
 namespace tpc
@@ -211,7 +217,26 @@ inline const T CalDet<T>::getValue(const ROC roc, const size_t row, const size_t
     }
     case PadSubset::Region: {
       const auto globalRow = roc.isOROC() ? mappedRow + mapper.getNumberOfRowsROC(ROC(0)) : mappedRow;
-      return mData[Mapper::REGION[globalRow] + roc.getSector() * Mapper::NREGIONS].getValue(Mapper::OFFSETCRUGLOBAL[globalRow] + mappedPad);
+      const auto dataRow = Mapper::REGION[globalRow] + roc.getSector() * Mapper::NREGIONS;
+      const auto index = Mapper::OFFSETCRUGLOBAL[globalRow] + mappedPad;
+      assert(dataRow < mData.size());
+      if (index >= mData[dataRow].getData().size()) {
+        // S. Wenzel: We shouldn't come here but we do. For instance for CalDet calibrations loaded from
+        // creator.loadIDCPadFlags(1731274461770);
+
+        // In this case there is an index overflow, leading to invalid reads and potentially a segfault.
+        // To increase stability, for now returning a trivial answer. This can be removed once either the algorithm
+        // or the calibration data has been fixed.
+#ifndef GPUCA_ALIGPUCODE // hide from GPU standalone compilation
+        static bool printMsg = true;
+        if (printMsg) {
+          LOG(error) << "Out of bound access in TPC CalDet ROC " << roc << " row " << row << " pad " << pad << " (no more messages printed)";
+        }
+        printMsg = false;
+#endif
+        return T{};
+      }
+      return mData[dataRow].getValue(index);
       break;
     }
   }

Detectors/TPC/base/src/TPCFlagsMemberCustomStreamer.cxx

@@ -69,7 +69,9 @@ static __attribute__((used)) int _R__dummyStreamer_3 =
   ([]() {
     auto cl = TClass::GetClass<o2::tpc::CalArray<o2::tpc::PadFlags>>();
     if (cl) {
-      cl->AdoptMemberStreamer("mData", new TMemberStreamer(MemberVectorPadFlagsStreamer));
+      if (!getenv("TPC_PADFLAGS_STREAMER_OFF")) {
+        cl->AdoptMemberStreamer("mData", new TMemberStreamer(MemberVectorPadFlagsStreamer));
+      }
     } else {
       // we should never come here ... and if we do we should assert/fail
       assert(false);

Detectors/TPC/base/test/testTPCCalDet.cxx

@@ -24,6 +24,7 @@
 #include "TPCBase/CalDet.h"
 #include "TFile.h"
 #include "Framework/TypeTraits.h"
+#include "TPCBase/DeadChannelMapCreator.h"
 
 namespace o2::tpc
 {
@@ -344,4 +345,12 @@ BOOST_AUTO_TEST_CASE(CalDetTypeTest)
   BOOST_CHECK(testDict == true);
 }
 
+BOOST_AUTO_TEST_CASE(CalDetStreamerTest)
+{
+  // simple code executing the TPC IDCPadFlags loading in a standalone env --> easy to valgrind
+  o2::tpc::DeadChannelMapCreator creator{};
+  creator.init("https://alice-ccdb.cern.ch");
+  creator.loadIDCPadFlags(1731274461770);
+}
+
 } // namespace o2::tpc

Framework/Core/test/test_MessageSet.cxx

@@ -93,3 +93,43 @@ TEST_CASE("MessageSetAddPartRef") {
   REQUIRE(msgSet.pairMap[0].partIndex == 0);
   REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
 }
+
+TEST_CASE("MessageSetAddMultiple")
+{
+  std::vector<fair::mq::MessagePtr> ptrs;
+  std::unique_ptr<fair::mq::Message> msg(nullptr);
+  std::unique_ptr<fair::mq::Message> msg2(nullptr);
+  ptrs.emplace_back(std::move(msg));
+  ptrs.emplace_back(std::move(msg2));
+  PartRef ref{std::move(msg), std::move(msg2)};
+  o2::framework::MessageSet msgSet;
+  msgSet.add(std::move(ref));
+  PartRef ref2{std::move(msg), std::move(msg2)};
+  msgSet.add(std::move(ref2));
+  std::vector<fair::mq::MessagePtr> msgs;
+  msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
+  msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
+  msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
+  msgSet.add([&msgs](size_t i) {
+    return std::move(msgs[i]);
+  }, 3);
+
+  REQUIRE(msgSet.messages.size() == 7);
+  REQUIRE(msgSet.messageMap.size() == 3);
+  REQUIRE(msgSet.pairMap.size() == 4);
+  REQUIRE(msgSet.messageMap[0].position == 0);
+  REQUIRE(msgSet.messageMap[0].size == 1);
+  REQUIRE(msgSet.messageMap[1].position == 2);
+  REQUIRE(msgSet.messageMap[1].size == 1);
+  REQUIRE(msgSet.messageMap[2].position == 4);
+  REQUIRE(msgSet.messageMap[2].size == 2);
+
+  REQUIRE(msgSet.pairMap[0].partIndex == 0);
+  REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
+  REQUIRE(msgSet.pairMap[1].partIndex == 1);
+  REQUIRE(msgSet.pairMap[1].payloadIndex == 0);
+  REQUIRE(msgSet.pairMap[2].partIndex == 2);
+  REQUIRE(msgSet.pairMap[2].payloadIndex == 0);
+  REQUIRE(msgSet.pairMap[3].partIndex == 2);
+  REQUIRE(msgSet.pairMap[3].payloadIndex == 1);
+}

run/O2HitMerger.h

@@ -88,6 +88,29 @@ namespace o2
 namespace devices
 {
 
+// Function communicating to primary particle server that it is now safe to shutdown.
+// From the perspective of o2-sim, this is the case when all configs have been propagated and the system
+// is running ok: For instance after the HitMerger is initialized and got it's first data from Geant workers.
+bool primaryServer_sendShutdownPermission(fair::mq::Channel& channel)
+{
+  std::unique_ptr<fair::mq::Message> request(channel.NewSimpleMessage((int)o2::O2PrimaryServerInfoRequest::AllowShutdown));
+  std::unique_ptr<fair::mq::Message> reply(channel.NewMessage());
+
+  int timeoutinMS = 100;
+  if (channel.Send(request, timeoutinMS) > 0) {
+    LOG(info) << "Sending Shutdown permission to particle server";
+    if (channel.Receive(reply, timeoutinMS) > 0) {
+      // the answer is a simple ack with a status code
+      LOG(info) << "Shutdown permission was acknowledged";
+    } else {
+      LOG(error) << "No answer received within " << timeoutinMS << "ms\n";
+      return false;
+    }
+    return true;
+  }
+  return false;
+}
+
 class O2HitMerger : public fair::mq::Device
 {
 
@@ -129,6 +152,9 @@ class O2HitMerger : public fair::mq::Device
     if (o2::devices::O2SimDevice::querySimConfig(GetChannels().at("o2sim-primserv-info").at(0))) {
       outfilename = o2::base::NameConf::getMCKinematicsFileName(o2::conf::SimConfig::Instance().getOutPrefix().c_str());
       mNExpectedEvents = o2::conf::SimConfig::Instance().getNEvents();
+    } else {
+      // we didn't manage to get a configuration --> better to fail
+      LOG(fatal) << "No configuration received. Aborting";
     }
     mAsService = o2::conf::SimConfig::Instance().asService();
     mForwardKine = o2::conf::SimConfig::Instance().forwardKine();
@@ -354,6 +380,13 @@ class O2HitMerger : public fair::mq::Device
       // for the next batch
       return waitForControlInput();
     }
+
+    static bool initAcknowledged = false;
+    if (!initAcknowledged) {
+      primaryServer_sendShutdownPermission(GetChannels().at("o2sim-primserv-info").at(0));
+      initAcknowledged = true;
+    }
+
     return more;
   }
 
@@ -413,10 +446,6 @@ class O2HitMerger : public fair::mq::Device
         };
       }
     }
-    if (!expectmore) {
-      // somehow FairMQ has difficulties shutting down; helping manually
-      // raise(SIGINT);
-    }
     return expectmore;
   }
 

run/O2PrimaryServerDevice.h

@@ -247,11 +247,10 @@ class O2PrimaryServerDevice final : public fair::mq::Device
     }
   }
 
-  // launches a thread that listens for status requests from outside asynchronously
+  // launches a thread that listens for status/config/shutdown requests from outside asynchronously
   void launchInfoThread()
   {
     static std::vector<std::thread> threads;
-
     auto sendErrorReply = [](fair::mq::Channel& channel) {
       LOG(error) << "UNKNOWN REQUEST";
       std::unique_ptr<fair::mq::Message> reply(channel.NewSimpleMessage((int)(404)));
@@ -260,7 +259,9 @@ class O2PrimaryServerDevice final : public fair::mq::Device
 
     LOG(info) << "LAUNCHING STATUS THREAD";
     auto lambda = [this, sendErrorReply]() {
-      while (mState != O2PrimaryServerState::Stopped) {
+      bool canShutdown{false};
+      // Exit only when both: serving stopped and allowed from outside.
+      while (!(mState == O2PrimaryServerState::Stopped && canShutdown)) {
         auto& channel = GetChannels().at("o2sim-primserv-info").at(0);
         if (!channel.IsValid()) {
           LOG(error) << "channel primserv-info not valid";
@@ -285,6 +286,11 @@ class O2PrimaryServerDevice final : public fair::mq::Device
             }
           } else if (request_payload == (int)O2PrimaryServerInfoRequest::Config) {
             HandleConfigRequest(channel);
+          } else if (request_payload == (int)O2PrimaryServerInfoRequest::AllowShutdown) {
+            LOG(info) << "Got info that we may shutdown";
+            std::unique_ptr<fair::mq::Message> ack(channel.NewSimpleMessage(200));
+            channel.Send(ack);
+            canShutdown = true;
           } else {
             sendErrorReply(channel);
           }
@@ -518,10 +524,13 @@ class O2PrimaryServerDevice final : public fair::mq::Device
 
   void PostRun() override
   {
+    // We shouldn't shut down immediately when all events have been served
+    // Instead we also need to wait until the info thread running some communication server
+    // with other processes is finished.
     while (!mInfoThreadStopped) {
       LOG(info) << "Waiting info thread";
       using namespace std::chrono_literals;
-      std::this_thread::sleep_for(100ms);
+      std::this_thread::sleep_for(1000ms);
     }
   }
 
@@ -534,7 +543,7 @@ class O2PrimaryServerDevice final : public fair::mq::Device
     if (mEventCounter >= mMaxEvents && mNeedNewEvent) {
       workavailable = false;
     }
-    if (!(mState == O2PrimaryServerState::ReadyToServe || mState == O2PrimaryServerState::WaitingEvent)) {
+    if (!(mState.load() == O2PrimaryServerState::ReadyToServe || mState.load() == O2PrimaryServerState::WaitingEvent)) {
       // send a zero answer
       workavailable = false;
     }

run/O2SimDevice.h

@@ -95,9 +95,6 @@ class O2SimDevice final : public fair::mq::Device
   // returns true if successful / false if not
   static bool querySimConfig(fair::mq::Channel& channel)
   {
-    // auto text = new std::string("configrequest");
-    // std::unique_ptr<fair::mq::Message> request(channel.NewMessage(const_cast<char*>(text->c_str()),
-    //                                                           text->length(), CustomCleanup, text));
     std::unique_ptr<fair::mq::Message> request(channel.NewSimpleMessage((int)O2PrimaryServerInfoRequest::Config));
     std::unique_ptr<fair::mq::Message> reply(channel.NewMessage());
 

run/PrimaryServerState.h

@@ -25,10 +25,11 @@ enum class O2PrimaryServerState {
 };
 static const char* PrimStateToString[5] = {"INIT", "SERVING", "WAITEVENT", "IDLE", "STOPPED"};
 
-/// enum class for type of info request
+/// enum class for request to o2sim-primserv-info channel of the O2PrimaryServerDevice
 enum class O2PrimaryServerInfoRequest {
-  Status = 1,
-  Config = 2
+  Status = 1,       // asks to retrieve current status of O2PrimaryServerDevice --> will send O2PrimaryServerState
+  Config = 2,       // asks for o2-sim config reply
+  AllowShutdown = 3 // can be used to let particle server know that shutdown is now safe (once all components initialized)
 };
 
 /// Struct to be used as payload when making a request