#ifndef NUMA_H_INCLUDED
#define NUMA_H_INCLUDED
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <sstream>
#include <string>
#include <thread>
#include <utility>
#include <vector>
#include <cstring>
#include "memory.h"
#if defined(__linux__) && !defined(__ANDROID__)
#if !defined(_GNU_SOURCE)
#define _GNU_SOURCE
#endif
#include <sched.h>
#elif defined(_WIN64)
#if _WIN32_WINNT < 0x0601
#undef _WIN32_WINNT
#define _WIN32_WINNT 0x0601
#endif
static constexpr size_t WIN_PROCESSOR_GROUP_SIZE = 64;
#if !defined(NOMINMAX)
#define NOMINMAX
#endif
#include <windows.h>
#if defined small
#undef small
#endif
using SetThreadSelectedCpuSetMasks_t = BOOL (*)(HANDLE, PGROUP_AFFINITY, USHORT);
using GetThreadSelectedCpuSetMasks_t = BOOL (*)(HANDLE, PGROUP_AFFINITY, USHORT, PUSHORT);
#endif
#include "misc.h"
namespace Stockfish {
using CpuIndex = size_t;
using NumaIndex = size_t;
inline CpuIndex get_hardware_concurrency() {
CpuIndex concurrency = std::thread::hardware_concurrency();
#ifdef _WIN64
concurrency = std::max<CpuIndex>(concurrency, GetActiveProcessorCount(ALL_PROCESSOR_GROUPS));
#endif
return concurrency;
}
inline const CpuIndex SYSTEM_THREADS_NB = std::max<CpuIndex>(1, get_hardware_concurrency());
#if defined(_WIN64)
struct WindowsAffinity {
std::optional<std::set<CpuIndex>> oldApi;
std::optional<std::set<CpuIndex>> newApi;
bool isNewDeterminate = true;
bool isOldDeterminate = true;
std::optional<std::set<CpuIndex>> get_combined() const {
if (!oldApi.has_value())
return newApi;
if (!newApi.has_value())
return oldApi;
std::set<CpuIndex> intersect;
std::set_intersection(oldApi->begin(), oldApi->end(), newApi->begin(), newApi->end(),
std::inserter(intersect, intersect.begin()));
return intersect;
}
bool likely_used_old_api() const { return oldApi.has_value() || !isOldDeterminate; }
};
inline std::pair<BOOL, std::vector<USHORT>> get_process_group_affinity() {
static constexpr size_t GroupArrayMinimumAlignment = 4;
static_assert(GroupArrayMinimumAlignment >= alignof(USHORT));
static constexpr int MAX_TRIES = 2;
USHORT GroupCount = 1;
for (int i = 0; i < MAX_TRIES; ++i)
{
auto GroupArray = std::make_unique<USHORT[]>(
GroupCount + (GroupArrayMinimumAlignment / alignof(USHORT) - 1));
USHORT* GroupArrayAligned = align_ptr_up<GroupArrayMinimumAlignment>(GroupArray.get());
const BOOL status =
GetProcessGroupAffinity(GetCurrentProcess(), &GroupCount, GroupArrayAligned);
if (status == 0 && GetLastError() != ERROR_INSUFFICIENT_BUFFER)
{
break;
}
if (status != 0)
{
return std::make_pair(status,
std::vector(GroupArrayAligned, GroupArrayAligned + GroupCount));
}
}
return std::make_pair(0, std::vector<USHORT>());
}
inline WindowsAffinity get_process_affinity() {
HMODULE k32 = GetModuleHandle(TEXT("Kernel32.dll"));
auto GetThreadSelectedCpuSetMasks_f = GetThreadSelectedCpuSetMasks_t(
(void (*)()) GetProcAddress(k32, "GetThreadSelectedCpuSetMasks"));
BOOL status = 0;
WindowsAffinity affinity;
if (GetThreadSelectedCpuSetMasks_f != nullptr)
{
USHORT RequiredMaskCount;
status = GetThreadSelectedCpuSetMasks_f(GetCurrentThread(), nullptr, 0, &RequiredMaskCount);
if (status == 0 && GetLastError() != ERROR_INSUFFICIENT_BUFFER)
{
affinity.isNewDeterminate = false;
}
else if (RequiredMaskCount > 0)
{
auto groupAffinities = std::make_unique<GROUP_AFFINITY[]>(RequiredMaskCount);
status = GetThreadSelectedCpuSetMasks_f(GetCurrentThread(), groupAffinities.get(),
RequiredMaskCount, &RequiredMaskCount);
if (status == 0)
{
affinity.isNewDeterminate = false;
}
else
{
std::set<CpuIndex> cpus;
for (USHORT i = 0; i < RequiredMaskCount; ++i)
{
const size_t procGroupIndex = groupAffinities[i].Group;
for (size_t j = 0; j < WIN_PROCESSOR_GROUP_SIZE; ++j)
{
if (groupAffinities[i].Mask & (KAFFINITY(1) << j))
cpus.insert(procGroupIndex * WIN_PROCESSOR_GROUP_SIZE + j);
}
}
affinity.newApi = std::move(cpus);
}
}
}
DWORD_PTR proc, sys;
status = GetProcessAffinityMask(GetCurrentProcess(), &proc, &sys);
if (status == 0 || proc == 0)
{
affinity.isOldDeterminate = false;
return affinity;
}
std::vector<USHORT> groupAffinity;
std::tie(status, groupAffinity) = get_process_group_affinity();
if (status == 0)
{
affinity.isOldDeterminate = false;
return affinity;
}
if (groupAffinity.size() == 1)
{
if (GetActiveProcessorGroupCount() != 1 || proc != sys)
{
std::set<CpuIndex> cpus;
const size_t procGroupIndex = groupAffinity[0];
const uint64_t mask = static_cast<uint64_t>(proc);
for (size_t j = 0; j < WIN_PROCESSOR_GROUP_SIZE; ++j)
{
if (mask & (KAFFINITY(1) << j))
cpus.insert(procGroupIndex * WIN_PROCESSOR_GROUP_SIZE + j);
}
affinity.oldApi = std::move(cpus);
}
}
else
{
if (GetThreadSelectedCpuSetMasks_f != nullptr)
{
std::thread th([&]() {
std::set<CpuIndex> cpus;
bool isAffinityFull = true;
for (auto procGroupIndex : groupAffinity)
{
const int numActiveProcessors =
GetActiveProcessorCount(static_cast<WORD>(procGroupIndex));
uint64_t procCombined = std::numeric_limits<uint64_t>::max();
uint64_t sysCombined = std::numeric_limits<uint64_t>::max();
for (int i = 0; i < std::min(numActiveProcessors, 2); ++i)
{
GROUP_AFFINITY GroupAffinity;
std::memset(&GroupAffinity, 0, sizeof(GROUP_AFFINITY));
GroupAffinity.Group = static_cast<WORD>(procGroupIndex);
GroupAffinity.Mask = static_cast<KAFFINITY>(1) << i;
status =
SetThreadGroupAffinity(GetCurrentThread(), &GroupAffinity, nullptr);
if (status == 0)
{
affinity.isOldDeterminate = false;
return;
}
SwitchToThread();
DWORD_PTR proc2, sys2;
status = GetProcessAffinityMask(GetCurrentProcess(), &proc2, &sys2);
if (status == 0)
{
affinity.isOldDeterminate = false;
return;
}
procCombined &= static_cast<uint64_t>(proc2);
sysCombined &= static_cast<uint64_t>(sys2);
}
if (procCombined != sysCombined)
isAffinityFull = false;
for (size_t j = 0; j < WIN_PROCESSOR_GROUP_SIZE; ++j)
{
if (procCombined & (KAFFINITY(1) << j))
cpus.insert(procGroupIndex * WIN_PROCESSOR_GROUP_SIZE + j);
}
}
if (!isAffinityFull)
{
affinity.oldApi = std::move(cpus);
}
});
th.join();
}
}
return affinity;
}
#endif
#if defined(__linux__) && !defined(__ANDROID__)
inline std::set<CpuIndex> get_process_affinity() {
std::set<CpuIndex> cpus;
[[maybe_unused]] auto set_to_all_cpus = [&]() {
for (CpuIndex c = 0; c < SYSTEM_THREADS_NB; ++c)
cpus.insert(c);
};
static constexpr CpuIndex MaxNumCpus = 1024 * 64;
cpu_set_t* mask = CPU_ALLOC(MaxNumCpus);
if (mask == nullptr)
std::exit(EXIT_FAILURE);
const size_t masksize = CPU_ALLOC_SIZE(MaxNumCpus);
CPU_ZERO_S(masksize, mask);
const int status = sched_getaffinity(0, masksize, mask);
if (status != 0)
{
CPU_FREE(mask);
std::exit(EXIT_FAILURE);
}
for (CpuIndex c = 0; c < MaxNumCpus; ++c)
if (CPU_ISSET_S(c, masksize, mask))
cpus.insert(c);
CPU_FREE(mask);
return cpus;
}
#endif
#if defined(__linux__) && !defined(__ANDROID__)
inline static const auto STARTUP_PROCESSOR_AFFINITY = get_process_affinity();
#elif defined(_WIN64)
inline static const auto STARTUP_PROCESSOR_AFFINITY = get_process_affinity();
inline static const auto STARTUP_USE_OLD_AFFINITY_API =
STARTUP_PROCESSOR_AFFINITY.likely_used_old_api();
#endif
class NumaReplicatedAccessToken {
public:
NumaReplicatedAccessToken() :
n(0) {}
explicit NumaReplicatedAccessToken(NumaIndex idx) :
n(idx) {}
NumaIndex get_numa_index() const { return n; }
private:
NumaIndex n;
};
class NumaConfig {
public:
NumaConfig() :
highestCpuIndex(0),
customAffinity(false) {
const auto numCpus = SYSTEM_THREADS_NB;
add_cpu_range_to_node(NumaIndex{0}, CpuIndex{0}, numCpus - 1);
}
static NumaConfig from_system([[maybe_unused]] bool respectProcessAffinity = true) {
NumaConfig cfg = empty();
#if defined(__linux__) && !defined(__ANDROID__)
std::set<CpuIndex> allowedCpus;
if (respectProcessAffinity)
allowedCpus = STARTUP_PROCESSOR_AFFINITY;
auto is_cpu_allowed = [respectProcessAffinity, &allowedCpus](CpuIndex c) {
return !respectProcessAffinity || allowedCpus.count(c) == 1;
};
bool useFallback = false;
auto fallback = [&]() {
useFallback = true;
cfg = empty();
};
auto nodeIdsStr = read_file_to_string("/sys/devices/system/node/online");
if (!nodeIdsStr.has_value() || nodeIdsStr->empty())
{
fallback();
}
else
{
remove_whitespace(*nodeIdsStr);
for (size_t n : indices_from_shortened_string(*nodeIdsStr))
{
std::string path =
std::string("/sys/devices/system/node/node") + std::to_string(n) + "/cpulist";
auto cpuIdsStr = read_file_to_string(path);
if (!cpuIdsStr.has_value())
{
fallback();
break;
}
else
{
remove_whitespace(*cpuIdsStr);
for (size_t c : indices_from_shortened_string(*cpuIdsStr))
{
if (is_cpu_allowed(c))
cfg.add_cpu_to_node(n, c);
}
}
}
}
if (useFallback)
{
for (CpuIndex c = 0; c < SYSTEM_THREADS_NB; ++c)
if (is_cpu_allowed(c))
cfg.add_cpu_to_node(NumaIndex{0}, c);
}
#elif defined(_WIN64)
std::optional<std::set<CpuIndex>> allowedCpus;
if (respectProcessAffinity)
allowedCpus = STARTUP_PROCESSOR_AFFINITY.get_combined();
auto is_cpu_allowed = [&allowedCpus](CpuIndex c) {
return !allowedCpus.has_value() || allowedCpus->count(c) == 1;
};
WORD numProcGroups = GetActiveProcessorGroupCount();
for (WORD procGroup = 0; procGroup < numProcGroups; ++procGroup)
{
for (BYTE number = 0; number < WIN_PROCESSOR_GROUP_SIZE; ++number)
{
PROCESSOR_NUMBER procnum;
procnum.Group = procGroup;
procnum.Number = number;
procnum.Reserved = 0;
USHORT nodeNumber;
const BOOL status = GetNumaProcessorNodeEx(&procnum, &nodeNumber);
const CpuIndex c = static_cast<CpuIndex>(procGroup) * WIN_PROCESSOR_GROUP_SIZE
+ static_cast<CpuIndex>(number);
if (status != 0 && nodeNumber != std::numeric_limits<USHORT>::max()
&& is_cpu_allowed(c))
{
cfg.add_cpu_to_node(nodeNumber, c);
}
}
}
{
NumaConfig splitCfg = empty();
NumaIndex splitNodeIndex = 0;
for (const auto& cpus : cfg.nodes)
{
if (cpus.empty())
continue;
size_t lastProcGroupIndex = *(cpus.begin()) / WIN_PROCESSOR_GROUP_SIZE;
for (CpuIndex c : cpus)
{
const size_t procGroupIndex = c / WIN_PROCESSOR_GROUP_SIZE;
if (procGroupIndex != lastProcGroupIndex)
{
splitNodeIndex += 1;
lastProcGroupIndex = procGroupIndex;
}
splitCfg.add_cpu_to_node(splitNodeIndex, c);
}
splitNodeIndex += 1;
}
cfg = std::move(splitCfg);
}
#else
for (CpuIndex c = 0; c < SYSTEM_THREADS_NB; ++c)
cfg.add_cpu_to_node(NumaIndex{0}, c);
#endif
cfg.remove_empty_numa_nodes();
if (!respectProcessAffinity)
cfg.customAffinity = true;
return cfg;
}
static NumaConfig from_string(const std::string& s) {
NumaConfig cfg = empty();
NumaIndex n = 0;
for (auto&& nodeStr : split(s, ":"))
{
auto indices = indices_from_shortened_string(std::string(nodeStr));
if (!indices.empty())
{
for (auto idx : indices)
{
if (!cfg.add_cpu_to_node(n, CpuIndex(idx)))
std::exit(EXIT_FAILURE);
}
n += 1;
}
}
cfg.customAffinity = true;
return cfg;
}
NumaConfig(const NumaConfig&) = delete;
NumaConfig(NumaConfig&&) = default;
NumaConfig& operator=(const NumaConfig&) = delete;
NumaConfig& operator=(NumaConfig&&) = default;
bool is_cpu_assigned(CpuIndex n) const { return nodeByCpu.count(n) == 1; }
NumaIndex num_numa_nodes() const { return nodes.size(); }
CpuIndex num_cpus_in_numa_node(NumaIndex n) const {
assert(n < nodes.size());
return nodes[n].size();
}
CpuIndex num_cpus() const { return nodeByCpu.size(); }
bool requires_memory_replication() const { return customAffinity || nodes.size() > 1; }
std::string to_string() const {
std::string str;
bool isFirstNode = true;
for (auto&& cpus : nodes)
{
if (!isFirstNode)
str += ":";
bool isFirstSet = true;
auto rangeStart = cpus.begin();
for (auto it = cpus.begin(); it != cpus.end(); ++it)
{
auto next = std::next(it);
if (next == cpus.end() || *next != *it + 1)
{
if (!isFirstSet)
str += ",";
const CpuIndex last = *it;
if (it != rangeStart)
{
const CpuIndex first = *rangeStart;
str += std::to_string(first);
str += "-";
str += std::to_string(last);
}
else
str += std::to_string(last);
rangeStart = next;
isFirstSet = false;
}
}
isFirstNode = false;
}
return str;
}
bool suggests_binding_threads(CpuIndex numThreads) const {
if (customAffinity)
return true;
if (numThreads <= 1)
return false;
size_t largestNodeSize = 0;
for (auto&& cpus : nodes)
if (cpus.size() > largestNodeSize)
largestNodeSize = cpus.size();
auto is_node_small = [largestNodeSize](const std::set<CpuIndex>& node) {
static constexpr double SmallNodeThreshold = 0.6;
return static_cast<double>(node.size()) / static_cast<double>(largestNodeSize)
<= SmallNodeThreshold;
};
size_t numNotSmallNodes = 0;
for (auto&& cpus : nodes)
if (!is_node_small(cpus))
numNotSmallNodes += 1;
return (numThreads > largestNodeSize / 2 || numThreads >= numNotSmallNodes * 4)
&& nodes.size() > 1;
}
std::vector<NumaIndex> distribute_threads_among_numa_nodes(CpuIndex numThreads) const {
std::vector<NumaIndex> ns;
if (nodes.size() == 1)
{
ns.resize(numThreads, NumaIndex{0});
}
else
{
std::vector<size_t> occupation(nodes.size(), 0);
for (CpuIndex c = 0; c < numThreads; ++c)
{
NumaIndex bestNode{0};
float bestNodeFill = std::numeric_limits<float>::max();
for (NumaIndex n = 0; n < nodes.size(); ++n)
{
float fill =
static_cast<float>(occupation[n] + 1) / static_cast<float>(nodes[n].size());
if (fill < bestNodeFill)
{
bestNode = n;
bestNodeFill = fill;
}
}
ns.emplace_back(bestNode);
occupation[bestNode] += 1;
}
}
return ns;
}
NumaReplicatedAccessToken bind_current_thread_to_numa_node(NumaIndex n) const {
if (n >= nodes.size() || nodes[n].size() == 0)
std::exit(EXIT_FAILURE);
#if defined(__linux__) && !defined(__ANDROID__)
cpu_set_t* mask = CPU_ALLOC(highestCpuIndex + 1);
if (mask == nullptr)
std::exit(EXIT_FAILURE);
const size_t masksize = CPU_ALLOC_SIZE(highestCpuIndex + 1);
CPU_ZERO_S(masksize, mask);
for (CpuIndex c : nodes[n])
CPU_SET_S(c, masksize, mask);
const int status = sched_setaffinity(0, masksize, mask);
CPU_FREE(mask);
if (status != 0)
std::exit(EXIT_FAILURE);
sched_yield();
#elif defined(_WIN64)
HMODULE k32 = GetModuleHandle(TEXT("Kernel32.dll"));
auto SetThreadSelectedCpuSetMasks_f = SetThreadSelectedCpuSetMasks_t(
(void (*)()) GetProcAddress(k32, "SetThreadSelectedCpuSetMasks"));
if (SetThreadSelectedCpuSetMasks_f != nullptr)
{
const USHORT numProcGroups = USHORT(
((highestCpuIndex + 1) + WIN_PROCESSOR_GROUP_SIZE - 1) / WIN_PROCESSOR_GROUP_SIZE);
auto groupAffinities = std::make_unique<GROUP_AFFINITY[]>(numProcGroups);
std::memset(groupAffinities.get(), 0, sizeof(GROUP_AFFINITY) * numProcGroups);
for (WORD i = 0; i < numProcGroups; ++i)
groupAffinities[i].Group = i;
for (CpuIndex c : nodes[n])
{
const size_t procGroupIndex = c / WIN_PROCESSOR_GROUP_SIZE;
const size_t idxWithinProcGroup = c % WIN_PROCESSOR_GROUP_SIZE;
groupAffinities[procGroupIndex].Mask |= KAFFINITY(1) << idxWithinProcGroup;
}
HANDLE hThread = GetCurrentThread();
const BOOL status =
SetThreadSelectedCpuSetMasks_f(hThread, groupAffinities.get(), numProcGroups);
if (status == 0)
std::exit(EXIT_FAILURE);
SwitchToThread();
}
if (SetThreadSelectedCpuSetMasks_f == nullptr || STARTUP_USE_OLD_AFFINITY_API)
{
GROUP_AFFINITY affinity;
std::memset(&affinity, 0, sizeof(GROUP_AFFINITY));
const size_t forcedProcGroupIndex = *(nodes[n].begin()) / WIN_PROCESSOR_GROUP_SIZE;
affinity.Group = static_cast<WORD>(forcedProcGroupIndex);
for (CpuIndex c : nodes[n])
{
const size_t procGroupIndex = c / WIN_PROCESSOR_GROUP_SIZE;
const size_t idxWithinProcGroup = c % WIN_PROCESSOR_GROUP_SIZE;
if (procGroupIndex != forcedProcGroupIndex)
continue;
affinity.Mask |= KAFFINITY(1) << idxWithinProcGroup;
}
HANDLE hThread = GetCurrentThread();
const BOOL status = SetThreadGroupAffinity(hThread, &affinity, nullptr);
if (status == 0)
std::exit(EXIT_FAILURE);
SwitchToThread();
}
#endif
return NumaReplicatedAccessToken(n);
}
template<typename FuncT>
void execute_on_numa_node(NumaIndex n, FuncT&& f) const {
std::thread th([this, &f, n]() {
bind_current_thread_to_numa_node(n);
std::forward<FuncT>(f)();
});
th.join();
}
private:
std::vector<std::set<CpuIndex>> nodes;
std::map<CpuIndex, NumaIndex> nodeByCpu;
CpuIndex highestCpuIndex;
bool customAffinity;
static NumaConfig empty() { return NumaConfig(EmptyNodeTag{}); }
struct EmptyNodeTag {};
NumaConfig(EmptyNodeTag) :
highestCpuIndex(0),
customAffinity(false) {}
void remove_empty_numa_nodes() {
std::vector<std::set<CpuIndex>> newNodes;
for (auto&& cpus : nodes)
if (!cpus.empty())
newNodes.emplace_back(std::move(cpus));
nodes = std::move(newNodes);
}
bool add_cpu_to_node(NumaIndex n, CpuIndex c) {
if (is_cpu_assigned(c))
return false;
while (nodes.size() <= n)
nodes.emplace_back();
nodes[n].insert(c);
nodeByCpu[c] = n;
if (c > highestCpuIndex)
highestCpuIndex = c;
return true;
}
bool add_cpu_range_to_node(NumaIndex n, CpuIndex cfirst, CpuIndex clast) {
for (CpuIndex c = cfirst; c <= clast; ++c)
if (is_cpu_assigned(c))
return false;
while (nodes.size() <= n)
nodes.emplace_back();
for (CpuIndex c = cfirst; c <= clast; ++c)
{
nodes[n].insert(c);
nodeByCpu[c] = n;
}
if (clast > highestCpuIndex)
highestCpuIndex = clast;
return true;
}
static std::vector<size_t> indices_from_shortened_string(const std::string& s) {
std::vector<size_t> indices;
if (s.empty())
return indices;
for (const auto& ss : split(s, ","))
{
if (ss.empty())
continue;
auto parts = split(ss, "-");
if (parts.size() == 1)
{
const CpuIndex c = CpuIndex{str_to_size_t(std::string(parts[0]))};
indices.emplace_back(c);
}
else if (parts.size() == 2)
{
const CpuIndex cfirst = CpuIndex{str_to_size_t(std::string(parts[0]))};
const CpuIndex clast = CpuIndex{str_to_size_t(std::string(parts[1]))};
for (size_t c = cfirst; c <= clast; ++c)
{
indices.emplace_back(c);
}
}
}
return indices;
}
};
class NumaReplicationContext;
class NumaReplicatedBase {
public:
NumaReplicatedBase(NumaReplicationContext& ctx);
NumaReplicatedBase(const NumaReplicatedBase&) = delete;
NumaReplicatedBase(NumaReplicatedBase&& other) noexcept;
NumaReplicatedBase& operator=(const NumaReplicatedBase&) = delete;
NumaReplicatedBase& operator=(NumaReplicatedBase&& other) noexcept;
virtual void on_numa_config_changed() = 0;
virtual ~NumaReplicatedBase();
const NumaConfig& get_numa_config() const;
private:
NumaReplicationContext* context;
};
template<typename T>
class NumaReplicated: public NumaReplicatedBase {
public:
using ReplicatorFuncType = std::function<T(const T&)>;
NumaReplicated(NumaReplicationContext& ctx) :
NumaReplicatedBase(ctx) {
replicate_from(T{});
}
NumaReplicated(NumaReplicationContext& ctx, T&& source) :
NumaReplicatedBase(ctx) {
replicate_from(std::move(source));
}
NumaReplicated(const NumaReplicated&) = delete;
NumaReplicated(NumaReplicated&& other) noexcept :
NumaReplicatedBase(std::move(other)),
instances(std::exchange(other.instances, {})) {}
NumaReplicated& operator=(const NumaReplicated&) = delete;
NumaReplicated& operator=(NumaReplicated&& other) noexcept {
NumaReplicatedBase::operator=(*this, std::move(other));
instances = std::exchange(other.instances, {});
return *this;
}
NumaReplicated& operator=(T&& source) {
replicate_from(std::move(source));
return *this;
}
~NumaReplicated() override = default;
const T& operator[](NumaReplicatedAccessToken token) const {
assert(token.get_numa_index() < instances.size());
return *(instances[token.get_numa_index()]);
}
const T& operator*() const { return *(instances[0]); }
const T* operator->() const { return instances[0].get(); }
template<typename FuncT>
void modify_and_replicate(FuncT&& f) {
auto source = std::move(instances[0]);
std::forward<FuncT>(f)(*source);
replicate_from(std::move(*source));
}
void on_numa_config_changed() override {
auto source = std::move(instances[0]);
replicate_from(std::move(*source));
}
private:
std::vector<std::unique_ptr<T>> instances;
void replicate_from(T&& source) {
instances.clear();
const NumaConfig& cfg = get_numa_config();
if (cfg.requires_memory_replication())
{
for (NumaIndex n = 0; n < cfg.num_numa_nodes(); ++n)
{
cfg.execute_on_numa_node(
n, [this, &source]() { instances.emplace_back(std::make_unique<T>(source)); });
}
}
else
{
assert(cfg.num_numa_nodes() == 1);
instances.emplace_back(std::make_unique<T>(std::move(source)));
}
}
};
template<typename T>
class LazyNumaReplicated: public NumaReplicatedBase {
public:
using ReplicatorFuncType = std::function<T(const T&)>;
LazyNumaReplicated(NumaReplicationContext& ctx) :
NumaReplicatedBase(ctx) {
prepare_replicate_from(T{});
}
LazyNumaReplicated(NumaReplicationContext& ctx, T&& source) :
NumaReplicatedBase(ctx) {
prepare_replicate_from(std::move(source));
}
LazyNumaReplicated(const LazyNumaReplicated&) = delete;
LazyNumaReplicated(LazyNumaReplicated&& other) noexcept :
NumaReplicatedBase(std::move(other)),
instances(std::exchange(other.instances, {})) {}
LazyNumaReplicated& operator=(const LazyNumaReplicated&) = delete;
LazyNumaReplicated& operator=(LazyNumaReplicated&& other) noexcept {
NumaReplicatedBase::operator=(*this, std::move(other));
instances = std::exchange(other.instances, {});
return *this;
}
LazyNumaReplicated& operator=(T&& source) {
prepare_replicate_from(std::move(source));
return *this;
}
~LazyNumaReplicated() override = default;
const T& operator[](NumaReplicatedAccessToken token) const {
assert(token.get_numa_index() < instances.size());
ensure_present(token.get_numa_index());
return *(instances[token.get_numa_index()]);
}
const T& operator*() const { return *(instances[0]); }
const T* operator->() const { return instances[0].get(); }
template<typename FuncT>
void modify_and_replicate(FuncT&& f) {
auto source = std::move(instances[0]);
std::forward<FuncT>(f)(*source);
prepare_replicate_from(std::move(*source));
}
void on_numa_config_changed() override {
auto source = std::move(instances[0]);
prepare_replicate_from(std::move(*source));
}
private:
mutable std::vector<std::unique_ptr<T>> instances;
mutable std::mutex mutex;
void ensure_present(NumaIndex idx) const {
assert(idx < instances.size());
if (instances[idx] != nullptr)
return;
assert(idx != 0);
std::unique_lock<std::mutex> lock(mutex);
if (instances[idx] != nullptr)
return;
const NumaConfig& cfg = get_numa_config();
cfg.execute_on_numa_node(
idx, [this, idx]() { instances[idx] = std::make_unique<T>(*instances[0]); });
}
void prepare_replicate_from(T&& source) {
instances.clear();
const NumaConfig& cfg = get_numa_config();
if (cfg.requires_memory_replication())
{
assert(cfg.num_numa_nodes() > 0);
cfg.execute_on_numa_node(
0, [this, &source]() { instances.emplace_back(std::make_unique<T>(source)); });
instances.resize(cfg.num_numa_nodes());
}
else
{
assert(cfg.num_numa_nodes() == 1);
instances.emplace_back(std::make_unique<T>(std::move(source)));
}
}
};
class NumaReplicationContext {
public:
NumaReplicationContext(NumaConfig&& cfg) :
config(std::move(cfg)) {}
NumaReplicationContext(const NumaReplicationContext&) = delete;
NumaReplicationContext(NumaReplicationContext&&) = delete;
NumaReplicationContext& operator=(const NumaReplicationContext&) = delete;
NumaReplicationContext& operator=(NumaReplicationContext&&) = delete;
~NumaReplicationContext() {
if (!trackedReplicatedObjects.empty())
std::exit(EXIT_FAILURE);
}
void attach(NumaReplicatedBase* obj) {
assert(trackedReplicatedObjects.count(obj) == 0);
trackedReplicatedObjects.insert(obj);
}
void detach(NumaReplicatedBase* obj) {
assert(trackedReplicatedObjects.count(obj) == 1);
trackedReplicatedObjects.erase(obj);
}
void move_attached([[maybe_unused]] NumaReplicatedBase* oldObj, NumaReplicatedBase* newObj) {
assert(trackedReplicatedObjects.count(oldObj) == 1);
assert(trackedReplicatedObjects.count(newObj) == 0);
trackedReplicatedObjects.erase(oldObj);
trackedReplicatedObjects.insert(newObj);
}
void set_numa_config(NumaConfig&& cfg) {
config = std::move(cfg);
for (auto&& obj : trackedReplicatedObjects)
obj->on_numa_config_changed();
}
const NumaConfig& get_numa_config() const { return config; }
private:
NumaConfig config;
std::set<NumaReplicatedBase*> trackedReplicatedObjects;
};
inline NumaReplicatedBase::NumaReplicatedBase(NumaReplicationContext& ctx) :
context(&ctx) {
context->attach(this);
}
inline NumaReplicatedBase::NumaReplicatedBase(NumaReplicatedBase&& other) noexcept :
context(std::exchange(other.context, nullptr)) {
context->move_attached(&other, this);
}
inline NumaReplicatedBase& NumaReplicatedBase::operator=(NumaReplicatedBase&& other) noexcept {
context = std::exchange(other.context, nullptr);
context->move_attached(&other, this);
return *this;
}
inline NumaReplicatedBase::~NumaReplicatedBase() {
if (context != nullptr)
context->detach(this);
}
inline const NumaConfig& NumaReplicatedBase::get_numa_config() const {
return context->get_numa_config();
}
}
#endif