puzzled_frame.cc

Go to the documentation of this file.

#include "puzzled_frame.hh"

#include <DataFormats/clb_common_header.hh>
#include <DataFormats/infoword.hh>

#include "input_buffer_collector.hh"

#include <string.h>

#include <log.hh>

#include <utility> // for std::pair and std::make_pair
#include <stdexcept> // for runtime_error

/**
 * \author cpellegrino
 */

boost::atomic<unsigned int> PuzzledFrame::n_obj(0);

uint32_t PuzzledFrame::frameDomStatus(int i) const
{
  uint32_t dom_status = 0;
  for (
      container_t::const_iterator it = m_container.begin(), et = m_container.end();
      it != et;
      ++it
  ) {
    const CLBCommonHeader* const clb_header = it->second->getCLBCommonHeader();
    dom_status = dom_status | clb_header->domStatus(i);
  }
  return dom_status;
}

uint32_t calculate_number_of_items(Frame const& frame)
{
  uint32_t nitems = 0;
  uint32_t const frame_length = frame.size();

  uint32_t const data_type = frame.getHeader()->DataType;

  if (is_optical(data_type)) {
    nitems = (frame_length - sizeof(DAQCommonHeader)) / 6; // 6 is sizeof(hit)
  } else if (is_acoustic(data_type)) {
    InfoWord const*const iw = static_cast<const InfoWord*>(
        static_cast<const void*>(frame.getPayload()));

    const unsigned int audiowordsize = iw->audioWordSize();

    nitems = (frame_length
        - sizeof(DAQCommonHeader)
        - sizeof(InfoWord))
        / audiowordsize;
  } else {
    throw std::runtime_error("Unknown data type");
  }

  return nitems;
}

bool PuzzledFrame::insert(CLBDataGram* datagram)
{
  //Test collision: is m_container[idx] empty?
  unsigned int const idx = datagram->getUDPSequenceNumber();

  container_t::const_iterator const it = m_container.find(idx);

  if (it != m_container.end()) {
    LOG_DEBUG << "Datagram collision. idx: " << idx << '\n'
              << "container.size(): " << m_container.size() << '\n'
              << "INFO:\n" << *datagram->getCLBCommonHeader();

    Log::Counter::get().add("collision");
    InBufferCollector::getCollector().putDataGram(datagram);
    return false;
  }

  if (!m_totalsize) {
    benchmark::chrono_swap(*datagram, *this);
  }

  m_container.insert(m_container.end(), container_t::value_type(idx, datagram));

  m_totalsize += datagram->size();

  return true;
}

std::pair<unsigned int, unsigned int> opt_boundaries(
    CLBDataGram const& dg,
    std::size_t max_size)
{
  std::size_t const previous_packets_size =
      dg.getCLBCommonHeader()->udpSequenceNumber()
      * (max_size - sizeof(CLBCommonHeader));

  std::size_t const chunk_in_previous = previous_packets_size % 6;

  unsigned int const head =
      chunk_in_previous
        ? 6 - chunk_in_previous
        : 0;

  unsigned int const tail =
      (dg.size() - sizeof(CLBCommonHeader) - head) % 6;

  return std::make_pair(head, tail);
}

// Preconditions:
// - Frame's memory has been already allocated elsewhere.
// - The header has been already written.
// - The container has more than 1 datagram, or it contains only the first one
//   UDPSequenceNumber == 0.

void buildOpto(
    PuzzledFrame::container_t const& c,
    frame_idx_t frame_idx,
    Frame& frame)
{
  assert(
      (
          c.size() > 1
          || c.begin()->second->getCLBCommonHeader()->udpSequenceNumber() == 0
      )
      && "Precondition failed."
  );

  uint32_t current_seq_number = 0;
  unsigned int previous_chunk = 0;

  std::size_t const max_size = c.begin()->second->size();

  std::size_t size = sizeof(DAQCommonHeader);

  for (
      PuzzledFrame::container_t::const_iterator it = c.begin(), et = c.end();
      it != et;
      ++it
  ) {

    CLBDataGram const& dg = *it->second;
    std::pair<unsigned int, unsigned int> const bounds = opt_boundaries(
        dg,
        max_size
    );

    if (dg.getCLBCommonHeader()->udpSequenceNumber() != current_seq_number) {
      LOG_DEBUG << "Datagram "
                << current_seq_number
                << " not present in optical frame "
                << frame_idx
                << ". Packet loss(?)";

      Log::Counter::get().add("missing odgram");

      char* const dest = &frame.front() + size - previous_chunk;

      std::size_t const write_size = dg.getPayloadSize() - bounds.first;

      std::memcpy(dest, dg.getPayload() + bounds.first, write_size);

      size += write_size - previous_chunk;
    } else {
      char* const dest = &frame.front() + size;

      std::size_t const write_size = dg.getPayloadSize();

      std::memcpy(dest, dg.getPayload(), write_size);

      size += write_size;
    }

    current_seq_number = dg.getCLBCommonHeader()->udpSequenceNumber() + 1;
    previous_chunk = bounds.second;
  }

  // check whether size is compatible an integer number of items
  // and if not strip away the boundary

  if (previous_chunk != 0) {
    LOG_DEBUG << "Missing last datagram(s) in optical frame "
              << frame_idx
              << ". Packet loss (?).";

    Log::Counter::get().add("missing odgram");
    size -= previous_chunk;
  }

  // check the presence of the trailer

  if (!isTrailer(*c.rbegin()->second->getCLBCommonHeader())) {
    LOG_DEBUG << "Missing trailer in optical frame "
              << frame_idx;

    Log::Counter::get().add("missing trailer");
  }

  frame.resize(size);
}

std::pair<unsigned int, unsigned int> acou_boundaries(
    CLBDataGram const& dg,
    std::size_t max_size,
    std::size_t item_size)
{
  unsigned int head = 0;
  unsigned int tail = 0;

  if (dg.getCLBCommonHeader()->udpSequenceNumber() != 0) {
    std::size_t const previous_audio_size =
        dg.getCLBCommonHeader()->udpSequenceNumber()
        * (max_size - sizeof(CLBCommonHeader))
        - sizeof(InfoWord);

    std::size_t const chunk_in_previous = previous_audio_size % item_size;

    head = chunk_in_previous ? item_size - chunk_in_previous : 0;

    tail = (dg.size() - sizeof(CLBCommonHeader) - head) % item_size;
  } else {
    head = 0;
    tail = (dg.size() - sizeof(CLBCommonHeader) - sizeof(InfoWord)) % item_size;
  }

  return std::make_pair(head, tail);
}

// Preconditions:
// - Frame's memory has been already allocated elsewhere.
// - The header has been already written.
// - The first datagram in the container must have UDPSequenceNumber == 0.

void buildAcou(
    PuzzledFrame::container_t const& c,
    frame_idx_t frame_idx,
    Frame& frame)
{
  uint32_t current_seq_number = 0;
  unsigned int previous_chunk = 0;

  std::size_t const max_size = c.begin()->second->size();

  std::size_t size = sizeof(DAQCommonHeader);

  InfoWord const& iw = *static_cast<const InfoWord*>(
      static_cast<const void*>(c.begin()->second->getPayload()));

  unsigned int const audio_word_size = iw.audioWordSize();

  for (
      PuzzledFrame::container_t::const_iterator it = c.begin(), et = c.end();
      it != et;
      ++it
  ) {

    CLBDataGram const& dg = *it->second;

    std::pair<unsigned int, unsigned int> const bounds = acou_boundaries(
        dg,
        max_size,
        audio_word_size);

    if (dg.getCLBCommonHeader()->udpSequenceNumber() != current_seq_number) {
      LOG_DEBUG << "Datagram "
                << current_seq_number
                << " not present in acoustic frame "
                << frame_idx
                << ". Packet loss(?)";

      Log::Counter::get().add("missing adgram");

      size -= previous_chunk;
      break;
    } else {
      char* const dest = &frame.front() + size;

      std::size_t const write_size = dg.getPayloadSize();

      memcpy(dest, dg.getPayload(), write_size);

      size += write_size;
    }

    ++current_seq_number;
    previous_chunk = bounds.second;
  }

  // check whether size is compatible an integer number of items
  // and if not strip away the boundary

  if (previous_chunk != 0) {
    LOG_DEBUG << "Missing last datagram(s) in acoustic frame "
              << frame_idx
              << ". Packet loss (?).";

    Log::Counter::get().add("missing adgram");
    size -= previous_chunk;
  }

  frame.resize(size);
}

void PuzzledFrame::getFrame(Frame& frame)
{
  benchmark::chrono_set(*this);
  benchmark::chrono_swap(*this, frame);

  CLBCommonHeader const*const clb_header
      = m_container.begin()->second->getCLBCommonHeader();

  int const frame_length = frameLength();

  // test the frame length
  // must be greater than 0
  if (frame_length < 0) {
    LOG_DEBUG << "Anomalous frame size: " << frame_length       << '\n'
              << "Read total size: "      << m_totalsize        << '\n'
              << "container.size(): "     << m_container.size();

    Log::Counter::get().add("wrong size");
    frame.clear();
    return;
  }

  frame.m_seqnumber = m_seqnumber;
  frame.resize(frame_length);

  DAQCommonHeader& daq_header = *static_cast<DAQCommonHeader*>(
      static_cast<void*>(
          &frame.front()));

  // initialise the parts of the DAQCommonHeader that are already known
  // before the assembling of the data.

  daq_header.DataType = clb_header->dataType();

  daq_header.DetectorIdentifier = m_detector_id;

  daq_header.RunNumber = clb_header->runNumber();

  daq_header.SequenceNumber = m_seqnumber;

  daq_header.Timestamp.Sec = clb_header->timeStamp().sec();
  daq_header.Timestamp.Tics = clb_header->timeStamp().tics();

  daq_header.DOMIdentifier = clb_header->domIdentifier();

  daq_header.NumberOfPackets = m_container.size();
  daq_header.HighestPacketNumber
    = m_container.rbegin()
    ->second
    ->getCLBCommonHeader()
    ->udpSequenceNumber();

  daq_header.DOMStatus1 = frameDomStatus(1);
  daq_header.DOMStatus2 = frameDomStatus(2);
  daq_header.DOMStatus3 = 0;
  daq_header.DOMStatus4 = 0;

  if (is_optical(daq_header.DataType)) {
    if (
        !(
            m_container.size() > 1
            || m_container.begin()->second->getCLBCommonHeader()->udpSequenceNumber() == 0
        )
    ) {
      LOG_DEBUG << "Optical frame "
                << m_frame_idx
                << " contains too few data to be assembled. Ignored.";

      Log::Counter::get().add("skipd oframe");
      frame.resize(0);
    } else {
      buildOpto(m_container, m_frame_idx, frame);
    }
  } else if (is_acoustic(daq_header.DataType)) {
    if (m_container.begin()->second->getUDPSequenceNumber() != 0) {
      LOG_DEBUG << "Acoustic frame "
                << m_frame_idx
                << " without infoword. Ignored.";

      Log::Counter::get().add("skipd aframe");
      frame.resize(0);
    } else if (m_container.begin()->second->size() <= sizeof(CLBCommonHeader)) {
      frame.resize(0);
    } else {
      buildAcou(m_container, m_frame_idx, frame);
    }
  } else {
    throw std::runtime_error("Unknown data type");
  }

  daq_header.FrameLength = frame.size();
  daq_header.NumberOfItems = calculate_number_of_items(frame);
}

void PuzzledFrame::reset()
{
  for (
      container_t::const_iterator it = m_container.begin(), et = m_container.end();
      it != et;
      ++it
  )
  {
    InBufferCollector::getCollector().putDataGram(it->second);
  }

  m_container.clear();

  m_totalsize = 0;
  m_seqnumber = 0;
  m_detector_id = 0;
  m_datatype = 0;
  m_frame_idx = 0;
}