DFEF::EFInterfaceEmulator Class Reference

#include <EFInterfaceEmulator.h>

Inheritance diagram for DFEF::EFInterfaceEmulator:

Collaboration diagram for DFEF::EFInterfaceEmulator:

Public Member Functions
	EFInterfaceEmulator (const boost::property_tree::ptree &args)
virtual void	open () override
	Opens the Session.
virtual void	close () override
	Closes the Session.
virtual std::future< std::unique_ptr< uint32_t[]> >	getNext () override
	Returns a pointer to the next Event object to be processed.
virtual std::future< void >	accept (uint64_t l0id, std::unique_ptr< uint32_t[]> hltr) override
	Marks the event as accepted by the High-Level Trigger.
virtual std::future< void >	reject (uint64_t l0id) override
	Marks the event as rejected by the High-Level Trigger.
virtual	~EFInterfaceEmulator ()

Private Member Functions
void	inputThreadCallback ()
	Compression level of built event.
void	outputThreadCallback ()
	The method executed by the output handling thread.

Private Attributes
std::string	m_name
std::unique_ptr< FileReaderWriter >	m_file_rw
eformat::Compression	m_comp
unsigned int	m_compLevel
	Compression type of built event.
std::unique_ptr< HLT::LoopThread >	m_inputThread
	Input handling thread (triggers reading new events)
std::unique_ptr< HLT::LoopThread >	m_outputThread
	Output handling thread (triggers post-processing of finished events)
std::queue< std::promise< std::unique_ptr< uint32_t[]> > >	m_inputQueue
	queue of promises for getNext() calls
std::mutex	m_iQMutex
	queue mutex
std::queue< std::pair< uint64_t, std::unique_ptr< uint32_t[]> > >	m_outputQueue
	queue of promises for accept() calls
std::mutex	m_oQMutex
	queue mutex
std::unordered_map< uint64_t, std::vector< std::unique_ptr< uint32_t[]> > >	m_events
	Map of events read from the input (stores duplicates per L1ID)
std::mutex	m_mMutex
	Map mutex.
std::mutex	m_RWMutex
	ReadWrite mutex for the file reader/writer.

Detailed Description

Definition at line 39 of file EFInterfaceEmulator.h.

Constructor & Destructor Documentation

EFInterfaceEmulator()

EFInterfaceEmulator::EFInterfaceEmulator

(

const boost::property_tree::ptree &

args

)

Definition at line 15 of file EFInterfaceEmulator.cxx.

{
  ERS_DEBUG(2, "Initializing EFInterfaceEmulator");
  try {
    m_name = cargs.get<std::string>("name");
  } 
  catch (const boost::property_tree::ptree_error &e) {
    throw std::runtime_error("Failed to get 'name' from configuration: " + std::string(e.what()));
  }
  {
    std::lock_guard<std::mutex> lock(m_RWMutex);
    m_file_rw = std::make_unique<FileReaderWriter>(cargs);
  }
  // TODO: read ptree to get the compression type and level
  m_comp = eformat::UNCOMPRESSED;
  m_compLevel = 0;
}

~EFInterfaceEmulator()

EFInterfaceEmulator::~EFInterfaceEmulator ( )

virtual

Definition at line 96 of file EFInterfaceEmulator.cxx.

{}

Member Function Documentation

accept()

std::future< void > EFInterfaceEmulator::accept ( uint64_t l0id, std::unique_ptr< uint32_t[]> hltr )

overridevirtual

Marks the event as accepted by the High-Level Trigger.

Definition at line 70 of file EFInterfaceEmulator.cxx.

                                                                                             {
  // TODO: here I have to keep track of the events that I read. When event done is received, I have to assemble the full event and write it to the output.
  std::promise<void> p;
  // Create a pair, push the pair in the inputThreadQueue, and notify the input thread
  std::pair<uint64_t, std::unique_ptr<uint32_t[]>> eventPair(l0id, std::move(hltr));
  {
    std::lock_guard<std::mutex> lock(m_oQMutex);
    m_outputQueue.push(std::move(eventPair));
  } 
  m_outputThread->cond().notify_one();
  p.set_value();
  return p.get_future();
}

close()

void EFInterfaceEmulator::close ( )

overridevirtual

Closes the Session.

Definition at line 43 of file EFInterfaceEmulator.cxx.

                                   {
  ERS_DEBUG(2, "Closing EFInterfaceEmulator");
  m_inputThread->stop();
  m_outputThread->stop();
  ERS_DEBUG(2, "Threads stopped, waiting for them to finish"); 
  m_inputThread->wait();
  m_outputThread->wait();
  ERS_DEBUG(2, "EFInterfaceEmulator closed and threads stopped");
  {
    std::lock_guard<std::mutex> lock(m_RWMutex);
    m_file_rw.reset();
  }
}

getNext()

std::future< std::unique_ptr< uint32_t[]> > EFInterfaceEmulator::getNext ( )

overridevirtual

Returns a pointer to the next Event object to be processed.

Definition at line 57 of file EFInterfaceEmulator.cxx.

                                                                      {
  ERS_DEBUG(2,  "EFInterfaceEmulator getNext called");
  std::promise<std::unique_ptr<uint32_t[]>> p;
  auto fut = p.get_future();
  {
    std::lock_guard<std::mutex> lock(m_iQMutex);
    m_inputQueue.push(std::move(p));
  }
  ERS_DEBUG(2,  "Notifying input thread");
  m_inputThread->cond().notify_one(); 
  return fut;
}

inputThreadCallback()

void EFInterfaceEmulator::inputThreadCallback ( )

private

Compression level of built event.

The method executed by the input handling thread

Definition at line 98 of file EFInterfaceEmulator.cxx.

                                                  {
  ERS_DEBUG(2, "Input thread started");
  while (true) {
    std::promise<std::unique_ptr<uint32_t[]>> p;
    {
      std::lock_guard<std::mutex> lock(m_iQMutex);
      ERS_DEBUG(2, "m_inputQueue size: " << m_inputQueue.size());
      if (m_inputQueue.empty()) {
        ERS_DEBUG(2, "Input queue is empty, breaking...");
        break; 
      }
      p = std::move(m_inputQueue.front());
      m_inputQueue.pop();
    }
    // Get the event from file, create a copy of it to store in the map, and return the original pointer to the promise
    ERS_DEBUG(2, "Reading next event from file");
    std::unique_ptr<uint32_t[]> event = nullptr;
    {
      std::lock_guard<std::mutex> lock(m_RWMutex);
      try {
        //FIXME: Sleep for random duration, only for debugging
        //auto sleepDuration = (std::rand() % 100) * 10;
        //ERS_DEBUG(2, "Sleeping for " << sleepDuration << " milliseconds");
        //std::this_thread::sleep_for(std::chrono::milliseconds(sleepDuration));
        //coverity[SLEEP]
        event = m_file_rw->getNextEvent();
      } catch (const DFEF::NoMoreEventsInFile &ex) {
        ERS_DEBUG(2, "No more events in file");  
        p.set_exception(std::make_exception_ptr(daq::df_ef_interface::NoMoreEvents("No more events in file")));
        continue;
      } catch (DFEF::ProblemReadingFromFile &ex) {
        ERS_DEBUG(2, "Problem reading from file: " << ex.what());
        p.set_exception(std::make_exception_ptr(daq::df_ef_interface::CommunicationError("Problem reading from file")));
        continue;
      }
    }
    ERS_DEBUG(2, "Event read");
    eformat::read::FullEventFragment fullEvent(event.get());
    uint32_t nwords = event[1];
    ERS_DEBUG(2, "Copying full event with L0ID: " << fullEvent.lvl1_id() << ", size: " << nwords * sizeof(uint32_t) << " bytes");
    std::unique_ptr<uint32_t[]> eventCopy(new uint32_t[nwords]);
    std::memcpy(eventCopy.get(), event.get(), nwords * sizeof(uint32_t));
    {
      std::lock_guard<std::mutex> lock(m_mMutex);
      auto& bucket = m_events[fullEvent.lvl1_id()];
      bucket.push_back(std::move(eventCopy));
    }
    p.set_value(std::move(event));
  }
}

open()

void EFInterfaceEmulator::open ( )

overridevirtual

Opens the Session.

Definition at line 33 of file EFInterfaceEmulator.cxx.

                                  {
  //Starting input and output threads
    ERS_DEBUG(2, "Starting input and output threads");
    m_inputThread = std::make_unique<HLT::LoopThread>([this]{return inputThreadCallback();},-1);
    m_outputThread = std::make_unique<HLT::LoopThread>([this]{return outputThreadCallback();}, -1);
    m_inputThread->start();
    m_outputThread->start();
    ERS_DEBUG(2, "Input and output threads started");
}

outputThreadCallback()

void EFInterfaceEmulator::outputThreadCallback ( )

private

The method executed by the output handling thread.

Definition at line 149 of file EFInterfaceEmulator.cxx.

                                                   {
  ERS_DEBUG(2, "Output thread started");
  while (true) {
    std::pair<uint64_t, std::unique_ptr<uint32_t[]>> eventPair;
    {
      std::lock_guard<std::mutex> lock(m_oQMutex);
      ERS_DEBUG(2, "m_outputQueue size: " << m_outputQueue.size());
      if (m_outputQueue.empty())
      {
        ERS_DEBUG(2, "Output queue is empty, breaking...");
        break; 
      }
      eventPair = std::move(m_outputQueue.front());
      m_outputQueue.pop();
    }
    std::unique_ptr<uint32_t[]> fullEvent = nullptr;
    {
      std::lock_guard<std::mutex> lock(m_mMutex);
      auto it = m_events.find(eventPair.first);
      if (it != m_events.end() && !it->second.empty()) {
        fullEvent = std::move(it->second.back());
        it->second.pop_back();
        if (it->second.empty()) {
          m_events.erase(it);
        }
      } else {
       throw std::runtime_error("Missing the FullEvent copy for event accepted with L0ID: " + std::to_string(eventPair.first)); 
      }
    }
    //FIXME Remove this, just for debugging
    std::unique_ptr<uint32_t[]> hltResult = std::move(eventPair.second);
    uint32_t nwords = hltResult[1];
    eformat::read::FullEventFragment fullEventFragment(fullEvent.get());
    ERS_DEBUG(2, "Processing HLTResult with L0ID: " << fullEventFragment.lvl1_id() << ", size: " << nwords * 4 << " bytes");
    // Merge HLTResult
    auto finalEvent = DFEF::merge_hltresult_with_input_event(hltResult.get(), fullEvent.get(), m_comp, m_compLevel);
    auto finalSize = finalEvent.size();
    auto sizeInBytes = finalSize*4; 
    {
      ERS_DEBUG(2, "Writing fullEventFragment with L0ID: " << eventPair.first << " of size: " << sizeInBytes << " bytes");
      std::lock_guard<std::mutex> lock(m_RWMutex);
      m_file_rw->writeEvent(sizeInBytes, finalEvent.data());
    }
    ERS_DEBUG(2, "Event with L0ID: " << eventPair.first << " written to file");
    ERS_DEBUG(2, "m_events size after erase: " << m_events.size());
  }
  ERS_DEBUG(2, "Output thread finished processing.");
}

reject()

std::future< void > EFInterfaceEmulator::reject ( uint64_t l0id )

overridevirtual

Marks the event as rejected by the High-Level Trigger.

Definition at line 84 of file EFInterfaceEmulator.cxx.

                                                            {
    ERS_DEBUG(2, "Rejecting event with L1ID: " << l0id);
    ERS_DEBUG(2, "m_events size before erase: " << m_events.size());
    {
      std::lock_guard<std::mutex> lock(m_mMutex);
      m_events.erase(l0id);
    }
    std::promise<void> p;
    p.set_value();
    return p.get_future(); 
}

Member Data Documentation

m_comp

eformat::Compression DFEF::EFInterfaceEmulator::m_comp

private

Definition at line 72 of file EFInterfaceEmulator.h.

m_compLevel

unsigned int DFEF::EFInterfaceEmulator::m_compLevel

private

Compression type of built event.

Definition at line 73 of file EFInterfaceEmulator.h.

m_events

std::unordered_map<uint64_t, std::vector<std::unique_ptr<uint32_t[]> > > DFEF::EFInterfaceEmulator::m_events

private

Map of events read from the input (stores duplicates per L1ID)

Definition at line 96 of file EFInterfaceEmulator.h.

m_file_rw

std::unique_ptr<FileReaderWriter> DFEF::EFInterfaceEmulator::m_file_rw

private

Definition at line 71 of file EFInterfaceEmulator.h.

m_inputQueue

std::queue<std::promise<std::unique_ptr<uint32_t[]> > > DFEF::EFInterfaceEmulator::m_inputQueue

private

queue of promises for getNext() calls

Definition at line 86 of file EFInterfaceEmulator.h.

m_inputThread

std::unique_ptr<HLT::LoopThread> DFEF::EFInterfaceEmulator::m_inputThread

private

Input handling thread (triggers reading new events)

Definition at line 81 of file EFInterfaceEmulator.h.

m_iQMutex

std::mutex DFEF::EFInterfaceEmulator::m_iQMutex

private

queue mutex

Definition at line 88 of file EFInterfaceEmulator.h.

m_mMutex

std::mutex DFEF::EFInterfaceEmulator::m_mMutex

private

Map mutex.

Definition at line 98 of file EFInterfaceEmulator.h.

m_name

std::string DFEF::EFInterfaceEmulator::m_name

private

Definition at line 70 of file EFInterfaceEmulator.h.

m_oQMutex

std::mutex DFEF::EFInterfaceEmulator::m_oQMutex

private

queue mutex

Definition at line 93 of file EFInterfaceEmulator.h.

m_outputQueue

std::queue<std::pair<uint64_t, std::unique_ptr<uint32_t[]> > > DFEF::EFInterfaceEmulator::m_outputQueue

private

queue of promises for accept() calls

Definition at line 91 of file EFInterfaceEmulator.h.

m_outputThread

std::unique_ptr<HLT::LoopThread> DFEF::EFInterfaceEmulator::m_outputThread

private

Output handling thread (triggers post-processing of finished events)

Definition at line 83 of file EFInterfaceEmulator.h.

m_RWMutex

std::mutex DFEF::EFInterfaceEmulator::m_RWMutex

private

ReadWrite mutex for the file reader/writer.

Definition at line 101 of file EFInterfaceEmulator.h.

---

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

• EFInterfaceEmulator.h
• EFInterfaceEmulator.cxx