Partitioning a Workload So Every Thread Reads Local Memory

The preceding articles each describe one mechanism: how a fault picks a node, how a mask restricts a thread, how a policy attaches to a range, what a migration costs. A working placement design is the combination of those mechanisms into a single layout where the mechanisms agree with each other. When they agree, the steady state has no remote access at all and no migration activity. When they disagree, the machine spends its time correcting placements that keep being made wrong.

The invariant a design is trying to hold

The goal is one sentence: for every thread, the memory it touches repeatedly is on the node the thread runs on, and the threads that touch the same memory run on the same node.

Both halves are necessary. A thread on the right node reading memory that lives elsewhere is the first-touch problem. Memory on the right node read by threads scattered across all nodes is the sharing problem, and it is not fixable by placement because there is no single correct node for a page that every node reads.

The design work is therefore to arrange that the second condition holds first, by partitioning the data so that each page has exactly one set of users, and then to place the threads of each set together. Data partitioning is the load-bearing decision; placement policy only implements it.

Shard by node, then bind

The straightforward structure is one shard per node, each with its own thread or thread group, each allocated on its node and pinned there.

The allocation is the part that needs care. Allocating the shard from the thread that will use it works, because first-touch places the pages correctly, but only if the thread is already pinned when it faults them, and only if the initialization loop actually writes every page. A calloc that the allocator satisfies from already-zeroed pages does not fault every page at that moment, and the faults then happen later, possibly from a different thread on a different node.

The reliable form makes the placement explicit rather than incidental, so that the code’s intent is visible and does not depend on when a page happens to be touched.

#include <numa.h>
#include <string.h>

struct shard {
    int      node;
    size_t   bytes;
    char    *data;
};

/* Allocate a shard on a specific node and touch every page there. */
static int shard_init(struct shard *s, int node, size_t bytes)
{
    s->node  = node;
    s->bytes = bytes;
    s->data  = numa_alloc_onnode(bytes, node);
    if (!s->data)
        return -1;
    /* Pre-fault on this thread, which the caller pins to the shard node. */
    memset(s->data, 0, bytes);
    return 0;
}

numa_alloc_onnode maps the range and then restricts it to one node with a bind policy, returning null if that restriction cannot be applied. The restriction is what a partition wants: a shard whose pages may silently land on another node is worse than a failure at startup, because the failure is visible and the mixed placement is not.

The thread that performs the initialization should be pinned, using the mask mechanism described in CPU Affinity Masks and What Pinning Actually Costs, to the shard’s node before the memset, so that the pre-faulting and the explicit allocation agree. If a startup thread initializes all shards and workers are spawned afterward, the pages are still placed correctly because the allocation named the node, but any page the allocator could not supply from that node falls back to the initializing thread’s node, which is why the failure return is worth checking.

The cost of the boundary

Partitioning is never complete. Some data is shared: a routing table, a global counter, a set of parameters, a queue of work items that moves between shards.

Every shared object is a decision. If it is read-only after initialization, the cheapest placement is to replicate it per node, one copy per shard each allocated on its node, so that every access is local and no cache line is shared. The cost is memory proportional to the number of nodes, which for a small configuration table is nothing and for a large index is unacceptable.

If it is writable and small, such as a counter, the cache line becomes the unit of contention rather than the page. A line written by threads on two nodes is invalidated across the fabric on every write, and the fabric traffic is proportional to the write rate, not the data size. The fix is not placement but padding: one counter per node, each on its own 64-byte line, summed only when the value is read. The padding wastes 56 bytes per counter and removes the cross-node invalidation entirely.

If it is writable and large, such as a work queue, it has to be sharded like the data, with items assigned to shards by a rule that keeps the producer and consumer of each item close. The rule matters more than the queue implementation: a queue that hands every item to the next available shard produces a stream of cross-node transfers, while a queue that assigns items by a key that correlates with the data’s shard keeps the handoff local.

When the node count and the thread count disagree

Real machines rarely have exactly one core per node, and the interesting cases are the two mismatches.

More threads than nodes is the common case. Two threads on the same node can share a shard only if they read it; if both write, they contend for the same cache lines within the node, which is cheaper than contending across nodes but not free. The usual resolution is to shard finer than the node: by thread, with each thread’s shard allocated on the node that thread is pinned to. Two shards on the same node share its DRAM channels, which is the resource that matters at that point.

More nodes than work is the awkward case. A machine with eight nodes running a workload with four shards has four idle nodes, and interleaving or spreading the four shards over eight nodes gains nothing because the threads are on four of them. The reverse case, eight shards on four nodes, puts two shards per node and halves the bandwidth available to each.

The number of shards should follow the number of nodes that will actually run threads, not the number of nodes the machine reports. Nodes with no CPUs, which firmware can describe and which appear when memory is attached to a device, cannot run a worker and should not receive a shard unless the workload is purely bandwidth-bound and wants their memory channels.

Verifying that the design holds

Three checks establish whether the layout is doing what it claims.

The first is residency per shard: the per-node page counts in /proc/<pid>/numa_maps for each shard’s area should be concentrated on the shard’s node, with a small tail elsewhere only if the node was full. A shard whose pages are split evenly between two nodes means the allocation fell back, and the node was over-committed.

The second is CPU placement: each worker’s last-run CPU should be on its shard’s node. A worker that migrated once and stayed migrated shows up as a shard whose pages are local to a node with no worker on it.

The third is traffic: the cross-node allocation counters should be near zero in steady state, since a correct layout allocates nothing remotely after startup. A steady-state cross-node allocation count that keeps rising means something in the workload is allocating on the wrong node, which for a partitioned design is always a bug in the partitioning rather than a policy failure.

These three checks correspond to the three quantities described in Measuring Remote Access with Hardware Counters, and the reason a shard can end up split despite an explicit allocation is the fallback path described in Where a Memory Policy Attaches and How It Inherits.

What the design gives up

A partitioned layout is not free, and the price is paid in flexibility. The partition is fixed at startup in most designs, because changing it means reallocating and re-touching everything, which is a migration of the whole dataset. A workload whose access pattern shifts, such as a shard that becomes hot because its key range became popular, cannot rebalance without that cost. Consistent hashing over shards reduces the amount of data that must move when a shard is added, but the pages still have to be faulted on the new node.

Capacity is also fixed. A shard sized to fit one node cannot grow past that node’s memory without spilling, and the spill goes to another node by fallback, which reintroduces exactly the remote access the design eliminated. Sizing shards to leave headroom is the mitigation, and headroom is capacity that the workload cannot use.

The third cost is complexity at every boundary. Each shared object needs its own decision, each decision needs its own justification, and the resulting code has more places where a placement assumption is embedded than a design that simply allocates and lets the kernel balance. That is the trade: a partitioned layout converts a runtime heuristic into a static invariant, and a static invariant is only worth having when the workload’s structure is stable enough to make it true.