frame_farm.cc

Go to the documentation of this file.

#include "frame_farm.hh"
#include "input_buffer_collector.hh"
#include <log.hh>
 
/**
 * \author cpellegrino
 */
 
static const useconds_t sc_sleep   = 100000; // 100000 = 100 ms, 10000 = 10 ms
 
static const int        sc_expsec  =      0;
 
static const int        sc_expusec = 3000000;
 
FrameFarm::~FrameFarm()
{
  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;
  }
}
 
void FrameFarm::getFrame(Frame& frame)
{
  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);
    }
  }
}
 
void FrameFarm::operator ()()
{
  /** 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);
  }
}
 
bool FrameFarm::insert(CLBDataGram* datagram)
{
  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;
}
 
PuzzledFrame* FrameFarm::getEmptyPuzzledFrame()
{
  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;
}