Skip to content

Keeping every update

Level: 101 · anyone who has just watched a shared total lose additions

One line: total += n keeps every number in one of three ways — a lock held across all three steps, one atomic add in place of the three steps, or a single owner of the total that every number is sent to. All six languages have the lock; all but Python have the atomic add; all but C++ have a standard way to hand the numbers to one owner.

The previous lesson lost additions because a handler's load and its store could have another handler's store between them. Each fix below closes that gap differently. Every program has eight handlers add the numbers 1 to 10,000 each — 80,000 additions of different sizes, with a correct total of 400,040,000 — once per fix, and prints the totals.

A lock around load, add and store One atomic add One owner, sent every number
Rust Mutex<u64> AtomicU64::fetch_add an mpsc channel, summed by one thread
Go sync.Mutex atomic.Int64.Add a channel, summed by one goroutine
C pthread_mutex_t atomic_fetch_add on an atomic_long a pipe, read by one thread
C++ std::mutex, held by a std::scoped_lock std::atomic<long>::fetch_add no channel in the standard library
Java a synchronized block AtomicLong.addAndGet a single-thread ExecutorService
Python threading.Lock no atomic integer in the standard library a queue.Queue, read by one thread

Rust

Verified output of every_update_rs.rs — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with a Mutex:         400040000
with fetch_add:       400040000
with one owner:       400040000

A Rust Mutex owns the value it guards. The only way to reach the number is lock(), which returns a guard; += goes through the guard, and the lock is released when the guard is dropped at the end of that statement. So the program cannot add to the total without holding the lock — unlike the Go, C, C++, Java and Python versions, where the lock and the variable are tied together only by the programmer's habit.

fetch_add ↗ replaces load, add and store with one operation. In the channel version, the owner's total is a local variable that no other thread can name, and for n in received ends when the last Sender is gone — each handler's clone is dropped when its thread finishes, and the original is dropped by hand, which is why the program says drop(numbers). The Rust library's RwLock and atomics ↗ compares the three for different access patterns, and its Channels ↗ goes further into the third.

every_update_rs.rs

every_update_rs.rs in full — pasted here by tools/run_examples.py from the file CI runs.

//! Keeping every update, three ways. Eight handlers each add the numbers 1 to
//! 10,000 to one total: under a Mutex, with an atomic add, and by sending every
//! number to the one thread that owns the total.
//!
//!   rustc --edition 2024 every_update_rs.rs -o every_update_rs && ./every_update_rs

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, mpsc};
use std::thread;

const HANDLERS: u64 = 8;
const LAST: u64 = 10_000;

fn with_a_mutex() -> u64 {
    let total = Mutex::new(0);
    thread::scope(|s| {
        for _ in 0..HANDLERS {
            s.spawn(|| {
                for n in 1..=LAST {
                    // load, add and store, all while this thread holds the lock
                    *total.lock().unwrap() += n;
                }
            });
        }
    });
    total.into_inner().unwrap()
}

fn with_an_atomic_add() -> u64 {
    let total = AtomicU64::new(0);
    thread::scope(|s| {
        for _ in 0..HANDLERS {
            s.spawn(|| {
                for n in 1..=LAST {
                    total.fetch_add(n, Ordering::SeqCst); // one indivisible operation
                }
            });
        }
    });
    total.into_inner()
}

fn with_one_owner() -> u64 {
    let (numbers, received) = mpsc::channel();
    let owner = thread::spawn(move || {
        let mut total = 0; // no other thread can reach this variable
        for n in received {
            total += n;
        }
        total
    });
    thread::scope(|s| {
        for _ in 0..HANDLERS {
            let numbers = numbers.clone();
            s.spawn(move || {
                for n in 1..=LAST {
                    numbers.send(n).unwrap();
                }
            });
        }
    });
    drop(numbers); // the last sender is gone, so the owner's loop ends
    owner.join().unwrap()
}

fn main() {
    println!(
        "{HANDLERS} handlers each add 1 to {LAST}; the total should be {}",
        HANDLERS * LAST * (LAST + 1) / 2
    );
    println!("with a Mutex:         {}", with_a_mutex());
    println!("with fetch_add:       {}", with_an_atomic_add());
    println!("with one owner:       {}", with_one_owner());
}

Go

Verified output of every_update_go.go — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with a sync.Mutex:    400040000
with atomic Add:      400040000
with one owner:       400040000

Nothing in Go ties mu to total: the lock works because every line that touches total takes it first. The Go library's A mutex guards a counter ↗ shows the unguarded version losing a different number on every run, and its Atomic counters ↗ takes atomic.Int64 further, to CompareAndSwap.

The third version is Go's proverb in Effective Go ↗ — do not communicate by sharing memory; share memory by communicating. The total is a local variable of one goroutine, and the handlers send it numbers. close(numbers) ends the owner's range loop once every handler has finished sending.

every_update_go.go

every_update_go.go in full — pasted here by tools/run_examples.py from the file CI runs.

// Keeping every update, three ways. Eight handlers each add the numbers 1 to
// 10,000 to one total: under a sync.Mutex, with an atomic add, and by sending
// every number to the one goroutine that owns the total.
//
//  go build every_update_go.go && ./every_update_go
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

const (
    handlers = 8
    last     = 10_000
)

func withAMutex() int64 {
    var mu sync.Mutex
    var total int64
    var wg sync.WaitGroup
    for range handlers {
        wg.Go(func() {
            for n := int64(1); n <= last; n++ {
                mu.Lock()
                total += n // load, add and store, all while holding the lock
                mu.Unlock()
            }
        })
    }
    wg.Wait()
    return total
}

func withAnAtomicAdd() int64 {
    var total atomic.Int64
    var wg sync.WaitGroup
    for range handlers {
        wg.Go(func() {
            for n := int64(1); n <= last; n++ {
                total.Add(n) // one indivisible operation
            }
        })
    }
    wg.Wait()
    return total.Load()
}

func withOneOwner() int64 {
    numbers := make(chan int64)
    result := make(chan int64)
    go func() {
        var total int64 // no other goroutine can reach this variable
        for n := range numbers {
            total += n
        }
        result <- total
    }()

    var wg sync.WaitGroup
    for range handlers {
        wg.Go(func() {
            for n := int64(1); n <= last; n++ {
                numbers <- n
            }
        })
    }
    wg.Wait()
    close(numbers) // the owner's range loop ends
    return <-result
}

func main() {
    fmt.Printf("%d handlers each add 1 to %d; the total should be %d\n",
        handlers, last, handlers*last*(last+1)/2)
    fmt.Println("with a sync.Mutex:   ", withAMutex())
    fmt.Println("with atomic Add:     ", withAnAtomicAdd())
    fmt.Println("with one owner:      ", withOneOwner())
}

C

Verified output of every_update_c.c — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with a pthread mutex:  400040000
with atomic_fetch_add: 400040000
with one owner:        400040000

The mutex version is the one most C code uses. atomic_fetch_add comes from C11's <stdatomic.h> and is one indivisible operation on an atomic_long.

C has no channel, but POSIX has the pipe, and a pipe can serve as one between threads. Each handler writes one long at a time into the pipe; POSIX promises ↗ that a write of no more than PIPE_BUF bytes is not interleaved with other writes to the same pipe, so the owner reads whole longs and nothing else. When main closes the write end after every handler has finished, the owner's next read returns 0 and it hands the total back through pthread_join.

every_update_c.c

every_update_c.c in full — pasted here by tools/run_examples.py from the file CI runs.

/* Keeping every update, three ways. Eight handlers each add the numbers 1 to
 * 10,000 to one total: under a pthread mutex, with a C11 atomic add, and by
 * writing every number into a pipe that one thread reads and adds up.
 *
 *   cc -std=c17 -Wall -Wextra -pedantic -pthread every_update_c.c -o every_update_c && ./every_update_c
 */
#define _POSIX_C_SOURCE 200809L

#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

enum { HANDLERS = 8, LAST = 10000 };

static void run_handlers(void *(*handler)(void *), void *arg) {
    pthread_t threads[HANDLERS];
    for (int i = 0; i < HANDLERS; i++) {
        pthread_create(&threads[i], NULL, handler, arg);
    }
    for (int i = 0; i < HANDLERS; i++) {
        pthread_join(threads[i], NULL);
    }
}

/* --- with a mutex --- */

static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static long locked_total;

static void *add_under_the_lock(void *arg) {
    (void)arg;
    for (long n = 1; n <= LAST; n++) {
        pthread_mutex_lock(&lock);
        locked_total += n; /* load, add and store, all while holding the lock */
        pthread_mutex_unlock(&lock);
    }
    return NULL;
}

/* --- with an atomic add --- */

static atomic_long atomic_total;

static void *add_atomically(void *arg) {
    (void)arg;
    for (long n = 1; n <= LAST; n++) {
        atomic_fetch_add(&atomic_total, n); /* one indivisible operation */
    }
    return NULL;
}

/* --- with one owner, reading a pipe --- */

static void *send_numbers(void *arg) {
    int write_end = *(int *)arg;
    for (long n = 1; n <= LAST; n++) {
        /* A write of at most PIPE_BUF bytes is never interleaved with another. */
        if (write(write_end, &n, sizeof n) != sizeof n) {
            abort();
        }
    }
    return NULL;
}

static void *own_the_total(void *arg) {
    int read_end = *(int *)arg;
    long *total = malloc(sizeof *total); /* no other thread can reach this */
    *total = 0;
    long n;
    ssize_t got;
    /* Each write put a whole long into the pipe, so a read of one long gets a
       whole long, and 0 means every write end is closed. */
    while ((got = read(read_end, &n, sizeof n)) == sizeof n) {
        *total += n;
    }
    if (got != 0) {
        abort();
    }
    return total;
}

static long with_one_owner(void) {
    int ends[2];
    if (pipe(ends) != 0) {
        abort();
    }
    pthread_t owner;
    pthread_create(&owner, NULL, own_the_total, &ends[0]);
    run_handlers(send_numbers, &ends[1]);
    close(ends[1]); /* the owner's next read returns 0 */

    void *result;
    pthread_join(owner, &result);
    long total = *(long *)result;
    free(result);
    close(ends[0]);
    return total;
}

int main(void) {
    printf("%d handlers each add 1 to %d; the total should be %ld\n", HANDLERS, LAST,
           (long)HANDLERS * LAST * (LAST + 1) / 2);

    run_handlers(add_under_the_lock, NULL);
    printf("with a pthread mutex:  %ld\n", locked_total);

    run_handlers(add_atomically, NULL);
    printf("with atomic_fetch_add: %ld\n", atomic_load(&atomic_total));

    printf("with one owner:        %ld\n", with_one_owner());
    return 0;
}

C++

Verified output of every_update_cpp.cpp — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with a std::mutex:    400040000
with fetch_add:       400040000

std::scoped_lock takes the mutex when it is constructed and releases it when it goes out of scope, at the end of each pass through the loop, so the unlock cannot be forgotten on any path out of the block. fetch_add is the atomic add.

There is no third version, because the C++ standard library's concurrency support ↗ has futures and promises for handing over one value but no queue that threads can share. The usual answer is a std::queue behind a std::mutex and a std::condition_variable, which puts a lock back in the middle. (Not machine-checked here.)

every_update_cpp.cpp

every_update_cpp.cpp in full — pasted here by tools/run_examples.py from the file CI runs.

// Keeping every update, two ways. Eight handlers each add the numbers 1 to
// 10,000 to one total: under a std::mutex, and with an atomic add.
//
//   c++ -std=c++20 -O2 -Wall -Wextra -Wpedantic -pthread every_update_cpp.cpp -o every_update_cpp && ./every_update_cpp

#include <atomic>
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

constexpr int handlers = 8;
constexpr long last = 10'000;

static void run_handlers(const std::function<void()>& handler) {
    std::vector<std::jthread> threads;
    for (int i = 0; i < handlers; ++i) {
        threads.emplace_back(handler);
    }
} // each std::jthread joins as it is destroyed

static long with_a_mutex() {
    std::mutex lock;
    long total = 0;
    run_handlers([&] {
        for (long n = 1; n <= last; ++n) {
            std::scoped_lock held{lock};
            total += n; // load, add and store, all while holding the lock
        }
    });
    return total;
}

static long with_an_atomic_add() {
    std::atomic<long> total{0};
    run_handlers([&] {
        for (long n = 1; n <= last; ++n) {
            total.fetch_add(n); // one indivisible operation
        }
    });
    return total.load();
}

int main() {
    std::cout << handlers << " handlers each add 1 to " << last << "; the total should be "
              << handlers * last * (last + 1) / 2 << "\n";
    std::cout << "with a std::mutex:    " << with_a_mutex() << "\n";
    std::cout << "with fetch_add:       " << with_an_atomic_add() << "\n";
}

Java

Verified output of every_update_java.java — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with synchronized:    400040000
with AtomicLong:      400040000
with one owner:       400040000

synchronized (lock) holds the object's monitor for the block. AtomicLong.addAndGet is the atomic add.

The third version needs no queue of its own: a single-thread executor is one. Every execute puts a task on the executor's queue, and its one thread runs the tasks in order, so ownedTotal += number never runs on two threads at once. close() ↗ waits until the queued tasks have all run, which is what makes it safe for main to read ownedTotal afterwards.

every_update_java.java

every_update_java.java in full — pasted here by tools/run_examples.py from the file CI runs.

// Keeping every update, three ways. Eight handlers each add the numbers 1 to
// 10,000 to one total: inside a synchronized block, with an AtomicLong, and by
// handing every addition to a single-thread executor, the one thread that owns
// the total.
//
//   java every_update_java.java      (Java 25: a compact source file, which imports java.base)

final int HANDLERS = 8;
final long LAST = 10_000;

final Object lock = new Object();
long lockedTotal = 0;
long ownedTotal = 0; // touched only by the owner thread, and read after it has finished

void runHandlers(Runnable handler) throws InterruptedException {
    List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < HANDLERS; i++) {
        threads.add(Thread.ofPlatform().start(handler));
    }
    for (Thread t : threads) {
        t.join();
    }
}

long withSynchronized() throws InterruptedException {
    runHandlers(() -> {
        for (long n = 1; n <= LAST; n++) {
            synchronized (lock) {
                lockedTotal += n; // load, add and store, all while holding the lock
            }
        }
    });
    return lockedTotal;
}

long withAnAtomicAdd() throws InterruptedException {
    AtomicLong total = new AtomicLong();
    runHandlers(() -> {
        for (long n = 1; n <= LAST; n++) {
            total.addAndGet(n); // one indivisible operation
        }
    });
    return total.get();
}

long withOneOwner() throws InterruptedException {
    try (ExecutorService owner = Executors.newSingleThreadExecutor()) {
        runHandlers(() -> {
            for (long n = 1; n <= LAST; n++) {
                long number = n;
                owner.execute(() -> ownedTotal += number); // queued, run one at a time
            }
        });
    } // close() waits until the owner has run every queued addition
    return ownedTotal;
}

void main() throws InterruptedException {
    IO.println(HANDLERS + " handlers each add 1 to " + LAST + "; the total should be "
            + HANDLERS * LAST * (LAST + 1) / 2);
    IO.println("with synchronized:    " + withSynchronized());
    IO.println("with AtomicLong:      " + withAnAtomicAdd());
    IO.println("with one owner:       " + withOneOwner());
}

Python

Verified output of every_update_py.py — regenerated by tools/run_examples.py, never hand-typed.

8 handlers each add 1 to 10000; the total should be 400040000
with a threading.Lock: 400040000
with one owner:        400040000

with lock: holds the lock for the += and releases it even if the body raises. threading ↗ offers locks, conditions, semaphores, events and barriers, and no atomic integer, so there is no second version.

In the third, the handlers put numbers on a queue.Queue ↗, which does its own locking, and one thread takes them off and adds them up. None on the queue is the signal that no more numbers are coming.

every_update_py.py

every_update_py.py in full — pasted here by tools/run_examples.py from the file CI runs.

"""Keeping every update, two ways. Eight handlers each add the numbers 1 to
10,000 to one total: under a threading.Lock, and by putting every number on a
queue that one thread reads and adds up.

    python3 every_update_py.py
"""

import queue
import threading

HANDLERS = 8
LAST = 10_000


def run_handlers(handler) -> None:
    threads = [threading.Thread(target=handler) for _ in range(HANDLERS)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()


def with_a_lock() -> int:
    lock = threading.Lock()
    total = 0

    def handler() -> None:
        nonlocal total
        for n in range(1, LAST + 1):
            with lock:
                total += n  # load, add and store, all while holding the lock

    run_handlers(handler)
    return total


def with_one_owner() -> int:
    numbers: queue.Queue[int | None] = queue.Queue()
    result = []

    def owner() -> None:
        total = 0  # no other thread can reach this variable
        while (n := numbers.get()) is not None:
            total += n
        result.append(total)

    def handler() -> None:
        for n in range(1, LAST + 1):
            numbers.put(n)

    owning_thread = threading.Thread(target=owner)
    owning_thread.start()
    run_handlers(handler)
    numbers.put(None)  # every handler has finished: tell the owner to stop
    owning_thread.join()
    return result[0]


print(f"{HANDLERS} handlers each add 1 to {LAST}; the total should be {HANDLERS * LAST * (LAST + 1) // 2}")
print(f"with a threading.Lock: {with_a_lock()}")
print(f"with one owner:        {with_one_owner()}")

Which one

  • One number that only ever changes by adding: the atomic add. It has no lock to forget, and nothing to wait for.
  • Several values that must change together — a total and a count, a balance and its history: a lock around the whole update. An atomic per value leaves a moment where one has changed and the other has not, which is the lost update's gap again.
  • State that has a natural owner — a server's table of sessions, one account: an owner, and messages to it. This is what a server that handles one request at a time already is, and why its clients may be as asynchronous as they like without losing a number.
  • Do not choose by speed folklore. Which is faster depends on how many threads contend and how long the critical section is. Measure it on your own workload.

See also