ReadourCard interface change (#26)

* added routines to enable/disable data, so that equipment buffers can be emptied at end of run

* ReadoutCard interface change - AliceO2Group/ReadoutCard#21

Author: sy-c

src/ReadoutEquipment.cxx

@@ -232,3 +232,12 @@ Thread::CallbackResult  ReadoutEquipment::threadCallback(void *arg) {
   }
   return Thread::CallbackResult::Ok;
 }
+
+
+void ReadoutEquipment::setDataOn() {
+  isDataOn=true;
+}
+
+void ReadoutEquipment::setDataOff() {
+  isDataOn=false;
+}

src/ReadoutEquipment.h

@@ -29,6 +29,10 @@ class ReadoutEquipment {
   void stop();
   const std::string & getName();
 
+  // enable / disable data production by the equipment
+  virtual void setDataOn();
+  virtual void setDataOff();
+
 // protected:
 // todo: give direct access to output FIFO?
   std::shared_ptr<AliceO2::Common::Fifo<DataBlockContainerReference>> dataOut;
@@ -44,9 +48,11 @@ class ReadoutEquipment {
   virtual Thread::CallbackResult prepareBlocks() {return Thread::CallbackResult::Idle;};
   virtual DataBlockContainerReference getNextBlock() {return nullptr;};
 
-
   protected:
-    
+  // data enabled ? controlled by setDataOn/setDataOff
+  bool isDataOn=false;
+
+   
   // Definition of performance counters for readout statistics.
   // Each counter is assigned a unique integer index (incremental, starting 0).
   // The last element can be used to get the number of counters defined.

src/ReadoutEquipmentRORC.cxx

@@ -27,10 +27,12 @@ class ReadoutEquipmentRORC : public ReadoutEquipment {
     ReadoutEquipmentRORC(ConfigFile &cfg, std::string name="rorcReadout");
     ~ReadoutEquipmentRORC();
 
+  private:
     Thread::CallbackResult prepareBlocks();
     DataBlockContainerReference getNextBlock(); 
-  
-  private:
+    void setDataOn();
+    void setDataOff();
+ 
     Thread::CallbackResult  populateFifoOut(); // the data readout loop function
 
     AliceO2::roc::ChannelFactory::DmaChannelSharedPtr channel;    // channel to ROC device
@@ -172,9 +174,6 @@ ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile &cfg, std::string name) :
        baseAddress, blockSize
     });
 
-    // clear locks if necessary
-    params.setForcedUnlockEnabled(true);
-
     // define link mask
     // this is harmless for C-RORC
     params.setLinkMask(AliceO2::roc::Parameters::linkMaskFromString(cfgLinkMask));
@@ -199,14 +198,6 @@ ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile &cfg, std::string name) :
 
     // todo: log parameters ?
 
-    // start DMA    
-    theLog.log("Starting DMA for ROC %s:%d",cardId.c_str(),cfgChannelNumber);
-    channel->startDma();    
-    
-    // get FIFO depth (it should be fully empty when starting)
-    RocFifoSize=channel->getTransferQueueAvailable();
-    theLog.log("ROC input queue size = %d pages",RocFifoSize);
-    if (RocFifoSize==0) {RocFifoSize=1;}
 
     // reset timeframe id
     currentTimeframe=0;
@@ -233,10 +224,6 @@ ReadoutEquipmentRORC::ReadoutEquipmentRORC(ConfigFile &cfg, std::string name) :
 
 
 ReadoutEquipmentRORC::~ReadoutEquipmentRORC() {
-  if (isInitialized) {
-    channel->stopDma();
-  }
-
   if (cfgRdhCheckEnabled) {
     theLog.log("Equipment %s : %llu timeframes, %llu pages, RDH checks %llu ok, %llu errors",name.c_str(),statsNumberOfTimeframes,statsNumberOfPages,statsRdhCheckOk,statsRdhCheckErr);  
   }
@@ -245,6 +232,8 @@ ReadoutEquipmentRORC::~ReadoutEquipmentRORC() {
 
 Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks(){
   if (!isInitialized) return  Thread::CallbackResult::Error;
+  if (!isDataOn) return  Thread::CallbackResult::Idle;
+  
   int isActive=0;
 
   // keep track of situations where the queue is completely empty
@@ -266,9 +255,9 @@ Thread::CallbackResult ReadoutEquipmentRORC::prepareBlocks(){
     if (newPage!=nullptr) {   
       // todo: check page is aligned as expected      
       AliceO2::roc::Superpage superpage;
-      superpage.offset=(char *)newPage-(char *)mp->getBaseBlockAddress()+pageSpaceReserved;
-      superpage.size=superPageSize;
-      superpage.userData=newPage;
+      superpage.setOffset((char *)newPage-(char *)mp->getBaseBlockAddress()+pageSpaceReserved);
+      superpage.setSize(superPageSize);
+      superpage.setUserData(newPage);
       channel->pushSuperpage(superpage);      
       isActive=1;
       nPushed++;
@@ -317,16 +306,21 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
 
   DataBlockContainerReference nextBlock=nullptr;
 
+  // ensure the initialization was fine in the main thread
+  if (!isInitialized) {
+  	return nullptr;
+  }  
   //channel->fillSuperpages();
 
   // check for completed page
   if ((channel->getReadyQueueSize()>0)) {
     auto superpage = channel->getSuperpage(); // this is the first superpage in FIFO ... let's check its state
     if (superpage.isFilled()) {
       std::shared_ptr<DataBlockContainer>d=nullptr;
+//      printf ("received a page with %d bytes - isFilled=%d isREady=%d\n",(int)superpage.getReceived(),(int)superpage.isFilled(),(int)superpage.isReady());
       try {
         if (pageSpaceReserved>=sizeof(DataBlock)) {
-          d=mp->getNewDataBlockContainer((void *)(superpage.userData));
+          d=mp->getNewDataBlockContainer((void *)(superpage.getUserData()));
         } else {
           // todo: allocate data block container elsewhere than beginning of page
           //d=mp->getNewDataBlockContainer(nullptr);        
@@ -412,6 +406,8 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
                 errorDescription.clear();
               }
               statsRdhCheckErr++;
+	      // stop on first RDH error (should distinguich valid/invalid block length)
+	      break;
             } else {
               statsRdhCheckOk++;
 
@@ -454,3 +450,26 @@ DataBlockContainerReference ReadoutEquipmentRORC::getNextBlock() {
 std::unique_ptr<ReadoutEquipment> getReadoutEquipmentRORC(ConfigFile &cfg, std::string cfgEntryPoint) {
   return std::make_unique<ReadoutEquipmentRORC>(cfg,cfgEntryPoint);
 }
+
+
+void ReadoutEquipmentRORC::setDataOn() {
+  if (isInitialized) {
+    // start DMA    
+    theLog.log("Starting DMA for ROC %s",getName().c_str());
+    channel->startDma();
+
+    // get FIFO depth (it should be fully empty when starting)
+    RocFifoSize=channel->getTransferQueueAvailable();
+    theLog.log("ROC input queue size = %d pages",RocFifoSize);
+    if (RocFifoSize==0) {RocFifoSize=1;}
+  }
+  ReadoutEquipment::setDataOn();
+}
+
+void ReadoutEquipmentRORC::setDataOff() {
+  if (isInitialized) {
+    theLog.log("Stopping DMA for ROC %s",getName().c_str());
+    channel->stopDma();
+  }
+  ReadoutEquipment::setDataOff();
+}

src/mainReadout.cxx

@@ -392,6 +392,9 @@ int main(int argc, char* argv[])
       readoutDevice->start();
   }
 
+  for (auto && readoutDevice : readoutDevices) {
+      readoutDevice->setDataOn();
+  }
   theLog.log("Running");
 
   // reset exit timeout, if any
@@ -412,11 +415,11 @@ int main(int argc, char* argv[])
     if (isRunning) {
       if (((cfgExitTimeout>0)&&(t.isTimeout()))||(ShutdownRequest)) {
         isRunning=0;
-        theLog.log("Stopping readout");
-        for (auto && readoutDevice : readoutDevices) {
-          readoutDevice->stop();
+        theLog.log("Stopping data readout");
+	for (auto && readoutDevice : readoutDevices) {
+          readoutDevice->setDataOff();	  
         }
-        theLog.log("Readout stopped");
+	
         t.reset(1000000);  // add a delay before stopping aggregator - continune to empty FIFOs
 
         // notify consumers of imminent data flow stop
@@ -452,7 +455,11 @@ int main(int argc, char* argv[])
     theLog.log("Stopping callgrind instrumentation");
   #endif
 
-
+  for (auto && readoutDevice : readoutDevices) {
+    readoutDevice->stop();
+  }
+  theLog.log("Readout stopped");
+  	
   theLog.log("Stopping aggregator");
   agg->stop();
 

src/testROC.cxx

@@ -95,7 +95,6 @@ ROCdevice::ROCdevice(std::string id) {
       mp->getBaseBlockAddress(), mp->getBaseBlockSize()
   });
 
-  params.setForcedUnlockEnabled(true);
   params.setLinkMask(AliceO2::roc::Parameters::linkMaskFromString(cfgLinkMask));
 
 
@@ -152,10 +151,10 @@ int ROCdevice::doLoop() {
     while ((channel->getReadyQueueSize()>0)) {
       auto superpage = channel->getSuperpage(); // this is the first superpage in FIFO ... let's check its state
       if (superpage.isFilled()) {
-        mp->releasePage(superpage.userData);
+        mp->releasePage(superpage.getUserData());
         channel->popSuperpage();
         nPop++;
-        nBytes+=superpage.size;
+        nBytes+=superpage.getSize();
       } else {
         break;
       }
@@ -170,9 +169,9 @@ int ROCdevice::doLoop() {
       void *newPage=mp->getPage();
       if (newPage!=nullptr) {
         AliceO2::roc::Superpage superpage;
-        superpage.offset=(char *)newPage-(char *)mp->getBaseBlockAddress();
-        superpage.size=superPageSize;
-        superpage.userData=newPage;
+        superpage.setOffset((char *)newPage-(char *)mp->getBaseBlockAddress());
+        superpage.setSize(superPageSize);
+        superpage.setUserData(newPage);
         channel->pushSuperpage(superpage);
         nPush++;
       } else {