Try to reinit dogma transports if there are several "BAD" inputs over long time

Author: linev

plugins/dogma/dogma/CombinerModule.h

@@ -127,6 +127,13 @@ namespace dogma {
             fResortIndx = -1;
          }
 
+         void ResetCounters()
+         {
+            fLastTotalDataSize = 0;
+            fLastLostTrig = 0;
+            fBadStateCount = 0;
+         }
+
          std::string TriggerRingAsStr(int RingSize)
          {
             std::string sbuf;
@@ -282,7 +289,7 @@ namespace dogma {
          bool ShiftToNextBuffer(unsigned ninp);
 
          /* cleanup input buffers in case of too large eventnumber mismatch*/
-         bool DropAllInputBuffers();
+         bool DropAllInputBuffers(bool reinit_transports = false);
 
          //uint32_t CurrEventId(unsigned int ninp) const { return fCfg[ninp].curr_evnt_num; }
 

plugins/dogma/src/CombinerModule.cxx

@@ -245,6 +245,8 @@ void dogma::CombinerModule::ProcessTimerEvent(unsigned timer)
    }
 
    if (timer_name == "BadInputs") {
+
+      int num_really_bad = 0;
       for (auto &cfg : fCfg) {
          // lost rate in 1000..1000000
          // datarate in 1M..100M
@@ -258,7 +260,15 @@ void dogma::CombinerModule::ProcessTimerEvent(unsigned timer)
 
          cfg.fLastLostTrig = cfg.fLostTrig;
          cfg.fLastTotalDataSize = cfg.fTotalDataSize;
+
+         if (cfg.fBadStateCount > 10)
+            num_really_bad++;
       }
+
+      if ((num_really_bad > 2) && fLastDropTm.Expired((fEventBuildTimeout > 0) ? 1.5*fEventBuildTimeout : 5.))
+         DropAllInputBuffers(true);
+
+      return;
    }
 
    if ((fFlushTimeout > 0) && (++fFlushCounter > 2)) {
@@ -891,7 +901,7 @@ bool dogma::CombinerModule::ShiftToNextSubEvent(unsigned ninp, bool fast, bool d
    return true;
 }
 
-bool dogma::CombinerModule::DropAllInputBuffers()
+bool dogma::CombinerModule::DropAllInputBuffers(bool reinit_transports)
 {
    DOUT0("DropAllInputBuffers()...");
 
@@ -902,6 +912,12 @@ bool dogma::CombinerModule::DropAllInputBuffers()
    if (fBNETsend)
       fCheckBNETProblems = chkActive; // activate testing again
 
+   if (reinit_transports)
+      for (unsigned n = 0; n < NumInputs(); n++) {
+         dabc::Command subcmd("ReinitTransport");
+         SubmitCommandToTransport(InputName(n), subcmd);
+      }
+
    unsigned maxnumsubev = 0, droppeddata = 0;
 
    for (unsigned ninp = 0; ninp < fCfg.size(); ninp++) {
@@ -917,6 +933,7 @@ bool dogma::CombinerModule::DropAllInputBuffers()
 
       fCfg[ninp].Reset();
       fCfg[ninp].Close();
+      fCfg[ninp].ResetCounters();
       while (SkipInputBuffers(ninp, 100)); // drop input port queue buffers until no more there
    }
 
@@ -1470,6 +1487,8 @@ int dogma::CombinerModule::ExecuteCommand(dabc::Command cmd)
             fCfg[n].fHubSizeTmCnt = 0;
             fCfg[n].fHubLastSize = 0;
             fCfg[n].fHubPrevSize = 0;
+            fCfg[n].fTotalDataSize = 0;
+            fCfg[n].ResetCounters();
             dabc::Command subcmd("ResetTransportStat");
             SubmitCommandToTransport(InputName(n), subcmd);
          }