Merge pull request #227 from sy-c/master

v2.13.0

Author: sy-c

doc/configurationParameters.md

@@ -106,6 +106,7 @@ The parameters related to 3rd-party libraries are described here for convenience
 | equipment-* | memoryPoolNumberOfPages | int | | Number of pages to be created for this equipment, taken from the chosen memory bank. The bank should have enough free space to accomodate (memoryPoolNumberOfPages + 1) * memoryPoolPageSize bytes. |
 | equipment-* | memoryPoolPageSize | bytes | | Size of each memory page to be created. Some space might be kept in each page for internal readout usage. |
 | equipment-* | name | string| | Name used to identify this equipment (in logs). By default, it takes the name of the configuration section, equipment-xxx |
+| equipment-* | numaNode | string | auto | If set, memory / thread will try to use given NUMA node. If "auto", will try to guess it for given equipment (eg ROC). |
 | equipment-* | outputFifoSize | int | -1 | Size of output fifo (number of pages). If -1, set to the same value as memoryPoolNumberOfPages (this ensures that nothing can block the equipment while there are free pages). |
 | equipment-* | rdhCheckDetectorField | int | 0 | If set, the detector field is checked and changes reported. |
 | equipment-* | rdhCheckEnabled | int | 0 | If set, data pages are parsed and RDH headers checked. Errors are reported in logs. |

doc/releaseNotes.md

@@ -477,3 +477,6 @@ This file describes the main feature changes for each readout.exe released versi
   - added consumer-FairMQChannel-*.enablePackedCopy. If set, the same superpage may be reused (space allowing) for the copy of multiple HBF (instead of a separate one for each copy). This allows a reduced memoryPoolNumberOfPages.
 - ROC fifo health check enabled, a warning message is logged when in/out ROC FIFOs are empty/full.
 
+## v2.13.0 - 11/08/2022
+- Updated configuration parameters:
+  - added equipment-*.numaNode. If set, memory / thread will try to use given NUMA node. If "auto", will try to guess it for given equipment (eg ROC). "auto" is default.

src/ConsumerFMQchannel.cxx

@@ -255,7 +255,7 @@ class ConsumerFMQchannel : public Consumer
           decDataBlockStats((*blockRef)->getData());
           delete blockRef;
         }
-      },"",0,fair::mq::RegionConfig{true, true});  // lock / zero
+      },fair::mq::RegionConfig{false,false});  // lock / zero - done later
 
       theLog.log(LogInfoDevel_(3008), "Got FMQ unmanaged memory buffer size %lu @ %p", memoryBuffer->GetSize(), memoryBuffer->GetData());
     }

src/MemoryBankManager.cxx

@@ -10,6 +10,9 @@
 // or submit itself to any jurisdiction.
 
 #include "MemoryBankManager.h"
+#include "ReadoutUtils.h"
+#include <unistd.h>
+#include <sys/mman.h>
 
 #include "readoutInfoLogger.h"
 
@@ -35,7 +38,7 @@ int MemoryBankManager::addBank(std::shared_ptr<MemoryBank> bankPtr, std::string
   return 0;
 }
 
-std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize, size_t pageNumber, std::string bankName, size_t firstPageOffset, size_t blockAlign)
+std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize, size_t pageNumber, std::string bankName, size_t firstPageOffset, size_t blockAlign, int numaNode)
 {
 
   void* baseAddress = nullptr; // base address of bank from which the block is taken
@@ -92,6 +95,13 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
       }
     }
 
+    // align at least to memory page
+    int systemPageSize = getpagesize();
+    if ((int)blockAlign < systemPageSize) {
+      blockAlign = systemPageSize;
+      theLog.log(LogInfoDevel,"Aligning memory block by default on system page size = %d bytes", systemPageSize);
+    }
+
     // align beginning of block as specified
     if (blockAlign > 0) {
       size_t bytesExcess = (((size_t)baseAddress) + offset) % blockAlign;
@@ -114,6 +124,36 @@ std::shared_ptr<MemoryPagesPool> MemoryBankManager::getPagedPool(size_t pageSize
   }
   // end of locked block
 
+  if (numaNode >= 0) {
+      // actual memory assignment is done on first write, in particular for FMQ
+      // so set NUMA node and zero the memory to lock it
+      // or try to move the block ?
+      numaBind(numaNode);
+  }
+
+  theLog.log(LogInfoDevel, "Zero memory");
+  void *blockAddress = &(((char*)baseAddress)[offset]);
+  // ensure pages stay in RAM
+  // no need to lock them now, we write the full range on next line. keep them locked then. this is faster.
+  mlock2(blockAddress, blockSize, MLOCK_ONFAULT);
+  bzero(blockAddress, blockSize);
+  theLog.log(LogInfoDevel, "Zero memory done");
+
+  if (numaNode >= 0) {
+    numaBind(-1);
+  }
+
+  int ptrNumaNode = -1;
+  if (numaGetNodeFromAddress(blockAddress, ptrNumaNode) == 0) {
+    theLog.log(LogInfoDevel, "Memory at %p is at node %d", blockAddress, ptrNumaNode);
+    if (numaNode >= 0) {
+      if (ptrNumaNode != numaNode) {
+        theLog.log(LogWarningDevel, "Warning, could not allocate memory pool on requested NUMA node");
+        // todo: try to move ?
+      }
+    }
+  }
+
   // create pool of pages from new block
   return std::make_shared<MemoryPagesPool>(pageSize, pageNumber, &(((char*)baseAddress)[offset]), blockSize, nullptr, firstPageOffset, newId);
 }

src/MemoryBankManager.h

@@ -43,8 +43,9 @@ class MemoryBankManager
   // - bankName: name of the bank from which to create the pool. If not specified, using the first bank.
   // - firstPageOffset: to control alignment of first page in pool. With zero, start from beginning of big block.
   // - blockAlign: alignment of beginning of big memory block from which pool is created. Pool will start at a multiple of this value.
+  // - numaNode: if >= 0, try to allocate the pool on given NUMA node.
   // NB: trivial implementation, once a region from a bank has been used, it can not be reused after the corresponding pool of pages has been release ... don't want to deal with fragmentation etc
-  std::shared_ptr<MemoryPagesPool> getPagedPool(size_t pageSize, size_t pageNumber, std::string bankName = "", size_t firstPageOffset = 0, size_t blockAlign = 0);
+  std::shared_ptr<MemoryPagesPool> getPagedPool(size_t pageSize, size_t pageNumber, std::string bankName = "", size_t firstPageOffset = 0, size_t blockAlign = 0, int numaNode = -1);
 
   // a struct to define a memory range
   struct memoryRange {

src/ReadoutEquipmentRORC.cxx

@@ -10,6 +10,7 @@
 // or submit itself to any jurisdiction.
 
 #include <Common/Timer.h>
+#include <ReadoutCard/CardFinder.h>
 #include <ReadoutCard/ChannelFactory.h>
 #include <ReadoutCard/DmaChannelInterface.h>
 #include <ReadoutCard/Exception.h>
@@ -515,3 +516,12 @@ void ReadoutEquipmentRORC::finalCounters()
   }
 }
 
+int getPreferredROCNumaNode(ConfigFile& cfg, std::string name) {
+  try {
+    std::string cardId = cfg.getValue<std::string>(name + ".cardId");
+    return AliceO2::roc::findCard(cardId).numaNode;
+  }
+  catch (...) {
+  }
+  return -1;
+}

src/ReadoutUtils.cxx

@@ -18,6 +18,12 @@
 #include <filesystem>
 
 #include "RAWDataHeader.h"
+#include "readoutInfoLogger.h"
+
+#ifdef WITH_NUMA
+#include <numa.h>
+#include <numaif.h>
+#endif
 
 // function to convert a string to a 64-bit integer value
 // allowing usual "base units" in suffix (k,M,G,T)
@@ -256,3 +262,39 @@ int getStatsFilesystem(unsigned long long &freeBytes, const std::string &path) {
   return 0; 
 }
 
+int numaBind(int numaNode) {
+  #ifdef WITH_NUMA
+    struct bitmask* nodemask = nullptr;
+    if (numaNode>=0) {
+      nodemask = numa_allocate_nodemask();
+      if (nodemask != nullptr) {
+        theLog.log(LogInfoDevel, "Enforcing memory allocated on NUMA node %d", numaNode);
+        numa_bitmask_clearall(nodemask);
+        numa_bitmask_setbit(nodemask, numaNode);
+        numa_bind(nodemask);
+        numa_free_nodemask(nodemask);
+	return 0;
+      }
+    } else {
+      nodemask = numa_get_mems_allowed();
+      numa_bind(nodemask);
+      theLog.log(LogInfoDevel, "Releasing memory NUMA node enforcement");
+      return 0;
+    }
+    //numa_set_membind(nodemask);
+  #endif
+  return -1;
+}
+
+int numaGetNodeFromAddress(void *ptr, int &node) {
+  #ifdef WITH_NUMA
+    int err;
+    node = -1;
+    err = move_pages(0, 1, &ptr, NULL, &node, 0);
+    if (!err) {
+      return 0;
+    }
+    // EFAULT -> memory not mapped yet
+  #endif
+  return -1;
+}

src/ReadoutUtils.h

@@ -71,6 +71,15 @@ int getStatsMemory(unsigned long long &freeBytes, const std::string& keyword);
 // returns 0 on success, -1 on error
 int getStatsFilesystem(unsigned long long &freeBytes, const std::string& path);
 
+// function to bind CPU / MEM to specific numa node
+// Use numaNode = -1 to bind to any / unbind.
+// returns 0 on success, or an error code.
+int numaBind(int numaNode);
+
+// function to get NUMA node of where given pointer is allocated
+// returns 0 on success, or an error code
+int numaGetNodeFromAddress(void *ptr, int &node);
+
 // end of _READOUTUTILS_H
 #endif
 

src/ReadoutVersion.h

@@ -9,5 +9,5 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-#define READOUT_VERSION "2.12.0"
+#define READOUT_VERSION "2.13.0"
 

src/ZmqClient.cxx

@@ -14,10 +14,7 @@
 #include <functional>
 #include <zmq.h>
 
-#include <InfoLogger/InfoLogger.hxx>
-#include <InfoLogger/InfoLoggerMacros.hxx>
-using namespace AliceO2::InfoLogger;
-extern InfoLogger theLog;
+#include "readoutInfoLogger.h"
 
 ZmqClient::ZmqClient(const std::string& url, int maxMsgSize, int zmqMaxQueue)
 {