kcenon::thread::numa_work_stealer Class Reference

NUMA-aware work stealer with enhanced victim selection policies. More...

#include <numa_work_stealer.h>

Collaboration diagram for kcenon::thread::numa_work_stealer:

Public Types
using	deque_accessor_fn = std::function<lockfree::work_stealing_deque<job*>*(std::size_t)>
	Function type for accessing worker's local deque.
using	cpu_accessor_fn = std::function<int(std::size_t)>
	Function type for getting worker's CPU affinity.

Public Member Functions
	numa_work_stealer (std::size_t worker_count, deque_accessor_fn deque_accessor, cpu_accessor_fn cpu_accessor, enhanced_work_stealing_config config={})
	Construct a NUMA-aware work stealer.
	~numa_work_stealer ()=default
	Destructor.
	numa_work_stealer (const numa_work_stealer &)=delete
numa_work_stealer &	operator= (const numa_work_stealer &)=delete
	numa_work_stealer (numa_work_stealer &&)=delete
numa_work_stealer &	operator= (numa_work_stealer &&)=delete
auto	steal_for (std::size_t worker_id) -> job *
	Attempt to steal work for a worker.
auto	steal_batch_for (std::size_t worker_id, std::size_t max_count) -> std::vector< job * >
	Attempt to steal multiple jobs for a worker.
auto	get_stats () const -> const work_stealing_stats &
	Get the current statistics.
auto	get_stats_snapshot () const -> work_stealing_stats_snapshot
	Get a snapshot of current statistics.
void	reset_stats ()
	Reset all statistics to zero.
auto	get_topology () const -> const numa_topology &
	Get the NUMA topology information.
auto	get_config () const -> const enhanced_work_stealing_config &
	Get the current configuration.
void	set_config (const enhanced_work_stealing_config &config)
	Update the configuration.

Private Member Functions
auto	select_victims (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victim workers based on the configured policy.
auto	select_victims_random (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using random policy.
auto	select_victims_round_robin (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using round-robin policy.
auto	select_victims_adaptive (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using adaptive (queue-size based) policy.
auto	select_victims_numa_aware (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using NUMA-aware policy.
auto	select_victims_locality_aware (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using locality-aware policy.
auto	select_victims_hierarchical (std::size_t requester_id, std::size_t count) -> std::vector< std::size_t >
	Select victims using hierarchical policy.
auto	calculate_batch_size (std::size_t victim_queue_size) const -> std::size_t
	Calculate batch size based on configuration and victim queue depth.
auto	get_worker_cpu (std::size_t worker_id) const -> int
	Get the CPU ID for a worker.
auto	workers_on_same_node (std::size_t worker_a, std::size_t worker_b) const -> bool
	Check if two workers are on the same NUMA node.
void	record_steal (std::size_t thief_id, std::size_t victim_id)
	Record a successful steal for affinity tracking.

Private Attributes
std::size_t	worker_count_
deque_accessor_fn	deque_accessor_
cpu_accessor_fn	cpu_accessor_
enhanced_work_stealing_config	config_
numa_topology	topology_
work_stealing_stats	stats_
std::unique_ptr< work_affinity_tracker >	affinity_tracker_
std::unique_ptr< backoff_calculator >	backoff_calculator_
std::mt19937_64	rng_
std::atomic< std::size_t >	round_robin_index_ {0}

Detailed Description

NUMA-aware work stealer with enhanced victim selection policies.

This class implements advanced work-stealing strategies with NUMA awareness, locality tracking, batch stealing, and comprehensive statistics collection. It coordinates stealing across multiple workers using configurable policies.

Design Goals

• Minimize cross-NUMA node memory access
• Maximize cache locality through affinity tracking
• Reduce contention through intelligent victim selection
• Provide detailed statistics for performance analysis

Thread Safety

All public methods are thread-safe and can be called concurrently from multiple worker threads. Statistics updates use atomic operations.

Memory Model

• Victim selection: sequential consistency for correctness
• Statistics: relaxed ordering for performance
• Topology access: read-only after construction

Usage Example

// Create workers
std::vector<std::unique_ptr<thread_worker>> workers;
// ... initialize workers ...
 
// Configure NUMA-aware stealing
auto config = enhanced_work_stealing_config::numa_optimized();
 
// Create accessor function
auto get_worker_deque = [&](std::size_t id) {
    return workers[id]->get_local_deque();
};
auto get_worker_cpu = [&](std::size_t id) {
    return workers[id]->get_policy().preferred_cpu;
};
 
// Create work stealer
numa_work_stealer stealer(workers.size(), get_worker_deque, get_worker_cpu, config);
 
// Steal work for worker 0
if (auto* stolen_job = stealer.steal_for(0)) {
    // Process stolen job
}
 
// Batch steal
auto batch = stealer.steal_batch_for(0, 4);
for (auto* j : batch) {
    // Process jobs
}
 
// Get statistics
auto stats = stealer.get_stats();
std::cout << "Success rate: " << stats.steal_success_rate() * 100 << "%\n";

Definition at line 98 of file numa_work_stealer.h.

Member Typedef Documentation

cpu_accessor_fn

using kcenon::thread::numa_work_stealer::cpu_accessor_fn = std::function<int(std::size_t)>

Function type for getting worker's CPU affinity.

Parameters

worker_id

The worker ID

Returns

Preferred CPU for the worker, or -1 if no preference

Definition at line 113 of file numa_work_stealer.h.

deque_accessor_fn

using kcenon::thread::numa_work_stealer::deque_accessor_fn = std::function<lockfree::work_stealing_deque<job*>*(std::size_t)>

Function type for accessing worker's local deque.

Parameters

worker_id

The worker ID

Returns

Pointer to the worker's local deque, or nullptr if not available

Definition at line 106 of file numa_work_stealer.h.

Constructor & Destructor Documentation

numa_work_stealer() [1/3]

kcenon::thread::numa_work_stealer::numa_work_stealer	(	std::size_t	worker_count,
		deque_accessor_fn	deque_accessor,
		cpu_accessor_fn	cpu_accessor,
		enhanced_work_stealing_config	config = {} )

Construct a NUMA-aware work stealer.

Parameters

worker_count	Number of workers in the pool
deque_accessor	Function to access worker deques
cpu_accessor	Function to get worker CPU affinity
config	Configuration for work stealing

Note

The accessor functions must remain valid for the lifetime of this object.

Definition at line 16 of file numa_work_stealer.cpp.

    : worker_count_(worker_count)
    , deque_accessor_(std::move(deque_accessor))
    , cpu_accessor_(std::move(cpu_accessor))
    , config_(config)
    , topology_(numa_topology::detect())
    , rng_(std::random_device{}())
{
    if (config_.track_locality && worker_count_ > 0)
    {
        affinity_tracker_ =
            std::make_unique<work_affinity_tracker>(worker_count_, config_.locality_history_size);
    }
 
    steal_backoff_config backoff_config;
    backoff_config.strategy = config_.backoff_strategy;
    backoff_config.initial_backoff = config_.initial_backoff;
    backoff_config.max_backoff = config_.max_backoff;
    backoff_config.multiplier = config_.backoff_multiplier;
    backoff_calculator_ = std::make_unique<backoff_calculator>(backoff_config);
}

References backoff_calculator_, kcenon::thread::enhanced_work_stealing_config::backoff_multiplier, kcenon::thread::enhanced_work_stealing_config::backoff_strategy, config_, kcenon::thread::enhanced_work_stealing_config::initial_backoff, kcenon::thread::enhanced_work_stealing_config::max_backoff, kcenon::thread::steal_backoff_config::strategy, kcenon::thread::enhanced_work_stealing_config::track_locality, and worker_count_.

~numa_work_stealer()

kcenon::thread::numa_work_stealer::~numa_work_stealer ( )

default

Destructor.

numa_work_stealer() [2/3]

kcenon::thread::numa_work_stealer::numa_work_stealer ( const numa_work_stealer & )

delete

numa_work_stealer() [3/3]

kcenon::thread::numa_work_stealer::numa_work_stealer ( numa_work_stealer && )

delete

Member Function Documentation

calculate_batch_size()

auto kcenon::thread::numa_work_stealer::calculate_batch_size ( std::size_t victim_queue_size ) const -> std::size_t

nodiscardprivate

Calculate batch size based on configuration and victim queue depth.

Definition at line 570 of file numa_work_stealer.cpp.

{
    if (!config_.adaptive_batch_size)
    {
        return config_.max_steal_batch;
    }
 
    // Adaptive batch size: steal at most half of victim's queue
    auto half_queue = victim_queue_size / 2;
    if (half_queue < config_.min_steal_batch)
    {
        return config_.min_steal_batch;
    }
 
    return std::min(half_queue, config_.max_steal_batch);
}

get_config()

auto kcenon::thread::numa_work_stealer::get_config ( ) const -> const enhanced_work_stealing_config&

nodiscard

Get the current configuration.

Returns

Reference to the work-stealing configuration

Definition at line 253 of file numa_work_stealer.cpp.

{
    return config_;
}

References config_.

get_stats()

auto kcenon::thread::numa_work_stealer::get_stats ( ) const -> const work_stealing_stats&

nodiscard

Get the current statistics.

Returns

Reference to the work-stealing statistics

Definition at line 233 of file numa_work_stealer.cpp.

{
    return stats_;
}

References stats_.

get_stats_snapshot()

auto kcenon::thread::numa_work_stealer::get_stats_snapshot ( ) const -> work_stealing_stats_snapshot

nodiscard

Get a snapshot of current statistics.

Returns

Non-atomic copy of statistics for safe reading

Definition at line 238 of file numa_work_stealer.cpp.

{
    return stats_.snapshot();
}

References kcenon::thread::work_stealing_stats::snapshot(), and stats_.

Here is the call graph for this function:

get_topology()

auto kcenon::thread::numa_work_stealer::get_topology ( ) const -> const numa_topology&

nodiscard

Get the NUMA topology information.

Returns

Reference to the detected NUMA topology

Definition at line 248 of file numa_work_stealer.cpp.

{
    return topology_;
}

References topology_.

get_worker_cpu()

auto kcenon::thread::numa_work_stealer::get_worker_cpu ( std::size_t worker_id ) const -> int

nodiscardprivate

Get the CPU ID for a worker.

Definition at line 587 of file numa_work_stealer.cpp.

{
    if (!cpu_accessor_)
    {
        return -1;
    }
    return cpu_accessor_(worker_id);
}

operator=() [1/2]

numa_work_stealer & kcenon::thread::numa_work_stealer::operator= ( const numa_work_stealer & )

delete

operator=() [2/2]

numa_work_stealer & kcenon::thread::numa_work_stealer::operator= ( numa_work_stealer && )

delete

record_steal()

void kcenon::thread::numa_work_stealer::record_steal ( std::size_t thief_id, std::size_t victim_id )

private

Record a successful steal for affinity tracking.

Definition at line 610 of file numa_work_stealer.cpp.

{
    if (config_.track_locality && affinity_tracker_)
    {
        affinity_tracker_->record_cooperation(thief_id, victim_id);
    }
}

References affinity_tracker_, config_, and kcenon::thread::enhanced_work_stealing_config::track_locality.

reset_stats()

void kcenon::thread::numa_work_stealer::reset_stats

(

)

Reset all statistics to zero.

Definition at line 243 of file numa_work_stealer.cpp.

{
    stats_.reset();
}

References kcenon::thread::work_stealing_stats::reset(), and stats_.

Here is the call graph for this function:

select_victims()

auto kcenon::thread::numa_work_stealer::select_victims ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victim workers based on the configured policy.

Parameters

requester_id	Worker requesting victims
count	Maximum number of victims to select

Returns

Vector of worker IDs to attempt stealing from

Definition at line 282 of file numa_work_stealer.cpp.

{
    switch (config_.policy)
    {
    case enhanced_steal_policy::random:
        return select_victims_random(requester_id, count);
 
    case enhanced_steal_policy::round_robin:
        return select_victims_round_robin(requester_id, count);
 
    case enhanced_steal_policy::adaptive:
        return select_victims_adaptive(requester_id, count);
 
    case enhanced_steal_policy::numa_aware:
        return select_victims_numa_aware(requester_id, count);
 
    case enhanced_steal_policy::locality_aware:
        return select_victims_locality_aware(requester_id, count);
 
    case enhanced_steal_policy::hierarchical:
        return select_victims_hierarchical(requester_id, count);
 
    default:
        return select_victims_random(requester_id, count);
    }
}

References kcenon::thread::adaptive, kcenon::thread::hierarchical, kcenon::thread::locality_aware, kcenon::thread::numa_aware, kcenon::thread::random, and kcenon::thread::round_robin.

select_victims_adaptive()

auto kcenon::thread::numa_work_stealer::select_victims_adaptive ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using adaptive (queue-size based) policy.

Definition at line 356 of file numa_work_stealer.cpp.

{
    // Score workers by queue size (larger queues are better targets)
    std::vector<std::pair<std::size_t, std::size_t>> scored_workers;
    scored_workers.reserve(worker_count_ - 1);
 
    for (std::size_t i = 0; i < worker_count_; ++i)
    {
        if (i == requester_id)
        {
            continue;
        }
 
        auto* deque = deque_accessor_(i);
        auto queue_size = deque ? deque->size() : 0;
        scored_workers.emplace_back(i, queue_size);
    }
 
    // Sort by queue size (descending)
    std::sort(scored_workers.begin(),
              scored_workers.end(),
              [](const auto& a, const auto& b) { return a.second > b.second; });
 
    // Take top 'count' workers with non-empty queues
    std::vector<std::size_t> victims;
    victims.reserve(count);
 
    for (const auto& [worker_id, queue_size] : scored_workers)
    {
        if (victims.size() >= count)
        {
            break;
        }
        if (queue_size > 0)
        {
            victims.push_back(worker_id);
        }
    }
 
    // If not enough with non-empty queues, add some randomly
    if (victims.size() < count)
    {
        for (const auto& [worker_id, queue_size] : scored_workers)
        {
            if (victims.size() >= count)
            {
                break;
            }
            if (std::find(victims.begin(), victims.end(), worker_id) == victims.end())
            {
                victims.push_back(worker_id);
            }
        }
    }
 
    return victims;
}

select_victims_hierarchical()

auto kcenon::thread::numa_work_stealer::select_victims_hierarchical ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using hierarchical policy.

Definition at line 505 of file numa_work_stealer.cpp.

{
    if (!config_.numa_aware || !topology_.is_numa_available())
    {
        return select_victims_adaptive(requester_id, count);
    }
 
    int requester_cpu = get_worker_cpu(requester_id);
    int requester_node = topology_.get_node_for_cpu(requester_cpu);
 
    // First: workers on same NUMA node
    std::vector<std::size_t> same_node_victims;
    // Second: workers on other NUMA nodes
    std::vector<std::size_t> other_node_victims;
 
    for (std::size_t i = 0; i < worker_count_; ++i)
    {
        if (i == requester_id)
        {
            continue;
        }
 
        int victim_cpu = get_worker_cpu(i);
        int victim_node = topology_.get_node_for_cpu(victim_cpu);
 
        if (victim_node == requester_node)
        {
            same_node_victims.push_back(i);
        }
        else
        {
            other_node_victims.push_back(i);
        }
    }
 
    // Shuffle within each group
    std::shuffle(same_node_victims.begin(), same_node_victims.end(), rng_);
    std::shuffle(other_node_victims.begin(), other_node_victims.end(), rng_);
 
    // Combine: same node first, then other nodes
    std::vector<std::size_t> victims;
    victims.reserve(count);
 
    for (auto victim_id : same_node_victims)
    {
        if (victims.size() >= count)
        {
            break;
        }
        victims.push_back(victim_id);
    }
 
    for (auto victim_id : other_node_victims)
    {
        if (victims.size() >= count)
        {
            break;
        }
        victims.push_back(victim_id);
    }
 
    return victims;
}

select_victims_locality_aware()

auto kcenon::thread::numa_work_stealer::select_victims_locality_aware ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using locality-aware policy.

Definition at line 474 of file numa_work_stealer.cpp.

{
    if (!config_.track_locality || !affinity_tracker_)
    {
        return select_victims_adaptive(requester_id, count);
    }
 
    // Get preferred victims from affinity tracker
    auto preferred = affinity_tracker_->get_preferred_victims(requester_id, count);
 
    // If not enough preferred victims, fill with adaptive selection
    if (preferred.size() < count)
    {
        auto additional = select_victims_adaptive(requester_id, count - preferred.size());
        for (auto victim_id : additional)
        {
            if (std::find(preferred.begin(), preferred.end(), victim_id) == preferred.end())
            {
                preferred.push_back(victim_id);
            }
            if (preferred.size() >= count)
            {
                break;
            }
        }
    }
 
    return preferred;
}

select_victims_numa_aware()

auto kcenon::thread::numa_work_stealer::select_victims_numa_aware ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using NUMA-aware policy.

Definition at line 415 of file numa_work_stealer.cpp.

{
    if (!config_.numa_aware || !topology_.is_numa_available())
    {
        return select_victims_adaptive(requester_id, count);
    }
 
    // Get requester's NUMA node
    int requester_cpu = get_worker_cpu(requester_id);
    int requester_node = topology_.get_node_for_cpu(requester_cpu);
 
    // Score workers: same node workers get higher priority
    std::vector<std::pair<std::size_t, double>> scored_workers;
    scored_workers.reserve(worker_count_ - 1);
 
    for (std::size_t i = 0; i < worker_count_; ++i)
    {
        if (i == requester_id)
        {
            continue;
        }
 
        int victim_cpu = get_worker_cpu(i);
        int victim_node = topology_.get_node_for_cpu(victim_cpu);
 
        auto* deque = deque_accessor_(i);
        auto queue_size = deque ? static_cast<double>(deque->size()) : 0.0;
 
        // Apply NUMA penalty for cross-node workers
        double score = queue_size;
        if (requester_node != victim_node && requester_node >= 0 && victim_node >= 0)
        {
            score /= config_.numa_penalty_factor;
        }
 
        scored_workers.emplace_back(i, score);
    }
 
    // Sort by score (descending)
    std::sort(scored_workers.begin(),
              scored_workers.end(),
              [](const auto& a, const auto& b) { return a.second > b.second; });
 
    std::vector<std::size_t> victims;
    victims.reserve(count);
 
    for (const auto& [worker_id, score] : scored_workers)
    {
        if (victims.size() >= count)
        {
            break;
        }
        victims.push_back(worker_id);
    }
 
    return victims;
}

select_victims_random()

auto kcenon::thread::numa_work_stealer::select_victims_random ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using random policy.

Definition at line 310 of file numa_work_stealer.cpp.

{
    std::vector<std::size_t> victims;
    victims.reserve(count);
 
    std::vector<std::size_t> candidates;
    candidates.reserve(worker_count_ - 1);
 
    for (std::size_t i = 0; i < worker_count_; ++i)
    {
        if (i != requester_id)
        {
            candidates.push_back(i);
        }
    }
 
    // Shuffle and take first 'count' elements
    std::shuffle(candidates.begin(), candidates.end(), rng_);
 
    auto num_victims = std::min(count, candidates.size());
    victims.insert(victims.end(), candidates.begin(), candidates.begin() + num_victims);
 
    return victims;
}

select_victims_round_robin()

auto kcenon::thread::numa_work_stealer::select_victims_round_robin ( std::size_t requester_id, std::size_t count ) -> std::vector<std::size_t>

nodiscardprivate

Select victims using round-robin policy.

Definition at line 336 of file numa_work_stealer.cpp.

{
    std::vector<std::size_t> victims;
    victims.reserve(count);
 
    auto start_index = round_robin_index_.fetch_add(1, std::memory_order_relaxed) % worker_count_;
 
    for (std::size_t i = 0; i < worker_count_ && victims.size() < count; ++i)
    {
        auto victim_id = (start_index + i) % worker_count_;
        if (victim_id != requester_id)
        {
            victims.push_back(victim_id);
        }
    }
 
    return victims;
}

set_config()

void kcenon::thread::numa_work_stealer::set_config

(

const enhanced_work_stealing_config &

config

)

Update the configuration.

Parameters

config

New configuration to use

Note

Changes take effect immediately. Be cautious when changing configuration while workers are actively stealing.

Definition at line 258 of file numa_work_stealer.cpp.

{
    config_ = config;
 
    // Update backoff calculator
    steal_backoff_config backoff_config;
    backoff_config.strategy = config_.backoff_strategy;
    backoff_config.initial_backoff = config_.initial_backoff;
    backoff_config.max_backoff = config_.max_backoff;
    backoff_config.multiplier = config_.backoff_multiplier;
    backoff_calculator_->set_config(backoff_config);
 
    // Update affinity tracker if needed
    if (config_.track_locality && !affinity_tracker_ && worker_count_ > 0)
    {
        affinity_tracker_ =
            std::make_unique<work_affinity_tracker>(worker_count_, config_.locality_history_size);
    }
    else if (!config_.track_locality)
    {
        affinity_tracker_.reset();
    }
}

References affinity_tracker_, backoff_calculator_, kcenon::thread::enhanced_work_stealing_config::backoff_multiplier, kcenon::thread::enhanced_work_stealing_config::backoff_strategy, config_, kcenon::thread::enhanced_work_stealing_config::initial_backoff, kcenon::thread::steal_backoff_config::initial_backoff, kcenon::thread::enhanced_work_stealing_config::locality_history_size, kcenon::thread::enhanced_work_stealing_config::max_backoff, kcenon::thread::steal_backoff_config::max_backoff, kcenon::thread::steal_backoff_config::multiplier, kcenon::thread::steal_backoff_config::strategy, kcenon::thread::enhanced_work_stealing_config::track_locality, and worker_count_.

steal_batch_for()

auto kcenon::thread::numa_work_stealer::steal_batch_for ( std::size_t worker_id, std::size_t max_count ) -> std::vector<job*>

nodiscard

Attempt to steal multiple jobs for a worker.

Parameters

worker_id	The worker requesting work
max_count	Maximum number of jobs to steal

Returns

Vector of stolen job pointers (may be empty or smaller than max_count)

Batch stealing is more efficient when multiple jobs need to be transferred. The actual batch size is determined by configuration and victim queue depth.

Thread Safety:

• Safe to call concurrently from multiple workers
• Statistics are updated atomically

Definition at line 129 of file numa_work_stealer.cpp.

{
    if (!config_.enabled || worker_count_ <= 1 || max_count == 0)
    {
        return {};
    }
 
    auto start_time = std::chrono::steady_clock::now();
    std::vector<job*> stolen_jobs;
 
    auto victims = select_victims(worker_id, config_.max_steal_attempts);
 
    std::size_t attempt = 0;
    for (auto victim_id : victims)
    {
        if (config_.collect_statistics)
        {
            stats_.steal_attempts.fetch_add(1, std::memory_order_relaxed);
        }
 
        auto* victim_deque = deque_accessor_(victim_id);
        if (victim_deque == nullptr)
        {
            continue;
        }
 
        auto queue_size = victim_deque->size();
        if (queue_size == 0)
        {
            if (config_.collect_statistics)
            {
                stats_.failed_steals.fetch_add(1, std::memory_order_relaxed);
            }
            continue;
        }
 
        auto batch_size = calculate_batch_size(queue_size);
        batch_size = std::min(batch_size, max_count);
 
        auto batch = victim_deque->steal_batch(batch_size);
        if (!batch.empty())
        {
            stolen_jobs = std::move(batch);
 
            if (config_.collect_statistics)
            {
                stats_.successful_steals.fetch_add(1, std::memory_order_relaxed);
                stats_.jobs_stolen.fetch_add(stolen_jobs.size(), std::memory_order_relaxed);
                stats_.batch_steals.fetch_add(1, std::memory_order_relaxed);
                stats_.total_batch_size.fetch_add(stolen_jobs.size(), std::memory_order_relaxed);
 
                if (workers_on_same_node(worker_id, victim_id))
                {
                    stats_.same_node_steals.fetch_add(1, std::memory_order_relaxed);
                }
                else
                {
                    stats_.cross_node_steals.fetch_add(1, std::memory_order_relaxed);
                }
            }
 
            record_steal(worker_id, victim_id);
            break;
        }
        else
        {
            if (config_.collect_statistics)
            {
                stats_.failed_steals.fetch_add(1, std::memory_order_relaxed);
            }
 
            // Apply backoff after failed steal
            if (++attempt < victims.size())
            {
                auto backoff_start = std::chrono::steady_clock::now();
                auto delay = backoff_calculator_->calculate(attempt);
                std::this_thread::sleep_for(delay);
 
                if (config_.collect_statistics)
                {
                    auto backoff_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                                          std::chrono::steady_clock::now() - backoff_start)
                                          .count();
                    stats_.total_backoff_time_ns.fetch_add(static_cast<std::uint64_t>(backoff_ns),
                                                           std::memory_order_relaxed);
                }
            }
        }
    }
 
    if (config_.collect_statistics)
    {
        auto elapsed_ns =
            std::chrono::duration_cast<std::chrono::nanoseconds>(
                std::chrono::steady_clock::now() - start_time)
                .count();
        stats_.total_steal_time_ns.fetch_add(static_cast<std::uint64_t>(elapsed_ns),
                                             std::memory_order_relaxed);
    }
 
    return stolen_jobs;
}

References kcenon::thread::delay.

steal_for()

auto kcenon::thread::numa_work_stealer::steal_for ( std::size_t worker_id ) -> job*

nodiscard

Attempt to steal work for a worker.

Parameters

worker_id

The worker requesting work

Returns

Stolen job pointer, or nullptr if no work available

This method selects victims based on the configured policy and attempts to steal a single job. NUMA awareness and affinity are considered when selecting victims.

Thread Safety:

• Safe to call concurrently from multiple workers
• Statistics are updated atomically

Definition at line 41 of file numa_work_stealer.cpp.

{
    if (!config_.enabled || worker_count_ <= 1)
    {
        return nullptr;
    }
 
    auto start_time = std::chrono::steady_clock::now();
    job* stolen_job = nullptr;
 
    auto victims = select_victims(worker_id, config_.max_steal_attempts);
 
    std::size_t attempt = 0;
    for (auto victim_id : victims)
    {
        if (config_.collect_statistics)
        {
            stats_.steal_attempts.fetch_add(1, std::memory_order_relaxed);
        }
 
        auto* victim_deque = deque_accessor_(victim_id);
        if (victim_deque == nullptr)
        {
            continue;
        }
 
        auto result = victim_deque->steal();
        if (result.has_value())
        {
            stolen_job = result.value();
 
            if (config_.collect_statistics)
            {
                stats_.successful_steals.fetch_add(1, std::memory_order_relaxed);
                stats_.jobs_stolen.fetch_add(1, std::memory_order_relaxed);
 
                if (workers_on_same_node(worker_id, victim_id))
                {
                    stats_.same_node_steals.fetch_add(1, std::memory_order_relaxed);
                }
                else
                {
                    stats_.cross_node_steals.fetch_add(1, std::memory_order_relaxed);
                }
            }
 
            record_steal(worker_id, victim_id);
            break;
        }
        else
        {
            if (config_.collect_statistics)
            {
                stats_.failed_steals.fetch_add(1, std::memory_order_relaxed);
            }
 
            // Apply backoff after failed steal
            if (++attempt < victims.size())
            {
                auto backoff_start = std::chrono::steady_clock::now();
                auto delay = backoff_calculator_->calculate(attempt);
                std::this_thread::sleep_for(delay);
 
                if (config_.collect_statistics)
                {
                    auto backoff_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                                          std::chrono::steady_clock::now() - backoff_start)
                                          .count();
                    stats_.total_backoff_time_ns.fetch_add(static_cast<std::uint64_t>(backoff_ns),
                                                           std::memory_order_relaxed);
                }
            }
        }
    }
 
    if (config_.collect_statistics)
    {
        auto elapsed_ns =
            std::chrono::duration_cast<std::chrono::nanoseconds>(
                std::chrono::steady_clock::now() - start_time)
                .count();
        stats_.total_steal_time_ns.fetch_add(static_cast<std::uint64_t>(elapsed_ns),
                                             std::memory_order_relaxed);
    }
 
    return stolen_job;
}

References kcenon::thread::delay, kcenon::thread::result< T >::has_value(), and kcenon::thread::result< T >::value().

Here is the call graph for this function:

workers_on_same_node()

auto kcenon::thread::numa_work_stealer::workers_on_same_node ( std::size_t worker_a, std::size_t worker_b ) const -> bool

nodiscardprivate

Check if two workers are on the same NUMA node.

Definition at line 596 of file numa_work_stealer.cpp.

{
    if (!topology_.is_numa_available())
    {
        return true; // On single-node systems, all workers are on the same node
    }
 
    int cpu_a = get_worker_cpu(worker_a);
    int cpu_b = get_worker_cpu(worker_b);
 
    return topology_.is_same_node(cpu_a, cpu_b);
}

Member Data Documentation

affinity_tracker_

std::unique_ptr<work_affinity_tracker> kcenon::thread::numa_work_stealer::affinity_tracker_

private

Definition at line 282 of file numa_work_stealer.h.

Referenced by record_steal(), and set_config().

backoff_calculator_

std::unique_ptr<backoff_calculator> kcenon::thread::numa_work_stealer::backoff_calculator_

private

Definition at line 283 of file numa_work_stealer.h.

Referenced by numa_work_stealer(), and set_config().

config_

enhanced_work_stealing_config kcenon::thread::numa_work_stealer::config_

private

Definition at line 279 of file numa_work_stealer.h.

Referenced by get_config(), numa_work_stealer(), record_steal(), and set_config().

cpu_accessor_

cpu_accessor_fn kcenon::thread::numa_work_stealer::cpu_accessor_

private

Definition at line 278 of file numa_work_stealer.h.

deque_accessor_

deque_accessor_fn kcenon::thread::numa_work_stealer::deque_accessor_

private

Definition at line 277 of file numa_work_stealer.h.

rng_

std::mt19937_64 kcenon::thread::numa_work_stealer::rng_

mutableprivate

Definition at line 286 of file numa_work_stealer.h.

round_robin_index_

std::atomic<std::size_t> kcenon::thread::numa_work_stealer::round_robin_index_ {0}

mutableprivate

Definition at line 289 of file numa_work_stealer.h.

{0};

stats_

work_stealing_stats kcenon::thread::numa_work_stealer::stats_

private

Definition at line 281 of file numa_work_stealer.h.

Referenced by get_stats(), get_stats_snapshot(), and reset_stats().

topology_

numa_topology kcenon::thread::numa_work_stealer::topology_

private

Definition at line 280 of file numa_work_stealer.h.

Referenced by get_topology().

worker_count_

std::size_t kcenon::thread::numa_work_stealer::worker_count_

private

Definition at line 276 of file numa_work_stealer.h.

Referenced by numa_work_stealer(), and set_config().

---

The documentation for this class was generated from the following files:

• include/kcenon/thread/stealing/numa_work_stealer.h
• src/stealing/numa_work_stealer.cpp