Changeset 11082

Timestamp:

06/20/11 18:26:11 (13 years ago)

Author:

tbretz

Message:

Added commands DUMP_STREAM and DUMP_RECV to have better access to the behaviour of the sockets.

Location:

trunk/FACT++/src

Files:

• EventBuilder.c (modified) (16 diffs)
• EventBuilderWrapper.h (modified) (5 diffs)
• fadctrl.cc (modified) (2 diffs)

trunk/FACT++/src/EventBuilder.c

@@ -10 +10 @@
 #include <unistd.h>
 #include <stdio.h>
-#include <time.h>
+#include <sys/time.h>
 #include <arpa/inet.h>
 #include <string.h>
@@ -43 +43 @@
 extern void factStat(int severity, int err, char* message ) ;
 
+
+extern void debugRead(int isock, int ibyte, int32_t event, int state, 
+                      uint32_t tsec, uint32_t tusec ) ;
+extern void debugStream(int isock, void *buf, int len) ;
+
+
+
 int g_actTime   =  0 ;
 int g_runStat   = 40 ;
@@ -129 +136 @@
   int16_t S  ;
 } SHORT_BYTE ;
+
+
+
 
 
@@ -395 +405 @@
   int nokCnt[MAX_SOCK],loopCnt=0;
   int sokCnt[MAX_SOCK];
+
+  struct timeval  *tv, atv;
+  tv=&atv;
+  uint32_t tsec, tusec ;
 
 
@@ -466 +480 @@
 
   head_len = sizeof(PEVNT_HEADER) ;
-  frst_len = head_len + 36 * 12 ;   //fad_header plus 36*pix_header
+//frst_len = head_len + 36 * 12 ;   //fad_header plus 36*pix_header
+  frst_len = head_len ;             //fad_header only, so each event must be longer, even for roi=0
+
 
   numok = numok2   = 0 ;
@@ -490 +506 @@
 
     for (i=0; i<gi_maxSocks; i++) {      //check all sockets if something to read
+
+gettimeofday( tv, NULL);
+tsec = atv.tv_sec ;
+tusec= atv.tv_usec ; 
+
       if (rd[i].sockStat <0 ) {         //try to connect if not yet done
         rd[i].sockStat=connect(rd[i].socket,
@@ -509 +530 @@
         jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT);
 
+
+if (jrd >0 ) {
+  debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ;
+}
+
+
+
         if (jrd == 0) {                 //connection has closed ...
            rd[i].sockStat = -1 ;        //flag (try to reopen next round)
@@ -528 +556 @@
              rd[i].bufPos += jrd ;        //==> prepare for continuation
              rd[i].bufLen -= jrd ;
+debugRead(i,jrd,rd[i].evtID, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data
            } else {                     //full dataset read
              rd[i].bufLen  = rd[i].bufPos + j ;
@@ -539 +568 @@
 
              } 
+debugRead(i,jrd,rd[i].evtID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
 
              //we have a complete buffer, copy to WORK area
@@ -675 +705 @@
                  rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ;
                  if (rd[i].bufLen <=0 ) rd[i].bufLen = 100000 ;
+debugRead(i,jrd,rd[i].evtID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event
               }
+  else {
+debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
+  }
+
            }
+
+  else {
+debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
+  }
+
         } //end interpreting last read
       } //end of successful read anything
@@ -953 +993 @@
 /* close run runId (all all runs if runId=0) */
 /* return: 0=close scheduled / >0 already closed / <0 does not exist */
-  int j ;
+  int i, j ;
 
   if (runId == 0 ) {
      for ( j=0; j<MAX_RUN; j++) {
         if ( runCtrl[j].fileId == 0 ) {  //run is open
+           i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
+           if (i<0) {
+              snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
+              factOut(kError,506, str ) ;
+              runCtrl[j].fileId = 888 ;
+           } else {
+              snprintf(str,MXSTR,"closing run %d ok  AAA",runCtrl[j].runId);
+              factOut(kInfo,507, str ) ;
+              runCtrl[j].fileId = 7777 ;
+           }
            runCtrl[j].closeTime = closeTime ;
         }
@@ -968 +1018 @@
      if ( runCtrl[j].runId == runId ) {
         if ( runCtrl[j].fileId == 0 ) {  //run is open
+           i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
+           if (i<0) {
+              snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
+              factOut(kError,506, str ) ;
+              runCtrl[j].fileId = 888 ;
+           } else {
+              snprintf(str,MXSTR,"closing run %d ok  AAA",runCtrl[j].runId);
+              factOut(kInfo,507, str ) ;
+              runCtrl[j].fileId = 7777 ;
+           }
            runCtrl[j].closeTime = closeTime ;
            return 0;
         } else if ( runCtrl[j].fileId <0 ) { //run not yet opened
            runCtrl[j].closeTime = closeTime ;
-           return 0;
+           return +1;
         } else {     // run already closed
-           return +1;
+           return +2;
         }
      }
@@ -1134 +1194 @@
   factOut(kInfo,-1, str ) ;
   for ( j=0; j<MAX_RUN; j++) 
-     if ( runCtrl[j].runId >0 ) {
+     if ( runCtrl[j].fileId ==0 ) {
         int i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
         if (i<0) {
@@ -1212 +1272 @@
 
 
-/*
+
 
      xwait.tv_sec = 20;;
@@ -1233 +1293 @@
 
 
- */
+
 
 
@@ -1250 +1310 @@
 } /*-----------------------------------------------------------------*/
 
+
+
+
+
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+  /*-----------------------------------------------------------------*/
+
+/*
+
+
+FileHandle_t  runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) 
+{ return 1; } ;
+
+int  runWrite(FileHandle_t fileHd ,  EVENT    *event, size_t len ) 
+{ return 1; } ;
+
+int  runClose(FileHandle_t fileHd ,  RUN_TAIL *runth, size_t len ) 
+{ return 1; } ;
+
+
+void factStat(int severity, int err, char* message ) {
+  printf("%3d %3d : %s\n",severity,err,message) ;
+}
+
+
+void debugRead(int isock, int ibyte, int32_t event, int state, uint32_t tsec, uint32_t tusec ) {
+  printf("%3d %5d %9d %3d %12d\n",isock, ibyte, event, state, tusec) ;
+}
+
+
+
+void debugStream(int isock, void *buf, int len) {
+}
+
+
+void factOut(int severity, int err, char* message ) {
+static FILE * fd ;
+static int  file=0 ;
+
+  if (file==0) {
+     printf("open file\n");
+     fd=fopen("x.out","w+") ;
+     file=999; 
+  }
+
+  fprintf(fd,"%3d %3d | %s \n",severity,err,message) ;
+
+  if (severity != kDebug) 
+  printf("%3d %3d | %s\n",severity,err,message) ;
+}
+
+
+
+int main() {
+  int i,b,c,p ;
+  char     ipStr[100] ;
+  struct in_addr IPaddr ;
+
+  g_maxMem = 1024*1024    ;        //MBytes
+  g_maxMem = g_maxMem * 1024 *10 ; //10GBytes
+
+  g_maxBoards = 40 ;
+
+  g_actBoards = g_maxBoards; 
+
+  g_runStat = 40 ;
+
+  i=0 ;
+
+// version for standard crates
+//for (c=0; c<4,c++) {
+//   for (b=0; b<10; b++) {
+//      sprintf(ipStr,"10.0.%d.%d",128+c,128+b)
+//
+//      inet_pton(PF_INET, ipStr, &IPaddr) ;
+//
+//      g_port[i].sockAddr.sin_family = PF_INET;
+//      g_port[i].sockAddr.sin_port = htons(5000) ;
+//      g_port[i].sockAddr.sin_addr = IPaddr ;
+//      g_port[i].sockDef = 1 ;
+//      i++ ;
+//   }
+//}
+//
+//version for PC-test
+  for (c=0; c<4; c++) {
+     for (b=0; b<10; b++) {
+        sprintf(ipStr,"10.0.%d.11",128+c) ;
+        if (c==0) sprintf(ipStr,"10.0.100.11") ;
+
+        inet_pton(PF_INET, ipStr, &IPaddr) ;
+        p = 31919+100*c+10*b; 
+
+
+        g_port[i].sockAddr.sin_family = PF_INET;
+        g_port[i].sockAddr.sin_port = htons(p) ;
+        g_port[i].sockAddr.sin_addr = IPaddr ;
+        g_port[i].sockDef = 1 ;
+
+        i++ ;
+     }
+  }
+
+
+//g_port[17].sockDef =-1 ;
+//g_actBoards-- ; 
+
+  StartEvtBuild() ;
+
+  return 0;
+
+}
+   
+   */

trunk/FACT++/src/EventBuilderWrapper.h

@@ -720 +720 @@
     DimDescribedService fDimCurrentEvent;
 
+    bool fDebugStream;
+    bool fDebugRead;
+
     int Write(const Time &time, const std::string &txt, int qos)
     {
@@ -731 +734 @@
         fDimRuns  ("FAD_CONTROL/RUNS",          "I:1", ""),
         fDimEvents("FAD_CONTROL/EVENTS",        "I:2", ""),
-        fDimCurrentEvent("FAD_CONTROL/CURRENT_EVENT", "I:1", "")
+        fDimCurrentEvent("FAD_CONTROL/CURRENT_EVENT", "I:1", ""),
+        fDebugStream(false), fDebugRead(false)
     {
         if (This)
@@ -847 +851 @@
     bool IsIgnored(int i) const { return g_port[i].sockDef==0; }
 
+    void SetDebugStream(bool b) { fDebugStream = b; if (!b) for (int i=0; i<40; i++) fDumpStream[i].close(); }
+    void SetDebugRead(bool b)   { fDebugRead = b;   if (!b) fDumpRead.close(); }
+
     size_t GetUsedMemory() const { return gi_usedMem; }
 
@@ -984 +991 @@
         return rc ? 0 : -1;
     }
+
+    ofstream fDumpStream[40];
+
+    void debugStream(int isock, void *buf, int len)
+    {
+        if (!fDebugStream)
+            return;
+
+        const int slot = isock/7;
+        if (slot<0 || slot>39)
+            return;
+
+        if (!fDumpStream[slot].is_open())
+        {
+            ostringstream name;
+            name << "socket_dump-" << setfill('0') << setw(2) << slot << ".bin";
+
+            fDumpStream[slot].open(name.str().c_str(), ios::app);
+            if (!fDumpStream[slot])
+            {
+                //ostringstream str;
+                //str << "Open file " << name << ": " << strerror(errno) << " (errno=" << errno << ")";
+                //Error(str);
+
+                return;
+            }
+        }
+
+        fDumpStream[slot].write(reinterpret_cast<const char*>(buf), len);
+    }
+
+    ofstream fDumpRead; // Stream to possibly dump docket events
+
+    void debugRead(int isock, int ibyte, int32_t event, int state, uint32_t tsec, uint32_t tusec)
+    {
+        //   isock = socketID (0-279)
+        //   ibyte = #bytes gelesen
+        //   event = eventId (oder 0 wenn noch nicht bekannt)
+        //   state : 0=reading data
+        //           1=finished reading data
+        //          -1=start reading data (header)
+        //          -2=start reading data,
+        //             eventId not known yet (too little data)
+        //   tsec, tusec = time when reading seconds, microseconds
+        //
+        if (!fDebugRead)
+            return;
+
+        if (!fDumpRead.is_open())
+        {
+            fDumpRead.open("socket_events.txt", ios::app);
+            if (!fDumpRead)
+            {
+                //ostringstream str;
+                //str << "Open file " << name << ": " << strerror(errno) << " (errno=" << errno << ")";
+                //Error(str);
+
+                return;
+            }
+        }
+
+        fDumpRead
+            << setw(2) << state   << " "
+            << setw(8) << tsec    << " "
+            << setw(9) << tusec   << " "
+            << setw(3) << isock   << " "
+            << setw(2) << isock/7 << " "
+            << event << " "
+            << ibyte << endl;
+    }
+
 };
 
@@ -1039 +1117 @@
     }
 
+    void debugStream(int isock, void *buf, int len)
+    {
+        return EventBuilderWrapper::This->debugStream(isock, buf, len);
+    }
+
+    void debugRead(int isock, int ibyte, int32_t event, int state, uint32_t tsec, uint32_t tusec)
+    {
+        return EventBuilderWrapper::This->debugRead(isock, ibyte, event, state, tsec, tusec);
+    }
+
     /*
     void message(int severity, const char *msg)

trunk/FACT++/src/fadctrl.cc

@@ -947 +947 @@
         for (int i=slot[0]; i<=slot[1]; i++)
             SetIgnore(i, block);
+
+        return T::GetCurrentState();
+    }
+
+    int SetDumpStream(const EventImp &evt)
+    {
+        if (!CheckEventSize(evt.GetSize(), "SetDumpStream", 1))
+            return T::kSM_FatalError;
+
+        SetDebugStream(evt.Get<uint8_t>());
+
+        return T::GetCurrentState();
+    }
+
+    int SetDumpRecv(const EventImp &evt)
+    {
+        if (!CheckEventSize(evt.GetSize(), "SetDumpRecv", 1))
+            return T::kSM_FatalError;
+
+        SetDebugRead(evt.Get<uint8_t>());
 
         return T::GetCurrentState();
@@ -1334 +1354 @@
 
         // Verbosity commands
-        T::AddEvent("SET_VERBOSE", "B")
+        T::AddEvent("SET_VERBOSE", "B:1")
             (boost::bind(&StateMachineFAD::SetVerbosity, this, _1))
             ("set verbosity state"
              "|verbosity[bool]:disable or enable verbosity for received data (yes/no), except dynamic data");
 
-        T::AddEvent("SET_HEX_OUTPUT", "B")
+        T::AddEvent("SET_HEX_OUTPUT", "B:1")
             (boost::bind(&StateMachineFAD::SetHexOutput, this, _1))
             ("enable or disable hex output for received data"
              "|hexout[bool]:disable or enable hex output for received data (yes/no)");
 
-        T::AddEvent("SET_DATA_OUTPUT", "B")
+        T::AddEvent("SET_DATA_OUTPUT", "B:1")
             (boost::bind(&StateMachineFAD::SetDataOutput, this, _1))
             ("");
 
-        T::AddEvent("PRINT_EVENT", "I")
+        T::AddEvent("PRINT_EVENT", "I:1")
             (boost::bind(&StateMachineFAD::PrintEvent, this, _1))
             ("Print (last) event"
              "|board[int]:slot from which the event should be printed (-1 for all)");
+
+        T::AddEvent("DUMP_STREAM", "B:1")
+            (boost::bind(&StateMachineFAD::SetDumpStream, this, _1))
+            ("For debugging purpose: the binary data stream read from the sockets 0-7 can be dumped to files."
+             "|switch[bool]:Enable (yes) or disable (no)");
+
+        T::AddEvent("DUMP_RECV", "B:1")
+            (boost::bind(&StateMachineFAD::SetDumpRecv, this, _1))
+            ("For debugging purpose: the times when data has been receives are dumped to a file."
+             "|switch[bool]:Enable (yes) or disable (no)");
 
         T::AddEvent("BLOCK_TRANSMISSION", "S:1;B:1")