trigbs_orderedMerge.cxx File Reference

Standalone executable to order and decompress events from multiple files and write them into a single file. More...

#include "eformat/FullEventFragmentNoTemplates.h"
#include "eformat/ROBFragmentNoTemplates.h"
#include "eformat/write/FullEventFragment.h"
#include "eformat/write/ROBFragment.h"
#include "eformat/compression.h"
#include "EventStorage/DataReader.h"
#include "EventStorage/pickDataReader.h"
#include "EventStorage/DataWriter.h"
#include "EventStorage/DRError.h"
#include "EventStorage/DWError.h"
#include "EventStorage/EventStorageRecords.h"
#include "EventStorage/ESCompression.h"
#include <iostream>
#include <string>
#include <vector>
#include <memory>
#include <optional>
#include <functional>
#include <unordered_set>

Go to the source code of this file.

Classes
struct	Event
class	WriteEvent
class	Buffer

Functions
std::unique_ptr< EventStorage::DataWriter >	createWriter (const EventStorage::DataReader &reader)
int	main (int argc, char **argv)

Detailed Description

Standalone executable to order and decompress events from multiple files and write them into a single file.

This is intended mainly for merging files from same run and LB but different streams. The reordering ensures uniform distribution of different kinds of events from different streams and removes duplicates. Using this script with files from different LBs is pointless since all events in LB N+1 are always after all events from LB N. The reordering is not strict and the level of out-of-order events is tied to the buffer size. The smaller the buffer, the larger the chance of finding lower-ID events still to be read from input files.

Definition in file trigbs_orderedMerge.cxx.

Function Documentation

createWriter()

std::unique_ptr< EventStorage::DataWriter > createWriter

(

const EventStorage::DataReader &

reader

)

Definition at line 167 of file trigbs_orderedMerge.cxx.

                                                                                         {
  EventStorage::run_parameters_record rp{};
  rp.run_number = reader.runNumber();
  rp.max_events = 0;
  rp.rec_enable = 0;
  rp.trigger_type = reader.triggerType();
  constexpr static std::bitset<128> bitMask64{0xffffffffffffffff};
  rp.detector_mask_LS = (reader.detectorMask() & bitMask64).to_ulong();
  rp.detector_mask_MS = ((reader.detectorMask() >> 64) & bitMask64).to_ulong();
  rp.beam_type = reader.beamType();
  rp.beam_energy = reader.beamEnergy();
 
  return std::make_unique<EventStorage::DataWriter>(
    /* writingPath  = */ ".",
    /* fileNameCore = */ "orderedMerge",
    /* rPar         = */ rp,
    /* fmdStrings   = */ reader.freeMetaDataStrings(),
    /* startIndex   = */ 1,
    /* compression  = */ EventStorage::CompressionType::NONE,
    /* compLevel    = */ 0
  );
}

main()

int main	(	int	argc,
		char **	argv )

Definition at line 190 of file trigbs_orderedMerge.cxx.

                                {
 
  if (argc < 3) {
    std::cout << "Usage: " << argv[0] << " FILE FILE [FILE]" << std::endl
              << "At least 2 input files are required" << std::endl;
    return 1;
  }
 
  std::vector<Buffer> buffers;
  size_t total_events{0}, read_events{0};
  std::unordered_set<uint64_t> written_event_ids;
  uint64_t max_event_id{0};
 
  std::cout << "Reading " << argc-1 << " files:" << std::endl;
  for (int i=1; i<argc; ++i) {
    std::cout << " - " << argv[i] << std::endl;
    buffers.emplace_back(std::unique_ptr<EventStorage::DataReader>(pickDataReader(argv[i])), s_defaultBufferSize);
    total_events += buffers.back().eventsInFile();
  }
 
  std::cout << "Input files contain " << total_events << " events in total" << std::endl;
 
  std::unique_ptr<EventStorage::DataWriter> writer = createWriter(buffers[0].reader());
 
  while (read_events < total_events) {
    std::optional<std::reference_wrapper<Buffer>> next_buffer;
    uint64_t lowest_event_id{std::numeric_limits<uint64_t>::max()};
    int buffer_id{-1}, next_buffer_id{-1};
    for (Buffer& buffer : buffers) {
      ++buffer_id;
      std::optional<std::reference_wrapper<const Event>> event = buffer.peek();
      if (not event.has_value()) {continue;}
      const uint64_t event_id{event.value().get().frag.global_id()};
      if (event_id < lowest_event_id) {
        lowest_event_id = event_id;
        next_buffer = buffer;
        if (s_debugLogging) {
          next_buffer_id = buffer_id;
        }
      }
    }
    if (not next_buffer.has_value()) {
      std::cout << "End of inputs reached, read " << read_events << " out of " << total_events << " events" << std::endl;
      break;
    }
 
    const Event event = next_buffer.value().get().next();
    ++read_events;
    const uint64_t event_id{event.frag.global_id()};
    if (not written_event_ids.insert(event_id).second) {
      std::cout << "Duplicate event with global_id " << event_id << ", skipping" << std::endl;
      continue;
    }
    if (event_id > max_event_id) {
      max_event_id = event_id;
    } else if (s_debugLogging) {
      std::cout << "The current event global_id " << event_id << " is lower than previously written event "
                << max_event_id << " - event ordering not strictly preserved" << std::endl;
    }
 
    if (s_debugLogging) {
      std::cout << "Writing event " << written_event_ids.size() << " with global_id " << event_id
                << " read from file " << next_buffer_id << std::endl;
    } else if (read_events % s_printInterval == 0) {
      std::cout << "Read " << read_events << " out of " << total_events << " events and wrote "
                << written_event_ids.size() << " events" << std::endl;
    }
 
    const WriteEvent writeEvent{event};
    if (writeEvent.status() != WriteEvent::Status::OK) {
      std::cerr << "Error in event serialisation" << std::endl;
      return 1;
    }
 
    const uint32_t sizeInBytes = writeEvent.size() * sizeof(uint32_t);
    uint32_t writeSizeOnDisk{0};
    const EventStorage::DWError err_code = writer->putData(sizeInBytes, writeEvent.blob().get(), writeSizeOnDisk);
    if (err_code != EventStorage::DWError::DWOK or not writer->good()) {
      std::cerr << "Error writing to file, exiting..." << std::endl;
      return 1;
    }
    if (writeSizeOnDisk != sizeInBytes) {
      std::cerr << "Error writing to file, wrote " << writeSizeOnDisk << " instead of " << sizeInBytes << " words" << std::endl;
      return 1;
    }
  }
 
  std::cout << "Merging successful, read " << read_events << " out of " << total_events << " events and wrote "
            << written_event_ids.size() << " events" << std::endl;
 
  return 0;
}