FullHistoryMerger.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 ALICEO2_FULLHISTORYMERGER_H
#define ALICEO2_FULLHISTORYMERGER_H
 
 
#include "Mergers/MergerConfig.h"
#include "Mergers/ObjectStore.h"
 
#include <Framework/Task.h>
 
namespace o2::monitoring
{
class Monitoring;
}
 
namespace o2::mergers
{
 
class FullHistoryMerger : public framework::Task
{
 public:
  FullHistoryMerger(const MergerConfig&, const header::DataHeader::SubSpecificationType&);
  ~FullHistoryMerger() override;
 
  void init(framework::InitContext& ctx) override;
  void run(framework::ProcessingContext& ctx) override;
 
  void endOfStream(framework::EndOfStreamContext& eosContext) override;
 
 private:
  header::DataHeader::SubSpecificationType mSubSpec;
 
  ObjectStore mMergedObject = std::monostate{};
  std::pair<std::string, framework::DataRef> mFirstObjectSerialized;
  std::unordered_map<std::string, ObjectStore> mCache;
 
  MergerConfig mConfig;
  std::unique_ptr<monitoring::Monitoring> mCollector;
  int mCyclesSinceReset = 0;
 
  // stats
  int mTotalObjectsMerged = 0;
  int mObjectsMerged = 0;
  int mTotalUpdatesReceived = 0;
  int mUpdatesReceived = 0;
 
 private:
  void updateCache(const framework::DataRef& ref);
  void mergeCache();
  void publish(framework::DataAllocator& allocator);
  void clear();
  bool shouldFinishCycle(const framework::InputRecord& inputs) const;
};
 
} // namespace o2::mergers
 
#endif //ALICEO2_FULLHISTORYMERGER_H