CombinerModule.h

Go to the documentation of this file.

// $Id: CombinerModule.h 4707 2021-02-26 11:29:21Z linev $
 
/************************************************************
 * The Data Acquisition Backbone Core (DABC)                *
 ************************************************************
 * Copyright (C) 2009 -                                     *
 * GSI Helmholtzzentrum fuer Schwerionenforschung GmbH      *
 * Planckstr. 1, 64291 Darmstadt, Germany                   *
 * Contact:  http://dabc.gsi.de                             *
 ************************************************************
 * This software can be used under the GPL license          *
 * agreements as stated in LICENSE.txt file                 *
 * which is part of the distribution.                       *
 ************************************************************/
 
#ifndef HADAQ_CombinerModule
#define HADAQ_CombinerModule
 
#ifndef DABC_ModuleAsync
#include "dabc/ModuleAsync.h"
#endif
 
#ifndef DABC_Profiler
#include "dabc/Profiler.h"
#endif
 
#ifndef HADAQ_HadaqTypeDefs
#include "hadaq/HadaqTypeDefs.h"
#endif
 
#ifndef HADAQ_Iterator
#include "hadaq/Iterator.h"
#endif
 
#define HADAQ_NEVTIDS 64UL             /* must be 2^n */
#define HADAQ_NEVTIDS_IN_FILE 0UL      /* must be 2^n */
#define HADAQ_RINGSIZE 100
 
namespace hadaq {
 
   class TerminalModule;
 
   class CombinerModule: public dabc::ModuleAsync {
 
      friend class TerminalModule;
 
      protected:
 
      struct InputCfg {
         hadaq::RawSubevent *subevnt; 
         hadaq::RawEvent  *evnt; 
         bool     has_data;      
         uint32_t data_size;     
         uint32_t fTrigNr;       
         uint32_t fLastTrigNr;   
         uint32_t fTrigTag;      
         uint32_t fTrigType;     
         unsigned fErrorBitsCnt{0}; 
         unsigned fHubId{0};     
         unsigned fUdpPort{0};    
         float fQueueLevel;      
         uint32_t fLastEvtBuildTrigId; 
         bool fDataError;        
         bool fEmpty;            
         void* fInfo;            
         int fQueueCapacity;     
         int fNumCanRecv;        
         unsigned fLostTrig;     
         unsigned fDroppedTrig;  
         uint32_t  fTrigNumRing[HADAQ_RINGSIZE]; 
         unsigned  fRingCnt;     
         unsigned  fResort;       
         ReadIterator fIter;      
         ReadIterator fResortIter; 
         int          fResortIndx;  
         std::string fCalibr;      
         bool        fCalibrReq;   
         int         fCalibrProgr; 
         std::string fCalibrState; 
         double      fCalibrQuality; 
         uint64_t    fHubLastSize; 
         uint64_t    fHubPrevSize; 
         int         fHubSizeTmCnt; 
 
 
         InputCfg() :
            subevnt(0),
            evnt(0),
            has_data(false),
            data_size(0),
            fTrigNr(0),
            fLastTrigNr(0xffffffff),
            fTrigTag(0),
            fTrigType(0),
            fQueueLevel(0),
            fLastEvtBuildTrigId(0),
            fDataError(false),
            fEmpty(true),
            fInfo(0),
            fQueueCapacity(0),
            fNumCanRecv(0),
            fLostTrig(0),
            fDroppedTrig(0),
            fRingCnt(0),
            fResort(false),
            fIter(),
            fResortIter(),
            fResortIndx(-1),
            fCalibr(),
            fCalibrReq(false),
            fCalibrProgr(0),
            fCalibrState(),
            fCalibrQuality(0.),
            fHubLastSize(0),
            fHubPrevSize(0),
            fHubSizeTmCnt(0)
         {
            for (int i=0;i<HADAQ_RINGSIZE;i++)
               fTrigNumRing[i]=0;
         }
 
         void Reset(bool complete = false)
         {
            // used to reset current subevent
            subevnt = 0;
            evnt = 0;
            has_data = false;
            data_size = 0;
            fTrigNr = 0;
            fTrigTag = 0;
            fTrigType = 0;
            fDataError = false;
            // fHubId = 0;
            fEmpty = true;
            // do not clear last fill level and last trig id
            if (complete) {
               fLastTrigNr = 0xffffffff;
               fUdpPort = 0;
            }
         }
 
         void Close()
         {
            // used to close all iterators
            fIter.Close();
            fResortIter.Close();
            fResortIndx = -1;
         }
 
         std::string TriggerRingAsStr(int RingSize) {
            std::string sbuf;
 
            unsigned cnt = fRingCnt, prev = 0;
            if ((int)cnt < RingSize) cnt += HADAQ_RINGSIZE;
            bool first = true;
            cnt -= RingSize;
            for (int n=0;n<RingSize;n++) {
               if (first && (fTrigNumRing[cnt]==0) && (n!=RingSize-1)) {
                  // ignore 0 in the beginning
               } else if (first) {
                  sbuf.append(dabc::format(" 0x%06x",(unsigned)fTrigNumRing[cnt]));
                  first = false;
               } else {
                  sbuf.append(dabc::format(" %d",(int)fTrigNumRing[cnt] - (int)fTrigNumRing[prev]));
               }
 
               prev = cnt; cnt = (cnt+1) % HADAQ_RINGSIZE;
            }
            return sbuf;
         }
      };
 
 
         /* master stream for event building*/
         //unsigned fMasterChannel;
 
         int fTriggerNrTolerance;
 
         double fEventBuildTimeout;
 
         bool fHadesTriggerType;
         unsigned fHadesTriggerHUB;
 
         uint32_t fLastTrigNr;  
 
         std::vector<InputCfg> fCfg; 
 
         WriteIterator      fOut;
 
         int                fFlushCounter;
         int32_t            fEBId; 
         pid_t              fPID; 
         bool               fIsTerminating;
         bool               fSkipEmpty;     
 
         bool               fRunToOracle;
 
         int                fNumReadBuffers; //< / SL: workaround counter, which indicates how many buffers were taken from queues
 
         unsigned           fSpecialItemId;  
         bool               fSpecialFired;  
         double             fLastEventRate; 
 
         bool               fCheckTag;
 
         bool               fBNETsend;  
         bool               fBNETrecv;  
         int                fBNETbunch; 
         int                fBNETNumSend; 
         int                fBNETNumRecv; 
         std::string        fBNETCalibrDir;   
         std::string        fBNETCalibrPackScript;  
         dabc::Command      fBnetCalibrCmd;  
 
         double             fFlushTimeout;
         dabc::Command      fBnetFileCmd;  
         dabc::Command      fBnetRefreshCmd; 
 
         std::string        fDataRateName;
         std::string        fDataDroppedRateName;
         std::string        fEventRateName;
         std::string        fLostEventRateName;
         std::string        fInfoName;
 
         uint64_t           fDataRateCnt;
         uint64_t           fDataDroppedRateCnt;
         uint64_t           fEventRateCnt;
         uint64_t           fLostEventRateCnt;
 
         uint64_t           fRunRecvBytes;
         uint64_t           fRunBuildEvents;   
         uint64_t           fRunDroppedData;
         uint64_t           fRunDiscEvents;
         uint64_t           fRunTagErrors;
         uint64_t           fRunDataErrors;
 
         uint64_t           fAllRecvBytes;
         uint64_t           fAllBuildEvents;       
         uint64_t           fAllBuildEventsLimit;  
         uint64_t           fAllDiscEvents;
         uint64_t           fAllDroppedData;
         uint64_t           fAllFullDrops;         
 
         unsigned           fEventIdCount[HADAQ_NEVTIDS];
 
         std::string        fRunInfoToOraFilename;
         std::string        fPrefix;
 
         uint32_t           fRunNumber;
 
         uint32_t           fEpicsRunNumber;
 
         uint32_t           fMaxHadaqTrigger;
         uint32_t           fTriggerRangeMask;
 
         bool               fEvnumDiffStatistics;
 
         bool               fExtraDebug;   
         dabc::TimeStamp    fLastDebugTm;  
         dabc::TimeStamp    fLastDropTm;   
         dabc::TimeStamp    fLastProcTm;   
         dabc::TimeStamp    fLastBuildTm;  
         double             fMaxProcDist;  
 
         std::string        fBnetInfo;     
         std::string        fBnetStat;     
 
         long               fBldCalls{0};   
         long               fInpCalls{0};   
         long               fOutCalls{0};   
         long               fBufCalls{0};   
         long               fTimerCalls{0}; 
         dabc::Profiler     fBldProfiler;   
 
         bool BuildEvent();
 
         bool FlushOutputBuffer();
 
         void DoInputSnapshot(unsigned ninp);
 
         void BeforeModuleStart() override;
 
         void AfterModuleStop() override;
 
         bool ShiftToNextHadTu(unsigned ninp);
 
         bool ShiftToNextEvent(unsigned ninp, bool fast = false, bool dropped = false);
 
         bool ShiftToNextSubEvent(unsigned ninp, bool fast = false, bool dropped = false);
 
         bool ShiftToNextBuffer(unsigned ninp);
 
         /* cleanup input buffers in case of too large eventnumber mismatch*/
         bool DropAllInputBuffers();
 
         //uint32_t CurrEventId(unsigned int ninp) const { return fCfg[ninp].curr_evnt_num; }
 
         void SetInfo(const std::string &info, bool forceinfo = false);
 
 
         /* Methods to export run begin to oracle via text file*/
         void StoreRunInfoStart();
 
         /* Methods to export run end and statistics to oracle via text file
              if this is called on eventbuilder exit, we use local time instead ioc/runnumber time*/
         void StoreRunInfoStop(bool onexit=false, unsigned newrunid = 0);
 
         void ResetInfoCounters();
 
         /* stolen from daqdata/hadaq/logger.c to keep oracle export output format of numbers*/
         char* Unit(unsigned long v);
 
         /* we synthetise old and new filenames ourselves, since all communication to hldoutput is faulty,
             because of timeshift between getting new runid and open/close of actual files*/
         std::string GenerateFileName(unsigned runid);
 
         void DoTerminalOutput();
 
         int DestinationPort(uint32_t trignr);
         bool CheckDestination(uint32_t trignr);
         void UpdateBnetInfo();
 
         void StartEventsBuilding();
 
      public:
         CombinerModule(const std::string &name, dabc::Command cmd = nullptr);
         virtual ~CombinerModule();
 
         void ModuleCleanup() override;
 
         void ProcessPoolEvent(unsigned) override { fBufCalls++; StartEventsBuilding(); }
         void ProcessInputEvent(unsigned) override { fInpCalls++; StartEventsBuilding(); }
         void ProcessOutputEvent(unsigned) override { fOutCalls++; StartEventsBuilding(); }
 
         void ProcessTimerEvent(unsigned timer) override;
 
         void ProcessUserEvent(unsigned item) override;
 
         int ExecuteCommand(dabc::Command cmd) override;
 
         bool ReplyCommand(dabc::Command cmd) override;
 
         int CalcTrigNumDiff(const uint32_t &prev, const uint32_t &next);
   };
 
}
 
#endif