FrameFarm Class Reference

#include <frame_farm.hh>

Public Member Functions
	FrameFarm (unsigned int DeltaTS, uint64_t run_start_time, std::size_t max_dump_size, const std::string &dump_prefix, const std::string &dump_postfix)
	~FrameFarm ()
void	operator() ()
void	stop ()
unsigned int	getTimeSliceDuration ()
bool	insert (CLBDataGram *datagram)
void	getFrame (Frame &frame)
int	runNumber () const
int	runNumber (int rn)
unsigned int	detectorId (unsigned int detector_id)

Private Types
typedef std::map< frame_idx_t, PuzzledFrame * >	container_t
typedef std::deque< PuzzledFrame * >	buffers_collector_t

Private Member Functions
PuzzledFrame *	getEmptyPuzzledFrame ()

Private Attributes
container_t	m_container
ptrDispatcher	m_frames
buffers_collector_t	m_empty_buffers
boost::mutex	m_mutex
boost::mutex	m_ebf_mutex
boost::atomic< bool >	m_running
unsigned int	m_timeslice_duration
uint64_t	m_start_run_ms
std::size_t	m_max_dump_size
int	m_run_number
unsigned int	m_detector_id
DumpFile	m_dumpfile

Detailed Description

Author

cpellegrino

Definition at line 23 of file frame_farm.hh.

Member Typedef Documentation

container_t

typedef std::map<frame_idx_t, PuzzledFrame*> FrameFarm::container_t

private

Definition at line 25 of file frame_farm.hh.

buffers_collector_t

typedef std::deque<PuzzledFrame*> FrameFarm::buffers_collector_t

private

Definition at line 27 of file frame_farm.hh.

Constructor & Destructor Documentation

FrameFarm()

FrameFarm::FrameFarm ( unsigned int  DeltaTS, uint64_t  run_start_time, std::size_t  max_dump_size, const std::string &  dump_prefix, const std::string &  dump_postfix  )

inline

Definition at line 54 of file frame_farm.hh.

    :
    m_timeslice_duration(DeltaTS),
    m_start_run_ms(run_start_time * 1000),
    m_max_dump_size(max_dump_size),
    m_run_number(0),
    m_detector_id(0),
    m_dumpfile(FilenameGenerator(dump_prefix, dump_postfix), m_max_dump_size)
  {}

~FrameFarm()

FrameFarm::~FrameFarm

(

)

Definition at line 15 of file frame_farm.cc.

{
  for (
      container_t::iterator it = m_container.begin(), et = m_container.end();
      it != et;
      ++it
  ) {
    delete it->second;
  }
 
  for (
      buffers_collector_t::iterator it = m_empty_buffers.begin(), et = m_empty_buffers.end();
      it != et;
      ++it
  ) {
    delete *it;
  }
}

Member Function Documentation

operator()()

void FrameFarm::operator()

(

)

This method runs on an independent thread the polling of completed or expired frames.

Definition at line 55 of file frame_farm.cc.

{
  /** This method runs on an independent thread the polling of
   *  completed or expired frames.
   *
   *
   **/
 
  m_running = true;
 
  while (m_running.load()) {
    {
      boost::mutex::scoped_lock lock(m_mutex);
 
      container_t::iterator it = m_container.begin();
 
      container_t::iterator et = m_container.end();
 
      m_frames.lock();
 
      Timer const t(0, 0);
 
      while (it != et) {
        PuzzledFrame* pframe = it->second;
        if (pframe->hasExpiredWrt(t)) {
          // Signal the completion!
          m_container.erase(it++);
          et = m_container.end();
 
          benchmark::chrono_set(*pframe);
 
          m_frames.put_nolock(pframe);
        } else {
          ++it;
        }
      }
      m_frames.signal();
      m_frames.unlock();
    }
    usleep(sc_sleep);
  }
}

stop()

void FrameFarm::stop ( )

inline

Definition at line 72 of file frame_farm.hh.

  {
    m_running = false;
    m_frames.close();
  }

getTimeSliceDuration()

unsigned int FrameFarm::getTimeSliceDuration ( )

inline

Definition at line 78 of file frame_farm.hh.

  {
    return m_timeslice_duration;
  }

insert()

bool FrameFarm::insert

(

CLBDataGram *

datagram

)

Definition at line 98 of file frame_farm.cc.

{
  CLBCommonHeader const*const header = datagram->getCLBCommonHeader();
 
  if (!datagram->hasValidTimeStamp()) {
    LOG_DEBUG
        << "Time-unreliable datagram detected: "
        << "DOM ID: "          << header->domIdentifier()    << ", "
        << "time stamp secs: " << header->timeStamp().sec()  << ", "
        << "time stamp tics: " << header->timeStamp().tics() << ", "
        << (m_max_dump_size ? "dumping to file" : "not dumping to file");
 
    if (m_max_dump_size) {
      bool const ok = m_dumpfile.write(datagram, m_detector_id);
      if (!ok) {
        LOG_ERROR << "Impossible to write time-unreliable data to dump file";
      }
    }
 
    Log::Counter::get().add("dumped dgrams");
    InBufferCollector::getCollector().putDataGram(datagram);
    return false;
  }
 
  std::pair<uint32_t, bool> const seq_num = seq_number(
      *header,
      m_start_run_ms,
      m_timeslice_duration);
 
  if (!seq_num.second) {
    LOG_DEBUG << "Discarding data which timestamp is before the run start time. "
              << "DOM ID: "  << header->domIdentifier()            << ", "
              << "seq_num: " << seq_num.first                      << ", "
              << "run start time in ms: " << m_start_run_ms        << ", "
              << "time stamp secs: " << header->timeStamp().sec()  << ", "
              << "time stamp tics: " << header->timeStamp().tics() << ", "
              << "timeslice duration in ms: " << m_timeslice_duration;
 
    Log::Counter::get().add("old data");
    InBufferCollector::getCollector().putDataGram(datagram);
    return false;
  }
 
  if (
      m_run_number != -1
      && header->runNumber() != static_cast<unsigned int>(m_run_number)
  ) {
    LOG_DEBUG << "Discarding data which run number ("
              << header->runNumber()
              << ") is different from the one of the current run: "
              << runNumber();
    Log::Counter::get().add("wrong runno");
    InBufferCollector::getCollector().putDataGram(datagram);
    return false;
  }
 
  frame_idx_t const idx = data2idx(
      *header,
      m_start_run_ms,
      m_timeslice_duration);
 
  bool insert_value;
 
  boost::mutex::scoped_lock lock(m_mutex);
 
  benchmark::chrono_set(*datagram);
 
  container_t::const_iterator const it = m_container.find(idx);
 
  benchmark::chrono_set(*datagram);
 
  if (it != m_container.end()) {
    insert_value = it->second->insert(datagram);
  } else {
    PuzzledFrame* const pframe = getEmptyPuzzledFrame();
 
    pframe->setSeqNumber(seq_num.first);
    pframe->setFrameIndex(idx);
 
    pframe->setDataType(header->dataType());
    pframe->setDetectorId(m_detector_id);
    insert_value = pframe->insert(datagram);
 
    container_t::value_type const pair(idx, pframe);
 
    m_container.insert(pair);
  }
 
  return insert_value;
}

getFrame()

void FrameFarm::getFrame

(

Frame &

frame

)

Definition at line 34 of file frame_farm.cc.

{
  frame.clear();
  // Block until data is coming
  PuzzledFrame* pframe = static_cast<PuzzledFrame*>(m_frames.get());
 
  if (pframe) {
    pframe->getFrame(frame);
 
    pframe->reset();
 
    boost::mutex::scoped_lock lock(m_ebf_mutex);
 
    if (m_empty_buffers.size() > 1000) {
      delete pframe;
    } else {
      m_empty_buffers.push_back(pframe);
    }
  }
}

runNumber() [1/2]

int FrameFarm::runNumber ( ) const

inline

Definition at line 87 of file frame_farm.hh.

  {
    return m_run_number;
  }

runNumber() [2/2]

int FrameFarm::runNumber ( int  rn )

inline

Definition at line 92 of file frame_farm.hh.

  {
    return m_run_number = rn;
  }

detectorId()

unsigned int FrameFarm::detectorId ( unsigned int  detector_id )

inline

Definition at line 97 of file frame_farm.hh.

  {
    return m_detector_id = detector_id;
  }

getEmptyPuzzledFrame()

PuzzledFrame * FrameFarm::getEmptyPuzzledFrame ( )

private

Definition at line 189 of file frame_farm.cc.

{
  PuzzledFrame* pframe;
 
  boost::mutex::scoped_lock lock(m_ebf_mutex);
 
  if (m_empty_buffers.size()) {
    pframe = m_empty_buffers.front();
 
    m_empty_buffers.pop_front();
    lock.unlock();
  } else {
    lock.unlock();
    pframe = new PuzzledFrame;
  }
 
  pframe->setTimer(Timer(sc_expsec, sc_expusec));
 
  return pframe;
}

Member Data Documentation

m_container

container_t FrameFarm::m_container

private

Definition at line 29 of file frame_farm.hh.

m_frames

ptrDispatcher FrameFarm::m_frames

private

Definition at line 31 of file frame_farm.hh.

m_empty_buffers

buffers_collector_t FrameFarm::m_empty_buffers

private

Definition at line 33 of file frame_farm.hh.

m_mutex

boost::mutex FrameFarm::m_mutex

private

Definition at line 35 of file frame_farm.hh.

m_ebf_mutex

boost::mutex FrameFarm::m_ebf_mutex

private

Definition at line 37 of file frame_farm.hh.

m_running

boost::atomic<bool> FrameFarm::m_running

private

Definition at line 39 of file frame_farm.hh.

m_timeslice_duration

unsigned int FrameFarm::m_timeslice_duration

private

Definition at line 41 of file frame_farm.hh.

m_start_run_ms

uint64_t FrameFarm::m_start_run_ms

private

Definition at line 43 of file frame_farm.hh.

m_max_dump_size

std::size_t FrameFarm::m_max_dump_size

private

Definition at line 45 of file frame_farm.hh.

m_run_number

int FrameFarm::m_run_number

private

Definition at line 47 of file frame_farm.hh.

m_detector_id

unsigned int FrameFarm::m_detector_id

private

Definition at line 48 of file frame_farm.hh.

m_dumpfile

DumpFile FrameFarm::m_dumpfile

private

Definition at line 50 of file frame_farm.hh.

---

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

• frame_farm.hh
• frame_farm.cc