How to change a field from several processes¶
Several processes can write the same field
of a box, and each write is safe: no reader
sees half a value, and the last writer wins. A write that you compute from
what the box holds is different, because reading and writing are two steps,
and another process can write in between. This guide shows how to keep such
a change whole with locked.
1. See the lost update¶
job.count += 1 reads count, adds 1 and writes the result back. Two
processes running it at the same time can both read 5 and both write 6:
| time | process A | process B | count |
|---|---|---|---|
| 1 | reads 5 | 5 | |
| 2 | reads 5 | 5 | |
| 3 | writes 6 | 6 | |
| 4 | writes 6 | 6, not 7 |
Each write was fine, but B computed its 6 from a 5 that A had already replaced, so A's increment is gone. Two workers counting 10000 times each often end below 20000:
import multiprocessing as mp
from collections.abc import Callable
from typing import Annotated
from sharedbox import Capacity, SharedBox
class Job(SharedBox, name="example-job"):
count: int = 0
state: Annotated[str, Capacity(16)] = "idle"
owner: Annotated[str, Capacity(16)] = ""
def count_unlocked(job: Job) -> None:
for _ in range(10_000):
job.count += 1
job.close()
update(count=job.count + 1) doesn't help,
because Python computes job.count + 1 before update runs, so the read is
still a separate step.
2. Read and write inside locked¶
Every write takes the box's lock, the
sequence lock, which makes other
writers wait and readers retry until the write is done.
locked holds that lock for a whole with
block, so no other process can write between your read and your write:
def count_locked(job: Job) -> None:
for _ in range(10_000):
with job.locked() as j:
j.count += 1
job.close()
Inside the block you use j, the object that as j names. It has the same
fields as job, and a type checker sees the same types. Now two workers end
at 20000 every time:
run_twice(job, count_unlocked)
print(job.count <= 20_000) # True, and often below 20000
job.count = 0
run_twice(job, count_locked)
print(job.count) # 20000
When the block ends, other processes see all of its changes at the same
moment, as they see an update. A watcher
or a signal of events reports each field the
block assigned once, with its value from after the block, however often the
block assigned it.
3. Check a value before you change it¶
The block can decide what to write from what it reads. Here two callers try to claim the same job, and only the first one gets it:
def claim(job: Job, me: str) -> bool:
with job.locked() as j:
if j.state != "idle":
return False
j.state = "busy"
j.owner = me
return True
4. Leave the box as it was when the block fails¶
If the block raises, every field it wrote goes back to its value from before the block, nobody is told about a change, and the exception carries on:
try:
with job.locked() as j:
j.count = 0
raise RuntimeError("changed my mind")
except RuntimeError:
pass
print(job.count) # 20000
Every process waits while the block runs
While the block holds the lock, reads and writes of the box wait in every
process, and one that waits longer than its lock timeout, 5 s unless the
class sets lock_timeout, raises
LockTimeoutError. A block that
reads, decides and writes takes a couple of microseconds. Do any slow
work, and every await, before or after the block, never inside it.
Use only the block's object inside the block
In the thread that holds the lock, reading or writing job itself
raises RuntimeError, and so does using j after the block or from
another thread. Methods that wait, watch or close, such as
watch or
close, raise TypeError on j.
Only job and j know that your thread holds the lock. A second box
you open on the same name in the same thread waits for the lock like
any other process, and raises LockTimeoutError. So does a box you
reach through a reference field
that points back at job. Don't start a child process with fork
inside the block either: the child would get a copy of a lock it
doesn't hold.
Locking two boxes needs one order. If one process locks box A and then box
B, while another locks B and then A, each ends up waiting for the box the
other holds, and neither moves until one of them raises LockTimeoutError.
Lock boxes in the same order in every process.
The whole script
"""The script of the guide "How to change a field from several processes"."""
import multiprocessing as mp
from collections.abc import Callable
from typing import Annotated
from sharedbox import Capacity, SharedBox
class Job(SharedBox, name="example-job"):
count: int = 0
state: Annotated[str, Capacity(16)] = "idle"
owner: Annotated[str, Capacity(16)] = ""
def count_unlocked(job: Job) -> None:
for _ in range(10_000):
job.count += 1
job.close()
def count_locked(job: Job) -> None:
for _ in range(10_000):
with job.locked() as j:
j.count += 1
job.close()
def claim(job: Job, me: str) -> bool:
with job.locked() as j:
if j.state != "idle":
return False
j.state = "busy"
j.owner = me
return True
def run_twice(job: Job, target: Callable[[Job], object]) -> None:
context = mp.get_context("spawn")
workers = [context.Process(target=target, args=(job,)) for _ in range(2)]
for worker in workers:
worker.start()
for worker in workers:
worker.join()
if __name__ == "__main__":
with Job() as job:
run_twice(job, count_unlocked)
print(job.count <= 20_000) # True, and often below 20000
job.count = 0
run_twice(job, count_locked)
print(job.count) # 20000
print(claim(job, "first"), claim(job, "second")) # True False
print(job.owner) # first
try:
with job.locked() as j:
j.count = 0
raise RuntimeError("changed my mind")
except RuntimeError:
pass
print(job.count) # 20000
Job.unlink()