Atomics and threads¶
ppy.atomic and ppy.concurrent give you shared memory one operation at a
time, and threads with what keeps them apart, over memory the program owns.
In this example four workers each add 500 to a shared counter and 500 × 2
to a shared total behind a mutex.
Run it¶
What it prints¶
python counters.ppy, ppy run counters.ppy
Atomics with C11's orders¶
def worker(counter: native.ptr[int], mutex: native.ptr[int], total: native.ptr[int], rounds: int) -> None:
for _ in range(rounds):
atomic.fetch_add(counter, 1, order="relaxed")
concurrent.lock(mutex)
native.store(total, native.load(total) + 2)
concurrent.unlock(mutex)
load, store, exchange, compare_exchange, the fetch_* family, and
fence each take an order: relaxed, acquire, release, acq_rel, or
seq_cst. The checker holds what C11 holds:
- a load is never
release - a store is never
acquire - a relaxed fence orders nothing (
E1641)
Natively each operation lowers to the instruction of that order. Under CPython the operations serialize under one lock, which implements every order.
Threads and what keeps them apart¶
concurrent.spawn(f, *args) runs a function of the module on a new thread
and hands back a handle; join waits.
The synchronization objects are memory the program owns:
- a mutex is one
intslot - a condition is one more, counting notifications
wait/notifyspin under the CPU's pause hint over the same atomics on every path
signal shows the pattern: a waiter blocks on a condition until the main
thread sets a flag and notifies.
Failures and targets¶
A thread that fails a guard fails its joiner, and the function falls back as a whole. Native code links pthreads; a target without them is refused with the reason.
Read on: Atomics and threads · Threads · Parallel range
counters.ppy is hand-written; there is no .py source and no conversion
step.
34_atomics_and_threads/counters.ppy¶
from ppy import atomic, concurrent, native
def worker(
counter: native.ptr[int], mutex: native.ptr[int], total: native.ptr[int], rounds: int
) -> None:
for _ in range(rounds):
atomic.fetch_add(counter, 1, order="relaxed")
concurrent.lock(mutex)
native.store(total, native.load(total) + 2)
concurrent.unlock(mutex)
def waiter(
flag: native.ptr[int], mutex: native.ptr[int], condition: native.ptr[int], out: native.ptr[int]
) -> None:
concurrent.lock(mutex)
while native.load(flag) == 0:
concurrent.wait(condition, mutex)
native.store(out, native.load(flag) * 10)
concurrent.unlock(mutex)
def run(
counter: native.ptr[int], mutex: native.ptr[int], total: native.ptr[int], rounds: int
) -> int:
first = concurrent.spawn(worker, counter, mutex, total, rounds)
second = concurrent.spawn(worker, counter, mutex, total, rounds)
third = concurrent.spawn(worker, counter, mutex, total, rounds)
fourth = concurrent.spawn(worker, counter, mutex, total, rounds)
concurrent.join(first)
concurrent.join(second)
concurrent.join(third)
concurrent.join(fourth)
atomic.fence()
return atomic.load(counter)
def signal(
flag: native.ptr[int], mutex: native.ptr[int], condition: native.ptr[int], out: native.ptr[int]
) -> int:
waiting = concurrent.spawn(waiter, flag, mutex, condition, out)
concurrent.lock(mutex)
native.store(flag, 4)
concurrent.notify(condition)
concurrent.unlock(mutex)
concurrent.join(waiting)
return native.load(out)
def main() -> None:
counter = native.stack_alloc[int](1)
mutex = native.stack_alloc[int](1)
total = native.stack_alloc[int](1)
print(run(counter, mutex, total, 500), native.load(total))
old, swapped = atomic.compare_exchange(counter, 2000, 7)
print(old, swapped, atomic.exchange(counter, 1), atomic.load(counter, order="acquire"))
flag = native.stack_alloc[int](1)
condition = native.stack_alloc[int](1)
out = native.stack_alloc[int](1)
print(signal(flag, mutex, condition, out))
main()
Source: examples/34_atomics_and_threads.