beamtime/cern-oct12/sorter/oct12-sorter.cxx (r4864/r3699)

Go to the documentation of this file.

// Small read and merge program, for FASP data

#include "dabc/logging.h"
#include "dabc/timing.h"
#include "dabc/ModuleSync.h"
#include "dabc/Port.h"
#include "dabc/Timer.h"
#include "dabc/Command.h"
#include "dabc/Manager.h"
#include "dabc/MemoryPool.h"
#include "dabc/PoolHandle.h"
#include "dabc/threads.h"
#include "dabc/Application.h"
#include "dabc/SocketDevice.h"
#include "dabc/statistic.h"
#include "dabc/Factory.h"
#include "dabc/Configuration.h"

#include "mbs/MbsTypeDefs.h"
#include "mbs/Iterator.h"

#include <string>
#include <vector>

enum ERocMbsTypes {
   proc_RocEvent     =  1,   
   proc_ErrEvent     =  2,   
   proc_MergedEvent  =  3,   
   proc_RawData      =  4,   
   proc_Triglog      =  5,   
   proc_TRD_MADC     =  6,   
   proc_TRD_Spadic   =  7,   
   proc_CERN_Oct11   =  8,   
   proc_SlaveMbs     =  9,   
   proc_EPICS        = 10,   
   proc_COSY_Nov11   = 11,   
   proc_SpadicV10Raw = 12,   
   proc_SpadicV10Event = 13, 
   proc_CERN_Oct12   = 14,    
   proc_FASP         = 15,    
   proc_TRBEvent     = 31     
};

std::string GetHeaderInfo(uint32_t id) 
{
   switch (id & 0xff) {
   case proc_RocEvent    : return dabc::format("ROC%02d", (id >> 16) & 0xff);
   case proc_ErrEvent    : return "ErrEvent";
   case proc_MergedEvent : return "Merged";
   case proc_RawData     : return "RawData";
   case proc_Triglog     : return "Triglog-M";
   case proc_TRD_MADC    : return "Cern-nov10";
   case proc_TRD_Spadic  : return dabc::format("SPADIC%02d", (id >> 16) & 0xff);
   case proc_CERN_Oct11  : return dabc::format("CERN-%s", (id >> 24) == 0x9 ? "M" : "S");
   case proc_SlaveMbs    : return "Triglog-S";
   case proc_EPICS       : return "EPICS";
   case proc_CERN_Oct12  : return dabc::format("CERN-%s", (id >> 24) == 0x9 ? "M" : "S");
   case proc_FASP        : return "FASP";
   case proc_TRBEvent    : return "TRB";
   }
   return "undef";
}

int last_master(0), last_slave(0), last_mastertm(0), mbseventid(0);

int GetEventNumber(uint32_t id, void* dataptr, int datalen)
{
   uint8_t* data = (uint8_t*) dataptr;
   switch (id & 0xff) {
   case proc_RocEvent: {
      int ownid =  (*((uint32_t*)  (data+8)) >> 6) & 0xffffff;
      int ownid2 =  (*((uint32_t*)  (data+datalen - 4)) >> 6) & 0xffffff;
      if (ownid+1 != ownid2) EOUT(("ROC SYNC PANICK"));
      return ownid;
   }
   case proc_TRD_Spadic: {
      int ownid = data[360] << 16 | data[361] << 8 | data[362];
      return ownid;
   }
   case proc_Triglog: {
      int ownid = *((uint32_t*)dataptr);
      last_master = ownid;
      last_mastertm = *((uint32_t*) (data + 8));
      return ownid;
   }
   case proc_SlaveMbs: {
      int ownid = *((uint32_t*)dataptr);
      last_slave = ownid;
      return ownid;
   }
   case proc_CERN_Oct12  :
      return (id >> 24) == 0x9 ? last_master : last_slave;

   case proc_EPICS:
      return last_master;

   case proc_FASP:
      return last_master;

   case proc_TRBEvent:
      return last_master;
   }

   EOUT(("Cannot define event ID"));

   return 0;
}


#include <map> 

typedef std::map<uint32_t,int> SubeventsMap;

class CernOct12SorterModule : public dabc::ModuleSync {
protected:

   int fBufferSize;
   bool fDoFASP;
   bool fDoReading;
   bool fDoWriting;  // write merged file
   bool fOnlyFull;  //  produce only full events


public:
   CernOct12SorterModule(const char* name, dabc::Command cmd) :
      dabc::ModuleSync(name, cmd)
   {
      fBufferSize = CreatePar(dabc::xmlBufferSize).AsInt(16*1024);

      fDoFASP = CreatePar("DoFASP").AsBool(false);
      fDoReading = CreatePar("DoReading").AsBool(false);
      fDoWriting = CreatePar("DoWriting").AsBool(false);
      fOnlyFull = CreatePar("OnlyFull").AsBool(false);

      CreatePar("RunName");

      CreatePoolHandle("Pool");

      if (fDoReading) CreateInput("Input", Pool());
      if (fDoFASP) CreateInput("FASP", Pool());
      if (fDoWriting) CreateOutput("Output", Pool());
   }

   virtual void MainLoop()
   {
      if (!fDoReading && fDoWriting && fDoFASP) {
         FaspLoop();
         return;
      }


      if (!fDoWriting) 
         if (fDoFASP ^ fDoReading) {
            NormalLoop();
            return;
         }

      if (fDoFASP && fDoReading)
         CombinerLoop();

      DOUT0(("Nothing to do"));
   }

   void FaspLoop()
   {
      // this is just for converting of the FASP files to LMD

      dabc::Buffer buf;

      int fullsz(0), fullcnt(0);

      bool iseof(false);

      while (!iseof) {
         buf = Recv(Input(), 5.);

         fullsz += buf.GetTotalSize();

         if (buf.GetTypeId() == mbs::mbt_MbsEvents)
            fullcnt += mbs::ReadIterator::NumEvents(buf);

         iseof = (buf.GetTypeId() == dabc::mbt_EOF);

         if (fDoWriting) {
            // DOUT0(("Write buffer"));
            Send(Output(), buf.HandOver());
         }
      }

      if (iseof)
         DOUT0(("END-OF-FILE detected!!"));

      DOUT0(("Totally received bytes %d  events %d", fullsz, fullcnt));

      if (fDoWriting)
         while (Output()->OutputPending() > 0) {
            DOUT0(("Waiting until data is written %d", Output()->OutputPending()));
            WorkerSleep(1.);
         }
   }


   void NormalLoop()
   {
      // this is to read normal LMD files and count statistic

      dabc::Buffer buf;

      // just loop over all buffers, taken from file

      int bufcnt(0), evntcnt(0),  errcnt(0);
      int64_t totalsz(0);

      SubeventsMap fMap;
      std::vector<int> triggers(16);

      int first_tm(0), last_tm(0);

      while (true) {

         buf = Recv(Input(), 5.);

         if (buf.GetTypeId() == dabc::mbt_EOF) {
            DOUT3(("Found EOF buffer len = %d", buf.GetTotalSize()));
            break;
         }

         if (buf.null()) {
            EOUT(("Get empty buffer id = %d", buf.GetTypeId()));
            break;
         }

         totalsz+=buf.GetTotalSize();

         mbs::ReadIterator iter(buf);
         while (iter.NextEvent()) {
            evntcnt++;

            //DOUT0(("============ NEXT EVENT %d tr %d ========== ", iter.evnt()->iEventNumber, iter.evnt()->iTrigger));

            if (iter.evnt()->iTrigger < 16)
               triggers[iter.evnt()->iTrigger]++;

            mbseventid = iter.evnt()->EventNumber();

            int maxid(-1), minid(-1);

            while (iter.NextSubEvent()) {
               mbs::SubeventHeader* sub = iter.subevnt();

               uint32_t fullid = sub->fFullId;

               fMap[fullid]++;

               int evid = GetEventNumber(fullid, sub->RawData(), sub->RawDataSize());

               // DOUT0(("%10s  %7d", GetHeaderInfo(fullid).c_str(), evid));

               //DOUT0(("   Next subevent %0x08x  evid %6d", fullid, evid));

               if (maxid<0) { maxid = evid; } else if (evid>maxid) maxid=evid;
               if (minid<0) minid = evid; else if (evid<minid) minid=evid;
            }

            last_tm = last_mastertm;
            if (first_tm==0) first_tm = last_tm;

            if ((maxid!=minid) && !fDoWriting) {
               errcnt++;
               if (errcnt<100) EOUT(("EventID missmatch %d %d master:%d slave:%d", minid, maxid, last_master, last_slave));
            }
         }

         bufcnt++;
      }

      DOUT0(("Found %d events in %d buffers totalsz %d", evntcnt, bufcnt, totalsz));
      for (unsigned n=0;n<triggers.size();n++)
         if (triggers[n]>0)
            DOUT1(("   Trigger %2u found %6d times", n, triggers[n]));

      DOUT1(("Found %u different subevents", fMap.size()));

      SubeventsMap:: iterator it = fMap.begin();
      while (it != fMap.end()) {

         std::string info = GetHeaderInfo(it->first);

         DOUT1(("   Subevent 0x%08x  %10s found %6d times", it->first,  info.c_str(), it->second));

         it++;
      }

      if (errcnt>0) EOUT(("TOTAL ERRORS COUNT = %d", errcnt));

      int tm = last_tm-first_tm;

      DOUT0(("HEAD: |              *Name* | *Time,s* | *Size,MB* | *Events* |  *TR1* |  *TR2* |  *TR8* | *EPICS* |"));
      DOUT0(("STAT: |   %17s |  %5d   |    %5.1f  |  %6d  | %6d | %6d | %6d |   %4d  |",
            Par("RunName").AsStr("---"), tm , totalsz/1e6, evntcnt, triggers[1], triggers[2], triggers[8], fMap[0xa]));

      WorkerSleep(1.);
   }


   void AccountLoop(int kind, int eventid)
   {
      static int lastkind = kind;
      static int left = eventid;
      static int right = eventid;

      if (kind == lastkind) {
         right = eventid;
         return;
      }

      std::string info, sinfo;
      if (left==right) dabc::formats(info, "event %d", left);
                  else dabc::formats(info, "events from %d to %d", left, right);

      switch (lastkind) {
         case 0: sinfo = "MERGE"; break;
         case 1: sinfo = "KEEP MAIN"; break;
         case 2: sinfo = "SKIP FASP"; break;
         case 3: sinfo = "SKIP MAIN"; break;
         default: sinfo = "?? WHAT ??"; break;
      }

      DOUT1(("%s %s", sinfo.c_str(), info.c_str()));

      lastkind = kind;
      left = eventid;
      right = eventid;
   }

   void VerifyPlace(mbs::WriteIterator& iter, unsigned size)
   {
      if ((size>0) && iter.IsPlaceForEvent(size-sizeof(mbs::EventHeader))) return;

      dabc::Buffer buf = iter.Close();

      if (fDoWriting) Send(Output(), buf);
                 else buf.Release();

      if (size>0)
         iter.Reset(Pool()->TakeBuffer(fBufferSize));
   }

   void CombinerLoop()
   {
      // this is to read and combine FASP and normal LMD files

      // Be aware - in FASP sync number is in event header,
      //            in LMD files - in

      mbs::ReadIterator iter1, iter2;

      int currid1(-1), currid2(-1);

      mbs::WriteIterator out_iter;

      // just loop over all buffers, taken from file

      int bufcnt(0), evntcnt(0),  errcnt(0);
      int64_t totalsz(0);

      SubeventsMap fMap;
      std::vector<int> triggers(16);

      int first_tm(0), last_tm(0);

      bool end_of_fasp_file(false), end_of_lmd_file(false);

      while (!end_of_lmd_file || !end_of_fasp_file) {

         if ((currid1<0) && !end_of_lmd_file) {
            iter1.NextEvent();

            if (!iter1.IsData()) {
               dabc::Buffer buf1 = Recv(Input(0), 5.);
               bufcnt++;
               if (buf1.GetTypeId() == dabc::mbt_EOF) {
                  end_of_lmd_file = true;
               } else
               if (buf1.null()) {
                  EOUT(("Get empty buffer id = %d", buf1.GetTypeId()));
                  break;
               } else
               if (!iter1.Reset(buf1.HandOver())) {
                  EOUT(("Problem to reset iter1"));
                  break;
               }
               if (!end_of_lmd_file)
                  iter1.NextEvent();
            }

            if (iter1.IsData()) {
               int maxid(-1), minid(-1);

               while (iter1.NextSubEvent()) {
                  mbs::SubeventHeader* sub = iter1.subevnt();
                  uint32_t fullid = sub->fFullId;
                  fMap[fullid]++;

                  int evid = GetEventNumber(fullid, sub->RawData(), sub->RawDataSize());
                  //DOUT0(("   Next subevent %0x08x  evid %6d", fullid, evid));
                  if (maxid<0) { maxid = evid; } else if (evid>maxid) maxid=evid;
                  if (minid<0) minid = evid; else if (evid<minid) minid=evid;
               }

               if (maxid!=minid) EOUT(("ID missmatch"));
               currid1 = minid;
            }
         }

         if ((currid2<0) && !end_of_fasp_file) {
            iter2.NextEvent();

            if (!iter2.IsData()) {
               dabc::Buffer buf2 = Recv(Input(1), 5.);
               if (buf2.GetTypeId() == dabc::mbt_EOF) {
                  end_of_fasp_file = true;
               } else
               if (buf2.null()) {
                  EOUT(("Get empty FASP buffer id = %d", buf2.GetTypeId()));
                  break;
               } else
               if (!iter2.Reset(buf2.HandOver())) {
                  EOUT(("Problem to reset iter2"));
                  break;
               }
               iter2.NextEvent();
            }

            if (!end_of_fasp_file)
               currid2 = iter2.evnt()->EventNumber();
         }

         if ((currid2<0) && (currid1<0)) {
            break;
         }

         if ((currid1<0) && (currid2>=0)) {
            // shift FASP
            AccountLoop(2, currid2);
            currid2 = -1;
            continue;
         }

         if ((currid2<0) && (currid1>=0)) {
            if (fOnlyFull) {
               // skip main data
               AccountLoop(3, currid1);
               currid1 = -1;
               continue;
            } else {
               // just copy of main data and continue
               VerifyPlace(out_iter, iter1.GetEventSize());
               out_iter.CopyEventFrom(iter1.GetEventPointer());

               AccountLoop(1, currid1);
               currid1 = -1;
               evntcnt++;
               continue;
            }
         }

         unsigned diff = ((unsigned) currid1 - (unsigned) currid2) & 0xffffff;

         if (diff == 0) {
            // very good, both events are the same

            VerifyPlace(out_iter, iter1.GetEventSize() + iter2.GetEventSize());

            fMap[proc_FASP]++;

            out_iter.CopyEventFrom(iter1.GetEventPointer(), false);
            out_iter.AddSubevent(iter2.GetSubeventsPointer());
            out_iter.FinishEvent();

            AccountLoop(0, currid1);
            currid1 = -1;
            currid2 = -1;
            evntcnt++;
         } else
         if (diff<0x800000) {
            // FASP id is smaller, shift FASP
            AccountLoop(2, currid2);
            currid2 = -1;

         } else {
            // Normal data is smaller,

            if (fOnlyFull) {
               // skip main data
               AccountLoop(3, currid1);
               currid1 = -1;
               continue;

            } else {
               //just store normal data
               VerifyPlace(out_iter, iter1.GetEventSize());
               out_iter.CopyEventFrom(iter1.GetEventPointer());
               AccountLoop(1, currid1);
               currid1 = -1;
               evntcnt++;
            }
         }
      }

      VerifyPlace(out_iter, 0); // flush last data
      AccountLoop(-1, -1);

      DOUT0(("Found %d events totalsz %d", evntcnt, totalsz));
      for (unsigned n=0;n<triggers.size();n++)
         if (triggers[n]>0)
            DOUT1(("   Trigger %2u found %6d times", n, triggers[n]));

      DOUT1(("Found %u different subevents", fMap.size()));

      SubeventsMap:: iterator it = fMap.begin();
      while (it != fMap.end()) {
         std::string info = GetHeaderInfo(it->first);
         DOUT1(("   Subevent 0x%08x  %10s found %6d times", it->first,  info.c_str(), it->second));
         it++;
      }

      if (errcnt>0) EOUT(("TOTAL ERRORS COUNT = %d", errcnt));

      int tm = last_tm-first_tm;

      DOUT0(("HEAD: |              *Name* | *Time,s* | *Size,MB* | *Events* |  *TR1* |  *TR2* |  *TR8* | *EPICS* |"));
      DOUT0(("STAT: | %17s |  %5d   |    %5.1f  |  %6d  | %6d | %6d | %6d |   %4d  |",
            Par("RunName").AsStr("---"), tm , totalsz/1e6, evntcnt, triggers[1], triggers[2], triggers[8], fMap[0xa]));

      if (fDoWriting)
         while (Output()->OutputPending() > 0) {
            DOUT0(("Waiting until data is written %d", Output()->OutputPending()));
            WorkerSleep(1.);
         }

      WorkerSleep(1.);
   }


   virtual void AfterModuleStop()
   {
      //         DOUT0(("After module stop"));

      dabc::mgr.StopApplication();
   }
};


class CernOct12Factory : public dabc::Factory  {
public:

   CernOct12Factory(const char* name) : dabc::Factory(name) {}

   virtual dabc::Module* CreateModule(const char* classname, const char* modulename, dabc::Command cmd)
   {
      if (strcmp(classname,"CernOct12SorterModule")==0)
         return new CernOct12SorterModule(modulename, cmd);

      return 0;
   }
};

dabc::FactoryPlugin cernoct12(new CernOct12Factory("cern-oct12"));

const int FaspBlockSize = 44;
const int FaspSyncPos = 34;



extern "C" void RunSorter()
{
   dabc::mgr.CreateMemoryPool("Pool", 5000000, 50);

   dabc::mgr.CreateModule("CernOct12SorterModule", "Sorter", "SorterThrd");

   if (dabc::mgr.FindModule("Sorter").Par("DoReading").AsBool())
      dabc::mgr.CreateTransport("Sorter/Input", mbs::typeLmdInput, "SorterThrd");
   else
      DOUT0(("Without normal reading"));

   if (dabc::mgr.FindModule("Sorter").Par("DoFASP").AsBool())
      dabc::mgr.CreateTransport("Sorter/FASP", "fasp::FileInput", "SorterThrd");
   else
      DOUT0(("Without FASP reading"));

   if (dabc::mgr.FindModule("Sorter").Par("DoWriting").AsBool())
      dabc::mgr.CreateTransport("Sorter/Output", mbs::typeLmdOutput, "SorterThrd");
   else
      DOUT0(("Without writing"));
}