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

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

private

Definition at line 25 of file frame_farm.hh.

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

private

Definition at line 27 of file frame_farm.hh.

Constructor & Destructor Documentation

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

(

)

Definition at line 17 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

void FrameFarm::operator()

(

)

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

Definition at line 57 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);
  }
}

void FrameFarm::stop ( )

inline

Definition at line 72 of file frame_farm.hh.

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

unsigned int FrameFarm::getTimeSliceDuration ( )

inline

Definition at line 78 of file frame_farm.hh.

  {
    return m_timeslice_duration;
  }

bool FrameFarm::insert

(

CLBDataGram *

datagram

)

Definition at line 100 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) {
    if((m_start_run_ms - header->timeStamp().inMilliSeconds())>=error_tf)
        LOG_ERROR << "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;

        else 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;
}

void FrameFarm::getFrame

(

Frame &

frame

)

Definition at line 36 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);
    }
  }
}

int FrameFarm::runNumber ( ) const

inline

Definition at line 87 of file frame_farm.hh.

  {
    return m_run_number;
  }

int FrameFarm::runNumber ( int  rn )

inline

Definition at line 92 of file frame_farm.hh.

  {
    return m_run_number = rn;
  }

unsigned int FrameFarm::detectorId ( unsigned int  detector_id )

inline

Definition at line 97 of file frame_farm.hh.

  {
    return m_detector_id = detector_id;
  }

PuzzledFrame * FrameFarm::getEmptyPuzzledFrame ( )

private

Definition at line 200 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

container_t FrameFarm::m_container

private

Definition at line 29 of file frame_farm.hh.

ptrDispatcher FrameFarm::m_frames

private

Definition at line 31 of file frame_farm.hh.

buffers_collector_t FrameFarm::m_empty_buffers

private

Definition at line 33 of file frame_farm.hh.

boost::mutex FrameFarm::m_mutex

private

Definition at line 35 of file frame_farm.hh.

boost::mutex FrameFarm::m_ebf_mutex

private

Definition at line 37 of file frame_farm.hh.

boost::atomic<bool> FrameFarm::m_running

private

Definition at line 39 of file frame_farm.hh.

unsigned int FrameFarm::m_timeslice_duration

private

Definition at line 41 of file frame_farm.hh.

uint64_t FrameFarm::m_start_run_ms

private

Definition at line 43 of file frame_farm.hh.

std::size_t FrameFarm::m_max_dump_size

private

Definition at line 45 of file frame_farm.hh.

int FrameFarm::m_run_number

private

Definition at line 47 of file frame_farm.hh.

unsigned int FrameFarm::m_detector_id

private

Definition at line 48 of file frame_farm.hh.

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