fixed timeout problem

jmyrcha · jmyrcha · commit 291f5deddbc0 · 2025-07-23T13:11:17.000+02:00
diff --git a/EventVisualisation/DataConverter/include/EventVisualisationDataConverter/Location.h b/EventVisualisation/DataConverter/include/EventVisualisationDataConverter/Location.h
@@ -26,6 +26,7 @@ namespace o2::event_visualisation
 struct LocationParams {
   std::string fileName;
   int port = -1;
+  int timeout = 100;
   std::string host = "localhost";
   bool toFile = true;
   bool toSocket = true;
@@ -38,6 +39,7 @@ class Location
   bool mToSocket;
   std::string mFileName;
   int mPort;
+  int mTimeout;
   std::string mHostName;
 
  public:
@@ -50,6 +52,7 @@ class Location
     this->mPort = params.port;
     this->mHostName = params.host;
     this->mClientSocket = -1;
+    this->mTimeout = params.timeout;
   }
   ~Location()
   {
diff --git a/EventVisualisation/DataConverter/src/Location.cxx b/EventVisualisation/DataConverter/src/Location.cxx
@@ -19,10 +19,78 @@
 #include <sys/socket.h>
 #include <unistd.h>
 #include <netdb.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <ctime>
 
 using namespace std;
+
 namespace o2::event_visualisation
 {
+
+int connect_with_timeout(const int socket, const struct sockaddr* addr, socklen_t addrlen, const unsigned int timeout_ms)
+{
+  int connection = 0;
+  // Setting O_NONBLOCK
+  int socket_flags_before;
+  if ((socket_flags_before = fcntl(socket, F_GETFL, 0) < 0)) {
+    return -1;
+  }
+  if (fcntl(socket, F_SETFL, socket_flags_before | O_NONBLOCK) < 0) {
+    return -1;
+  }
+  do {
+    if (connect(socket, addr, addrlen) < 0) {
+      if ((errno != EWOULDBLOCK) && (errno != EINPROGRESS)) {
+        connection = -1; // error
+      } else {           // wait for complete
+        // deadline 'timeout' ms from now
+        timespec now; // NOLINT(*-pro-type-member-init)
+        if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
+          connection = -1;
+          break;
+        }
+        const timespec deadline = {.tv_sec = now.tv_sec,
+                                   .tv_nsec = now.tv_nsec + timeout_ms * 1000000l};
+        do {
+          if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
+            connection = -1;
+            break;
+          }
+          // compute remaining deadline
+          const int ms_until_deadline = static_cast<int>((deadline.tv_sec - now.tv_sec) * 1000l + (deadline.tv_nsec - now.tv_nsec) / 1000000l);
+          if (ms_until_deadline < 0) {
+            connection = 0;
+            break;
+          }
+          pollfd connectionPool[] = {{.fd = socket, .events = POLLOUT}};
+          connection = poll(connectionPool, 1, ms_until_deadline);
+
+          if (connection > 0) { // confirm the success
+            int error = 0;
+            socklen_t len = sizeof(error);
+            if (getsockopt(socket, SOL_SOCKET, SO_ERROR, &error, &len) == 0) {
+              errno = error;
+            }
+            if (error != 0) {
+              connection = -1;
+            }
+          }
+        } while (connection == -1 && errno == EINTR); // If interrupted, try again.
+        if (connection == 0) {
+          errno = ETIMEDOUT;
+          connection = -1;
+        }
+      }
+    }
+  } while (false);
+  // Restore socket state
+  if (fcntl(socket, F_SETFL, socket_flags_before) < 0) {
+    return -1;
+  }
+  return connection;
+}
+
 void Location::open()
 {
   if (this->mToFile) {
@@ -37,7 +105,7 @@ void Location::open()
     // ask once
     static auto server = gethostbyname(this->mHostName.c_str());
     if (server == nullptr) {
-      fprintf(stderr, "ERROR, no such host\n");
+      LOGF(info, "Error no such host %s", this->mHostName.c_str());
       return;
     };
 
@@ -52,8 +120,8 @@ void Location::open()
       return;
     }
 
-    if (connect(this->mClientSocket, (struct sockaddr*)&serverAddress,
-                sizeof(serverAddress)) == -1) {
+    if (connect_with_timeout(this->mClientSocket, (sockaddr*)&serverAddress,
+                             sizeof(serverAddress), this->mTimeout) == -1) {
       LOGF(info, "Error connecting to %s:%d", this->mHostName.c_str(), this->mPort);
       ::close(this->mClientSocket);
       this->mClientSocket = -1;
@@ -97,6 +165,7 @@ void Location::write(char* buf, std::streamsize size)
     this->mOut->write(buf, size);
   }
   if (this->mToSocket && this->mClientSocket != -1) {
+    LOGF(info, "Location::write() socket %s ++++++++++++++++++++++", fileName());
     try {
       auto real = send(this->mClientSocket, buf, size, 0);
       if (real != size) {
@@ -110,4 +179,4 @@ void Location::write(char* buf, std::streamsize size)
   }
 }
 
-} // namespace o2::event_visualisation
+} // namespace o2::event_visualisation
diff --git a/EventVisualisation/Workflow/include/EveWorkflow/EveWorkflowHelper.h b/EventVisualisation/Workflow/include/EveWorkflow/EveWorkflowHelper.h
@@ -183,7 +183,7 @@ class EveWorkflowHelper
   bool isInsideITSROF(float t);
   bool isInsideTimeBracket(float t);
 
-  void save(const std::string& jsonPath, const std::string& ext, int numberOfFiles, const std::string& receiverHostname, int receiverPort, bool useOnlyFiles, bool useOnlySockets);
+  void save(const std::string& jsonPath, const std::string& ext, int numberOfFiles, const std::string& receiverHostname, int receiverPort, int receiverTimeout, bool useOnlyFiles, bool useOnlySockets);
 
   bool mUseTimeBracket = false;
   bool mUseEtaBracketTPC = false;
diff --git a/EventVisualisation/Workflow/include/EveWorkflow/O2DPLDisplay.h b/EventVisualisation/Workflow/include/EveWorkflow/O2DPLDisplay.h
@@ -50,7 +50,8 @@ class TPCFastTransform;
 class O2DPLDisplaySpec : public o2::framework::Task
 {
  public:
-  static constexpr auto allowedTracks = "ITS,TPC,MFT,MCH,MID,ITS-TPC,TPC-TRD,ITS-TPC-TOF,ITS-TPC-TRD,ITS-TPC-TRD-TOF,MCH-MID,MFT-MCH,MFT-MCH-MID,PHS,EMC,HMP";
+  static constexpr auto allowedTracks =
+    "ITS,TPC,MFT,MCH,MID,ITS-TPC,TPC-TRD,ITS-TPC-TOF,ITS-TPC-TRD,ITS-TPC-TRD-TOF,MCH-MID,MFT-MCH,MFT-MCH-MID,PHS,EMC,HMP";
   static constexpr auto allowedClusters = "ITS,TPC,TRD,TOF,MFT,MCH,MID,PHS,EMC,HMP";
 
   O2DPLDisplaySpec(bool disableWrite, bool useMC, o2::dataformats::GlobalTrackID::mask_t trkMask,
@@ -63,9 +64,26 @@ class O2DPLDisplaySpec : public o2::framework::Task
                    bool eveHostNameMatch,
                    const std::string& receiverHostname,
                    int receiverPort,
+                   int receiverTimeout,
                    bool useOnlyFiles,
                    bool useOnlySockets)
-    : mDisableWrite(disableWrite), mUseMC(useMC), mTrkMask(trkMask), mClMask(clMask), mDataRequest(dataRequest), mGGCCDBRequest(gr), mEMCALCalibLoader(emcCalibLoader), mJsonPath(jsonPath), mExt(ext), mTimeInterval(timeInterval), mEveHostNameMatch(eveHostNameMatch), mRunType(o2::parameters::GRPECS::NONE), mReceiverHostname(receiverHostname), mReceiverPort(receiverPort), mUseOnlyFiles(useOnlyFiles), mUseOnlySockets(useOnlySockets)
+    : mDisableWrite(disableWrite),
+      mUseMC(useMC),
+      mTrkMask(trkMask),
+      mClMask(clMask),
+      mDataRequest(dataRequest),
+      mGGCCDBRequest(gr),
+      mEMCALCalibLoader(emcCalibLoader),
+      mJsonPath(jsonPath),
+      mExt(ext),
+      mTimeInterval(timeInterval),
+      mEveHostNameMatch(eveHostNameMatch),
+      mRunType(o2::parameters::GRPECS::NONE),
+      mReceiverHostname(receiverHostname),
+      mReceiverPort(receiverPort),
+      mReceiverTimeout(receiverTimeout),
+      mUseOnlyFiles(useOnlyFiles),
+      mUseOnlySockets(useOnlySockets)
   {
     this->mTimeStamp = std::chrono::high_resolution_clock::now() - timeInterval; // first run meets condition
   }
@@ -85,7 +103,8 @@ class O2DPLDisplaySpec : public o2::framework::Task
   std::string mJsonPath;                   // folder where files are stored
   std::string mExt;                        // extension of created files (".json" or ".root")
   std::chrono::milliseconds mTimeInterval; // minimal interval between files in milliseconds
-  bool mPrimaryVertexTriggers;             // instead of drawing vertices with tracks (and maybe calorimeter triggers), draw vertices with calorimeter triggers (and maybe tracks)
+  bool mPrimaryVertexTriggers;
+  // instead of drawing vertices with tracks (and maybe calorimeter triggers), draw vertices with calorimeter triggers (and maybe tracks)
   int mEventCounter = 0;
   std::chrono::time_point<std::chrono::high_resolution_clock> mTimeStamp;
 
@@ -101,10 +120,10 @@ class O2DPLDisplaySpec : public o2::framework::Task
 
   std::string mReceiverHostname;
   int mReceiverPort;
+  int mReceiverTimeout;
   bool mUseOnlyFiles;
   bool mUseOnlySockets;
 };
-
 } // namespace o2::event_visualisation
 
-#endif
+#endif
diff --git a/EventVisualisation/Workflow/src/AO2DConverter.cxx b/EventVisualisation/Workflow/src/AO2DConverter.cxx
@@ -75,7 +75,7 @@ struct AO2DConverter {
     mHelper->mEvent.setFirstTForbit(mTfOrbit);
     mHelper->mEvent.setCreationTime(collision.collisionTime());
     const std::string hostname("localhost");
-    mHelper->save(jsonPath, ".root", -1, hostname, -1, true, true);
+    mHelper->save(jsonPath, ".root", -1, hostname, -1, 100, true, true);
     mHelper->clear();
   }
 };
diff --git a/EventVisualisation/Workflow/src/EveWorkflowHelper.cxx b/EventVisualisation/Workflow/src/EveWorkflowHelper.cxx
@@ -164,9 +164,10 @@ void EveWorkflowHelper::selectTracks(const CalibObjectsConst* calib,
       t0 *= this->mTPCBin2MUS;
       terr *= this->mTPCBin2MUS;
     } else if constexpr (isITSTrack<decltype(_tr)>()) {
-      t0 += 0.5f * this->mITSROFrameLengthMUS;          // ITS time is supplied in \mus as beginning of ROF
-      terr *= this->mITSROFrameLengthMUS;               // error is supplied as a half-ROF duration, convert to \mus
-    } else if constexpr (isMFTTrack<decltype(_tr)>()) { // Same for MFT
+      t0 += 0.5f * this->mITSROFrameLengthMUS; // ITS time is supplied in \mus as beginning of ROF
+      terr *= this->mITSROFrameLengthMUS;      // error is supplied as a half-ROF duration, convert to \mus
+    } else if constexpr (isMFTTrack<decltype(_tr)>()) {
+      // Same for MFT
       t0 += 0.5f * this->mMFTROFrameLengthMUS;
       terr *= this->mMFTROFrameLengthMUS;
     } else if constexpr (!(isMCHTrack<decltype(_tr)>() || isMIDTrack<decltype(_tr)>() ||
@@ -264,7 +265,8 @@ void EveWorkflowHelper::selectTracks(const CalibObjectsConst* calib,
           if (gid.getSource() == o2::dataformats::GlobalTrackID::TPC && checkTPCDCA) {
             const auto& tpcTr = mRecoCont->getTPCTrack(gid);
             o2::track::TrackPar trc{tpcTr};
-            if (!tpcTr.hasBothSidesClusters()) { // need to correct track Z with this vertex time
+            if (!tpcTr.hasBothSidesClusters()) {
+              // need to correct track Z with this vertex time
               float dz = (tpcTr.getTime0() * mTPCTimeBins2MUS -
                           (pv.getTimeStamp().getTimeStamp() + mTPCVDrift->getTimeOffset())) *
                          mTPCVDrift->getVDrift();
@@ -427,10 +429,12 @@ void EveWorkflowHelper::draw(std::size_t primaryVertexIdx, bool sortTracks)
           break;
         case GID::TPC: {
           float dz = 0.f;
-          if (conf.PVMode) { // for TPC the nominal time (center of the bracket) is stored but in the PVMode we correct it by the PV time
+          if (conf.PVMode) {
+            // for TPC the nominal time (center of the bracket) is stored but in the PVMode we correct it by the PV time
             tim = pvTime;
             const auto& tpcTr = mRecoCont->getTPCTrack(gid);
-            if (!tpcTr.hasBothSidesClusters()) { // need to correct track Z with this vertex time
+            if (!tpcTr.hasBothSidesClusters()) {
+              // need to correct track Z with this vertex time
               float dz = (tpcTr.getTime0() * mTPCTimeBins2MUS - (pvTime + mTPCVDrift->getTimeOffset())) *
                          mTPCVDrift->getVDrift();
               if (tpcTr.hasCSideClustersOnly()) {
@@ -513,7 +517,7 @@ void EveWorkflowHelper::draw(std::size_t primaryVertexIdx, bool sortTracks)
 }
 
 void EveWorkflowHelper::save(const std::string& jsonPath, const std::string& ext, int numberOfFiles,
-                             const std::string& receiverHostname, int receiverPort, bool useOnlyFiles,
+                             const std::string& receiverHostname, int receiverPort, int receiverTimeout, bool useOnlyFiles,
                              bool useOnlySockets)
 {
   mEvent.setEveVersion(o2_eve_version);
@@ -522,6 +526,7 @@ void EveWorkflowHelper::save(const std::string& jsonPath, const std::string& ext
 
   VisualisationEventSerializer::getInstance(ext)->toFile(mEvent, Location({.fileName = producer.newFileName(),
                                                                            .port = receiverPort,
+                                                                           .timeout = receiverTimeout,
                                                                            .host = receiverHostname,
                                                                            .toFile = !useOnlySockets,
                                                                            .toSocket = !useOnlyFiles}));
@@ -550,7 +555,8 @@ std::vector<PNT>
   auto tp = trc;
   float dxmin = std::abs(xMin - tp.getX()), dxmax = std::abs(xMax - tp.getX());
 
-  if (dxmin > dxmax) { // start from closest end
+  if (dxmin > dxmax) {
+    // start from closest end
     std::swap(xMin, xMax);
     dx = -dx;
   }
@@ -763,7 +769,7 @@ void EveWorkflowHelper::drawTPCTRD(GID gid, float trackTime, GID::Source source)
   const auto& tpcTrdTrack = mRecoCont->getTPCTRDTrack<o2::trd::TrackTRD>(gid);
   addTrackToEvent(tpcTrdTrack, gid, trackTime, 0., source);
   drawTPCClusters(tpcTrdTrack.getRefGlobalTrackId(), trackTime * mMUS2TPCTimeBins);
-  drawTRDClusters(tpcTrdTrack);                       // tracktime
+  drawTRDClusters(tpcTrdTrack); // tracktime
 }
 
 void EveWorkflowHelper::drawITSTPCTRD(GID gid, float trackTime, GID::Source source)
@@ -799,7 +805,7 @@ void EveWorkflowHelper::drawTPCTOF(GID gid, float trackTime)
   const auto& match = mRecoCont->getTPCTOFMatch(gid.getIndex());
   addTrackToEvent(trTPCTOF, gid, trackTime, 0);
   drawTPCClusters(match.getTrackRef(), trackTime * mMUS2TPCTimeBins);
-  drawTOFClusters(gid);                 // trackTime
+  drawTOFClusters(gid); // trackTime
 }
 
 void EveWorkflowHelper::drawMFTMCH(GID gid, float trackTime)
@@ -978,7 +984,8 @@ void EveWorkflowHelper::drawTOFClusters(GID gid)
 
 void EveWorkflowHelper::drawITSClusters(GID gid) // float trackTime
 {
-  if (gid.getSource() == GID::ITS) { // this is for for full standalone tracks
+  if (gid.getSource() == GID::ITS) {
+    // this is for for full standalone tracks
     const auto& trc = mRecoCont->getITSTrack(gid);
     auto refs = mRecoCont->getITSTracksClusterRefs();
     int ncl = trc.getNumberOfClusters();
@@ -989,7 +996,8 @@ void EveWorkflowHelper::drawITSClusters(GID gid) // float trackTime
       float xyz[] = {glo.X(), glo.Y(), glo.Z()};
       drawPoint(xyz); // trackTime;
     }
-  } else if (gid.getSource() == GID::ITSAB) { // this is for ITS tracklets from ITS-TPC afterburner
+  } else if (gid.getSource() == GID::ITSAB) {
+    // this is for ITS tracklets from ITS-TPC afterburner
     const auto& trc = mRecoCont->getITSABRef(gid);
     const auto& refs = mRecoCont->getITSABClusterRefs();
     int ncl = trc.getNClusters();
@@ -1049,9 +1057,10 @@ void EveWorkflowHelper::drawTPC(GID gid, float trackTime, float dz)
   }
 
   addTrackToEvent(tr, gid, trackTime, dz, GID::TPC);
-  float clTime0 = EveConfParam::Instance().PVMode ? trackTime * mMUS2TPCTimeBins
-                                                  : -2e9;                                // in PVMode use supplied real time converted to TB, otherwise pass dummy time to use tpcTrack.getTime0
-  drawTPCClusters(gid, clTime0);                                                         // trackTime
+  float clTime0 = EveConfParam::Instance().PVMode
+                    ? trackTime * mMUS2TPCTimeBins
+                    : -2e9;      // in PVMode use supplied real time converted to TB, otherwise pass dummy time to use tpcTrack.getTime0
+  drawTPCClusters(gid, clTime0); // trackTime
 }
 
 void EveWorkflowHelper::drawITS(GID gid, float trackTime)
@@ -1198,16 +1207,18 @@ EveWorkflowHelper::EveWorkflowHelper()
   mTPCBin2MUS = elParams.ZbinWidth;
   const auto grp = o2::base::GRPGeomHelper::instance().getGRPECS();
   const auto& alpParamsITS = o2::itsmft::DPLAlpideParam<o2::detectors::DetID::ITS>::Instance();
-  mITSROFrameLengthMUS = grp->isDetContinuousReadOut(o2::detectors::DetID::ITS) ? alpParamsITS.roFrameLengthInBC *
-                                                                                    o2::constants::lhc::LHCBunchSpacingMUS
-                                                                                : alpParamsITS.roFrameLengthTrig *
-                                                                                    1.e-3;
+  mITSROFrameLengthMUS = grp->isDetContinuousReadOut(o2::detectors::DetID::ITS)
+                           ? alpParamsITS.roFrameLengthInBC *
+                               o2::constants::lhc::LHCBunchSpacingMUS
+                           : alpParamsITS.roFrameLengthTrig *
+                               1.e-3;
 
   const auto& alpParamsMFT = o2::itsmft::DPLAlpideParam<o2::detectors::DetID::MFT>::Instance();
-  mMFTROFrameLengthMUS = grp->isDetContinuousReadOut(o2::detectors::DetID::MFT) ? alpParamsMFT.roFrameLengthInBC *
-                                                                                    o2::constants::lhc::LHCBunchSpacingMUS
-                                                                                : alpParamsMFT.roFrameLengthTrig *
-                                                                                    1.e-3;
+  mMFTROFrameLengthMUS = grp->isDetContinuousReadOut(o2::detectors::DetID::MFT)
+                           ? alpParamsMFT.roFrameLengthInBC *
+                               o2::constants::lhc::LHCBunchSpacingMUS
+                           : alpParamsMFT.roFrameLengthTrig *
+                               1.e-3;
 
   mPVParams = &o2::vertexing::PVertexerParams::Instance();
 
diff --git a/EventVisualisation/Workflow/src/O2DPLDisplay.cxx b/EventVisualisation/Workflow/src/O2DPLDisplay.cxx
@@ -62,6 +62,7 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     {"jsons-folder", VariantType::String, "jsons", {"name of the folder to store json files"}},
     {"receiver-hostname", VariantType::String, "arcbs04.cern.ch", {"name of the host where visualisation data is transmitted (only eve format)"}},
     {"receiver-port", VariantType::Int, 8001, {"port number of the host where visualisation data is transmitted (only eve format)"}},
+    {"receiver-timeout", VariantType::Int, 300, {"socket connection timeout (ms)"}},
     {"use-only-files", VariantType::Bool, false, {"do not transmit visualisation data using sockets (only eve format)"}},
     {"use-only-sockets", VariantType::Bool, false, {"do not store visualisation data using filesystem"}},
     {"use-json-format", VariantType::Bool, false, {"instead of eve format (default) use json format"}},
@@ -190,7 +191,7 @@ void O2DPLDisplaySpec::run(ProcessingContext& pc)
         helper.mEvent.setRunType(this->mRunType);
         helper.mEvent.setPrimaryVertex(pv);
         helper.mEvent.setCreationTime(tinfo.creation);
-        helper.save(this->mJsonPath, this->mExt, conf.maxFiles, this->mReceiverHostname, this->mReceiverPort, this->mUseOnlyFiles, this->mUseOnlySockets);
+        helper.save(this->mJsonPath, this->mExt, conf.maxFiles, this->mReceiverHostname, this->mReceiverPort, this->mReceiverTimeout, this->mUseOnlyFiles, this->mUseOnlySockets);
         filesSaved++;
         currentTime = std::chrono::high_resolution_clock::now(); // time AFTER save
         this->mTimeStamp = currentTime;                          // next run AFTER period counted from last save
@@ -308,6 +309,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
 
   auto receiverHostname = cfgc.options().get<std::string>("receiver-hostname");
   auto receiverPort = cfgc.options().get<int>("receiver-port");
+  auto receiverTimeout = cfgc.options().get<int>("receiver-timeout");
   auto useOnlyFiles = cfgc.options().get<bool>("use-only-files");
   auto useOnlySockets = cfgc.options().get<bool>("use-only-sockets");
 
@@ -409,7 +411,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
     {},
     AlgorithmSpec{adaptFromTask<O2DPLDisplaySpec>(disableWrite, useMC, srcTrk, srcCl, dataRequest, ggRequest,
                                                   emcalCalibLoader, jsonFolder, ext, timeInterval, eveHostNameMatch,
-                                                  receiverHostname, receiverPort, useOnlyFiles, useOnlySockets)}});
+                                                  receiverHostname, receiverPort, receiverTimeout, useOnlyFiles, useOnlySockets)}});
 
   // configure dpl timer to inject correct firstTForbit: start from the 1st orbit of TF containing 1st sampled orbit
   o2::raw::HBFUtilsInitializer hbfIni(cfgc, specs);

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ struct AO2DConverter {`
`75`	`75`	`mHelper->mEvent.setFirstTForbit(mTfOrbit);`
`76`	`76`	`mHelper->mEvent.setCreationTime(collision.collisionTime());`
`77`	`77`	`const std::string hostname("localhost");`
`78`		`- mHelper->save(jsonPath, ".root", -1, hostname, -1, true, true);`
	`78`	`+ mHelper->save(jsonPath, ".root", -1, hostname, -1, 100, true, true);`
`79`	`79`	`mHelper->clear();`
`80`	`80`	`}`
`81`	`81`	`};`