Lock-Free Programming Basics

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// LOCK-FREE: At least one thread always succeeds
// The CAS loop is lock-free: if CAS fails, another thread succeeded
class LockFreeCounter {
    private value: Atomic<number> = new Atomic(0);
    
    increment(): void {
        while (true) {
            const current = this.value.load();
            if (this.value.compareAndSet(current, current + 1)) {
                return;  // Success!
            }
            // If we failed, someone else succeeded → progress was made
        }
    }
}
 
// WAIT-FREE: Every thread completes in bounded steps
// fetch_add is wait-free: completes in O(1) regardless of contention
class WaitFreeCounter {
    private value: Atomic<number> = new Atomic(0);
    
    increment(): number {
        // fetch_add always completes in one atomic operation
        // No retry loop, no possibility of spinning indefinitely
        return this.value.fetchAdd(1);
    }
}
 
// BLOCKING (for comparison): One slow thread blocks all
class BlockingCounter {
    private value: number = 0;
    private mutex: Mutex = new Mutex();
    
    increment(): void {
        this.mutex.lock();  // If holder is delayed, we wait here forever
        try {
            this.value++;
        } finally {
            this.mutex.unlock();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// HELPING: Other threads complete pending operations
 
class TwoPhaseSet<T> {
    // Pointer to current state (immutable once published)
    private state: AtomicReference<SetState<T>>;
    
    add(item: T): boolean {
        while (true) {
            // 1. Read current state
            const current = this.state.load();
            
            // 2. Check if another thread has a pending operation
            if (current.pendingOp !== null) {
                // HELP: Complete the pending operation first
                this.helpComplete(current.pendingOp);
                continue;  // Retry with fresh state
            }
            
            // 3. Check if item already exists
            if (current.items.has(item)) {
                return false;  // Already present
            }
            
            // 4. Create new state with our addition
            const newState = new SetState(
                new Set([...current.items, item]),
                null  // No pending operation
            );
            
            // 5. Try to CAS our new state in
            if (this.state.compareAndSet(current, newState)) {
                return true;
            }
            // CAS failed; someone else modified; retry
        }
    }
    
    private helpComplete(op: PendingOp): void {
        // Execute the pending operation on behalf of the original thread
        // This ensures the system makes progress even if the original
        // thread is delayed or crashed
    }
}
 
// Key insight: By helping other threads, we guarantee that
// the system never stalls waiting for any single thread.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

#include <atomic>
 
std::atomic<int> data{0};
std::atomic<bool> ready{false};
 
// WRONG: No ordering guarantees
void publish_wrong() {
    data.store(42, std::memory_order_relaxed);
    ready.store(true, std::memory_order_relaxed);
    // Compiler/CPU might reorder these!
}
 
// CORRECT: Release/Acquire ordering
void publish_correct() {
    data.store(42, std::memory_order_relaxed);  // Ordinary store is fine...
    ready.store(true, std::memory_order_release);  // ...because release prevents reordering
}
 
bool consume_correct() {
    if (ready.load(std::memory_order_acquire)) {  // Acquire matches release
        return data.load(std::memory_order_relaxed) == 42;  // Guaranteed true!
    }
    return false;
}
 
// SIMPLEST: Sequential consistency (default)
void publish_simple() {
    data.store(42);                // Default: seq_cst
    ready.store(true);             // Default: seq_cst
    // No reordering possible; easiest to reason about
}
 
bool consume_simple() {
    if (ready.load()) {            // Default: seq_cst
        return data.load() == 42;  // Guaranteed true!
    }
    return false;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// READ-COPY-UPDATE: Updates create new versions; readers see consistent snapshots
 
class RCUStyle<T> {
    private current: AtomicReference<T>;
    
    // Readers are lock-free and fast
    read(): T {
        return this.current.load();  // Always sees a consistent version
    }
    
    // Writers create new version and swap
    update(modifyFn: (old: T) => T): void {
        while (true) {
            const oldVersion = this.current.load();
            const newVersion = modifyFn(oldVersion);  // Copy + modify
            
            if (this.current.compareAndSet(oldVersion, newVersion)) {
                // Old version can be garbage collected when safe
                return;
            }
            // Contention: another writer; retry
        }
    }
}
 
// RCU is ideal for read-heavy workloads:
// - Readers never block
// - Writers pay the cost of copying
// - Old versions remain valid until readers finish

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// STATE MACHINE: Atomic state transitions
 
enum State {
    IDLE,
    STARTING,
    RUNNING,
    STOPPING,
    STOPPED
}
 
class LifecycleManager {
    private state: Atomic<State> = new Atomic(State.IDLE);
    
    start(): boolean {
        // Only IDLE → STARTING transition is valid
        if (this.state.compareAndSet(State.IDLE, State.STARTING)) {
            this.doStart();
            this.state.store(State.RUNNING);
            return true;
        }
        return false;  // Already started or stopping
    }
    
    stop(): boolean {
        // Only RUNNING → STOPPING transition is valid
        if (this.state.compareAndSet(State.RUNNING, State.STOPPING)) {
            this.doStop();
            this.state.store(State.STOPPED);
            return true;
        }
        return false;  // Not running
    }
    
    getState(): State {
        return this.state.load();
    }
    
    private doStart(): void { /* initialization */ }
    private doStop(): void { /* cleanup */ }
}
 
// CAS ensures exactly one thread performs each transition

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// SEQLOCK: Optimistic read for data that's rarely written
 
class SeqLock<T> {
    private sequence: Atomic<number> = new Atomic(0);
    private data: T;  // NOT atomic; protected by sequence
    
    // Writer: Acquire exclusive access via odd sequence
    write(newData: T): void {
        // 1. Increment to odd (signals write in progress)
        this.sequence.fetchAdd(1);
        
        // 2. Write the data (non-atomic is fine; writes are exclusive)
        this.data = newData;
        
        // 3. Increment to even (signals write complete)
        this.sequence.fetchAdd(1);
    }
    
    // Reader: Optimistic read with retry
    read(): T {
        while (true) {
            // 1. Read sequence before
            const seq1 = this.sequence.load();
            
            // 2. If odd, write in progress; spin
            if (seq1 % 2 !== 0) continue;
            
            // 3. Read the data
            const result = this.data;
            
            // 4. Read sequence after
            const seq2 = this.sequence.load();
            
            // 5. If sequence changed, data might be torn; retry
            if (seq1 === seq2) {
                return result;  // Consistent read!
            }
            // Sequence changed; retry
        }
    }
}
 
// SeqLock is ideal when:
// - Writes are rare
// - Reads are frequent
// - Data is small (fits in cache line)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

#include <atomic>
 
// BAD: These atomics share a cache line (likely 64 bytes)
struct BadCounters {
    std::atomic<long> counter1;  // Offset 0
    std::atomic<long> counter2;  // Offset 8
    // Both on same 64-byte cache line!
    // Updating counter1 invalidates counter2 in other caches
};
 
// GOOD: Pad each counter to its own cache line
struct alignas(64) PaddedCounter {
    std::atomic<long> value;
    char padding[64 - sizeof(std::atomic<long>)];
};
 
struct GoodCounters {
    PaddedCounter counter1;  // Offset 0, own cache line
    PaddedCounter counter2;  // Offset 64, own cache line
    // Updating one doesn't affect the other
};
 
// C++17 provides std::hardware_destructive_interference_size
#include <new>
struct ModernPaddedCounter {
    alignas(std::hardware_destructive_interference_size) 
    std::atomic<long> value;
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

// STRESS TEST: Hammer the data structure from many threads
 
async function stressTestStack(
    stack: LockFreeStack<number>,
    numThreads: number,
    opsPerThread: number
): Promise<void> {
    const promises: Promise<void>[] = [];
    const pushedCounts: number[] = new Array(numThreads).fill(0);
    const poppedSums: number[] = new Array(numThreads).fill(0);
    
    // Each thread pushes and pops randomly
    for (let t = 0; t < numThreads; t++) {
        promises.push((async () => {
            for (let i = 0; i < opsPerThread; i++) {
                if (Math.random() < 0.5) {
                    const value = t * opsPerThread + i;
                    stack.push(value);
                    pushedCounts[t]++;
                } else {
                    const value = stack.pop();
                    if (value !== null) {
                        poppedSums[t] += value;
                    }
                }
            }
        })());
    }
    
    await Promise.all(promises);
    
    // Drain remaining items
    let remaining = 0;
    let remainingSum = 0;
    while (true) {
        const value = stack.pop();
        if (value === null) break;
        remaining++;
        remainingSum += value;
    }
    
    // Verify: total pushed == total popped + remaining
    const totalPushed = pushedCounts.reduce((a, b) => a + b, 0);
    const totalPopped = (/* count from poppedSums */) + remaining;
    
    console.assert(
        stack.isEmpty(),
        "Stack should be empty after draining"
    );
    
    console.log(`Stress test: ${totalPushed} pushed, ${totalPopped} popped`);
}
 
// Run with: stressTestStack(new LockFreeStack(), 16, 100000);