eventbuilder::EventBuilder Class Reference

Public Member Functions
	EventBuilder (const std::string &name, framework::Process &proc)
	enableLogging ("EventBuilder")
void	configure (framework::config::Parameters &ps) override
	Callback for the EventProcessor to configure itself from the given set of parameters.
void	produce (framework::Event &event) override
	Process the event and put new data products into it.
Public Member Functions inherited from framework::Producer
	Producer (const std::string &name, Process &process)
	Class constructor.
virtual void	process (Event &event) final
	Processing an event for a Producer is calling produce.
Public Member Functions inherited from framework::EventProcessor
	DECLARE_FACTORY (EventProcessor, EventProcessor *, const std::string &, Process &)
	declare that we have a factory for this class
	EventProcessor (const std::string &name, Process &process)
	Class constructor.
virtual	~EventProcessor ()=default
	Class destructor.
virtual void	beforeNewRun (ldmx::RunHeader &run_header)
	Callback for Producers to add parameters to the run header before conditions are initialized.
virtual void	onNewRun (const ldmx::RunHeader &run_header)
	Callback for the EventProcessor to take any necessary action when the run being processed changes.
virtual void	onFileOpen (EventFile &event_file)
	Callback for the EventProcessor to take any necessary action when a new event input ROOT file is opened.
virtual void	onFileClose (EventFile &event_file)
	Callback for the EventProcessor to take any necessary action when a event input ROOT file is closed.
virtual void	onProcessStart ()
	Callback for the EventProcessor to take any necessary action when the processing of events starts, such as creating histograms.
virtual void	onProcessEnd ()
	Callback for the EventProcessor to take any necessary action when the processing of events finishes, such as calculating job-summary quantities.
template<class T >
const T &	getCondition (const std::string &condition_name)
	Access a conditions object for the current event.
TDirectory *	getHistoDirectory ()
	Access/create a directory in the histogram file for this event processor to create histograms and analysis tuples.
void	setStorageHint (framework::StorageControl::Hint hint)
	Mark the current event as having the given storage control hint from this module_.
void	setStorageHint (framework::StorageControl::Hint hint, const std::string &purposeString)
	Mark the current event as having the given storage control hint from this module and the given purpose string.
int	getLogFrequency () const
	Get the current logging frequency from the process.
int	getRunNumber () const
	Get the run number from the process.
std::string	getName () const
	Get the processor name.
void	createHistograms (const std::vector< framework::config::Parameters > &histos)
	Internal function which is used to create histograms passed from the python configuration @parma histos vector of Parameters that configure histograms to create.

Private Member Functions
PhysicsEventData	assemblePayload (const std::vector< DataFragment > &fragments)
void	writeEventBinary (const PhysicsEventData &ev, const std::string &path="events.bin")

Private Attributes
bool	m_verbose_parse_
unsigned int	m_event_id_
FragmentBuffer	m_event_buffer_
std::string	m_input_file_
packing::utility::Reader	m_reader_
std::string	m_output_name_ {"BuilderOutput"}
long long	m_coherence_window_ns_
int	m_min_subsystems_ {2}
std::string	m_binary_file_ {"events.bin"}
	binary event output, empty to disable
std::string	m_performance_file_ {"event_performance.csv"}
	performance metrics CSV, empty to disable
std::chrono::steady_clock::time_point	m_start_time_
std::chrono::steady_clock::time_point	m_event_start_time_
uint64_t	m_total_bytes_read_ {0}
uint64_t	m_total_events_built_ {0}
unsigned long	m_events_since_last_report_ {0}
std::chrono::steady_clock::time_point	m_window_start_time_
uint64_t	m_window_events_count_ {0}
uint64_t	m_window_bytes_read_ {0}

Static Private Attributes
static constexpr unsigned int	WINDOW_SIZE

Additional Inherited Members
Protected Member Functions inherited from framework::EventProcessor
void	abortEvent ()
	Abort the event immediately.
Protected Attributes inherited from framework::EventProcessor
HistogramPool	histograms_
	helper object for making and filling histograms
NtupleManager &	ntuple_ {NtupleManager::getInstance()}
	Manager for any ntuples.
logging::logger	the_log_
	The logger for this EventProcessor.

Detailed Description

Definition at line 14 of file EventBuilder.h.

Constructor & Destructor Documentation

EventBuilder()

eventbuilder::EventBuilder::EventBuilder ( const std::string & name, framework::Process & proc )

inline

Definition at line 16 of file EventBuilder.h.

      : framework::Producer(name, proc),
        m_verbose_parse_(false),
        m_event_id_(0) {}

Member Function Documentation

assemblePayload()

PhysicsEventData EventBuilder::assemblePayload ( const std::vector< DataFragment > & fragments )

private

Definition at line 540 of file EventBuilder.cxx.

                                              {
  PhysicsEventData event_data;
  if (fragments.empty()) return event_data;
  event_data.event_id_ = m_event_id_;
  event_data.timestamp_ = fragments.front().header_.timestamp_;
  for (const auto& fragment : fragments) {
    GenericDataBlock g;
    g.subsystem_id_ = fragment.header_.subsystem_id_;
    g.timestamp_ns_ = fragment.header_.timestamp_;
    g.data_ = fragment.payload_;
    if (fragment.trailer_.checksum_ != 0)
      g.checksum_ = fragment.trailer_.checksum_;
    else
      g.checksum_ = crc32(fragment.payload_);
    event_data.blocks_.push_back(std::move(g));
    event_data.systems_readout_.push_back(fragment.header_.subsystem_id_);
  }
  return event_data;
}

configure()

void EventBuilder::configure ( framework::config::Parameters & parameters )

overridevirtual

Callback for the EventProcessor to configure itself from the given set of parameters.

The parameters a processor has access to are the member variables of the python class in the sequence that has class_name equal to the EventProcessor class name.

For an example, look at MyProcessor.

Parameters

parameters

Parameters for configuration.

Reimplemented from framework::EventProcessor.

Definition at line 58 of file EventBuilder.cxx.

                                                          {
  if (ps.exists("verbose_parse")) {
    m_verbose_parse_ = ps.get<bool>("verbose_parse");
  }
  if (ps.exists("dat_file")) {
    m_input_file_ = ps.get<std::string>("dat_file");
  } else {
    const char* env_input = std::getenv("EVENTBUILDER_INPUT");
    if (env_input) m_input_file_ = env_input;
  }
  if (ps.exists("output_name"))
    m_output_name_ = ps.get<std::string>("output_name");
  if (ps.exists("coherence_window_ns")) {
    m_coherence_window_ns_ = ps.get<double>("coherence_window_ns");
  }
  if (ps.exists("min_subsystems")) {
    m_min_subsystems_ = ps.get<int>("min_subsystems");
  }
  if (ps.exists("binary_file")) {
    m_binary_file_ = ps.get<std::string>("binary_file");
  }
  if (ps.exists("performance_file")) {
    m_performance_file_ = ps.get<std::string>("performance_file");
  }
 
  ldmx_log(info) << "configure(): dat_file='" << m_input_file_
                 << "' output_name='" << m_output_name_
                 << "' verbose_parse=" << (m_verbose_parse_ ? "true" : "false")
                 << " coherence_window_ns=" << m_coherence_window_ns_
                 << " min_subsystems=" << m_min_subsystems_;
 
  if (!m_input_file_.empty()) {
    ldmx_log(info) << "configure(): opening file '" << m_input_file_ << "'";
    m_reader_.open(m_input_file_);
    if (!m_reader_) {
      ldmx_log(error) << "failed to open input file: '" << m_input_file_ << "'";
    } else {
      ldmx_log(info) << "configure(): file opened successfully";
    }
  } else {
    ldmx_log(error) << "no input file specified";
  }
 
  // Initialize performance tracking
  m_start_time_ = std::chrono::steady_clock::now();
  m_total_bytes_read_ = 0;
  m_total_events_built_ = 0;
  m_events_since_last_report_ = 0;
 
  // Initialize windowed metrics for real-time DAQ monitoring
  m_window_start_time_ = m_start_time_;
  m_window_events_count_ = 0;
  m_window_bytes_read_ = 0;
}

References framework::config::Parameters::exists(), framework::config::Parameters::get(), m_binary_file_, m_performance_file_, and packing::utility::Reader::open().

produce()

void EventBuilder::produce ( framework::Event & event )

overridevirtual

Process the event and put new data products into it.

Parameters

event

The Event to process.

Implements framework::Producer.

Definition at line 113 of file EventBuilder.cxx.

                                                {
  static int produce_call_count = 0;
  produce_call_count++;
  ldmx_log(debug) << "produce() called, count=" << produce_call_count;
  ldmx_log(debug) << "verbose_parse=" << (m_verbose_parse_ ? "true" : "false");
  if (m_verbose_parse_ || produce_call_count <= 3) {
    ldmx_log(info) << "produce() call #" << produce_call_count;
  }
 
  // Start timing for this event
  m_event_start_time_ = std::chrono::steady_clock::now();
 
  // Track errors encountered during event assembly
  uint32_t current_event_errors = 0;
 
  while (m_reader_ && !m_reader_.eof()) {
    // Try to read a RogueFrameHeader
    packing::RogueFrameHeader frame_header;
    frame_header.read(m_reader_);
 
    // Store location of end-of-frame for potential recovery
    const long long frame_end =
        static_cast<long long>(m_reader_.tell()) + frame_header.size();
 
    // Check if this is a data frame (channel 0) and not YAML
    if (frame_header.channel() != 0 || frame_header.probablyYaml()) {
      // Skip this frame
      if (m_verbose_parse_)
        ldmx_log(debug) << "skipping non-data frame (channel="
                        << frame_header.channel() << ")";
      m_reader_.seek(frame_end);
      continue;
    }
 
    // We have a valid data frame - try to parse it
    DataFragment fragment{{}, {}, {.checksum_ = 0}};
    bool parsed_ok = false;
 
    // Try RoR format first
    packing::LDMXRoRHeader ror_header;
    int pos_before_ror = m_reader_.tell();
 
    // Attempt to read as RoR header
    try {
      ror_header.read(m_reader_);
      // Successfully parsed RoR header
      fragment.header_.subsystem_id_ =
          static_cast<uint64_t>(ror_header.subsystem());
      fragment.header_.contributor_id_ =
          static_cast<uint64_t>(ror_header.contributor());
      fragment.header_.timestamp_ = ror_header.timestamp();
 
      if (m_verbose_parse_) {
        ldmx_log(debug) << "parsed RoR header subsys="
                        << (int)ror_header.subsystem()
                        << " contrib=" << (int)ror_header.contributor()
                        << " ts=" << ror_header.timestamp();
      }
 
      // Read remaining frame payload after RoR header
      std::vector<uint8_t> payload_data;
      long long payload_size = frame_end - m_reader_.tell();
      if (payload_size > 0) {
        payload_data.resize(payload_size);
        m_reader_.read(reinterpret_cast<char*>(payload_data.data()),
                       payload_size);
      }
      fragment.payload_ = std::move(payload_data);
      parsed_ok = true;
    } catch (...) {
      // RoR parse failed, try packing subsystem format
      if (m_verbose_parse_)
        ldmx_log(debug)
            << "RoR header parse failed, trying packing subsystem format";
      current_event_errors |= ldmx::EventSummary::ERROR_PARSE_FAILURE;
      m_reader_.seek(pos_before_ror);
    }
 
    if (!parsed_ok) {
      // Try packing subsystem format
      packing::rawdatafile::SubsystemPacket pkt;
      try {
        pkt.read(m_reader_);
        fragment.header_.subsystem_id_ = static_cast<uint64_t>(pkt.id());
        fragment.header_.contributor_id_ =
            0;  // Not available in packing subsystem format
        fragment.header_.timestamp_ = static_cast<uint64_t>(pkt.header()[1]) *
                                      1000000000ULL;  // event number to ns
 
        // Convert packet data to payload bytes
        const auto& data = pkt.data();
        fragment.payload_.resize(data.size() * 4);
        std::memcpy(fragment.payload_.data(),
                    reinterpret_cast<const char*>(data.data()),
                    data.size() * 4);
 
        if (m_verbose_parse_) {
          ldmx_log(debug) << "parsed packing subsystem pkt subsys=" << pkt.id()
                          << " data_size=" << data.size();
        }
        parsed_ok = true;
      } catch (...) {
        // Both parse attempts failed, skip this frame
        if (m_verbose_parse_)
          ldmx_log(debug) << "failed to parse as either format, skipping frame";
        current_event_errors |= ldmx::EventSummary::ERROR_PARSE_FAILURE;
        m_reader_.seek(frame_end);
        continue;
      }
    }
 
    if (!parsed_ok) {
      m_reader_.seek(frame_end);
      continue;
    }
 
    if (m_verbose_parse_)
      ldmx_log(debug) << "adding fragment subsys="
                      << fragment.header_.subsystem_id_
                      << " ts=" << fragment.header_.timestamp_
                      << " bytes=" << fragment.payload_.size();
 
    // Check if this fragment is outside the coherence window of the current
    // event batch If so, it means we should finalize the previous event before
    // adding this one
    long long fragment_ts = fragment.header_.timestamp_;
    long long buffer_ref_time = m_event_buffer_.getReferenceTime();
 
    if (buffer_ref_time > 0 &&
        (fragment_ts < buffer_ref_time - m_coherence_window_ns_ ||
         fragment_ts > buffer_ref_time + m_coherence_window_ns_)) {
      // This fragment is outside the current window - try to build the previous
      // event
      std::vector<DataFragment> assembled_event_fragments;
      if (m_event_buffer_.tryBuildEvent(m_coherence_window_ns_,
                                        m_min_subsystems_,
                                        assembled_event_fragments) &&
          !assembled_event_fragments.empty()) {
        // Output the complete event and return
        ++m_event_id_;
        // Add each subsystem's raw data as vector<uint8_t> directly
        uint64_t event_timestamp = 0;
        for (const auto& frag : assembled_event_fragments) {
          std::string subsys_name = packing::LDMXRoRHeader::getSubsystemName(
              static_cast<uint8_t>(frag.header_.subsystem_id_),
              static_cast<uint8_t>(frag.header_.contributor_id_));
          std::vector<uint8_t> payload_bytes(frag.payload_.begin(),
                                             frag.payload_.end());
          event.add(subsys_name, payload_bytes);
          if (event_timestamp == 0) {
            event_timestamp = frag.header_.timestamp_;
          }
        }
        event.getEventHeader().setIntParameter(
            "RoR Timestamp", static_cast<long long>(event_timestamp));
        event.getEventHeader().setRealData();
        // Still create PhysicsEventData for binary output file
        PhysicsEventData final_event =
            assemblePayload(assembled_event_fragments);
        writeEventBinary(final_event, m_binary_file_);
        ldmx_log(info) << "assembled event id=" << m_event_id_
                       << " timestamp=" << final_event.timestamp_
                       << " fragments=" << assembled_event_fragments.size()
                       << " systems=" << final_event.systems_readout_.size();
 
        ldmx::EventSummary summary;
        summary.setEventNumber(m_event_id_);
        summary.setTimestampNs(static_cast<uint64_t>(final_event.timestamp_));
        std::set<uint64_t> unique_sys, unique_sources;
        uint64_t total_payload = 0;
        for (const auto& f : assembled_event_fragments) {
          unique_sys.insert(f.header_.subsystem_id_);
          // a subsystem may have several contributors, a duplicate is the same
          // pair twice
          unique_sources.insert(f.header_.subsystem_id_ << 8 |
                                f.header_.contributor_id_);
          total_payload += f.payload_.size();
        }
        // Check for duplicate subsystems
        if (unique_sources.size() != assembled_event_fragments.size()) {
          current_event_errors |= ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM;
        }
        summary.setNSystems(static_cast<uint32_t>(unique_sys.size()));
        summary.setSystemIds(
            std::vector<uint64_t>(unique_sys.begin(), unique_sys.end()));
        summary.setPayloadSize(total_payload);
        summary.setErrorFlags(current_event_errors);
        event.add("EventSummary", summary);
 
        // Update performance metrics
        m_total_bytes_read_ += total_payload;
        m_total_events_built_++;
        m_events_since_last_report_++;
        std::chrono::steady_clock::time_point now =
            std::chrono::steady_clock::now();
        std::chrono::milliseconds event_build_time_ms =
            std::chrono::duration_cast<std::chrono::milliseconds>(
                now - m_event_start_time_);
        std::chrono::seconds total_elapsed_s =
            std::chrono::duration_cast<std::chrono::seconds>(now -
                                                             m_start_time_);
 
        double events_per_sec =
            (total_elapsed_s.count() > 0)
                ? static_cast<double>(m_total_events_built_) /
                      total_elapsed_s.count()
                : 0.0;
        double mb_per_sec = (total_elapsed_s.count() > 0)
                                ? static_cast<double>(m_total_bytes_read_) /
                                      (1024.0 * 1024.0) /
                                      total_elapsed_s.count()
                                : 0.0;
 
        // Update windowed metrics
        m_window_events_count_++;
        m_window_bytes_read_ += total_payload;
        std::chrono::seconds window_elapsed_s =
            std::chrono::duration_cast<std::chrono::seconds>(
                now - m_window_start_time_);
 
        // Reset window if we've processed WINDOW_SIZE events
        double window_events_per_sec = 0.0;
        double window_mb_per_sec = 0.0;
        if (m_window_events_count_ >= WINDOW_SIZE) {
          double window_time_sec =
              std::max(1.0, static_cast<double>(window_elapsed_s.count()));
          window_events_per_sec =
              static_cast<double>(m_window_events_count_) / window_time_sec;
          window_mb_per_sec = static_cast<double>(m_window_bytes_read_) /
                              (1024.0 * 1024.0) / window_time_sec;
 
          // Reset window
          m_window_start_time_ = now;
          m_window_events_count_ = 0;
          m_window_bytes_read_ = 0;
        }
 
        // Write performance metrics to CSV
        long long event_build_time_ms_val = event_build_time_ms.count();
        writePerformanceMetric(m_performance_file_, m_event_id_,
                               static_cast<double>(event_build_time_ms_val),
                               events_per_sec, mb_per_sec,
                               window_events_per_sec, window_mb_per_sec,
                               m_total_events_built_, m_total_bytes_read_);
 
        if (m_verbose_parse_ || m_events_since_last_report_ % 100 == 0) {
          ldmx_log(info) << "Performance: "
                         << "total_events=" << m_total_events_built_ << ", "
                         << "events_per_sec=" << std::fixed
                         << std::setprecision(2) << events_per_sec << ", "
                         << "mb_per_sec=" << std::fixed << std::setprecision(3)
                         << mb_per_sec;
        }
 
        current_event_errors = 0;  // Reset for next event
        // The fragment that tripped the window-close belongs to the NEXT
        // event, not the one we just emitted. Buffer it here (the buffer is
        // empty post-build, so this also resets the reference time) instead
        // of dropping it -- otherwise every event after the first loses its
        // triggering subsystem.
        m_event_buffer_.addFragment(std::move(fragment));
        return;  // Return the event to framework
      }
    }
 
    // Add fragment to buffer (may start a new event batch if buffer was empty)
    m_event_buffer_.addFragment(std::move(fragment));
 
    if (m_verbose_parse_)
      ldmx_log(debug) << "frame added to buffer, searching for more frames...";
  }
 
  // Reached EOF - flush any remaining events in the buffer
  if (m_verbose_parse_)
    ldmx_log(debug) << "reached EOF, flushing remaining events";
 
  std::vector<DataFragment> assembled_event_fragments;
  while (m_event_buffer_.tryBuildEvent(
      m_coherence_window_ns_, m_min_subsystems_, assembled_event_fragments)) {
    if (assembled_event_fragments.empty()) break;
 
    // Mark truncated events (those flushed at EOF)
    current_event_errors |= ldmx::EventSummary::ERROR_TRUNCATED_EVENT;
 
    ++m_event_id_;
    // Add each subsystem's raw data as vector<uint8_t> directly
    uint64_t event_timestamp = 0;
    for (const auto& frag : assembled_event_fragments) {
      std::string subsys_name = packing::LDMXRoRHeader::getSubsystemName(
          static_cast<uint8_t>(frag.header_.subsystem_id_),
          static_cast<uint8_t>(frag.header_.contributor_id_));
      std::vector<uint8_t> payload_bytes(frag.payload_.begin(),
                                         frag.payload_.end());
      event.add(subsys_name, payload_bytes);
      if (event_timestamp == 0) {
        event_timestamp = frag.header_.timestamp_;
      }
    }
    event.getEventHeader().setIntParameter(
        "RoR Timestamp", static_cast<long long>(event_timestamp));
    event.getEventHeader().setRealData();
    // Still create PhysicsEventData for binary output file
    PhysicsEventData final_event = assemblePayload(assembled_event_fragments);
    writeEventBinary(final_event, m_binary_file_);
    ldmx_log(info) << "assembled event id=" << m_event_id_
                   << " timestamp=" << final_event.timestamp_
                   << " fragments=" << assembled_event_fragments.size()
                   << " systems=" << final_event.systems_readout_.size();
 
    ldmx::EventSummary summary;
    summary.setEventNumber(m_event_id_);
    summary.setTimestampNs(static_cast<uint64_t>(final_event.timestamp_));
    std::set<uint64_t> unique_sys, unique_sources;
    uint64_t total_payload = 0;
    for (const auto& f : assembled_event_fragments) {
      unique_sys.insert(f.header_.subsystem_id_);
      // a subsystem may have several contributors, a duplicate is the same pair
      // twice
      unique_sources.insert(f.header_.subsystem_id_ << 8 |
                            f.header_.contributor_id_);
      total_payload += f.payload_.size();
    }
    // Check for duplicate subsystems
    if (unique_sources.size() != assembled_event_fragments.size()) {
      current_event_errors |= ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM;
    }
    summary.setNSystems(static_cast<uint32_t>(unique_sys.size()));
    summary.setSystemIds(
        std::vector<uint64_t>(unique_sys.begin(), unique_sys.end()));
    summary.setPayloadSize(total_payload);
    summary.setErrorFlags(current_event_errors);
    event.add("EventSummary", summary);
 
    // Update performance metrics
    m_total_bytes_read_ += total_payload;
    m_total_events_built_++;
    m_events_since_last_report_++;
    std::chrono::steady_clock::time_point now =
        std::chrono::steady_clock::now();
    std::chrono::milliseconds event_build_time_ms =
        std::chrono::duration_cast<std::chrono::milliseconds>(
            now - m_event_start_time_);
    std::chrono::seconds total_elapsed_s =
        std::chrono::duration_cast<std::chrono::seconds>(now - m_start_time_);
 
    double events_per_sec = (total_elapsed_s.count() > 0)
                                ? static_cast<double>(m_total_events_built_) /
                                      total_elapsed_s.count()
                                : 0.0;
    double mb_per_sec = (total_elapsed_s.count() > 0)
                            ? static_cast<double>(m_total_bytes_read_) /
                                  (1024.0 * 1024.0) / total_elapsed_s.count()
                            : 0.0;
 
    // Update windowed metrics
    m_window_events_count_++;
    m_window_bytes_read_ += total_payload;
    std::chrono::seconds window_elapsed_s =
        std::chrono::duration_cast<std::chrono::seconds>(now -
                                                         m_window_start_time_);
 
    // Calculate window metrics (for final event, may not have full window)
    double window_events_per_sec = 0.0;
    double window_mb_per_sec = 0.0;
    if (m_window_events_count_ >= WINDOW_SIZE || window_elapsed_s.count() > 0) {
      double window_time_sec =
          std::max(1.0, static_cast<double>(window_elapsed_s.count()));
      window_events_per_sec =
          static_cast<double>(m_window_events_count_) / window_time_sec;
      window_mb_per_sec = static_cast<double>(m_window_bytes_read_) /
                          (1024.0 * 1024.0) / window_time_sec;
    }
 
    // Write performance metrics to CSV
    long long event_build_time_ms_val = event_build_time_ms.count();
    writePerformanceMetric(m_performance_file_, m_event_id_,
                           static_cast<double>(event_build_time_ms_val),
                           events_per_sec, mb_per_sec, window_events_per_sec,
                           window_mb_per_sec, m_total_events_built_,
                           m_total_bytes_read_);
 
    if (m_verbose_parse_ || m_events_since_last_report_ % 100 == 0) {
      ldmx_log(info) << "Performance: "
                     << "total_events=" << m_total_events_built_ << ", "
                     << "events_per_sec=" << std::fixed << std::setprecision(2)
                     << events_per_sec << ", "
                     << "mb_per_sec=" << std::fixed << std::setprecision(3)
                     << mb_per_sec;
    }
 
    current_event_errors = 0;  // Reset for next event
 
    assembled_event_fragments.clear();
    return;
  }
 
  // No more events - print final summary statistics
  std::chrono::steady_clock::time_point final_time =
      std::chrono::steady_clock::now();
  std::chrono::seconds total_time_s =
      std::chrono::duration_cast<std::chrono::seconds>(final_time -
                                                       m_start_time_);
  double final_events_per_sec =
      (total_time_s.count() > 0)
          ? static_cast<double>(m_total_events_built_) / total_time_s.count()
          : 0.0;
  double final_mb_per_sec = (total_time_s.count() > 0)
                                ? static_cast<double>(m_total_bytes_read_) /
                                      (1024.0 * 1024.0) / total_time_s.count()
                                : 0.0;
 
  ldmx_log(info) << "\n===== FINAL STATISTICS =====";
  ldmx_log(info) << "Total events built: " << m_total_events_built_;
  ldmx_log(info) << "Total bytes read: "
                 << m_total_bytes_read_ / (1024.0 * 1024.0) << " MB";
  ldmx_log(info) << "Total time: " << total_time_s.count() << " seconds";
  ldmx_log(info) << "Average throughput: "
                 << "events_per_sec=" << std::fixed << std::setprecision(2)
                 << final_events_per_sec << ", "
                 << "mb_per_sec=" << std::fixed << std::setprecision(3)
                 << final_mb_per_sec;
  ldmx_log(info) << "=============================\n";
  ldmx_log(info) << "Event building complete";
 
  abortEvent();
}

References framework::EventProcessor::abortEvent(), packing::RogueFrameHeader::channel(), packing::LDMXRoRHeader::contributor(), packing::rawdatafile::SubsystemPacket::data(), packing::utility::Reader::eof(), ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM, ldmx::EventSummary::ERROR_PARSE_FAILURE, ldmx::EventSummary::ERROR_TRUNCATED_EVENT, packing::LDMXRoRHeader::getSubsystemName(), packing::rawdatafile::SubsystemPacket::header(), m_binary_file_, m_performance_file_, packing::RogueFrameHeader::probablyYaml(), packing::LDMXRoRHeader::read(), packing::rawdatafile::SubsystemPacket::read(), packing::RogueFrameHeader::read(), packing::utility::Reader::read(), packing::utility::Reader::seek(), ldmx::EventSummary::setErrorFlags(), ldmx::EventSummary::setEventNumber(), ldmx::EventSummary::setNSystems(), ldmx::EventSummary::setPayloadSize(), ldmx::EventSummary::setSystemIds(), ldmx::EventSummary::setTimestampNs(), packing::RogueFrameHeader::size(), packing::LDMXRoRHeader::subsystem(), packing::utility::Reader::tell(), and packing::LDMXRoRHeader::timestamp().

writeEventBinary()

void EventBuilder::writeEventBinary ( const PhysicsEventData & ev, const std::string & path = "events.bin" )

private

Definition at line 561 of file EventBuilder.cxx.

                                                           {
  static std::mutex g_out_mutex;
  std::lock_guard<std::mutex> lg(g_out_mutex);
  if (path.empty()) return;
  std::ofstream ofs(path, std::ios::binary | std::ios::app);
  if (!ofs) return;
  uint64_t event_id_u = static_cast<uint64_t>(ev.event_id_);
  uint64_t ts = static_cast<uint64_t>(ev.timestamp_);
  uint32_t nblocks = static_cast<uint32_t>(ev.blocks_.size());
  ofs.write(reinterpret_cast<const char*>(&event_id_u), sizeof(event_id_u));
  ofs.write(reinterpret_cast<const char*>(&ts), sizeof(ts));
  ofs.write(reinterpret_cast<const char*>(&nblocks), sizeof(nblocks));
  for (const auto& b : ev.blocks_) {
    uint64_t sid = b.subsystem_id_;
    uint64_t bts = b.timestamp_ns_;
    uint32_t psz = static_cast<uint32_t>(b.data_.size());
    uint32_t csum = b.checksum_;
    ofs.write(reinterpret_cast<const char*>(&sid), sizeof(sid));
    ofs.write(reinterpret_cast<const char*>(&bts), sizeof(bts));
    ofs.write(reinterpret_cast<const char*>(&psz), sizeof(psz));
    ofs.write(reinterpret_cast<const char*>(&csum), sizeof(csum));
    if (psz)
      ofs.write(reinterpret_cast<const char*>(b.data_.data()),
                static_cast<std::streamsize>(psz));
  }
  ofs.flush();
}

Member Data Documentation

m_binary_file_

std::string eventbuilder::EventBuilder::m_binary_file_ {"events.bin"}

private

binary event output, empty to disable

Definition at line 47 of file EventBuilder.h.

{"events.bin"};

Referenced by configure(), and produce().

m_coherence_window_ns_

long long eventbuilder::EventBuilder::m_coherence_window_ns_

private

Initial value:

{
      5000000}

Definition at line 41 of file EventBuilder.h.

                                  {
      5000000};  // 5 ms window for collecting fragments

m_event_buffer_

FragmentBuffer eventbuilder::EventBuilder::m_event_buffer_

private

Definition at line 37 of file EventBuilder.h.

m_event_id_

unsigned int eventbuilder::EventBuilder::m_event_id_

private

Definition at line 36 of file EventBuilder.h.

m_event_start_time_

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_event_start_time_

private

Definition at line 53 of file EventBuilder.h.

m_events_since_last_report_

unsigned long eventbuilder::EventBuilder::m_events_since_last_report_ {0}

private

Definition at line 56 of file EventBuilder.h.

{0};

m_input_file_

std::string eventbuilder::EventBuilder::m_input_file_

private

Definition at line 38 of file EventBuilder.h.

m_min_subsystems_

int eventbuilder::EventBuilder::m_min_subsystems_ {2}

private

Definition at line 45 of file EventBuilder.h.

{2};

m_output_name_

std::string eventbuilder::EventBuilder::m_output_name_ {"BuilderOutput"}

private

Definition at line 40 of file EventBuilder.h.

{"BuilderOutput"};

m_performance_file_

std::string eventbuilder::EventBuilder::m_performance_file_ {"event_performance.csv"}

private

performance metrics CSV, empty to disable

Definition at line 49 of file EventBuilder.h.

{"event_performance.csv"};

Referenced by configure(), and produce().

m_reader_

packing::utility::Reader eventbuilder::EventBuilder::m_reader_

private

Definition at line 39 of file EventBuilder.h.

m_start_time_

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_start_time_

private

Definition at line 52 of file EventBuilder.h.

m_total_bytes_read_

uint64_t eventbuilder::EventBuilder::m_total_bytes_read_ {0}

private

Definition at line 54 of file EventBuilder.h.

{0};

m_total_events_built_

uint64_t eventbuilder::EventBuilder::m_total_events_built_ {0}

private

Definition at line 55 of file EventBuilder.h.

{0};

m_verbose_parse_

bool eventbuilder::EventBuilder::m_verbose_parse_

private

Definition at line 35 of file EventBuilder.h.

m_window_bytes_read_

uint64_t eventbuilder::EventBuilder::m_window_bytes_read_ {0}

private

Definition at line 63 of file EventBuilder.h.

{0};

m_window_events_count_

uint64_t eventbuilder::EventBuilder::m_window_events_count_ {0}

private

Definition at line 62 of file EventBuilder.h.

{0};

m_window_start_time_

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_window_start_time_

private

Definition at line 61 of file EventBuilder.h.

WINDOW_SIZE

unsigned int eventbuilder::EventBuilder::WINDOW_SIZE

staticconstexprprivate

Initial value:

=
      100

Definition at line 59 of file EventBuilder.h.

---

The documentation for this class was generated from the following files:

• EventBuilder/include/EventBuilder/EventBuilder.h
• EventBuilder/src/EventBuilder/EventBuilder.cxx