We can't find the internet
Attempting to reconnect
Something went wrong!
Attempting to reconnect
← Parallel Python with Process Pools step 12 of 24
Collect Every Result Before You Join the Workers
A worker process sends its results back to the parent on a
multiprocessing.Queue. The parent calls worker.join() to wait for the
worker to exit, then reads the queue. With a few small results this works.
With many results, both processes hang forever.
Why join can hang
put() on a multiprocessing.Queue adds the object to a buffer and returns.
A background thread in the worker, the feeder thread, then pickles the
object and writes it into a pipe to the parent. A pipe holds a limited number
of bytes. When it is full, the feeder waits for the parent to read.
A process that has put items on a queue waits for its feeder to write them all before it exits. So this order deadlocks. Each one waits for the next:
Parent: waits in join() for the worker to exit, and reads nothing
Worker: waits for its feeder to write everything before it exits
Feeder: waits for the parent to read, to make room in the pipe
Small tests pass because their results fit in the pipe. The fix is to read every result first and join afterwards.
The parent also needs to know when it has read everything. qsize() and
empty() can't tell it, because a worker can put an item a moment after you
check. Instead, each worker sends a completion message after its last
result, and the parent reads until it has one from every worker. To stop the
workers, the parent puts one sentinel per worker on their input queue. A
sentinel is a value that means "no more work".
What to build
The grader can't start worker processes, so you build the same protocol with
threads and queue.SimpleQueue. Threads can't hit the pipe deadlock, but the
steps and their order are the same.
Fill in worker and solve in the starter:
def solve(*, items: list[int], workers: int) -> tuple[tuple[int, ...], int]:
- On each call, create a new inbox for jobs and a new outbox for results.
- Start exactly
workersthreads that runworkeron the shared inbox, with indices0toworkers - 1. - Put every integer from
itemson the inbox, duplicates included. Then put oneNoneper worker.Noneis the sentinel. - In
worker, read the inbox until you getNone. For each integer, put its square on the outbox. OnNone, put exactly oneDone(worker=index)on the outbox and return. - In
solve, read the outbox until you have receivedworkersDonemessages, and keep every integer you read. - Join the threads. Return the integers sorted, as a tuple, and the number
of
Donemessages you received.
Never use qsize() or empty() to decide when to stop reading.
For items=[-3, 0, 3] and workers=2, return ((0, 9, 9), 2). For an empty
list and four workers, return ((), 4). workers is always positive and can
be larger than the number of items. A worker that gets no integers still
sends its Done.
Why counting Done messages is enough
Each worker puts all its results before its Done. A queue hands items out
in the order they went in, so once you have every worker's Done, you have
every result. Results from different workers arrive mixed together, so you
sort them.
Going further
- If a worker can crash, waiting for its
Doneforever is another hang. Real code adds a timeout or checks the worker'sexitcode. multiprocessing.SimpleQueuehas no feeder thread, so itsput()itself waits when the pipe is full. See the guide to pipes and queues.
Your code, saved here
Sign in to access your accepted solutions and saved draft history. You can keep reading the lesson without an account.
Sign in to your accountStuck?
Python reference solution
Sign in to track your attempts toward unlocking the reference solution.