Line data    Source code

       1              : #include "TRACE/tracemf.h"
       2              : #define TRACE_NAME "ShmemWrapper"
       3              : 
       4              : #include "artdaq/ArtModules/detail/ShmemWrapper.hh"
       5              : 
       6              : #include "art/Framework/Services/Registry/ServiceHandle.h"
       7              : #include "artdaq/ArtModules/ArtdaqSharedMemoryServiceInterface.h"
       8              : #include "artdaq/DAQdata/NetMonHeader.hh"
       9              : 
      10            0 : art::ShmemWrapper::ShmemWrapper(fhicl::ParameterSet const& ps)
      11              : {
      12            0 :         init_timeout_s_ = ps.get<double>("init_fragment_timeout_seconds", 600.0);
      13              :         // Make sure the ArtdaqSharedMemoryService is available
      14            0 :         art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
      15            0 : }
      16              : 
      17            0 : std::shared_ptr<ArtdaqEvent> art::ShmemWrapper::receiveMessages()
      18              : {
      19            0 :         TLOG(TLVL_DEBUG + 34) << "Receiving Fragment from NetMonTransportService";
      20            0 :         TLOG(TLVL_DEBUG + 33) << "receiveMessage BEGIN";
      21            0 :         art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
      22            0 :         std::shared_ptr<ArtdaqEvent> output;
      23              : 
      24              :         // Do not process data until Init Fragment received!
      25            0 :         auto start = std::chrono::steady_clock::now();
      26            0 :         while (!init_received_ && artdaq::TimeUtils::GetElapsedTime(start) < init_timeout_s_)
      27              :         {
      28            0 :                 usleep(static_cast<unsigned>(init_timeout_s_ * 1000000 / 100));  // Check 100 times
      29            0 :                 if (eod_received_)
      30              :                 {
      31            0 :                         TLOG(TLVL_DEBUG + 32) << "Received EndOfData message while waiting for Init Fragment, returning";
      32            0 :                         return nullptr;
      33              :                 }
      34              :         }
      35            0 :         if (!init_received_)
      36              :         {
      37            0 :                 TLOG(TLVL_ERROR) << "Did not receive Init Fragment after " << init_timeout_s_ << " seconds.";
      38              :         }
      39              : 
      40            0 :         output = shm->ReceiveEvent(false);
      41              : 
      42            0 :         if (output == nullptr)
      43              :         {
      44            0 :                 TLOG(TLVL_DEBUG + 32) << "Did not receive event after timeout, returning from receiveMessage ";
      45            0 :                 return output;
      46              :         }
      47              : 
      48            0 :         TLOG(TLVL_DEBUG + 33) << "receiveMessage END";
      49              : 
      50            0 :         TLOG(TLVL_DEBUG + 34) << "Done Receiving Fragments from Shared Memory";
      51            0 :         return output;
      52            0 : }
      53              : 
      54            0 : artdaq::FragmentPtrs art::ShmemWrapper::receiveInitMessage()
      55              : {
      56            0 :         TLOG(TLVL_DEBUG + 34) << "Receiving Init Fragment from NetMonTransportService";
      57              : 
      58            0 :         TLOG(TLVL_DEBUG + 33) << "receiveInitMessage BEGIN";
      59            0 :         art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
      60            0 :         auto start = std::chrono::steady_clock::now();
      61            0 :         std::shared_ptr<ArtdaqEvent> eventMap;
      62            0 :         while (eventMap == nullptr)
      63              :         {
      64            0 :                 eventMap = shm->ReceiveEvent(true);
      65              : 
      66            0 :                 if (eventMap != nullptr)
      67              :                 {
      68            0 :                         auto type = eventMap->FirstFragmentType();
      69            0 :                         if (type == artdaq::Fragment::EndOfDataFragmentType)
      70              :                         {
      71            0 :                                 TLOG(TLVL_DEBUG + 32) << "Received shutdown message, returning";
      72            0 :                                 eod_received_ = true;
      73            0 :                                 artdaq::FragmentPtrs output;
      74            0 :                                 for (auto& frag : *eventMap->fragments[artdaq::Fragment::EndOfDataFragmentType])
      75              :                                 {
      76            0 :                                         output.emplace_back(new artdaq::Fragment(std::move(frag)));
      77              :                                 }
      78            0 :                                 return output;
      79            0 :                         }
      80            0 :                         if (type != artdaq::Fragment::InitFragmentType)
      81              :                         {
      82            0 :                                 TLOG(TLVL_WARNING) << "Did NOT receive Init Fragment as first broadcast! Type="
      83            0 :                                                    << artdaq::detail::RawFragmentHeader::SystemTypeToString(type);
      84            0 :                                 eventMap = nullptr;
      85              :                         }
      86              :                 }
      87            0 :                 else if (artdaq::TimeUtils::GetElapsedTime(start) > init_timeout_s_)
      88              :                 {
      89            0 :                         TLOG(TLVL_WARNING) << "Did not receive Init fragment after init_fragment_timeout_seconds (" << artdaq::TimeUtils::GetElapsedTime(start) << ")!";
      90            0 :                         eod_received_ = true;
      91            0 :                         return artdaq::FragmentPtrs();
      92              :                 }
      93              :         }
      94              : 
      95              :         // We return false, indicating we're done reading, if:
      96              :         //   1) we did not obtain an event, because we timed out and were
      97              :         //      configured NOT to keep trying after a timeout, or
      98              :         //   2) the event we read was the end-of-data marker: a null
      99              :         //      pointer
     100              : 
     101            0 :         TLOG(TLVL_DEBUG + 33) << "receiveInitMessage: Returning top Fragment";
     102            0 :         artdaq::FragmentPtrs output;
     103            0 :         for (auto& frag : *eventMap->fragments[artdaq::Fragment::InitFragmentType])
     104              :         {
     105            0 :                 output.emplace_back(new artdaq::Fragment(std::move(frag)));
     106              :         }
     107              : 
     108              : #if DUMP_RECEIVE_MESSAGE
     109              :         std::string fileName = "receiveInitMessage_" + std::to_string(getpid()) + ".bin";
     110              :         std::fstream ostream(fileName.c_str(), std::ios::out | std::ios::binary);
     111              :         ostream.write(buffer, header->data_length);
     112              :         ostream.close();
     113              : #endif
     114              : 
     115            0 :         TLOG(TLVL_DEBUG + 33) << "receiveInitMessage END";
     116            0 :         init_received_ = true;
     117              : 
     118            0 :         TLOG(TLVL_DEBUG + 34) << "Done Receiving Init Fragment from NetMonTransportService";
     119            0 :         return output;
     120            0 : }