Add transfer of remaining superpages for CRU backend

Author: PascalBoeschoten

include/ReadoutCard/DmaChannelInterface.h

@@ -30,10 +30,6 @@ class DmaChannelInterface
     /// Call this before pushing pages. May become unneeded in the future.
     virtual void startDma() = 0;
 
-    /// Stops DMA for the given channel
-    /// Called automatically on channel closure.
-    virtual void stopDma() = 0;
-
     /// Resets the channel. Requires the DMA to be stopped.
     /// \param resetLevel The depth of the reset
     virtual void resetChannel(ResetLevel::type resetLevel) = 0;
@@ -78,6 +74,11 @@ class DmaChannelInterface
     /// superpage can be inspected with getSuperpage() or popped with popSuperpage().
     virtual int getReadyQueueSize() = 0;
 
+    /// Stops DMA for the given channel.
+    /// Called automatically on channel closure.
+    /// This moves any remaining superpages to the "ready queue", even if they are not filled.
+    virtual void stopDma() = 0;
+
     /// Returns the type of the card this DmaChannelInterface is controlling
     /// \return The card type
     virtual CardType::type getCardType() = 0;

src/CommandLineUtilities/ProgramDmaBench.cxx

@@ -344,7 +344,11 @@ class ProgramDmaBench: public Program
       }
 
       // Lock-free queues. Usable size is (size-1), so we add 1
+      /// Queue for passing filled superpages from the push thread to the readout thread
       folly::ProducerConsumerQueue<size_t> readoutQueue {static_cast<uint32_t>(mMaxSuperpages) + 1};
+      /// Queue for free superpages. This starts out as full, then the readout thread consumes them. When superpages
+      /// arrive, they are passed via the readoutQueue to the readout thread. When the readout thread is done with it,
+      /// it is put back in the freeQueue.
       folly::ProducerConsumerQueue<size_t> freeQueue {static_cast<uint32_t>(mMaxSuperpages) + 1};
       for (size_t i = 0; i < mMaxSuperpages; ++i) {
         size_t offset = i * mSuperpageSize;
@@ -481,7 +485,7 @@ class ProgramDmaBench: public Program
             // Read out pages
             int pages = mSuperpageSize / mPageSize;
             for (int i = 0; i < pages; ++i) {
-              auto readoutCount = mReadoutCount.fetch_add(1, std::memory_order_relaxed);
+              auto readoutCount = fetchAddReadoutCount();
               readoutPage(mBufferBaseAddress + offset + i * mPageSize, mPageSize, readoutCount);
             }
 
@@ -501,22 +505,32 @@ class ProgramDmaBench: public Program
       lowPriorityFuture.get();
     }
 
+    /// Atomically fetch and increment the readout count. We do this because it is accessed by multiple threads.
+    /// Although there is currently only one writer at a time and a regular increment probably would be OK.
+    uint64_t fetchAddReadoutCount()
+    {
+      return mReadoutCount.fetch_add(1, std::memory_order_relaxed);
+    }
+
     /// Free the pages that were pushed in excess
     void freeExcessPages(std::chrono::milliseconds timeout)
     {
+      // First deal with the remaining filled superpages
       auto start = std::chrono::steady_clock::now();
       int popped = 0;
       while ((std::chrono::steady_clock::now() - start) < timeout) {
         if (mChannel->getReadyQueueSize() > 0) {
           auto superpage = mChannel->getSuperpage();
           if (superpage.isFilled()) {
+            readoutPage(mBufferBaseAddress + superpage.getOffset(), superpage.getSize(), fetchAddReadoutCount());
             mChannel->popSuperpage();
             popped += superpage.getReceived() / mPageSize;
           }
         }
       }
+
       std::cout << "\n\n";
-      getLogger() << "Popped " << popped << " excess pages" << endm;
+      getLogger() << "Popped " << popped << " excess filled pages" << endm;
     }
 
     uint32_t get32bitFromPage(uintptr_t pageAddress, size_t index)
@@ -545,9 +559,7 @@ class ProgramDmaBench: public Program
     void readoutPage(uintptr_t pageAddress, size_t pageSize, int64_t readoutCount)
     {
       // Read out to file
-      if (mOptions.fileOutputAscii || mOptions.fileOutputBin) {
-        printToFile(pageAddress, pageSize, readoutCount);
-      }
+      printToFile(pageAddress, pageSize, readoutCount);
 
       // Data error checking
       if (!mOptions.noErrorCheck) {
@@ -815,6 +827,7 @@ class ProgramDmaBench: public Program
       }
     }
 
+    /// Prints the page to a file in ASCII or binary format if such output is enabled
     void printToFile(uintptr_t pageAddress, size_t pageSize, int64_t pageNumber)
     {
       auto page = reinterpret_cast<const volatile uint32_t*>(pageAddress);

src/Cru/CruDmaChannel.cxx

@@ -79,6 +79,9 @@ auto CruDmaChannel::allowedChannels() -> AllowedChannels {
 CruDmaChannel::~CruDmaChannel()
 {
   setBufferNonReady();
+  if (mReadyQueue.size() > 0) {
+    getLogger() << "Remaining superpages in the ready queue: " << mReadyQueue.size() << InfoLogger::InfoLogger::endm;
+  }
 }
 
 void CruDmaChannel::deviceStartDma()
@@ -136,6 +139,14 @@ void CruDmaChannel::setBufferNonReady()
 void CruDmaChannel::deviceStopDma()
 {
   setBufferNonReady();
+  for (auto& link : mLinks) {
+    size_t size = link.queue.size();
+    for (size_t i = 0; i < size; ++i) {
+      transferSuperpageFromLinkToReady(link);
+    }
+    assert(link.queue.empty());
+  }
+  assert(mLinkQueuesTotalAvailable == LINK_QUEUE_CAPACITY * mLinks.size());
 }
 
 void CruDmaChannel::deviceResetChannel(ResetLevel::type resetLevel)
@@ -203,9 +214,7 @@ void CruDmaChannel::pushSuperpage(Superpage superpage)
   }
 
   // Once we've confirmed the link has a slot available, we push the superpage
-  mLinkQueuesTotalAvailable--;
-  link.queue.push_back(superpage);
-
+  pushSuperpageToLink(link, superpage);
   auto maxPages = superpage.getSize() / Cru::DMA_PAGE_SIZE;
   auto busAddress = getBusOffsetAddress(superpage.getOffset());
   getBar().pushSuperpageDescriptor(link.id, maxPages, busAddress);
@@ -229,6 +238,22 @@ auto CruDmaChannel::popSuperpage() -> Superpage
   return superpage;
 }
 
+void CruDmaChannel::pushSuperpageToLink(Link& link, const Superpage& superpage)
+{
+  mLinkQueuesTotalAvailable--;
+  link.queue.push_back(superpage);
+}
+
+void CruDmaChannel::transferSuperpageFromLinkToReady(Link& link)
+{
+  link.queue.front().ready = true;
+  link.queue.front().received = link.queue.front().size;
+  mReadyQueue.push_back(link.queue.front());
+  mLinkQueuesTotalAvailable++;
+  link.queue.pop_front();
+  link.superpageCounter++;
+}
+
 void CruDmaChannel::fillSuperpages()
 {
   // Check for arrivals & handle them
@@ -255,12 +280,7 @@ void CruDmaChannel::fillSuperpages()
         }
 
         // Front superpage has arrived
-        link.queue.front().ready = true;
-        link.queue.front().received = link.queue.front().size;
-        mReadyQueue.push_back(link.queue.front());
-        mLinkQueuesTotalAvailable++;
-        link.queue.pop_front();
-        link.superpageCounter++;
+        transferSuperpageFromLinkToReady(link);
       }
     }
   }

src/Cru/CruDmaChannel.h

@@ -100,6 +100,12 @@ class CruDmaChannel final : public DmaChannelPdaBase
     /// Gets index of next link to push
     LinkIndex getNextLinkIndex();
 
+    /// Push a superpage to a link
+    void pushSuperpageToLink(Link& link, const Superpage& superpage);
+
+    /// Mark the front superpage of a link ready and transfer it to the ready queue
+    void transferSuperpageFromLinkToReady(Link& link);
+
     /// BAR 0 is needed for DMA engine interaction and various other functions
     Pda::PdaBar mPdaBar;