numa.hpp

Go to the documentation of this file.

#pragma once
 
#include <cstdlib>
#include <jaffarCommon/exceptions.hpp>
#include <jaffarCommon/parallel.hpp>
#include <map>
#include <mutex>
#include <numa.h>
#include <vector>
 
namespace jaffarPlus
{
 
static int _numaCount; 
 
static int _threadCount; 
 
static std::vector<std::vector<size_t>> _numaDistanceMatrix; 
 
static std::vector<std::vector<size_t>>
    _numaPreferenceMatrix; 
 
static std::vector<ssize_t> _numaDelegateThreadId; 
 
static thread_local int _preferredNumaDomain; 
 
static thread_local int _myThreadId; 
 
__INLINE__ void initializeNUMA()
{
  // Getting number of threads
  _threadCount = jaffarCommon::parallel::getMaxThreadCount();
 
  // Getting number of numa domains (serial execution collapses to a single domain)
  if (_threadCount == 1)
    _numaCount = 1;
  else
    _numaCount = numa_max_node() + 1;
 
  // Checking whether the numa library calls are available
  const auto numaAvailable = numa_available();
  if (numaAvailable != 0) JAFFAR_THROW_RUNTIME("The system does not provide NUMA detection support.");
 
  // Overriding thread count, if requested
  if (auto* value = std::getenv("JAFFAR_ENGINE_OVERRIDE_MAX_THREAD_COUNT")) jaffarCommon::parallel::setThreadCount(std::stoul(value));
 
  // Check NUMA distances
  _numaDistanceMatrix.resize(_numaCount);
  for (ssize_t i = 0; i < _numaCount; i++) _numaDistanceMatrix[i].resize(_numaCount);
  for (ssize_t i = 0; i < _numaCount; i++)
    for (ssize_t j = 0; j < _numaCount; j++) _numaDistanceMatrix[i][j] = (size_t)numa_distance(i, j);
 
  // printf("NUMA Distances:\n");
  // for(ssize_t i = 0; i < _numaCount; i++)
  // {
  //   for(ssize_t j = 0; j < _numaCount; j++)
  //   {
  //     printf("%3lu ", _numaDistanceMatrix[i][j]);
  //   }
  //   printf("\n");
  // }
 
  // Establishing NUMA preference matrix
  _numaPreferenceMatrix.resize(_numaCount);
  for (ssize_t i = 0; i < _numaCount; i++)
  {
    // Putting NUMA domains in the same group based on distance
    std::map<size_t, std::vector<int>> _localPreferenceMap;
    for (ssize_t j = 0; j < _numaCount; j++) _localPreferenceMap[_numaDistanceMatrix[i][j]].push_back(j);
 
    // Assigning NUMA domains based on local preference
    for (const auto& entry : _localPreferenceMap)
    {
      const auto& localPreferenceGroup = entry.second;
      for (ssize_t j = 0; j < (ssize_t)localPreferenceGroup.size(); j++)
      {
        const size_t offset = (j + i) % localPreferenceGroup.size();
        _numaPreferenceMatrix[i].push_back(localPreferenceGroup[offset]);
      }
    }
  }
 
  // printf("NUMA Locality Preferences:\n");
  // for(ssize_t i = 0; i < _numaCount; i++)
  // {
  //   for(ssize_t j = 0; j < _numaCount; j++)
  //   {
  //     printf("%3lu ", _numaPreferenceMatrix[i][j]);
  //   }
  //   printf("\n");
  // }
 
  // Initializing numa delegate thread storage
  _numaDelegateThreadId.resize(_numaCount);
  for (int i = 0; i < _numaCount; i++) _numaDelegateThreadId[i] = -1;
 
  // Detecting thread-specific NUMA information
  std::mutex workMutex;
  JAFFAR_PARALLEL
  {
    // Identify this worker by its dense OpenMP thread id (unique 0..N-1), NOT sched_getcpu(): the CPU
    // id collides whenever threads share a core -- under oversubscription, or simply unpinned
    // scheduling on CI runners that don't set OMP_PROC_BIND. A colliding id corrupts the per-NUMA
    // delegate selection below (and StateDb's delegate-gated queue allocation), leaving a domain with
    // no queues so the search immediately "runs out of states". The preferred NUMA domain still comes
    // from the actual CPU the thread is currently on.
    _myThreadId          = jaffarCommon::parallel::getThreadId();
    _preferredNumaDomain = numa_node_of_cpu(sched_getcpu());
 
    // Setting myself as numa delegate, if I'm the first thread seen in this numa domain
    workMutex.lock();
    if (_numaDelegateThreadId[_preferredNumaDomain] == -1) _numaDelegateThreadId[_preferredNumaDomain] = _myThreadId;
    workMutex.unlock();
  }
}
 
} // namespace jaffarPlus