DPLRawParser.h

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
#ifndef FRAMEWORK_UTILS_DPLRAWPARSER_H
#define FRAMEWORK_UTILS_DPLRAWPARSER_H
 
 
#include "DPLUtils/RawParser.h"
#include "Framework/InputRecord.h"
#include "Framework/DataRef.h"
#include "Framework/DataRefUtils.h"
#include "Framework/Logger.h"
#include "Framework/DataProcessingHeader.h"
#include "Headers/DataHeader.h"
#include <utility> // std::declval
 
namespace o2::framework
{
 
template <bool BOUNDS_CHECKS = true>
class DPLRawParser
{
 public:
  using rawparser_type = RawParser<8192, BOUNDS_CHECKS>;
  using buffer_type = typename rawparser_type::buffer_type;
 
  DPLRawParser() = delete;
  DPLRawParser(InputRecord& inputs, std::vector<InputSpec> filterSpecs = {}, fair::Severity sev = fair::Severity::alarm) : mInputs(inputs), mFilterSpecs(filterSpecs), mSeverity(sev) {}
 
  void setMaxFailureMessages(size_t n) { mMaxFailureMessages = n; }
  void setExtFailureCounter(size_t* cnt) { mExtFailureCounter = cnt; }
  static void setCheckIncompleteHBF(bool v) { rawparser_type::setCheckIncompleteHBF(v); }
 
  // this is a dummy default buffer used to initialize the RawParser in the iterator
  // constructor
  static constexpr o2::header::RAWDataHeader initializer = o2::header::RAWDataHeader{.memorySize = sizeof(o2::header::RAWDataHeader)};
 
  template <typename T>
  class Iterator
  {
   public:
    using iterator_category = std::forward_iterator_tag;
    using self_type = Iterator;
    using value_type = T;
    using reference = T&;
    using pointer = T*;
    // the iterator over the input channels
    using input_iterator = decltype(std::declval<InputRecord>().begin());
    // the parser type
    using parser_type = rawparser_type const;
 
    Iterator() = delete;
 
    Iterator(InputRecord& parent, input_iterator it, input_iterator end, std::vector<InputSpec> const& filterSpecs, fair::Severity sev = fair::Severity::alarm, size_t maxErrMsg = -1, size_t* cntErrMsg = nullptr)
      : mParent(parent), mInputIterator(it), mEnd(end), mPartIterator(mInputIterator.begin()), mParser(std::make_unique<parser_type>(reinterpret_cast<const char*>(&initializer), sizeof(initializer))), mCurrent(mParser->begin()), mFilterSpecs(filterSpecs), mMaxFailureMessages(maxErrMsg), mExtFailureCounter(cntErrMsg), mSeverity(sev)
    {
      mParser.reset();
      next();
    }
 
    ~Iterator() = default;
 
    // prefix increment
    self_type& operator++()
    {
      next();
      return *this;
    }
    // postfix increment
    self_type operator++(int /*unused*/)
    {
      self_type copy(*this);
      operator++();
      return copy;
    }
    // return reference
    reference operator*()
    {
      return *mCurrent;
    }
    // comparison
    bool operator==(const self_type& other) const
    {
      bool result = mInputIterator == other.mInputIterator;
      result = result && mPartIterator == other.mPartIterator;
      if (mParser != nullptr && other.mParser != nullptr) {
        result = result && mCurrent == other.mCurrent;
      }
      return result;
    }
 
    bool operator!=(const self_type& rh) const
    {
      return not operator==(rh);
    }
 
    o2::header::DataHeader const* o2DataHeader() const
    {
      if (mInputIterator != mEnd) {
        return DataRefUtils::getHeader<o2::header::DataHeader*>(*mPartIterator);
      }
      return nullptr;
    }
 
    o2::framework::DataProcessingHeader const* o2DataProcessingHeader() const
    {
      if (mInputIterator != mEnd) {
        return DataRefUtils::getHeader<o2::framework::DataProcessingHeader*>(*mPartIterator);
      }
      return nullptr;
    }
 
    buffer_type const* raw() const
    {
      return mCurrent.raw();
    }
 
    buffer_type const* data() const
    {
      return mCurrent.data();
    }
 
    size_t offset() const
    {
      return mCurrent.offset();
    }
 
    size_t size() const
    {
      return mCurrent.size();
    }
 
    size_t sizeTotal() const
    {
      return mCurrent.sizeTotal();
    }
 
    template <typename U>
    U const* get_if() const
    {
      return mCurrent.template get_if<U>();
    }
 
    friend std::ostream& operator<<(std::ostream& os, self_type const& it)
    {
      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mInputIterator.end() && it.mParser != nullptr) {
        os << it.mCurrent;
      }
      return os;
    }
 
    // helper wrapper to control the format and content of the stream output
    template <raw_parser::FormatSpec FmtCtrl>
    struct Fmt {
      static constexpr raw_parser::FormatSpec format_control = FmtCtrl;
      Fmt(self_type const& _it) : it{_it} {}
      self_type const& it;
    };
 
    template <raw_parser::FormatSpec FmtCtrl>
    friend std::ostream& operator<<(std::ostream& os, Fmt<FmtCtrl> const& fmt)
    {
      auto const& it = fmt.it;
      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mInputIterator.end() && it.mParser != nullptr) {
        if constexpr (FmtCtrl == raw_parser::FormatSpec::Info) {
          // TODO: need to propagate the format spec also on the RawParser object
          // for now this operation prints the RDH version info and the table header
          os << *it.mParser;
        } else {
          os << it;
        }
      }
      return os;
    }
 
   private:
    // the iterator over the parts in one channel
    using part_iterator = typename input_iterator::const_iterator;
    // the iterator over the over the parser pages
    using parser_iterator = typename parser_type::const_iterator;
 
    bool next()
    {
      auto logFailure = [this](const std::string& msg, const std::runtime_error& e) {
        if (!this->mExtFailureCounter || (*this->mExtFailureCounter)++ < this->mMaxFailureMessages) {
          if (this->mSeverity == fair::Severity::alarm) {
            LOG(alarm) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else if (this->mSeverity == fair::Severity::warn) {
            LOG(warn) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else if (this->mSeverity == fair::Severity::fatal) {
            LOG(fatal) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else if (this->mSeverity == fair::Severity::critical) {
            LOG(critical) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else if (this->mSeverity == fair::Severity::error) {
            LOG(error) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else if (this->mSeverity == fair::Severity::info) {
            LOG(info) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          } else {
            LOG(debug) << msg << (*this->mInputIterator).spec->binding << " : " << e.what();
          }
        }
      };
 
      while (mInputIterator != mEnd) {
        bool isInitial = mParser == nullptr;
        while (mPartIterator != mInputIterator.end()) {
          // first increment on the parser level
          if (mParser && mCurrent != mParser->end() && ++mCurrent != mParser->end()) {
            // we have an active parser and there is still data at the incremented iterator
            return true;
          }
          // now increment on the level of one input
          mParser.reset();
          if (!isInitial && (mPartIterator == mInputIterator.end() || ++mPartIterator == mInputIterator.end())) {
            // no more parts, go to next input
            break;
          }
          isInitial = false;
          // check filter rules
          if (mFilterSpecs.size() > 0) {
            bool isSelected = false;
            for (auto const& spec : mFilterSpecs) {
              if ((isSelected = DataRefUtils::match(*mPartIterator, spec)) == true) {
                break;
              }
            }
            if (!isSelected) {
              continue;
            }
          }
          gsl::span<const char> raw;
          try {
            raw = mParent.get<gsl::span<char>>(*mPartIterator);
          } catch (const std::runtime_error& e) {
            logFailure("failed to read data from ", e);
          }
          if (raw.size() == 0) {
            continue;
          }
 
          try {
            mParser = std::make_unique<parser_type>(raw.data(), raw.size());
          } catch (const std::runtime_error& e) {
            logFailure("can not create raw parser from ", e);
          }
 
          if (mParser != nullptr) {
            mCurrent = mParser->begin();
            return true;
          }
        } // end loop over parts on one input
        ++mInputIterator;
        mPartIterator = mInputIterator.begin();
      } // end loop over inputs
      return false;
    }
 
    InputRecord& mParent;
    input_iterator mInputIterator;
    input_iterator mEnd;
    part_iterator mPartIterator;
    std::unique_ptr<parser_type> mParser;
    parser_iterator mCurrent;
    std::vector<InputSpec> const& mFilterSpecs;
    size_t mMaxFailureMessages = -1;
    size_t* mExtFailureCounter = nullptr; // external optionally provided counter to throttle error messages
    fair::Severity mSeverity = fair::Severity::alarm;
  };
 
  using const_iterator = Iterator<DataRef const>;
 
  const_iterator begin() const
  {
    return const_iterator(mInputs, mInputs.begin(), mInputs.end(), mFilterSpecs, mSeverity, mMaxFailureMessages, mExtFailureCounter);
  }
 
  const_iterator end() const
  {
    return const_iterator(mInputs, mInputs.end(), mInputs.end(), mFilterSpecs, mSeverity, mMaxFailureMessages, mExtFailureCounter);
  }
 
  using RDHInfo = typename o2::framework::DPLRawParser<BOUNDS_CHECKS>::const_iterator::template Fmt<raw_parser::FormatSpec::Info>;
 
 private:
  InputRecord& mInputs;
  std::vector<InputSpec> mFilterSpecs;
  size_t mMaxFailureMessages = -1;
  size_t* mExtFailureCounter = nullptr; // external optionally provided counter to throttle error messages
  fair::Severity mSeverity = fair::Severity::alarm;
};
 
} // namespace o2::framework
 
#endif //FRAMEWORK_UTILS_DPLRAWPARSER_H