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 = 500000;

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)
{
  if (m_max_dump_size && (!datagram->hasValidTimeStamp())) {
    if (!m_dumpfile.write(datagram, m_detector_id)) {
      static unsigned int n = 0;
      const unsigned int n_iter = 10000;
      if (n % n_iter == 0) {
        LOG_DEBUG
            << "Time-unreliable data detected. Dumped to file. "
            << "This message has a suppression factor of " << n_iter;
        n = 1;
      }
    }
    Log::Counter::get().add("dumped dgrams");
    InBufferCollector::getCollector().putDataGram(datagram);
    return false;
  }

  CLBCommonHeader const*const header = datagram->getCLBCommonHeader();

  int32_t const seq_num = seq_number(
      *header,
      m_start_run_ms,
      m_timeslice_duration);

  if (seq_num < 0) {
    LOG_DEBUG << "Discarding data which timestamp is before the run start time. "
              << "DOM ID: "  << header->domIdentifier()           << ", "
              << "seq_num: " << seq_num                           << ", "
              << "run start time in ms: " << m_start_run_ms       << ", "
              << "time stamp secs: " << header->timeStamp().sec() << ", "
              << "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);
    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;
}