//
// MIDAS analyzer
//
// K.Olchanski
//

#undef NDEBUG // this program requires working assert()

#include <stdio.h>
#include <unistd.h> // usleep()
#include <assert.h>
#include <sys/stat.h> // struct stat_buffer;

#include "manalyzer.h"
#include "midasio.h"

//////////////////////////////////////////////////////////

static bool gTrace = false;

//////////////////////////////////////////////////////////
//
// Methods of TARunInfo
//
//////////////////////////////////////////////////////////

TARunInfo::TARunInfo(int runno, const char* filename, const std::vector<std::string>& args)
{
   if (gTrace)
      printf("TARunInfo::ctor!\n");
   fRunNo = runno;
   if (filename)
      fFileName = filename;
   fOdb = NULL;
#ifdef HAVE_ROOT
   fRoot = new TARootHelper(this);
#else
   fRoot = NULL;
#endif
   fMtInfo = NULL;
   fArgs = args;
}

TARunInfo::~TARunInfo()
{
   if (gTrace)
      printf("TARunInfo::dtor!\n");
   fRunNo = 0;
   fFileName = "(deleted)";
   if (fOdb) {
      delete fOdb;
      fOdb = NULL;
   }
#ifdef HAVE_ROOT
   if (fRoot) {
      delete fRoot;
      fRoot = NULL;
   }
#endif
   int count = 0;
   while (1) {
      TAFlowEvent* flow = ReadFlowQueue();
      if (!flow)
         break;
      delete flow;
      count++;
   }
   if (gTrace) {
      printf("TARunInfo::dtor: deleted %d queued flow events!\n", count);
   }

   if (fMtInfo) {
      delete fMtInfo;
      fMtInfo = NULL;
   }
}

//////////////////////////////////////////////////////////
//
// Methods of TAFlowEvent
//
//////////////////////////////////////////////////////////

TAFlowEvent::TAFlowEvent(TAFlowEvent* flow) // ctor
{
   if (gTrace)
      printf("TAFlowEvent::ctor: chain %p\n", flow);
   fNext = flow;
}

TAFlowEvent::~TAFlowEvent() // dtor
{
   if (gTrace)
      printf("TAFlowEvent::dtor: this %p, next %p\n", this, fNext);
   if (fNext)
      delete fNext;
   fNext = NULL;
}

//////////////////////////////////////////////////////////
//
// Methods of TARunObject
//
//////////////////////////////////////////////////////////

TARunObject::TARunObject(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::ctor, run %d\n", runinfo->fRunNo);
}

void TARunObject::BeginRun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::BeginRun, run %d\n", runinfo->fRunNo);
}

void TARunObject::EndRun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::EndRun, run %d\n", runinfo->fRunNo);
}

void TARunObject::NextSubrun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::NextSubrun, run %d\n", runinfo->fRunNo);
}

void TARunObject::PauseRun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::PauseRun, run %d\n", runinfo->fRunNo);
}

void TARunObject::ResumeRun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::ResumeRun, run %d\n", runinfo->fRunNo);
}

void TARunObject::PreEndRun(TARunInfo* runinfo)
{
   if (gTrace)
      printf("TARunObject::PreEndRun, run %d\n", runinfo->fRunNo);
}

TAFlowEvent* TARunObject::Analyze(TARunInfo* runinfo, TMEvent* event, TAFlags* flags, TAFlowEvent* flow)
{
   if (gTrace)
      printf("TARunObject::Analyze!\n");

   // This default analyze function does no work, instruct the Profiler to not time / log this
   *flags|=TAFlag_SKIP_PROFILE;

   return flow;
}

TAFlowEvent* TARunObject::AnalyzeFlowEvent(TARunInfo* runinfo, TAFlags* flags, TAFlowEvent* flow)
{
   if (gTrace)
      printf("TARunObject::Analyze!\n");

   // This default analyze function does no work, instruct the Profiler to not time / log this
   *flags|=TAFlag_SKIP_PROFILE;

   return flow;
}

void TARunObject::AnalyzeSpecialEvent(TARunInfo* runinfo, TMEvent* event)
{
   if (gTrace)
      printf("TARunObject::AnalyzeSpecialEvent!\n");
}

//////////////////////////////////////////////////////////
//
// Methods of TAFactory
//
//////////////////////////////////////////////////////////

void TAFactory::Usage()
{
   if (gTrace)
      printf("TAFactory::Usage!\n");
}

void TAFactory::Init(const std::vector<std::string> &args)
{
   if (gTrace)
      printf("TAFactory::Init!\n");
}

void TAFactory::Finish()
{
   if (gTrace)
      printf("TAFactory::Finish!\n");
}

#ifdef HAVE_ROOT

//////////////////////////////////////////////////////////
//
// Methods of TARootHelper
//
//////////////////////////////////////////////////////////

std::string   TARootHelper::fOutputDirectory = "root_output_files";
std::string   TARootHelper::fOutputFileName = "";
TApplication* TARootHelper::fgApp = NULL;
TDirectory*   TARootHelper::fgDir = NULL;
THttpServer*  TARootHelper::fgHttpServer = NULL;

TARootHelper::TARootHelper(const TARunInfo* runinfo) // ctor
{
   if (gTrace)
      printf("TARootHelper::ctor!\n");

   std::string filename = fOutputFileName;

   if (filename.empty()) {
      char xfilename[256];
      snprintf(xfilename, sizeof(xfilename), "output%05d.root", runinfo->fRunNo);

      if (fOutputDirectory.empty()) {
         filename = xfilename;
      } else {
         filename = fOutputDirectory + "/" + xfilename;

         struct stat buffer;
         int status = stat(fOutputDirectory.c_str(), &buffer);

         if (status < 0 && errno == ENOENT) {
            fprintf(stdout, "TARootHelper::ctor: creating output directory \"%s\"\n", fOutputDirectory.c_str());
            status = mkdir(fOutputDirectory.c_str(), 0777);
            if (status == -1) {
               fprintf(stderr, "TARootHelper::ctor: Error: cannot output directory \"%s\", errno %d (%s)\n", fOutputDirectory.c_str(), errno, strerror(errno));
            }
         }
      }
   }

   fOutputFile = new TFile(filename.c_str(), "RECREATE");
   
   if (!fOutputFile->IsOpen()) {
      fprintf(stderr, "TARootHelper::ctor: Error: cannot open output ROOT file \"%s\"\n", filename.c_str());
      fOutputFile = new TFile("/dev/null", "UPDATE");
      assert(fOutputFile);
      assert(fOutputFile->IsOpen());
   }

   if (fOutputFile != NULL) {
      fOutputFile->cd();
   }
}

TARootHelper::~TARootHelper() // dtor
{
   if (gTrace)
      printf("TARootHelper::dtor!\n");
   
   if (fOutputFile != NULL) {
      fOutputFile->Write();
      fOutputFile->Close();
      fOutputFile = NULL;
   }

   if (fgDir)
      fgDir->cd();
}

#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <assert.h>
#include <signal.h>

#include "manalyzer.h"
#include "midasio.h"

#ifdef HAVE_THTTP_SERVER
#include "THttpServer.h"
#endif

#ifdef HAVE_ROOT
#include <TSystem.h>
#include <TROOT.h>
#endif

//////////////////////////////////////////////////////////
//
// Methods and Defaults of TAMultithreadHelper
//
//////////////////////////////////////////////////////////

static int gDefaultMultithreadQueueLength = 100;
static int gDefaultMultithreadWaitEmpty = 100; // microseconds
static int gDefaultMultithreadWaitFull = 100; // microseconds

static void WaitForAllQueuesEmpty(TAMultithreadHelper* mt)
{
   fprintf(stderr, "Waiting for all queues to empty out!\n");

   int count_all_empty = 0;

   for (int t=0; ; ) {
      int count_not_empty = 0;
      size_t count_events = 0;

      for (unsigned i=0; i<mt->fMtThreads.size(); i++) {
         std::lock_guard<std::mutex> lock(mt->fMtFlowQueueMutex[i]);
         if (!mt->fMtThreadIsRunning[i]) // skip threads that have shutdown
            continue;
         if (!mt->fMtFlowQueue[i].empty()) {
            count_not_empty++;
            count_events += mt->fMtFlowQueue[i].size();
            break;
         }
         if (mt->fMtThreadIsBusy[i]) {
            count_not_empty++;
            count_events += 1; // module is analyzing 1 event
            break;
         }
         // implicit unlock
      }

      if (count_not_empty == 0) {
         count_all_empty++;
      }

      if (count_all_empty > 1) {
         // must loop over "all empty" at least twice! K.O.
         break;
      }

      if (t > 10) {
         fprintf(stderr, "Timeout waiting for all queues to empty out, %d queues still have %d flow events!\n", count_not_empty, (int)count_events);
         //break;
      }

      ::sleep(1);
      t++;
   }
}

static void WaitForAllThreadsShutdown(TAMultithreadHelper* mt)
{
   fprintf(stderr, "Waiting for all threads to shut down!\n");
   
   mt->fMtShutdownRequested = true;

   for (int t=0; ; ) {
      int count_running = 0;

      for (unsigned i=0; i<mt->fMtThreads.size(); i++) {
         if (mt->fMtThreadIsRunning[i])
            count_running++;
      }

      if (count_running == 0) {
         break;
      }

      if (t > 10) {
         fprintf(stderr, "Timeout waiting for all threads to shut down, %d still running!\n", count_running);
         break;
      }

      ::sleep(1);
      t++;
   }
   
   fprintf(stderr, "Joining all threads!\n");
   for (unsigned i=0; i<mt->fMtThreads.size(); i++) {
      if (!mt->fMtThreadIsRunning[i] ) {
         // pointer is null if thread is already shutdown
         if (mt->fMtThreads[i]) {
            mt->fMtThreads[i]->join();
            delete mt->fMtThreads[i];
            mt->fMtThreads[i] = NULL;
         }
      } else {
         fprintf(stderr, "Thread %d failed to shut down!\n", i);
      }
   }
}


TAMultithreadHelper::TAMultithreadHelper(int nModules):
   fMtFlowQueueMutex(nModules),
   fMtFlowQueue(nModules),
   fMtFlagQueue(nModules),
   fMtThreads(nModules,NULL),
   fMtThreadIsRunning(nModules),
   fMtThreadIsBusy(nModules),
   fMtShutdownRequested(false),
   fMtQuitRequested(false)
   // ctor
{
   for (auto &b: fMtThreadIsRunning) { b = false; }
   for (auto &b: fMtThreadIsBusy) { b = false; }
   // default max queue size
   fMtQueueDepth = gDefaultMultithreadQueueLength;
   // queue settings
   fMtQueueFullUSleepTime  = gDefaultMultithreadWaitFull; //u seconds
   fMtQueueEmptyUSleepTime = gDefaultMultithreadWaitEmpty; //u seconds
}

TAMultithreadHelper::~TAMultithreadHelper() // dtor
{
   size_t nmodules = fMtFlowQueueMutex.size();

   // just for kicks, check that all queues have correct size
   assert(nmodules == fMtFlowQueue.size());
   assert(nmodules == fMtFlagQueue.size());
   assert(nmodules == fMtFlowQueueMutex.size());

   // should not come to the destructor while threads are still running
   WaitForAllThreadsShutdown(this);

   int count = 0;
   for (size_t i=0; i<nmodules; i++) {
      // check that the thread is stopped
      //assert(!fMtThread[i].joinable());
      // empty the thread queue
      std::lock_guard<std::mutex> lock(fMtFlowQueueMutex[i]);
      while (!fMtFlowQueue[i].empty()) {
         TAFlowEvent* flow = fMtFlowQueue[i].front();
         TAFlags* flag = fMtFlagQueue[i].front();
         fMtFlowQueue[i].pop_front();
         fMtFlagQueue[i].pop_front();
         delete flow;
         delete flag;
         count++;
      }
      // implicit unlock of mutex
   }
   if (gTrace) {
      printf("TAMultithreadInfo::dtor: deleted %d queued flow events!\n", count);
   }
}

bool TAMultithreadHelper::gfMultithread           = false;
int TAMultithreadHelper::gfMtMaxBacklog           = 100;
std::mutex TAMultithreadHelper::gfLock; //Lock for modules to execute code that is not thread safe (many root fitting libraries)

static void MtQueueFlowEvent(TAMultithreadHelper* mt, int i, TAFlags* flag, TAFlowEvent* flow, bool wait)
{
   assert(mt);

   if (flag == NULL) {
      flag = new TAFlags;
      *flag = 0;
   }

   //PrintQueueLength();

   while (1) {
      {
         //Lock and queue events
         std::lock_guard<std::mutex> lock(mt->fMtFlowQueueMutex[i]);
         
         if ((((int)mt->fMtFlowQueue[i].size()) < mt->fMtQueueDepth) || mt->fMtShutdownRequested || !wait) {
            mt->fMtFlowQueue[i].push_back(flow);
            mt->fMtFlagQueue[i].push_back(flag);
            return;
         }
         // Unlock when we go out of scope
      }
      
      usleep(mt->fMtQueueFullUSleepTime);
   }
}

#if 0
//Function to print the length of the flow queue when in multithread mode
//Maybe make root update a graphical window?
static void PrintMtQueueLength(TAMultithreadHelper* mt)
{
   printf("Multithread queue lengths:\n");
   for (unsigned i=0; i<mt->fMtFlowQueue.size(); i++) {
      printf("%d:\t%zu\n",i,mt->fMtFlowQueue[i].size());
   }
}
#endif

//////////////////////////////////////////////////////////
//
// Methods of TARegister
//
//////////////////////////////////////////////////////////

std::vector<TAFactory*> *gModules = NULL;

TARegister::TARegister(TAFactory* m)
{
   if (!gModules)
      gModules = new std::vector<TAFactory*>;
   gModules->push_back(m);
}

#if 0
static double GetTimeSec()
{
  struct timeval tv;
  gettimeofday(&tv,NULL);
  return tv.tv_sec + 0.000001*tv.tv_usec;
}
#endif

//////////////////////////////////////////////////////////
//
//                      Profiler class
//
//////////////////////////////////////////////////////////

TAUserProfilerFlow::TAUserProfilerFlow(TAFlowEvent* flow, const char* name, const TAClock& start) : TAFlowEvent(flow), fModuleName(name)
{
   fStart = start;
   fStop = TAClockNow();
}

TAUserProfilerFlow::~TAUserProfilerFlow() // dtor
{
}

double TAUserProfilerFlow::GetTimer() const
{
   TAClockDuration elapsed_seconds = fStop - fStart;
   return elapsed_seconds.count();
}

#ifdef HAVE_ROOT
#include "TH1D.h"
#endif
#include <map>

class Profiler
{
private:
   std::string fBinaryName;
   std::string fBinaryPath;
   clock_t fStartCPU;
   std::chrono::system_clock::time_point fStartUser;
   uint32_t fMIDASStartTime;
   uint32_t fMIDASStopTime;

   //Track Queue lengths when multithreading
#ifdef HAVE_ROOT
   int fNQueues=0;
   std::vector<TH1D*> fAnalysisQueue;
   std::atomic<int> fQueueIntervalCounter;
#endif

   // Track Analyse TMEvent time per module (main thread)
#ifdef HAVE_ROOT
   std::vector<TH1D*>  fAnalyzeEventTimeHistograms;
#endif
   std::vector<std::string> fModuleNames;
   std::vector<double> fAnalyzeEventMean;
   std::vector<double> fAnalyzeEventRMS;
   std::vector<int>    fAnalyzeEventEntries;

   std::vector<double> fAnalyzeEventTimeMax;
   std::vector<double> fAnalyzeEventTimeTotal;

   //Track Analyse flow event time per module (can be multiple threads)
#ifdef HAVE_ROOT
   std::vector<TH1D*>  fAnalyzeFlowEventTimeHistograms;
#endif
   std::vector<double> fAnalyzeFlowEventMean;
   std::vector<double> fAnalyzeFlowEventRMS;
   std::vector<int>    fAnalyzeFlowEventEntries;

   std::vector<double> fAnalyzeFlowEventTimeMax;
   std::vector<double> fAnalyzeFlowEventTimeTotal;
#ifdef HAVE_ROOT
   //Track user profiling
   std::map<unsigned int,int> fUserMap;
   std::vector<TH1D*> fUserHistograms;
   std::vector<double> fTotalUserTime;
   std::vector<double> fMaxUserTime;

   // Number of events between samples
   const int fQueueInterval = 100;
#endif

public:
   Profiler( const int queue_interval_check );
   ~Profiler();
   void Begin(TARunInfo* runinfo,const std::vector<TARunObject*> fRunRun );
   // Function for profiling the 'main' thread (that unpacks TMEvents)
   void LogAnalyzeEvent(TAFlags* flag, TAFlowEvent* flow, const int i, const TAClock& start);
   // Function for profiling module threads
   void LogAnalyzeFlowEvent(TAFlags* flag, TAFlowEvent* flow, const int i, const TAClock& start);
   // Extra function for users custom profiling windows
   void LogUserWindows(TAFlags* flag, TAFlowEvent* flow);
   void AddModuleMap( const char* UserProfileName, unsigned long hash);
   void LogMTQueueLength(TARunInfo* runinfo);
   void End(TARunInfo* runinfo);
   void Print() const;
};

Profiler::Profiler(const int queue_interval_check)
#ifdef HAVE_ROOT
   :
   fQueueIntervalCounter(0),
   fQueueInterval(queue_interval_check)
#endif
{
   if (gTrace)
      printf("Profiler::ctor\n");
   fMIDASStartTime = 0;
   fMIDASStopTime = 0;
   fStartCPU = clock();
   fStartUser =  std::chrono::system_clock::now();
}

Profiler::~Profiler()
{
   if (gTrace)
      printf("Profiler::dtor\n");
   
#ifdef HAVE_ROOT
   for (TH1D* h: fAnalyzeFlowEventTimeHistograms)
      delete h;
   fAnalyzeFlowEventTimeHistograms.clear();
   for (TH1D* h: fAnalyzeEventTimeHistograms)
      delete h;
   fAnalyzeEventTimeHistograms.clear();
   for( TH1D* h: fAnalysisQueue)
      delete h;
   fAnalysisQueue.clear();
#endif
}

void Profiler::Begin(TARunInfo* runinfo, const std::vector<TARunObject*> runrun)
{
   if (gTrace)
      printf("Profiler::begin\n");

   runinfo->fOdb->RU32("Runinfo/Start time binary",(uint32_t*) &fMIDASStartTime);

#ifdef HAVE_ROOT
   // Put Profiler histograms in their own folders in the output root file
   if (runinfo->fRoot->fOutputFile) {
      runinfo->fRoot->fOutputFile->cd(); // select correct ROOT directory
      gDirectory->mkdir("ProfilerReport")->cd();
      runinfo->fRoot->fOutputFile->cd("ProfilerReport");
      gDirectory->mkdir("AnalyzeFlowEventTime");
      gDirectory->mkdir("AnalyzeFlowTime");
      gDirectory->mkdir("MTQueueLength");
   }

   // Setup module processing time histograms
   // Number of bins
   Int_t Nbins=1000;
   // Array of bin edges
   Double_t bins[Nbins+1];
   // Processing time range (seconds)
   Double_t TimeRange = 10;
   // Set uneven binning to better sample fast modules with accuracy 
   // without having a large number of bins
   for (int i=0; i<Nbins+1; i++) {
      bins[i] = TimeRange*pow(1.1,i)/pow(1.1,Nbins);
   }

   if (runinfo->fMtInfo)
      fNQueues = runrun.size();
#endif

   // Per module metric setup
   for (size_t i = 0; i < runrun.size(); i++) {

      if (runrun[i]->fModuleName.empty())
         fModuleNames.push_back("Unnamed Module " + std::to_string(i));
      else
         fModuleNames.push_back(runrun[i]->fModuleName);

      // Metrics for the AnalyzeEvent function (main thread)
      fAnalyzeEventMean.push_back(0);
      fAnalyzeEventRMS.push_back(0);
      fAnalyzeEventEntries.push_back(0);
      fAnalyzeEventTimeMax.push_back(0);
      fAnalyzeEventTimeTotal.push_back(0);

      // Metric for the AnalyzeFlowEvent function (side threads)
      fAnalyzeFlowEventMean.push_back(0);
      fAnalyzeFlowEventRMS.push_back(0);
      fAnalyzeFlowEventEntries.push_back(0);
      fAnalyzeFlowEventTimeMax.push_back(0);
      fAnalyzeFlowEventTimeTotal.push_back(0);

#ifdef HAVE_ROOT
      if (runinfo->fRoot->fOutputFile) {
         runinfo->fRoot->fOutputFile->cd("ProfilerReport/AnalyzeFlowTime");
         TH1D* Histo = new TH1D( TString(std::to_string(i) + "_" + fModuleNames.at(i)),
                                 TString(fModuleNames.at(i) + " Event Proccessing Time; s"),
                                 Nbins, bins);
         fAnalyzeFlowEventTimeHistograms.push_back(Histo);
         
         runinfo->fRoot->fOutputFile->cd("ProfilerReport/AnalyzeFlowEventTime");
         TH1D* AnalyzeEventHisto = new TH1D(TString(std::to_string(i) + "_" + fModuleNames.at(i) + "_TMEvent"), 
                                            TString(fModuleNames.at(i) + " Flow Proccessing Time; s"),
                                            Nbins, bins);
         fAnalyzeEventTimeHistograms.push_back(AnalyzeEventHisto);
      } else {
         fAnalyzeFlowEventTimeHistograms.push_back(NULL);
         fAnalyzeEventTimeHistograms.push_back(NULL);
      }
         
      // Periodically (once every fQueueInterval events) record the 
      // flow queue size if running in multithreaded mode 
      if (runinfo->fMtInfo) {
         if (runinfo->fRoot->fOutputFile) {
            runinfo->fRoot->fOutputFile->cd("ProfilerReport/MTQueueLength");
            TH1D* QueueHisto = new TH1D(TString(std::to_string(i) + "_" + fModuleNames.at(i) + "_Queue"),
                                        TString(fModuleNames.at(i) + " Multithread Queue Length; Queue Depth"),
                                        runinfo->fMtInfo->fMtQueueDepth*1.2,
                                        0,
                                        runinfo->fMtInfo->fMtQueueDepth*1.2);
            fAnalysisQueue.push_back(QueueHisto);
         } else {
            fAnalysisQueue.push_back(NULL);
         }
      }
#endif
   }
}

void Profiler::LogAnalyzeFlowEvent(TAFlags* flag, TAFlowEvent* flow, const int i, const TAClock& start)
{
   if (gTrace)
      printf("Profiler::log\n");

   TAClock stop = TAClockNow();
   if ((*flag) & TAFlag_SKIP_PROFILE) {
      //Unset bit
      *flag -= TAFlag_SKIP_PROFILE;
      return;
   }

   std::chrono::duration<double> elapsed_seconds = stop - start;
   double dt = elapsed_seconds.count();
   fAnalyzeFlowEventTimeTotal[i] += dt;
   if (dt > fAnalyzeFlowEventTimeMax[i])
      fAnalyzeFlowEventTimeMax[i] = dt;

   fAnalyzeFlowEventMean[i]  +=dt;
   fAnalyzeFlowEventRMS[i]   +=dt*dt;
   fAnalyzeFlowEventEntries[i]++;

#ifdef HAVE_ROOT
   if (fAnalyzeFlowEventTimeHistograms[i])
      fAnalyzeFlowEventTimeHistograms[i]->Fill(dt);
#endif
}

void Profiler::LogAnalyzeEvent(TAFlags* flag, TAFlowEvent* flow, int i, const TAClock& start)
{
   if (gTrace)
      printf("Profiler::log_AnalyzeEvent_time\n");

   TAClock stop = TAClockNow();
   if ((*flag) & TAFlag_SKIP_PROFILE) {
      //Unset bit
      *flag -= TAFlag_SKIP_PROFILE;
      return;
   }

   std::chrono::duration<double> elapsed_seconds = stop - start;
   double dt = elapsed_seconds.count();
   fAnalyzeEventTimeTotal[i] += dt;
   if (dt > fAnalyzeEventTimeMax[i])
      fAnalyzeEventTimeMax[i] = dt;

   fAnalyzeEventMean[i]   +=dt;
   fAnalyzeEventRMS[i]    +=dt*dt;
   fAnalyzeEventEntries[i]++;

#ifdef HAVE_ROOT
   if (fAnalyzeEventTimeHistograms[i])
      fAnalyzeEventTimeHistograms[i]->Fill(dt);
#endif

}

void Profiler::LogMTQueueLength(TARunInfo* runinfo)
{
   if (gTrace)
      printf("Profiler::log_mt_queue_length\n");

#ifdef HAVE_ROOT
   fQueueIntervalCounter++;
   if (runinfo->fMtInfo && (fQueueIntervalCounter % fQueueInterval ==0 )) {
      for (int i=0; i<fNQueues; i++) {
         int j=0;
         {  //Lock guard
            std::lock_guard<std::mutex> lock(runinfo->fMtInfo->fMtFlowQueueMutex[i]);
            j=runinfo->fMtInfo->fMtFlowQueue[i].size();
         }
         fAnalysisQueue.at(i)->Fill(j);
      }
   }
#endif
}


void Profiler::LogUserWindows(TAFlags* flag, TAFlowEvent* flow)
{
   if (gTrace)
      printf("Profiler::LogUserWindows\n");

#ifdef HAVE_ROOT
   //Clocks unfold backwards... 
   std::vector<TAFlowEvent*> flowArray;
   int FlowEvents=0;
   TAFlowEvent* f = flow;
   while (f) {
      flowArray.push_back(f);
      f=f->fNext;
      FlowEvents++;
   }
   for (int ii=FlowEvents-1; ii>=0; ii--) {
      f=flowArray[ii];
      TAUserProfilerFlow* timer=dynamic_cast<TAUserProfilerFlow*>(f);
      if (timer) {
         const char* name = timer->fModuleName.c_str();
         unsigned int hash = std::hash<std::string>{}(timer->fModuleName);
         if (!fUserMap.count(hash))
            AddModuleMap(name,hash);
         double dt=999.;
         dt=timer->GetTimer();
         int i = fUserMap[hash];
         fTotalUserTime[i] += dt;
         if (dt > fMaxUserTime[i])
            fMaxUserTime.at(i) = dt;
         fUserHistograms.at(i)->Fill(dt);
      }
   }
#else
   fprintf(stderr, "manalyzer must be built with ROOT for using the user profiling tools\n");
#endif
}

void Profiler::AddModuleMap( const char* UserProfileName, unsigned long hash)
{
   if (gTrace)
     printf("Profiler::AddModuleMap\n");

   std::lock_guard<std::mutex> lock(TAMultithreadHelper::gfLock);

#ifdef HAVE_ROOT
   gDirectory->cd("/ProfilerReport");
   fUserMap[hash] = fUserHistograms.size();
   Int_t Nbins = 100;
   Double_t bins[Nbins+1];
   Double_t TimeRange = 10; //seconds
   for (int i=0; i<Nbins+1; i++) {
      bins[i] = TimeRange*pow(1.1,i)/pow(1.1,Nbins);
   }
   TH1D* Histo = new TH1D(UserProfileName,UserProfileName,Nbins,bins);
   fUserHistograms.push_back(Histo);
   fTotalUserTime.push_back(0.);
   fMaxUserTime.push_back(0.);
#endif
   return;
}

void Profiler::End(TARunInfo* runinfo)
{
   if (gTrace)
     printf("Profiler::End\n");

   for (size_t i=0; i<fAnalyzeEventMean.size(); i++) {
      fAnalyzeEventMean.at(i) = fAnalyzeEventMean.at(i) / fAnalyzeEventEntries.at(i);
      fAnalyzeEventRMS.at(i) = fAnalyzeEventRMS.at(i) / fAnalyzeEventEntries.at(i) - 
         ( fAnalyzeEventMean.at(i) / fAnalyzeEventEntries.at(i) ) *
         ( fAnalyzeEventMean.at(i) / fAnalyzeEventEntries.at(i) );
   }
   for (size_t i=0; i<fAnalyzeFlowEventMean.size(); i++) {
      fAnalyzeFlowEventMean.at(i) = fAnalyzeFlowEventMean.at(i) / fAnalyzeFlowEventEntries.at(i);
      fAnalyzeFlowEventRMS.at(i) = fAnalyzeFlowEventRMS.at(i) / fAnalyzeFlowEventEntries.at(i) - 
         ( fAnalyzeFlowEventMean.at(i) / fAnalyzeFlowEventEntries.at(i) ) * 
         ( fAnalyzeFlowEventMean.at(i) / fAnalyzeFlowEventEntries.at(i) );
   }
#ifdef HAVE_ROOT
  
   if (runinfo->fRoot->fOutputFile) {
      runinfo->fRoot->fOutputFile->cd("ProfilerReport/AnalyzeFlowTime");
      for (TH1D* h: fAnalyzeFlowEventTimeHistograms) {
         h->Write();
      }
      runinfo->fRoot->fOutputFile->cd("ProfilerReport/AnalyzeFlowEventTime");
      for (TH1D* h: fAnalyzeEventTimeHistograms) {
         h->Write();
      }
      if (runinfo->fMtInfo) {
         runinfo->fRoot->fOutputFile->cd("ProfilerReport/MTQueueLength");
         for (TH1D* h: fAnalysisQueue) {
            h->Write();
         }
      }
   }
#endif
}

void Profiler::Print() const
{
#ifdef HAVE_ROOT
   if (fAnalyzeFlowEventTimeHistograms.size()>0) {
#else
   if (fAnalyzeEventEntries.size()>0) {
#endif
      double AllAnalyzeFlowEventTime=0;
      for (auto& n : fAnalyzeFlowEventTimeTotal)
         AllAnalyzeFlowEventTime += n;
      double AllAnalyzeEventTime=0;
      for (auto& n : fAnalyzeEventTimeTotal)
         AllAnalyzeEventTime += n;
      //double max_AnalyzeEvent_time=*std::max_element(TotalAnalyzeEventTime.begin(),TotalAnalyzeEventTime.end());
      printf("Module average processing time\n");
      printf("      \t\t\t\tAnalyzeEvent (one thread)                \tAnalyzeFlowEvent (multithreadable)\n");
      printf("Module\t\t\t\tEntries\tMean(ms)RMS(ms)\tMax(ms)\tSum(s)\tEntries\tMean(ms)RMS(ms)\tMax(ms)\tSum(s)\n");
      printf("----------------------------------------------------------------------------------------------------------------\n");
      for (size_t i=0; i<fModuleNames.size(); i++) {
         printf("%-25s", fModuleNames.at(i).c_str());
         if (fAnalyzeEventEntries.at(i))
            printf("\t%d\t%.1f\t%.1f\t%.1f\t%.3f",
               fAnalyzeEventEntries.at(i),
               fAnalyzeEventMean.at(i)*1000.,
               fAnalyzeEventRMS.at(i)*1000.,
               fAnalyzeEventTimeMax.at(i)*1000., //ms
               fAnalyzeEventTimeTotal.at(i)); //s
         else
            printf("\t-\t-\t-\t-\t-");

         if (fAnalyzeFlowEventEntries.at(i))
            printf("\t%d\t%.1f\t%.1f\t%.1f\t%.3f",
               fAnalyzeFlowEventEntries.at(i),
               fAnalyzeFlowEventMean.at(i)*1000.,
               fAnalyzeFlowEventRMS.at(i)*1000.,
               fAnalyzeFlowEventTimeMax.at(i)*1000., //ms
               fAnalyzeFlowEventTimeTotal.at(i)); //s
         else
            printf("\t-\t-\t-\t-\t-");
         printf("\n");
      }
      printf("----------------------------------------------------------------------------------------------------------------\n");
      printf("                                   Analyse TMEvent total time   %f\n",AllAnalyzeEventTime);
      printf("                                                                           Analyse FlowEvent total time %f\n",AllAnalyzeFlowEventTime);
#ifdef HAVE_ROOT
      if (fUserHistograms.size()) {
         printf("Custom profiling windows\tEntries\tMean(ms)RMS(ms)\tMax(ms)\tSum(s)\n");
         printf("----------------------------------------------------------------------\n");
         for (size_t i=0; i<fUserHistograms.size(); i++) {
            printf("%-25s\t%d\t%.1f\t%.1f\t%.1f\t%.3f\t\n",fUserHistograms.at(i)->GetTitle(),
            (int)fUserHistograms.at(i)->GetEntries(),
            fUserHistograms.at(i)->GetMean()*1000., //ms
            fUserHistograms.at(i)->GetRMS()*1000., //ms
            fMaxUserTime.at(i)*1000., //ms
            fTotalUserTime.at(i)); //s
         }
         printf("----------------------------------------------------------------------\n");
      } else {
         printf("----------------------------------------------------------------------------------------------------------------\n");
      }
#else
      printf("To use custom profile windows, please build rootana with root\n");
#endif
   }

   //CPU and Wall clock time:
   double cputime = (double)(clock() - fStartCPU)/CLOCKS_PER_SEC;
   std::chrono::duration<double> usertime = std::chrono::system_clock::now() - fStartUser;
   printf("%s\tCPU time: %.2fs\tUser time: %.2fs\tAverage CPU Usage: ~%.1f%%\n",
      getenv("_"),
      cputime,
      usertime.count(),
      100.*cputime/usertime.count());
}

//////////////////////////////////////////////////////////
//
//                RunHandler class
//
//////////////////////////////////////////////////////////

class Profiler;

class RunHandler
{
public:
   TARunInfo* fRunInfo = NULL;
   std::vector<TARunObject*> fRunRun;
   std::vector<std::string>  fArgs;
   bool fMultithreadEnabled = false;
   Profiler* fProfiler = NULL;
   bool fProfilerEnabled = false;
   int fProfilerIntervalCheck;

   RunHandler(const std::vector<std::string>& args, bool multithread, bool profile, int queue_interval_check) // ctor
   {
      fRunInfo = NULL;
      fArgs = args;
      fMultithreadEnabled = multithread;
      fProfiler = NULL;
      fProfilerEnabled = profile;
      fProfilerIntervalCheck = queue_interval_check;
   }

   ~RunHandler() // dtor
   {
      if (fRunInfo) {
         delete fRunInfo;
         fRunInfo = NULL;
      }
      if (fProfiler) {
         delete fProfiler;
         fProfiler = NULL;
      }
   }

   void PerModuleThread(int i)
   {
      //bool data_processing=true;
      int nModules=(*gModules).size();

      TAMultithreadHelper* mt = fRunInfo->fMtInfo;

      assert(nModules == (int)mt->fMtFlowQueueMutex.size());
      assert(nModules == (int)mt->fMtFlowQueue.size());
      assert(nModules == (int)mt->fMtFlagQueue.size());

      { //Lock scope
         std::lock_guard<std::mutex> lock(mt->fMtFlowQueueMutex[i]);
         mt->fMtThreadIsRunning[i] = true;
      }

      while (!mt->fMtShutdownRequested) {
         TAFlowEvent* flow = NULL;
         TAFlags* flag = NULL;

         { //Lock scope
            std::lock_guard<std::mutex> lock(mt->fMtFlowQueueMutex[i]);
            if (!mt->fMtFlowQueue[i].empty()) {
               flow=mt->fMtFlowQueue[i].front();
               flag=mt->fMtFlagQueue[i].front();
               mt->fMtFlowQueue[i].pop_front();
               mt->fMtFlagQueue[i].pop_front();
            }

            if (flow == NULL)
               mt->fMtThreadIsBusy[i] = false; // we will sleep
            else
               mt->fMtThreadIsBusy[i] = true; // we will analyze an event

            // implicit unlock of mutex
         }

         if (flow == NULL) { // wait until queue not empty
            usleep(mt->fMtQueueEmptyUSleepTime);
            continue;
         }

         TAClock start_time = TAClockNow();
         
         flow = fRunRun[i]->AnalyzeFlowEvent(fRunInfo, flag, flow);

         if (fProfiler)
            fProfiler->LogAnalyzeFlowEvent(flag, flow, i, start_time);

         if ((*flag) & TAFlag_QUIT) { // shut down the analyzer
            delete flow;
            delete flag;
            flow = NULL;
            flag = NULL;
            mt->fMtQuitRequested = true;
            mt->fMtShutdownRequested = true;
            break; // stop the thread
         }

         if ((*flag) & TAFlag_SKIP) { // stop processing this event
            delete flow;
            delete flag;
            flow = NULL;
            flag = NULL;
            continue;
         }

         if (i==nModules-1) //If I am the last module... free memory, else queue up for next module to process
         {
            if (fProfiler) {
               fProfiler->LogUserWindows(flag, flow);
               fProfiler->LogMTQueueLength(fRunInfo);
            }
            delete flow;
            delete flag;
            flow = NULL;
            flag = NULL;
         }
         else 
         {
            MtQueueFlowEvent(mt, i+1, flag, flow, true);
            flow = NULL;
            flag = NULL;
         }
      }

      { //Lock scope
         std::lock_guard<std::mutex> lock(mt->fMtFlowQueueMutex[i]);
         mt->fMtThreadIsRunning[i] = false;
         mt->fMtThreadIsBusy[i] = false;
      }
   }

   void CreateRun(int run_number, const char* file_name)
   {
      assert(fRunInfo == NULL);
      assert(fRunRun.size() == 0);
      
      fRunInfo = new TARunInfo(run_number, file_name, fArgs);

      if (fProfilerEnabled)
         fProfiler = new Profiler( fProfilerIntervalCheck );

      int nModules = (*gModules).size();

      for (int i=0; i<nModules; i++)
         fRunRun.push_back((*gModules)[i]->NewRunObject(fRunInfo));

      if (fMultithreadEnabled) {
         TAMultithreadHelper* mt = new TAMultithreadHelper(nModules);
         fRunInfo->fMtInfo = mt;
         for (int i=0; i<nModules; i++) {
            printf("Create fMtFlowQueue thread %d\n",i);
            mt->fMtThreads[i]=new std::thread(&RunHandler::PerModuleThread,this,i);
         }
      }
   }

   void BeginRun()
   {
      assert(fRunInfo != NULL);
      assert(fRunInfo->fOdb != NULL);
      if (fProfiler)
         fProfiler->Begin(fRunInfo, fRunRun);
      for (unsigned i=0; i<fRunRun.size(); i++)
         fRunRun[i]->BeginRun(fRunInfo);

   }

   void EndRun(TAFlags* flags)
   {
      assert(fRunInfo);

      // make sure the shutdown sequence matches the description in the README file!
      
      // zeroth. Flush events queued for analysis before calling PreEndRun (insure 
      // deterministic behaviour thats the same as in single threaded mode)
      if (fRunInfo->fMtInfo) {
         WaitForAllQueuesEmpty(fRunInfo->fMtInfo);
      }
      // first, call PreEndRun() to tell analysis modules that there will be no more
      // MIDAS events, no more calls to Analyze(). PreEndRun() may generate more
      // flow events, they to into the flow queue or into the multithread queue

      for (unsigned i=0; i<fRunRun.size(); i++)
         fRunRun[i]->PreEndRun(fRunInfo);

      // if in single threaded mode, analyze all queued flow events - call AnalyzeFlowEvent()
      // this can generate additional flow events that will be queued in the queue.

      AnalyzeFlowQueue(flags);

      // if in multi threaded mode, allow all the queues to drain naturally
      // and shutdown the threads

      if (fRunInfo->fMtInfo) {
         WaitForAllQueuesEmpty(fRunInfo->fMtInfo);
         WaitForAllThreadsShutdown(fRunInfo->fMtInfo);
         if (fRunInfo->fMtInfo->fMtQuitRequested) {
            (*flags) |= TAFlag_QUIT;
         }
      }

      // all data analysis is complete
      
      for (unsigned i=0; i<fRunRun.size(); i++)
         fRunRun[i]->EndRun(fRunInfo);

      if (fProfiler) {
         fProfiler->End(fRunInfo);
         fProfiler->Print();
      }
   }

   void NextSubrun()
   {
      assert(fRunInfo);
      
      for (unsigned i=0; i<fRunRun.size(); i++)
         fRunRun[i]->NextSubrun(fRunInfo);
   }

   void DeleteRun()
   {
      assert(fRunInfo);

      for (unsigned i=0; i<fRunRun.size(); i++) {
         delete fRunRun[i];
         fRunRun[i] = NULL;
      }

      fRunRun.clear();
      assert(fRunRun.size() == 0);
      
      if (fProfiler) {
         delete fProfiler;
         fProfiler = NULL;
      }

      delete fRunInfo;
      fRunInfo = NULL;
   }

   void AnalyzeSpecialEvent(TMEvent* event)
   {
      for (unsigned i=0; i<fRunRun.size(); i++)
         fRunRun[i]->AnalyzeSpecialEvent(fRunInfo, event);
   }

   TAFlowEvent* AnalyzeFlowEvent(TAFlags* flags, TAFlowEvent* flow)
   {
      for (unsigned i=0; i<fRunRun.size(); i++) {
         TAClock start_time = TAClockNow();
         flow = fRunRun[i]->AnalyzeFlowEvent(fRunInfo, flags, flow);
         if (fProfiler)
            fProfiler->LogAnalyzeFlowEvent(flags, flow, i, start_time);
         if (!flow)
            break;
         if ((*flags) & TAFlag_SKIP)
               break;
         if ((*flags) & TAFlag_QUIT)
            break;
      }
      return flow;
   }

   void AnalyzeFlowQueue(TAFlags* ana_flags)
   {
      while (1) {
         if (fRunInfo->fMtInfo)
            if (fRunInfo->fMtInfo->fMtQuitRequested) {
               *ana_flags |= TAFlag_QUIT;
               break;
            }

         TAFlowEvent* flow = fRunInfo->ReadFlowQueue();
         if (!flow)
            break;

         int flags = 0;
         flow = AnalyzeFlowEvent(&flags, flow);
         if (flow)
            delete flow;
         if (flags & TAFlag_QUIT) {
            *ana_flags |= TAFlag_QUIT;
            break;
         }
      }
   }

   void AnalyzeEvent(TMEvent* event, TAFlags* flags, TMWriterInterface *writer)
   {
      assert(fRunInfo != NULL);
      assert(fRunInfo->fOdb != NULL);
      assert(event != NULL);
      assert(flags != NULL);

      if (fRunInfo->fMtInfo)
         if (fRunInfo->fMtInfo->fMtQuitRequested) {
            *flags |= TAFlag_QUIT;
            return;
         }
      
      TAFlowEvent* flow = NULL;
                  
      for (unsigned i=0; i<fRunRun.size(); i++) {
         TAClock start_time = TAClockNow();
         flow = fRunRun[i]->Analyze(fRunInfo, event, flags, flow);
         if (fProfiler)
            fProfiler->LogAnalyzeEvent(flags, flow, i, start_time);
         if (*flags & TAFlag_SKIP)
            break;
         if (*flags & TAFlag_QUIT)
            break;
      }

      if (flow) {
         if ((*flags & TAFlag_SKIP)||(*flags & TAFlag_QUIT)) {
            // skip further processing of this event
         } else {
            if (fRunInfo->fMtInfo) {
               MtQueueFlowEvent(fRunInfo->fMtInfo, 0, NULL, flow, true);
               flow = NULL; // ownership passed to the multithread event queue
            } else {
               flow = AnalyzeFlowEvent(flags, flow);
            }
         }
      }

      if (fProfiler && !fRunInfo->fMtInfo) {
         fProfiler->LogUserWindows(flags, flow);
      }

      if (*flags & TAFlag_WRITE)
         if (writer)
            TMWriteEvent(writer, event);
      
      if (flow)
         delete flow;

      if (*flags & TAFlag_QUIT)
         return;

      if (fRunInfo->fMtInfo)
         if (fRunInfo->fMtInfo->fMtQuitRequested) {
            *flags |= TAFlag_QUIT;
            return;
         }

      AnalyzeFlowQueue(flags);
   }
};

TAFlowEvent* TARunInfo::ReadFlowQueue()
{
   if (fFlowQueue.empty())
      return NULL;

   TAFlowEvent* flow = fFlowQueue.front();
   fFlowQueue.pop_front();
   return flow;
}

void TARunInfo::AddToFlowQueue(TAFlowEvent* flow)
{
   if (fMtInfo) {
      // call MtQueueFlowEvent with wait=false to avoid deadlock
      MtQueueFlowEvent(fMtInfo, 0, NULL, flow, false);
   } else {
      fFlowQueue.push_back(flow);
   }
}

#ifdef HAVE_MIDAS

#ifdef HAVE_TMFE

#ifdef HAVE_ROOT
#include "TSystem.h"
#endif

#include "tmfe.h"

static bool gRunStartRequested = false;
static bool gRunStopRequested = false;

class RpcHandler: public TMFeRpcHandlerInterface
{
   TMFeResult HandleBeginRun(int run_number)
   {
      printf("RpcHandler::HandleBeginRun(%d)\n", run_number);
      gRunStartRequested = true;
      gRunStopRequested = false;
      return TMFeOk();
   }

   TMFeResult HandleEndRun(int run_number)
   {
      printf("RpcHandler::HandleEndRun(%d)\n", run_number);
      gRunStartRequested = false;
      gRunStopRequested = true;
      return TMFeOk();
   }

   TMFeResult HandleStartAbortRun(int run_number)
   {
      printf("RpcHandler::HandleStartAbortRun(%d)\n", run_number);
      // run did not really start, pretend it started and immediately ended
      gRunStartRequested = false;
      gRunStopRequested = true;
      return TMFeOk();
   }

   TMFeResult HandleRpc(const char* cmd, const char* args, std::string& result)
   {
      return TMFeOk();
   }
};

TMFeResult ReceiveEvent(TMEventBuffer* b, TMEvent *e, int timeout_msec = 0)
{
   assert(b != NULL);
   assert(e != NULL);
   
   e->Reset();
   
   TMFeResult r = b->ReceiveEvent(&e->data, timeout_msec);
   
   if (r.error_flag)
      return r;

   if (e->data.size() == 0)
      return TMFeOk();
   
   e->ParseEvent();
   
   assert(e->data.size() == e->event_header_size + e->data_size);
   
   return TMFeOk();
}

static int ProcessMidasOnlineTmfe(const std::vector<std::string>& args, const char* progname, const char* hostname, const char* exptname, const char* bufname, int event_id, int trigger_mask, const char* sampling_type_string, int num_analyze, TMWriterInterface* writer, bool multithread, bool profiler, int queue_interval_check)
{
   TMFE *mfe = TMFE::Instance();

   TMFeResult r = mfe->Connect(progname, hostname, exptname);

   if (r.error_flag) {
      fprintf(stderr, "Cannot connect to MIDAS: %s\n", r.error_message.c_str());
      return -1;
   }

   //MVOdb* odb = mfe->fOdbRoot;

   TMEventBuffer *b = new TMEventBuffer(mfe);

   /* open event buffer */

   r = b->OpenBuffer(bufname);

   if (r.error_flag) {
      fprintf(stderr, "Cannot open event buffer \"%s\": %s\n", bufname, r.error_message.c_str());
      return -1;
   }

   /* request read cache */
   size_t cache_size = 100000;
   if(!strcmp(sampling_type_string,"GET_RECENT"))
      cache_size=0;
   r = b->SetCacheSize(cache_size, 0);

   if (r.error_flag) {
      fprintf(stderr, "Cannot set cache size on event buffer \"%s\": %s\n", bufname, r.error_message.c_str());
      return -1;
   }

   /* request events */
   
   r = b->AddRequest(event_id, trigger_mask, sampling_type_string);

   if (r.error_flag) {
      fprintf(stderr, "Cannot add event request on event buffer \"%s\": %s\n", bufname, r.error_message.c_str());
      return -1;
   }

   RpcHandler* h = new RpcHandler();

   mfe->AddRpcHandler(h);

   mfe->DeregisterTransitionPause();
   mfe->DeregisterTransitionResume();
   mfe->SetTransitionSequenceStart(300);
   mfe->SetTransitionSequenceStop(700);

   mfe->StartRpcThread();

   /* reqister event requests */

   RunHandler rh(args, multithread, profiler, queue_interval_check);

   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Init(args);

   if (mfe->fStateRunning) {
      rh.CreateRun(mfe->fRunNumber, NULL);
      rh.fRunInfo->fOdb = mfe->fOdbRoot;
      rh.BeginRun();
   }

   TMEvent e;

   while (!mfe->fShutdownRequested) {
      bool do_sleep = true;

      if (gRunStartRequested) {
         gRunStartRequested = false;

         if (rh.fRunInfo) {
            TAFlags flags = 0;
            rh.EndRun(&flags);
            rh.fRunInfo->fOdb = NULL;
            rh.DeleteRun();
         }
         
         rh.CreateRun(mfe->fRunNumber, NULL);
         rh.fRunInfo->fOdb = mfe->fOdbRoot;
         rh.BeginRun();

         continue;
      }

      if (gRunStopRequested) {
         gRunStopRequested = false;

         if (rh.fRunInfo) {
            TAFlags flags = 0;
            rh.EndRun(&flags);
            rh.fRunInfo->fOdb = NULL;
            rh.DeleteRun();
         }

         continue;
      }

      //r = buf.ReceiveEvent(&e, BM_NO_WAIT);
      //r = buf.ReceiveEvent(&e, BM_WAIT);
      //r = buf.ReceiveEvent(&e, 8000);
      //r = buf.ReceiveEvent(&e, 5000);
      r = ReceiveEvent(b, &e, 100);

      if (r.error_flag) {
         fprintf(stderr, "Cannot read event on event buffer \"%s\": %s\n", bufname, r.error_message.c_str());
         break;
      }

      //e.PrintHeader();
      //::sleep(1);

      if ((e.data_size > 0) && (rh.fRunInfo != NULL)) {

         //e.PrintHeader();
         //e.PrintBanks(2);

         TAFlags flags = 0;
      
         rh.AnalyzeEvent(&e, &flags, writer);

         if (flags & TAFlag_QUIT) {
            mfe->fShutdownRequested = true;
         }

         if (num_analyze > 0) {
            num_analyze--;
            if (num_analyze == 0) {
               mfe->fShutdownRequested = true;
            }
         }

         do_sleep = false;
      }
      
#ifdef HAVE_THTTP_SERVER
      if (TARootHelper::fgHttpServer) {
         int nreq = TARootHelper::fgHttpServer->ProcessRequests();
         if (nreq > 0) {
            do_sleep = false;
            //printf("ProcessRequests() returned %d\n", nreq);
         }
      }
#endif
#ifdef HAVE_ROOT
      if (TARootHelper::fgApp) {
         gSystem->DispatchOneEvent(kTRUE);
      }
#endif

      //printf("do_sleep %d\n", do_sleep);

      if (do_sleep) {
         mfe->Yield(0.010);
      } else {
         mfe->Yield(0);
      }
   }

   if (rh.fRunInfo) {
      TAFlags flags = 0;
      rh.EndRun(&flags);
      rh.fRunInfo->fOdb = NULL;
      rh.DeleteRun();
   }

   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Finish();

   /* close event buffer */
   r = b->CloseBuffer();

   delete b;
   b = NULL;

   if (r.error_flag) {
      fprintf(stderr,"Cannot close event buffer \"%s\": %s\n", bufname, r.error_message.c_str());
      //return -1;
   }
   
   /* disconnect from experiment */
   mfe->Disconnect();

   return 0;
}

#else

#include "TMidasOnline.h"

#ifdef HAVE_ROOT
#include "TSystem.h"
#endif

class OnlineHandler: public TMHandlerInterface
{
public:
   RunHandler fRun;
   int fNumAnalyze = 0;
   TMWriterInterface* fWriter = NULL;
   bool fQuit = false;
   MVOdb* fOdb = NULL;

   OnlineHandler(int num_analyze, TMWriterInterface* writer, MVOdb* odb, const std::vector<std::string>& args, bool multithread, bool profiler, int queue_interval_check) // ctor
      : fRun(args, multithread, profiler, queue_interval_check)
   {
      fNumAnalyze = num_analyze;
      fWriter = writer;
      fOdb = odb;
   }

   ~OnlineHandler() // dtor
   {
      fWriter = NULL;
      fOdb = NULL;
   }

   void StartRun(int run_number)
   {
      fRun.CreateRun(run_number, NULL);
      fRun.fRunInfo->fOdb = fOdb;
      fRun.BeginRun();
   }

   void Transition(int transition, int run_number, int transition_time)
   {
      //printf("OnlineHandler::Transtion: transition %d, run %d, time %d\n", transition, run_number, transition_time);
      
      if (transition == TR_START) {
         if (fRun.fRunInfo) {
            TAFlags flags = 0;
            fRun.EndRun(&flags);
            if (flags & TAFlag_QUIT)
               fQuit = true;
            fRun.fRunInfo->fOdb = NULL;
            fRun.DeleteRun();
         }
         assert(fRun.fRunInfo == NULL);

         StartRun(run_number);
         printf("Begin run: %d\n", run_number);
      } else if (transition == TR_STOP) {
         TAFlags flags = 0;
         fRun.EndRun(&flags);
         if (flags & TAFlag_QUIT)
            fQuit = true;
         fRun.fRunInfo->fOdb = NULL;
         fRun.DeleteRun();
         printf("End of run %d\n", run_number);
      }
   }

   void Event(const void* data, int data_size)
   {
      //printf("OnlineHandler::Event: ptr %p, size %d\n", data, data_size);

      if (!fRun.fRunInfo) {
         StartRun(0); // start fake run for events outside of a run
      }

      TMEvent* event = new TMEvent(data, data_size);

      TAFlags flags = 0;
      
      fRun.AnalyzeEvent(event, &flags, fWriter);

      if (flags & TAFlag_QUIT)
         fQuit = true;

      if (fNumAnalyze > 0) {
         fNumAnalyze--;
         if (fNumAnalyze == 0)
            fQuit = true;
      }

      if (event) {
         delete event;
         event = NULL;
      }
   }
};

static int ProcessMidasOnlineOld(const std::vector<std::string>& args, const char* hostname, const char* exptname, int num_analyze, TMWriterInterface* writer, bool multithread, bool profiler, int queue_interval_check)
{
   TMidasOnline *midas = TMidasOnline::instance();

   int err = midas->connect(hostname, exptname, "rootana");
   if (err != 0) {
      fprintf(stderr,"Cannot connect to MIDAS, error %d\n", err);
      return -1;
   }

   MVOdb* odb = MakeMidasOdb(midas->fDB);
   
   OnlineHandler* h = new OnlineHandler(num_analyze, writer, odb, args, multithread, profiler, queue_interval_check);

   midas->RegisterHandler(h);
   midas->registerTransitions();

   /* reqister event requests */

   midas->eventRequest("SYSTEM",-1,-1,(1<<1));

   int run_number = 0; // midas->odbReadInt("/runinfo/Run number");
   int run_state  = 0; // midas->odbReadInt("/runinfo/State");

   odb->RI("runinfo/run number", &run_number);
   odb->RI("runinfo/state", &run_state);

   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Init(args);

   if ((run_state == STATE_RUNNING)||(run_state == STATE_PAUSED)) {
      h->StartRun(run_number);
   }

   while (!h->fQuit) {
#ifdef HAVE_THTTP_SERVER
      if (TARootHelper::fgHttpServer) {
         TARootHelper::fgHttpServer->ProcessRequests();
      }
#endif
#ifdef HAVE_ROOT
      if (TARootHelper::fgApp) {
         gSystem->DispatchOneEvent(kTRUE);
      }
#endif
      if (!TMidasOnline::instance()->poll(10))
         break;
   }

   if (h->fRun.fRunInfo) {
      TAFlags flags = 0;
      h->fRun.EndRun(&flags);
      h->fRun.fRunInfo->fOdb = NULL;
      h->fRun.DeleteRun();
   }

   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Finish();

   delete h; h = NULL;
   delete odb; odb = NULL;
   
   /* disconnect from experiment */
   midas->disconnect();

   return 0;
}

#endif
#endif

std::vector<std::string> TARunInfo::fgFileList;
int TARunInfo::fgCurrentFileIndex = 0;

static int ProcessMidasFiles(const std::vector<std::string>& files, const std::vector<std::string>& args, int num_skip, int num_analyze, TMWriterInterface* writer, bool multithread, bool profiler, int queue_interval_check)
{
   int number_of_missing_files = 0;

   TARunInfo::fgFileList.clear();

   for (unsigned i=0; i<files.size(); i++)
      TARunInfo::fgFileList.push_back(files[i]);
   
   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Init(args);

   RunHandler run(args, multithread, profiler, queue_interval_check);

   bool done = false;

   for (TARunInfo::fgCurrentFileIndex = 0;
        TARunInfo::fgCurrentFileIndex < (int)TARunInfo::fgFileList.size();
        TARunInfo::fgCurrentFileIndex++) {
      std::string filename = TARunInfo::fgFileList[TARunInfo::fgCurrentFileIndex];

      TMReaderInterface *reader = TMNewReader(filename.c_str());

      if (reader->fError) {
         printf("Could not open \"%s\", error: %s\n", filename.c_str(), reader->fErrorString.c_str());
         delete reader;
         number_of_missing_files++;
         continue;
      }

      while (1) {
         TMEvent* event = TMReadEvent(reader);

         if (!event) // EOF
            break;

         if (event->error) {
            delete event;
            break;
         }

         if (event->event_id == 0x8000) // begin of run event
            {
               int runno = event->serial_number;

               if (run.fRunInfo) {
                  if (run.fRunInfo->fRunNo == runno) {
                     // next subrun file, nothing to do
                     run.fRunInfo->fFileName = filename;
                     run.NextSubrun();
                  } else {
                     // file with a different run number
                     TAFlags flags = 0;
                     run.EndRun(&flags);
                     if (flags & TAFlag_QUIT) {
                        done = true;
                     }
                     run.DeleteRun();
                  }
               }

               if (!run.fRunInfo) {
                  run.CreateRun(runno, filename.c_str());
                  run.fRunInfo->fOdb = MakeFileDumpOdb(event->GetEventData(), event->data_size);
                  run.BeginRun();
               }

               assert(run.fRunInfo);

               run.AnalyzeSpecialEvent(event);

               if (writer)
                  TMWriteEvent(writer, event);
            }
         else if (event->event_id == 0x8001) // end of run event
            {
               //int runno = event->serial_number;
               run.AnalyzeSpecialEvent(event);
               if (writer)
                  TMWriteEvent(writer, event);

               if (run.fRunInfo->fOdb) {
                  delete run.fRunInfo->fOdb;
                  run.fRunInfo->fOdb = NULL;
               }
               
               run.fRunInfo->fOdb = MakeFileDumpOdb(event->GetEventData(), event->data_size);
            }
         else if (event->event_id == 0x8002) // message event
            {
               run.AnalyzeSpecialEvent(event);
               if (writer)
                  TMWriteEvent(writer, event);
            }
         else
            {
               if (!run.fRunInfo) {
                  // create a fake begin of run
                  run.CreateRun(0, filename.c_str());
                  run.fRunInfo->fOdb = MakeNullOdb();
                  run.BeginRun();
               }

               if (num_skip > 0) {
                  num_skip--;
               } else {
                  TAFlags flags = 0;

                  run.AnalyzeEvent(event, &flags, writer);

                  if (flags & TAFlag_QUIT)
                     done = true;

                  if (num_analyze > 0) {
                     num_analyze--;
                     if (num_analyze == 0)
                        done = true;
                  }
               }
            }

         delete event;

         if (done)
            break;

#ifdef HAVE_ROOT
         if (TARootHelper::fgApp) {
            gSystem->DispatchOneEvent(kTRUE);
         }
#endif
      }

      reader->Close();
      delete reader;

      if (done)
         break;
   }

#ifdef HAVE_THTTP_SERVER
   if (0 && TARootHelper::fgHttpServer) {
      while (1) {
         gSystem->DispatchOneEvent(kTRUE);
         //sleep(1);
      }
   }
#endif

   if (run.fRunInfo) {
      TAFlags flags = 0;
      run.EndRun(&flags);
      if (flags & TAFlag_QUIT)
         done = true;
      run.DeleteRun();
   }
   
   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Finish();
   if (number_of_missing_files)
   {
      printf("%d midas files were not openable\n",number_of_missing_files);
      return number_of_missing_files;
   }
   return 0;
}

static int ProcessDemoMode(const std::vector<std::string>& args, int num_skip, int num_analyze, TMWriterInterface* writer, bool multithread, bool profiler, int queue_interval_check)
{
   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Init(args);

   RunHandler run(args, multithread, profiler, queue_interval_check);

   bool done = false;

   int runno = 1;
   
   for (unsigned i=0; true; i++) {
      char s[256];
      snprintf(s, sizeof(s), "%03d", i);
      std::string filename = std::string("demo_subrun_") + s;

      if (!run.fRunInfo) {
         run.CreateRun(runno, filename.c_str());
         run.fRunInfo->fOdb = MakeNullOdb();
         run.BeginRun();
      }

      // we do not generate a fake begin of run event...
      //run.AnalyzeSpecialEvent(event);

      // only switch subruns after the first subrun file
      if (i>0) {
         run.fRunInfo->fFileName = filename;
         run.NextSubrun();
      }

      TMEvent* event = new TMEvent();

      for (unsigned j=0; j<100; j++) {
         event->Init(0x0001, 0xFFFF, j+1, 0, 0);
         uint32_t test_data[] = { 0x11112222, 0x33334444, 0x55556666, 0x77778888 };
         event->AddBank("TEST", TID_DWORD, (const char*)test_data, sizeof(test_data));

         if (num_skip > 0) {
            num_skip--;
         } else {
            TAFlags flags = 0;
            
            run.AnalyzeEvent(event, &flags, writer);
            
            if (flags & TAFlag_QUIT)
               done = true;
            
            if (num_analyze > 0) {
               num_analyze--;
               if (num_analyze == 0)
                  done = true;
            }
         }

         if (done)
            break;

#ifdef HAVE_ROOT
         if (TARootHelper::fgApp) {
            gSystem->DispatchOneEvent(kTRUE);
         }
#endif
      }

      delete event;

      // we do not generate a fake end of run event...
      //run.AnalyzeSpecialEvent(event);

      if (done)
         break;
   }

   if (run.fRunInfo) {
      TAFlags flags = 0;
      run.EndRun(&flags);
      run.DeleteRun();
   }
   
   for (unsigned i=0; i<(*gModules).size(); i++)
      (*gModules)[i]->Finish();
   
   return 0;
}

static bool gEnableShowMem = false;

#if 0
static int ShowMem(const char* label)
{
  if (!gEnableShowMem)
    return 0;

  FILE* fp = fopen("/proc/self/statm","r");
  if (!fp)
    return 0;

  int mem = 0;
  fscanf(fp,"%d",&mem);
  fclose(fp);

  if (label)
    printf("memory at %s is %d\n", label, mem);

  return mem;
}
#endif

class EventDumpModule: public TARunObject
{
public:
   EventDumpModule(TARunInfo* runinfo)
      : TARunObject(runinfo)
   {
      if (gTrace)
         printf("EventDumpModule::ctor, run %d\n", runinfo->fRunNo);
   }
   
   ~EventDumpModule()
   {
      if (gTrace)
         printf("EventDumpModule::dtor!\n");
   }

   void BeginRun(TARunInfo* runinfo)
   {
      printf("EventDumpModule::BeginRun, run %d\n", runinfo->fRunNo);
   }

   void EndRun(TARunInfo* runinfo)
   {
      printf("EventDumpModule::EndRun, run %d\n", runinfo->fRunNo);
   }

   void NextSubrun(TARunInfo* runinfo)
   {
      printf("EventDumpModule::NextSubrun, run %d, file %s\n", runinfo->fRunNo, runinfo->fFileName.c_str());
   }

   void PauseRun(TARunInfo* runinfo)
   {
      printf("EventDumpModule::PauseRun, run %d\n", runinfo->fRunNo);
   }

   void ResumeRun(TARunInfo* runinfo)
   {
      printf("EventDumpModule::ResumeRun, run %d\n", runinfo->fRunNo);
   }

   TAFlowEvent* Analyze(TARunInfo* runinfo, TMEvent* event, TAFlags* flags, TAFlowEvent* flow)
   {
      printf("EventDumpModule::Analyze, run %d, ", runinfo->fRunNo);
      event->FindAllBanks();
      std::string h = event->HeaderToString();
      std::string b = event->BankListToString();
      printf("%s: %s\n", h.c_str(), b.c_str());
      return flow;
   }

   void AnalyzeSpecialEvent(TARunInfo* runinfo, TMEvent* event)
   {
      printf("EventDumpModule::AnalyzeSpecialEvent, run %d, event serno %d, id 0x%04x, data size %d\n", runinfo->fRunNo, event->serial_number, (int)event->event_id, event->data_size);
   }
};

class EventDumpModuleFactory: public TAFactory
{
public:

   void Init(const std::vector<std::string> &args)
   {
      if (gTrace)
         printf("EventDumpModuleFactory::Init!\n");
   }
   
   void Finish()
   {
      if (gTrace)
         printf("EventDumpModuleFactory::Finish!\n");
   }
   
   TARunObject* NewRunObject(TARunInfo* runinfo)
   {
      if (gTrace)
         printf("EventDumpModuleFactory::NewRunObject, run %d, file %s\n", runinfo->fRunNo, runinfo->fFileName.c_str());
      return new EventDumpModule(runinfo);
   }
};

#ifdef HAVE_ROOT
#include <TGMenu.h>
#include <TGButton.h>
#include <TBrowser.h>

#define CTRL_QUIT 1
#define CTRL_NEXT 2
#define CTRL_CONTINUE 3
#define CTRL_PAUSE    4
#define CTRL_NEXT_FLOW 5

#define CTRL_TBROWSER 11

class ValueHolder
{
public:
   int fValue;

   ValueHolder() // ctor
   {
      fValue = 0;
   }
};

class TextButton: public TGTextButton
{
public:
   ValueHolder* fHolder;
   int    fValue;
   
   TextButton(TGWindow*p, const char* text, ValueHolder* holder, int value) // ctor
      : TGTextButton(p, text)
   {
      fHolder = holder;
      fValue = value;
   }

#if 0
   void Pressed()
   {
      printf("Pressed!\n");
   }
   
   void Released()
   {
      printf("Released!\n");
   }
#endif
   
   void Clicked()
   {
      //printf("Clicked button %s, value %d!\n", GetString().Data(), fValue);
      if (fHolder)
         fHolder->fValue = fValue;
      //gSystem->ExitLoop();
   }
};

class MainWindow: public TGMainFrame
{
public:
   TGPopupMenu*		fMenu;
   TGMenuBar*		fMenuBar;
   TGLayoutHints*	fMenuBarItemLayout;

   TGCompositeFrame*    fButtonsFrame;

   ValueHolder* fHolder;

   TextButton* fNextButton;
   TextButton* fNextFlowButton;
   TextButton* fContinueButton;
   TextButton* fPauseButton;

   TextButton* fQuitButton;
  
public:
   MainWindow(const TGWindow*w, int s1, int s2, ValueHolder* holder) // ctor
      : TGMainFrame(w, s1, s2)
   {
      if (gTrace)
         printf("MainWindow::ctor!\n");

      fHolder = holder;
      //SetCleanup(kDeepCleanup);
   
      SetWindowName("ROOT Analyzer Control");

      // layout the gui
      fMenu = new TGPopupMenu(gClient->GetRoot());
      fMenu->AddEntry("New TBrowser", CTRL_TBROWSER);
      fMenu->AddEntry("-", 0);
      fMenu->AddEntry("Next",     CTRL_NEXT);
      fMenu->AddEntry("NextFlow", CTRL_NEXT_FLOW);
      fMenu->AddEntry("Continue", CTRL_CONTINUE);
      fMenu->AddEntry("Pause",    CTRL_PAUSE);
      fMenu->AddEntry("-", 0);
      fMenu->AddEntry("Quit",     CTRL_QUIT);

      fMenuBarItemLayout = new TGLayoutHints(kLHintsTop|kLHintsLeft, 0, 4, 0, 0);

      fMenu->Associate(this);

      fMenuBar = new TGMenuBar(this, 1, 1, kRaisedFrame);
      fMenuBar->AddPopup("&Rootana", fMenu, fMenuBarItemLayout);
      fMenuBar->Layout();

      AddFrame(fMenuBar, new TGLayoutHints(kLHintsTop|kLHintsLeft|kLHintsExpandX));

      fButtonsFrame = new TGVerticalFrame(this);

      fNextButton = new TextButton(fButtonsFrame, "Next", holder, CTRL_NEXT);
      fNextFlowButton = new TextButton(fButtonsFrame, "Next Flow Event", holder, CTRL_NEXT_FLOW);

      fButtonsFrame->AddFrame(fNextButton, new TGLayoutHints(kLHintsExpandX, 1, 1, 1, 1));
      fButtonsFrame->AddFrame(fNextFlowButton, new TGLayoutHints(kLHintsExpandX, 1, 1, 1, 1));

      TGHorizontalFrame *hframe = new TGHorizontalFrame(fButtonsFrame);

      fContinueButton = new TextButton(hframe, " Continue ", holder, CTRL_CONTINUE);
      fPauseButton = new TextButton(hframe, " Pause ", holder, CTRL_PAUSE);

      hframe->AddFrame(fContinueButton, new TGLayoutHints(kLHintsExpandX, 1, 1, 1, 1));
      hframe->AddFrame(fPauseButton, new TGLayoutHints(kLHintsExpandX, 1, 1, 1, 1));

      fButtonsFrame->AddFrame(hframe, new TGLayoutHints(kLHintsExpandX));

      fQuitButton = new TextButton(fButtonsFrame, "Quit ", holder, CTRL_QUIT);
      fButtonsFrame->AddFrame(fQuitButton, new TGLayoutHints(kLHintsExpandX, 1, 1, 1, 1));
   
      AddFrame(fButtonsFrame, new TGLayoutHints(kLHintsExpandX));

      MapSubwindows(); 
      Layout();
      Resize(GetDefaultSize());
      MapWindow();
   }

   ~MainWindow() // dtor // Closing the control window closes the whole program
   {
      if (gTrace)
         printf("MainWindow::dtor!\n");

      delete fMenu;
      delete fMenuBar;
      delete fMenuBarItemLayout;
   }

   void CloseWindow()
   {
      if (gTrace)
         printf("MainWindow::CloseWindow()\n");

      if (fHolder)
         fHolder->fValue = CTRL_QUIT;
      //gSystem->ExitLoop();
   }
  
   Bool_t ProcessMessage(Long_t msg, Long_t parm1, Long_t parm2)
   {
      //printf("GUI Message %d %d %d\n",(int)msg,(int)parm1,(int)parm2);
      switch (GET_MSG(msg))
         {
         default:
            break;
         case kC_COMMAND:
            switch (GET_SUBMSG(msg))
               {
               default:
                  break;
               case kCM_MENU:
                  //printf("parm1 %d\n", (int)parm1);
                  switch (parm1)
                     {
                     case CTRL_TBROWSER:
                        new TBrowser();
                        break;
                     default:
                        //printf("Control %d!\n", (int)parm1);
                        if (fHolder)
                           fHolder->fValue = parm1;
                        //gSystem->ExitLoop();
                        break;
                     }
                  break;
               }
            break;
         }

      return kTRUE;
   }
};
#endif

class InteractiveModule: public TARunObject
{
public:
   bool fContinue;
   bool fNextFlow;
   int  fSkip;
#ifdef HAVE_ROOT
   static ValueHolder* fgHolder;
   static MainWindow *fgCtrlWindow;
#endif
   
   InteractiveModule(TARunInfo* runinfo)
      : TARunObject(runinfo)
   {
      if (gTrace)
         printf("InteractiveModule::ctor, run %d\n", runinfo->fRunNo);
      fContinue = false;
      fNextFlow = false;
      fSkip = 0;
#ifdef HAVE_ROOT
      if (!fgHolder)
         fgHolder = new ValueHolder;
      if (!fgCtrlWindow && runinfo->fRoot->fgApp) {
         fgCtrlWindow = new MainWindow(gClient->GetRoot(), 200, 300, fgHolder);
      }
#endif
   }
   
   ~InteractiveModule()
   {
      if (gTrace)
         printf("InteractiveModule::dtor!\n");
   }

   void BeginRun(TARunInfo* runinfo)
   {
      printf("InteractiveModule::BeginRun, run %d\n", runinfo->fRunNo);
   }

   void EndRun(TARunInfo* runinfo)
   {
      printf("InteractiveModule::EndRun, run %d\n", runinfo->fRunNo);

#ifdef HAVE_ROOT
      if (fgCtrlWindow && runinfo->fRoot->fgApp) {
         fgCtrlWindow->fNextButton->SetEnabled(false);
         fgCtrlWindow->fNextFlowButton->SetEnabled(false);
         fgCtrlWindow->fContinueButton->SetEnabled(false);
         fgCtrlWindow->fPauseButton->SetEnabled(false);
         while (1) {
#ifdef HAVE_THTTP_SERVER
            if (TARootHelper::fgHttpServer) {
               TARootHelper::fgHttpServer->ProcessRequests();
            }
#endif
#ifdef HAVE_ROOT
            if (TARootHelper::fgApp) {
               gSystem->DispatchOneEvent(kTRUE);
            }
#endif
#ifdef HAVE_MIDAS
#ifdef HAVE_TMFE
            TMFE* mfe = TMFE::Instance();
            mfe->Yield(0.010);
            if (mfe->fShutdownRequested) {
               return;
            }
#else
            if (!TMidasOnline::instance()->sleep(10)) {
               // FIXME: indicate that we should exit the analyzer
               return;
            }
#endif
#else
            gSystem->Sleep(10);
#endif

            int ctrl = fgHolder->fValue;
            fgHolder->fValue = 0;

            switch (ctrl) {
            case CTRL_QUIT:
               return;
            case CTRL_NEXT:
               return;
            case CTRL_CONTINUE:
               return;
            }
         }
      }
#endif
   }

   void PauseRun(TARunInfo* runinfo)
   {
      printf("InteractiveModule::PauseRun, run %d\n", runinfo->fRunNo);
   }

   void ResumeRun(TARunInfo* runinfo)
   {
      printf("InteractiveModule::ResumeRun, run %d\n", runinfo->fRunNo);
   }

   void InteractiveLoop(TARunInfo* runinfo, TAFlags* flags)
   {
#ifdef HAVE_ROOT
      if (fgCtrlWindow && runinfo->fRoot->fgApp) {
         while (1) {
#ifdef HAVE_THTTP_SERVER
            if (TARootHelper::fgHttpServer) {
               TARootHelper::fgHttpServer->ProcessRequests();
            }
#endif
#ifdef HAVE_ROOT
            if (TARootHelper::fgApp) {
               gSystem->DispatchOneEvent(kTRUE);
            }
#endif
#ifdef HAVE_MIDAS
#ifdef HAVE_TMFE
            TMFE* mfe = TMFE::Instance();
            mfe->Yield(0.010);
            if (mfe->fShutdownRequested) {
               *flags |= TAFlag_QUIT;
               return;
            }
#else
            if (!TMidasOnline::instance()->sleep(10)) {
               *flags |= TAFlag_QUIT;
               return;
            }
#endif
#else
            gSystem->Sleep(10);
#endif

            int ctrl = fgHolder->fValue;
            fgHolder->fValue = 0;

            switch (ctrl) {
            case CTRL_QUIT:
               *flags |= TAFlag_QUIT;
               return;
            case CTRL_NEXT:
               return;
            case CTRL_NEXT_FLOW:
               fNextFlow = true;
               return;
            case CTRL_CONTINUE:
               fContinue = true;
               return;
            }
         }
      }
#endif

      while (1) {
         char str[256];
         fprintf(stdout, "manalyzer> "); fflush(stdout);
         char* s = fgets(str, sizeof(str)-1, stdin);

         if (s == NULL) {
            // EOF
            *flags |= TAFlag_QUIT;
            return;
         }
         
         printf("command [%s]\n", str);

         if (str[0] == 'h') { // "help"
            printf("Interactive manalyzer commands:\n");
            printf(" q - quit\n");
            printf(" h - help\n");
            printf(" c - continue until next TAFlag_DISPLAY event\n");
            printf(" n - next event\n");
            printf(" aNNN - analyze N events, i.e. \"a10\"\n");
         } else if (str[0] == 'q') { // "quit"
            *flags |= TAFlag_QUIT;
            return;
         } else if (str[0] == 'n') { // "next"
            return;
         } else if (str[0] == 'c') { // "continue"
            fContinue = true;
            return;
         } else if (str[0] == 'a') { // "analyze" N events
            int num = atoi(str+1);
            printf("Analyzing %d events\n", num);
            if (num > 0) {
               fSkip = num-1;
            }
            return;
         }
      }
   }

   TAFlowEvent* Analyze(TARunInfo* runinfo, TMEvent* event, TAFlags* flags, TAFlowEvent* flow)
   {
      printf("InteractiveModule::Analyze, run %d, %s\n", runinfo->fRunNo, event->HeaderToString().c_str());

#ifdef HAVE_ROOT
      if (fgHolder->fValue == CTRL_QUIT) {
         *flags |= TAFlag_QUIT;
         return flow;
      } else if (fgHolder->fValue == CTRL_PAUSE) {
         fContinue = false;
      }
#endif

      if ((fContinue||fNextFlow) && !(*flags & TAFlag_DISPLAY)) {
         return flow;
      } else {
         fContinue = false;
      }

      if (fSkip > 0) {
         fSkip--;
         return flow;
      }

      InteractiveLoop(runinfo, flags);

      return flow;
   }

   TAFlowEvent* AnalyzeFlowEvent(TARunInfo* runinfo, TAFlags* flags, TAFlowEvent* flow)
   {
      printf("InteractiveModule::AnalyzeFlowEvent, run %d\n", runinfo->fRunNo);

#ifdef HAVE_ROOT
      if (fgHolder->fValue == CTRL_QUIT) {
         *flags |= TAFlag_QUIT;
         return flow;
      } else if (fgHolder->fValue == CTRL_PAUSE) {
         fContinue = false;
      }
#endif

      if ((!fNextFlow) && !(*flags & TAFlag_DISPLAY)) {
         return flow;
      }

      fNextFlow = false;

      InteractiveLoop(runinfo, flags);
      
      return flow;
   }
   
   void AnalyzeSpecialEvent(TARunInfo* runinfo, TMEvent* event)
   {
      if (gTrace)
         printf("InteractiveModule::AnalyzeSpecialEvent, run %d, event serno %d, id 0x%04x, data size %d\n", runinfo->fRunNo, event->serial_number, (int)event->event_id, event->data_size);
   }
};

#ifdef HAVE_ROOT
MainWindow* InteractiveModule::fgCtrlWindow = NULL;
ValueHolder* InteractiveModule::fgHolder = NULL;
#endif

class InteractiveModuleFactory: public TAFactory
{
public:

   void Init(const std::vector<std::string> &args)
   {
      if (gTrace)
         printf("InteractiveModuleFactory::Init!\n");
   }
   
   void Finish()
   {
      if (gTrace)
         printf("InteractiveModuleFactory::Finish!\n");
   }
   
   TARunObject* NewRunObject(TARunInfo* runinfo)
   {
      if (gTrace)
         printf("InteractiveModuleFactory::NewRunObject, run %d, file %s\n", runinfo->fRunNo, runinfo->fFileName.c_str());
      return new InteractiveModule(runinfo);
   }
};

//////////////////////////////////////////////////////////
//
//                   main program
//
//////////////////////////////////////////////////////////

static void help()
{
  printf("\nUsage: ./manalyzer.exe [-h] [-R8081] [-oOutputfile.mid] [file1 file2 ...] [-- arguments passed to modules ...]\n");
  printf("\n");
  printf("-h: print this help message\n");
  printf("--demo: activate the demo mode, online connection or input file not needed, midas events are generated internally, add -e0 or -eNNN to set number of demo events \n");
  printf("\n");
  printf("-Hhostname: connect to MIDAS experiment on given host\n");
  printf("-Eexptname: connect to this MIDAS experiment\n");
  printf("--midas-progname SSS -- set analyzer's MIDAS program name, default is \"ana\"\n");
  printf("--midas-hostname HOSTNAME[:PORT] -- connect to MIDAS mserver on given host and port\n");
  printf("--midas-exptname EXPTNAME -- connect to given experiment\n");
  printf("--midas-buffer BUFZZZ -- connect to given MIDAS event buffer\n");
  printf("--midas-sampling SSS -- sample events from MIDAS event buffer: GET_ALL=get every event (will block the event writers, GET_NONBLOCKING=get as many as we can process, GET_RECENT=get recent events, see bm_receive_event(). Default is GET_NONBLOCKING\n");
  printf("--midas-event-id III -- receive only events with matching event ID\n");
  printf("--midas-trigger-mask 0xMASK -- receive only events with matching trigger mask\n");
  printf("\n");
  printf("-oOutputfile.mid: write selected events into this file\n");
  printf("-Rnnnn: Start the ROOT THttpServer HTTP server on specified tcp port, use -R8081, access by firefox http://localhost:8081\n");
  printf("-eNNN: Number of events to analyze, 0=unlimited\n");
  printf("-sNNN: Number of events to skip before starting analysis\n");
  printf("\n");
  printf("--dump: activate the event dump module\n");
  printf("\n");
  printf("-t: Enable tracing of constructors, destructors and function calls\n");
  printf("-m: Enable memory leak debugging\n");
  printf("-g: Enable graphics display when processing data files\n");
  printf("-i: Enable intractive mode\n");
  printf("\n");
  printf("--mt: Enable multithreaded mode. Extra multithread config settings:\n");
  printf("--mtqlNNN: Module thread queue length (buffer).              Default: %d\n", gDefaultMultithreadQueueLength);
  printf("--mtseNNN: Module thread sleep time with empty queue (usec). Default: %d\n", gDefaultMultithreadWaitEmpty);
  printf("--mtsfNNN: Module thread sleep time when next queue is full (usec). Default: %d\n", gDefaultMultithreadWaitFull);
  printf("\n");
  printf("--no-profiler: Turn off manalyzer module profiler\n");
  printf("--pqiNNN: Profile multithread queue lengths every NNN events \n");
#ifdef HAVE_ROOT
  printf("\n");
  printf("-Doutputdirectory: Specify output root file directory\n");
  printf("-Ooutputfile.root: Specify output root file filename\n");
#endif
  printf("\n");
  printf("--: All following arguments are passed to the analyzer modules Init() method\n");
  printf("\n");
  printf("Analyzer modules usage:\n");
  if (gModules)
     for (unsigned i=0; i<(*gModules).size(); i++)
        (*gModules)[i]->Usage();
  printf("\n");
  printf("Example1: analyze online data: ./manalyzer.exe -R9091\n");
  printf("Example2: analyze existing data: ./manalyzer.exe /data/alpha/current/run00500.mid\n");
  exit(1);
}

// duplicate c++20 std::string s.starts_with()

static bool starts_with(const std::string& s, const char* prefix)
{
   return (s.substr(0, strlen(prefix)) == prefix);
}

// Main function call

int manalyzer_main(int argc, char* argv[])
{
   setbuf(stdout, NULL);
   setbuf(stderr, NULL);
 
   signal(SIGILL,  SIG_DFL);
   signal(SIGBUS,  SIG_DFL);
   signal(SIGSEGV, SIG_DFL);
   signal(SIGPIPE, SIG_DFL);
   
   std::vector<std::string> args;
   for (int i=0; i<argc; i++) {
      if (strcmp(argv[i],"-h")==0)
         help(); // does not return
      args.push_back(argv[i]);
   }

   int  httpPort = 0;
   int num_skip = 0;
   int num_analyze = 0;

   TMWriterInterface *writer = NULL;

   bool event_dump = false;
   bool demo_mode = false;
#ifdef HAVE_ROOT
   bool root_graphics = false;
#endif
   bool interactive = false;

   bool multithread = false;

   bool performance_profiler = true;
   int snap_shot_queue_length = 100;

   std::vector<std::string> files;
   std::vector<std::string> modargs;

#ifdef HAVE_MIDAS
   std::string midas_hostname = "";
   std::string midas_exptname = "";
   std::string midas_progname = "ana";
   std::string midas_buffer   = "SYSTEM";
   //std::string midas_sampling = "GET_ALL";
   std::string midas_sampling = "GET_NONBLOCKING";
   int midas_event_id = -1;
   int midas_trigger_mask = -1;
#endif

   for (unsigned int i=1; i<args.size(); i++) { // loop over the commandline options
      std::string arg = args[i];
      //printf("argv[%d] is %s\n",i,arg);

      if (arg == "--") {
         for (unsigned j=i+1; j<args.size(); j++)
            modargs.push_back(args[j]);
         break;
      } else if (arg == "--dump") {
         event_dump = true;
      } else if (arg == "--demo") {
         demo_mode = true;
         num_analyze = 100;
#ifdef HAVE_ROOT
      } else if (arg == "-g") {
         root_graphics = true;
#endif
      } else if (arg == "-i") {
         interactive = true;
      } else if (arg == "-t") {
         gTrace = true;
         TMReaderInterface::fgTrace = true;
         TMWriterInterface::fgTrace = true;
      } else if (starts_with(arg, "-o")) {
         writer = TMNewWriter(arg.c_str()+2);
      } else if (starts_with(arg, "-s")) {
         num_skip = atoi(arg.c_str()+2);
      } else if (starts_with(arg, "-e")) {
         num_analyze = atoi(arg.c_str()+2);
      } else if (starts_with(arg, "-m")) { // Enable memory debugging
         gEnableShowMem = true;
      } else if (starts_with(arg, "-R")) { // Set the ROOT THttpServer HTTP port
         httpPort = atoi(arg.c_str()+2);
#ifdef HAVE_MIDAS
      } else if (starts_with(arg, "-H")) {
         midas_hostname = arg.c_str()+2;
      } else if (starts_with(arg, "-E")) {
         midas_exptname = arg.c_str()+2;
      } else if (arg == "--midas-progname") {
         midas_progname = args[i+1]; i++;
      } else if (arg == "--midas-hostname") {
         midas_hostname = args[i+1]; i++;
      } else if (arg == "--midas-exptname") {
         midas_exptname = args[i+1]; i++;
      } else if (arg == "--midas-buffer") {
         midas_buffer = args[i+1]; i++;
      } else if (arg == "--midas-sampling") {
         midas_sampling = args[i+1]; i++;
      } else if (arg == "--midas-event-id") {
         midas_event_id = atoi(args[i+1].c_str()); i++;
      } else if (arg == "--midas-trigger-mask") {
         midas_trigger_mask = strtoul(args[i+1].c_str(), NULL, 0); i++;
#endif
      } else if (starts_with(arg, "--mtql")) {
         gDefaultMultithreadQueueLength = atoi(arg.c_str()+6);
      } else if (starts_with(arg, "--mtse")) {
         gDefaultMultithreadWaitEmpty = atoi(arg.c_str()+6);
      } else if (starts_with(arg, "--mtsf")) {
         gDefaultMultithreadWaitFull = atoi(arg.c_str()+6);
      } else if (arg == "--mt") {
         multithread=true;
      } else if (arg == "--no-profiler") {
         performance_profiler = 0;
      } else if (starts_with(arg, "--pqi")) {
         snap_shot_queue_length = atoi(arg.c_str()+5);
#ifdef HAVE_ROOT
      } else if (starts_with(arg, "-O")) {
         TARootHelper::fOutputFileName = arg.c_str()+2;
      } else if (starts_with(arg, "-D")) {
         TARootHelper::fOutputDirectory = arg.c_str()+2;
#endif
      } else if (arg == "-h") {
         help(); // does not return
      } else if (arg[0] == '-') {
         help(); // does not return
      } else {
         files.push_back(args[i]);
      }
   }

   if (!gModules)
      gModules = new std::vector<TAFactory*>;

   if ((*gModules).size() == 0)
      event_dump = true;

   if (event_dump)
      (*gModules).push_back(new EventDumpModuleFactory);

   if (interactive)
      (*gModules).push_back(new InteractiveModuleFactory);

   printf("Registered modules: %d\n", (int)(*gModules).size());

#ifdef HAVE_ROOT
   if (multithread) {
      // see https://root.cern/manual/multi_threading/
      ROOT::EnableImplicitMT();
      ROOT::EnableThreadSafety();
   }

   if (root_graphics) {
      TARootHelper::fgApp = new TApplication("manalyzer", NULL, NULL, 0, 0);
   }

   TARootHelper::fgDir = new TDirectory("manalyzer", "location of histograms");
   TARootHelper::fgDir->cd();
#endif

   if (httpPort) {
#ifdef HAVE_THTTP_SERVER
      char str[256];
      snprintf(str, sizeof(str), "http:127.0.0.1:%d?cors", httpPort);
      THttpServer *s = new THttpServer(str);
      //s->SetTimer(100, kFALSE);
      TARootHelper::fgHttpServer = s;
#else
      fprintf(stderr,"ERROR: No support for the THttpServer!\n");
#endif
   }
   
   for (unsigned i=0; i<files.size(); i++) {
      printf("file[%d]: %s\n", i, files[i].c_str());
   }
   int exit_state = 0;
   if (demo_mode) {
      exit_state = ProcessDemoMode(modargs, num_skip, num_analyze, writer, multithread, performance_profiler, snap_shot_queue_length);
   } else if (files.size() > 0) {
      exit_state = ProcessMidasFiles(files, modargs, num_skip, num_analyze, writer, multithread, performance_profiler, snap_shot_queue_length);
   } else {
#ifdef HAVE_MIDAS
#ifdef HAVE_TMFE
      exit_state = ProcessMidasOnlineTmfe(modargs, midas_progname.c_str(), midas_hostname.c_str(), midas_exptname.c_str(), midas_buffer.c_str(), midas_event_id, midas_trigger_mask, midas_sampling.c_str(), num_analyze, writer, multithread, performance_profiler, snap_shot_queue_length);
#else
      exit_state = ProcessMidasOnlineOld(modargs, midas_hostname.c_str(), midas_exptname.c_str(), num_analyze, writer, multithread, performance_profiler, snap_shot_queue_length);
#endif
#endif
   }

   if (writer) {
      writer->Close();
      delete writer;
      writer = NULL;
   }

   return exit_state;
}

/* emacs
 * Local Variables:
 * tab-width: 8
 * c-basic-offset: 3
 * indent-tabs-mode: nil
 * End:
 */