Merge pull request #14 from sy-c/master

sy-c · web-flow · commit 9e4d5fe2fd77 · 2018-06-14T11:43:17.000+02:00
merging latest dev
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -53,7 +53,7 @@ ELSE ()
 ENDIF ()
 
 # Add compiler flags for warnings and (more importantly) fPIC and debug symbols
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -pedantic -Wextra -fPIC")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -pedantic -Wextra -fPIC -g")
 
 ####################################
 # Module, library and executable definition
diff --git a/readout.cfg b/readout.cfg
@@ -34,6 +34,21 @@ enabled=0
 class=MockInjector
 
 
+###################################
+# memory banks
+###################################
+# All section names should start with 'bank-' to be taken into account.
+# They define memory to be allocated to readout
+# If bank name not specified in each equipment, the first available bank (created first) will be used.
+# Types of memory banks include: malloc, MemoryMappedFile
+# NB: the FairMQChannel consumers may also create some banks, which will not be
+# listed here, and created before them.
+
+[bank-default]
+type=malloc
+size=128M
+
+
 
 ###################################
 # equipments
@@ -53,13 +68,18 @@ equipmentType=dummy
 enabled=1
 eventMaxSize=20000
 eventMinSize=10000
+memoryPoolNumberOfPages=100
+memoryPoolPageSize=128k
 
 [equipment-dummy-2]
 name=dummy-2
 equipmentType=dummy
 enabled=1
 eventMaxSize=30000
 eventMinSize=20000
+memoryPoolNumberOfPages=100
+memoryPoolPageSize=128k
+
 
 
 
@@ -103,10 +123,9 @@ cardId=86:00.0
 consumerType=stats
 enabled=1
 # this publishes stats, if enabled, to O2 monitoring system
-monitoringEnabled=0
+monitoringEnabled=1
 monitoringUpdatePeriod=5
-monitoringConfig=file:/etc/monitoring.cfg
-
+monitoringURI=infologger://
 
 
 # recording to file
diff --git a/src/ConsumerFMQchannel.cxx b/src/ConsumerFMQchannel.cxx
@@ -138,6 +138,11 @@ class ConsumerFMQchannel: public Consumer {
   }
 
   ~ConsumerFMQchannel() {
+    // release in reverse order
+    mp=nullptr;
+    memoryBuffer=nullptr;   // warning: data range may still be referenced in memory bank manager
+    sendingChannel=nullptr;
+    transportFactory=nullptr;
   }
 
   int pushData(DataBlockContainerReference &) {
diff --git a/src/ConsumerStats.cxx b/src/ConsumerStats.cxx
@@ -89,7 +89,7 @@ class ConsumerStats: public Consumer {
       // https://alice.its.cern.ch/jira/browse/FLPPROT-69
 
       monitoringCollector->send({counterBlocks, "readout.Blocks"});
-      monitoringCollector->send({counterBytesTotal, "readout.BytesTotal"});
+//      monitoringCollector->send({counterBytesTotal, "readout.BytesTotal"});
       monitoringCollector->send({counterBytesTotal, "readout.BytesTotal"}, DerivedMetricMode::RATE);
       monitoringCollector->send({counterBytesDiff, "readout.BytesInterval"});
 //      monitoringCollector->send({(counterBytesTotal/(1024*1024)), "readout.MegaBytesTotal"});
diff --git a/src/RAWDataHeader.h b/src/RAWDataHeader.h
@@ -114,23 +114,24 @@ typedef struct _RAWDataHeaderV3
     struct {
       uint32_t version:8;        /// bit 0 to 7: header version
       uint32_t headerSize:8;     /// bit 8 to 15: header size
-      uint32_t blockLength:16;   /// bit 16 to 32: block length
+      uint32_t blockLength:16;   /// bit 16 to 31: block length
     };
   };
 
   union {
     uint32_t word2 = 0x00ffffff;
     struct {
-      uint32_t feeId:8;           /// bit 0 to 15: FEE id
+      uint32_t feeId:16;          /// bit 0 to 15: FEE id
       uint32_t priorityBit:8;     /// bit 16 to 23: priority bit
-      uint32_t zero2:16;    /// bit 16 to 32: reserved
+      uint32_t zero2:8;           /// bit 24 to 31: reserved
     };
   };
 
   union {
     uint32_t word1 = 0x0;
     struct {
-      uint32_t zero1:32;     /// bit 0 to 31: reserved
+      uint32_t offsetNextPacket:16;  /// bit 0 to 15: offset of next block
+      uint32_t memorySize:16;        /// bit 16 to 31: size of block (in bytes) in memory
     };
   };
 
@@ -152,7 +153,7 @@ typedef struct _RAWDataHeaderV3
   union {
     uint32_t word6 = 0xffffffff;
     struct {
-      uint32_t heartbeatOrbit;     /// bit 0 to 31: TRG orbit
+      uint32_t heartbeatOrbit;     /// bit 0 to 31: HB orbit
     };
   };
 
@@ -173,9 +174,9 @@ typedef struct _RAWDataHeaderV3
   union {
     uint32_t word11 = 0x0;
     struct {
-      uint32_t triggerBC:12;     /// bit 0 to 11: trigger BC ID
+      uint32_t triggerBC:12;     /// bit 0 to 11: TRG BC ID
       uint32_t zero11_0:4;       /// bit 12 to 15: reserved
-      uint32_t heartbeatBC:12;   /// bit 16 to 27: heartbeat BC ID
+      uint32_t heartbeatBC:12;   /// bit 16 to 27: HB BC ID
       uint32_t zero11_1:4;       /// bit 28 to 31: reserved
     };
   };
@@ -212,9 +213,9 @@ typedef struct _RAWDataHeaderV3
   union {
     uint32_t word14 = 0x0;
     struct {
-      uint32_t stopBit:8;     /// bit 0 to 7: stop bit
+      uint32_t stopBit:8;          /// bit 0 to 7: stop bit
       uint32_t pagesCoutner:16;    /// bit 8 to 23: pages counter
-      uint32_t zero14:8;      /// bit 24 to 31: reserved
+      uint32_t zero14:8;           /// bit 24 to 31: reserved
     };
   };
   
diff --git a/src/RdhUtils.h b/src/RdhUtils.h
@@ -39,6 +39,9 @@ class RdhHandle {
   inline uint8_t getHeaderSize() {
     return rdhPtr->headerSize;
   }
+  inline uint32_t getHbOrbit() {
+    return (uint32_t) rdhPtr->heartbeatOrbit;
+  }  
   
   private:
   o2::Header::RAWDataHeader *rdhPtr;  // pointer to RDH in memory
diff --git a/src/ReadoutEquipmentRORC.cxx b/src/ReadoutEquipmentRORC.cxx
@@ -47,10 +47,21 @@ class ReadoutEquipmentRORC : public ReadoutEquipment {
 
     unsigned long long statsRdhCheckOk=0;   // number of RDH structs which have passed check ok
     unsigned long long statsRdhCheckErr=0;  // number of RDH structs which have not passed check    
+    unsigned long long statsNumberOfPages=0; // number of pages read out
+    unsigned long long statsNumberOfTimeframes=0; // number of timeframes read out
+    
     
     AliceO2::Common::Timer timeframeClock;	// timeframe id should be increased at each clock cycle
-    int currentTimeframe=0;	// id of current timeframe
+    int currentTimeframe=0;	                // id of current timeframe
+    bool usingSoftwareClock=false;              // if set, using internal software clock to generate timeframe id
+
+    const unsigned int LHCBunches=3564;    // number of bunches in LHC
+    const unsigned int LHCOrbitRate=11246; // LHC orbit rate, in Hz. 299792458 / 26659
+    const uint32_t timeframePeriodOrbits=256;   // timeframe interval duration in number of LHC orbits
     
+    uint32_t currentTimeframeHbOrbitBegin=0; // HbOrbit of beginning of timeframe 
+    uint32_t firstTimeframeHbOrbitBegin=0; // HbOrbit of beginning of first timeframe
+        
     size_t superPageSize=0; // usable size of a superpage
 };
 
@@ -202,9 +213,19 @@ ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile &cfg, std::string name) :
     theLog.log("ROC input queue size = %d pages",RocFifoSize);
     if (RocFifoSize==0) {RocFifoSize=1;}
 
-    // reset timeframe clock
-    timeframeClock.reset(1000000/50.0); // 50Hz rate
+    // reset timeframe id
     currentTimeframe=0;
+    if (!cfgRdhCheckEnabled) {
+      usingSoftwareClock=true; // if RDH disabled, use internal clock for TF id
+    }
+    if (usingSoftwareClock) {
+      // reset timeframe clock
+      double timeframeRate=LHCOrbitRate*1.0/timeframePeriodOrbits; // timeframe rate, in Hz
+      theLog.log("Timeframe IDs generated by software, %.2lf Hz",timeframeRate);
+      timeframeClock.reset(1000000/timeframeRate);
+    } else {
+      theLog.log("Timeframe IDs generated from RDH trigger counters");
+    }
 
   }
   catch (const std::exception& e) {
@@ -222,7 +243,7 @@ ReadoutEquipmentRORC::~ReadoutEquipmentRORC() {
   }
 
   if (cfgRdhCheckEnabled) {
-    theLog.log("Equipment %s : RDH checks %llu ok, %llu errors",name.c_str(),statsRdhCheckOk,statsRdhCheckErr);  
+    theLog.log("Equipment %s : %llu timeframes, %llu pages, RDH checks %llu ok, %llu errors",name.c_str(),statsNumberOfTimeframes,statsNumberOfPages,statsRdhCheckOk,statsRdhCheckErr);  
   }
 }
 
@@ -323,20 +344,22 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
         theLog.log("make_shared<DataBlock> failed");
       }
       if (d!=nullptr) {
+        statsNumberOfPages++;
+        
         d->getData()->header.dataSize=superpage.getReceived();
         d->getData()->header.linkId=0; // TODO
 
         channel->popSuperpage();
         nextBlock=d;
-
-	if (timeframeClock.isTimeout()) {
-	  currentTimeframe++;
-	  timeframeClock.increment();
-	}
-	d->getData()->header.id=currentTimeframe;
         
         // validate RDH structure, if configured to do so
         int linkId=-1;
+        int hbOrbit=-1;
+                
+        // checks to do:
+        // - HB clock consistent in all RDHs
+        // - increasing counters
+
         if (cfgRdhCheckEnabled) {
           std::string errorDescription;
           size_t blockSize=d->getData()->header.dataSize;
@@ -352,6 +375,26 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
               }
             }
             
+            if (hbOrbit==-1) {
+              hbOrbit=h.getHbOrbit();
+              if ((statsNumberOfPages==1) || ((uint32_t)hbOrbit>=currentTimeframeHbOrbitBegin+timeframePeriodOrbits)) {
+                if (statsNumberOfPages==1) {
+                  firstTimeframeHbOrbitBegin=hbOrbit;
+                }
+                statsNumberOfTimeframes++;
+                currentTimeframeHbOrbitBegin=hbOrbit-((hbOrbit-firstTimeframeHbOrbitBegin)%timeframePeriodOrbits); // keep it periodic and aligned to 1st timeframe
+                int newTimeframe=1+(currentTimeframeHbOrbitBegin-firstTimeframeHbOrbitBegin)/timeframePeriodOrbits;
+                if (newTimeframe!=currentTimeframe+1) {
+                  printf("Non-contiguous timeframe IDs %d ... %d\n",currentTimeframe,newTimeframe);
+                }
+                currentTimeframe=newTimeframe;
+                 //printf("Starting timeframe %d @ orbit %d (actual: %d)\n",currentTimeframe,(int)currentTimeframeHbOrbitBegin,(int)hbOrbit);
+              } else {
+                 //printf("HB orbit %d\n",hbOrbit);
+              }
+              
+            }           
+            
             //data format:
             // RDH v3 = https://docs.google.com/document/d/1otkSDYasqpVBDnxplBI7dWNxaZohctA-bvhyrzvtLoQ/edit?usp=sharing
             if (h.validateRdh(errorDescription)) {
@@ -383,7 +426,17 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
         if (linkId>=0) {
           d->getData()->header.linkId=linkId;
         }
-        
+
+        if (usingSoftwareClock) {
+	  if (timeframeClock.isTimeout()) {
+	    currentTimeframe++;
+            statsNumberOfTimeframes++;
+	    timeframeClock.increment();
+	  }
+        }
+
+        // set timeframe id
+        d->getData()->header.id=currentTimeframe;        
       }
       else {
         // no data block container... what to do???
diff --git a/src/mainReadout.cxx b/src/mainReadout.cxx
@@ -94,7 +94,7 @@ int main(int argc, char* argv[])
   sigaction(SIGINT,&signalSettings,NULL);
 
   // log startup and options
-  theLog.log("Readout process starting");
+  theLog.log("Readout process starting, pid %d",getpid());
   theLog.log("Optional built features enabled:");
   #ifdef WITH_FAIRMQ
     theLog.log("FAIRMQ : yes");

Original file line number	Diff line number	Diff line change
`@@ -138,6 +138,11 @@ class ConsumerFMQchannel: public Consumer {`
`138`	`138`	`}`
`139`	`139`
`140`	`140`	`~ConsumerFMQchannel() {`
	`141`	`+ // release in reverse order`
	`142`	`+ mp=nullptr;`
	`143`	`+ memoryBuffer=nullptr; // warning: data range may still be referenced in memory bank manager`
	`144`	`+ sendingChannel=nullptr;`
	`145`	`+ transportFactory=nullptr;`
`141`	`146`	`}`
`142`	`147`
`143`	`148`	`int pushData(DataBlockContainerReference &) {`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,9 @@ class RdhHandle {`
`39`	`39`	`inline uint8_t getHeaderSize() {`
`40`	`40`	`return rdhPtr->headerSize;`
`41`	`41`	`}`
	`42`	`+ inline uint32_t getHbOrbit() {`
	`43`	`+ return (uint32_t) rdhPtr->heartbeatOrbit;`
	`44`	`+ }`
`42`	`45`
`43`	`46`	`private:`
`44`	`47`	`o2::Header::RAWDataHeader *rdhPtr; // pointer to RDH in memory`