ZMQEventProcessor.h

/**************************************************************************
 * basf2 (Belle II Analysis Software Framework)                           *
 * Author: The Belle II Collaboration                                     *
 *                                                                        *
 * See git log for contributors and copyright holders.                    *
 * This file is licensed under LGPL-3.0, see LICENSE.md.                  *
 **************************************************************************/
#pragma once
 
#include <framework/core/Module.h>
#include <framework/core/EventProcessor.h>
#include <framework/core/Path.h>
#include <framework/pcore/ProcessMonitor.h>
 
namespace Belle2 {
  class ZMQEventProcessor : public EventProcessor {
  public:
 
    ZMQEventProcessor();
 
    virtual ~ZMQEventProcessor();
 
    void process(const PathPtr& spath, long maxEvent);
 
    void cleanup();
 
  private:
    void initialize(const ModulePtrList& moduleList, const ModulePtr& histogramManager);
 
    void forkAndRun(long maxEvent, const PathPtr& inputPath, const PathPtr& mainPath, const PathPtr& outputPath,
                    const ModulePtrList& terminateGlobally);
 
    void terminateAndCleanup(const ModulePtr& histogramManager);
 
    void runMonitoring(const PathPtr& inputPath, const PathPtr& mainPath, const ModulePtrList& terminateGlobally, long maxEvent);
 
    void runInput(const PathPtr& inputPath, const ModulePtrList& terminateGlobally, long maxEvent);
 
    void runOutput(const PathPtr& outputPath, const ModulePtrList& terminateGlobally, long maxEvent);
 
    void runWorker(unsigned int numProcesses, const PathPtr& inputPath, const PathPtr& mainPath, const ModulePtrList& terminateGlobally,
                   long maxEvent);
 
    void processPath(const PathPtr& localPath, const ModulePtrList& terminateGlobally, long maxEvent);
 
    void processCore(const PathPtr& startPath, const ModulePtrList& modulePathList, long maxEvent = 0, bool isInputProcess = true,
                     bool isWorkerProcess = false, bool isOutputProcess = false);
 
    bool processEvent(PathIterator moduleIter, bool skipMasterModule, bool Worker = false, bool output = false);
 
    void processBeginRun(bool skipDB = false);
 
    void processEndRun();
 
    ProcessMonitor m_processMonitor;
 
    EventMetaData m_previousEventMetaData;
 
  };
 
}