Avoid waits that hang

A join waits for a worker. If the worker is blocked on a channel endpoint or a lock that the waiting function still holds, or waits for a cancel that the function has not made yet, the join never returns. Plenty rejects such waits at compile time. These habits keep your programs clear of the whole class:

1. Move each endpoint into one worker as an argument: spawn(produce, sender).drop_job()?. A failed start then drops it.
2. Guard each endpoint you keep as the last manager: with spawn(produce, sender).drop_job()?, receiver as inbox:.
3. To wait inside the block, end the guard first. Open it before any statement that can leave the block, and wait after it ends.
4. Wait in a statement of its own. An endpoint moved into the same expression is still held while the wait runs.
5. Call drop_job() on a start you will not retry.
6. Start workers outside a lock block, and lock inside them.
7. Cancel a token on every exit when a worker waits for it: with spawn(serve, stop.share()).drop_job()?, stop.cancel_on_exit():.

The channel shutdown lesson showed habits 1 to 3, and the mutex lesson showed habit 6. Cooperative cancellation shows habit 7. Executor pools follow the same habits, as the future wait lesson shows. This lesson covers several producers and failed starts.

Several producers

Each producer needs its own sender, and the receiver must outlive none of them. One guard after several ? starts is not enough: when a later start fails, ? joins the earlier producers before the guard exists.

def produce(sender: Sender[i64]) -> Result[(), SendError[i64] | OverflowError[i64]]:
    for n in range(5):
        sender.send(n * n)?
    Ok(())

def drain(inbox: &Receiver[i64]) -> Result[i64, OverflowError[i64]]:
    mut total = 0
    while True:
        match inbox.recv():
            case Ok(value):
                total = total + value
            case Err(_):
                break
    Ok(total)

def main() -> Result[(), SpawnError[()] | AllocError | IoError | OverflowError[i64]]:
    sender, receiver = channel[i64](2)?
    second = sender.share()
    with spawn(produce, sender).drop_job()?, spawn(produce, second).drop_job()?, receiver as inbox:
        print(drain(inbox)?)?
    Ok(())

leaving this `with` block joins a task whose worker holds `Sender[i64]` while `receiver` still holds `Receiver[i64]`
`?` leaves the block here

Give the receiver to a consuming worker instead, and start it first. Then guard the sender and hand each producer a share of it:

def produce(sender: Sender[i64]) -> Result[(), SendError[i64] | OverflowError[i64]]:
    for n in range(5):
        sender.send(n * n)?
    Ok(())

def total(receiver: Receiver[i64]) -> Result[i64, OverflowError[i64]]:
    mut sum = 0
    while True:
        match receiver.recv():
            case Ok(value):
                sum = sum + value
            case Err(_):
                break
    Ok(sum)

def main() -> Result[(), SpawnError[()] | AllocError | IoError | OverflowError[i64]]:
    sender, receiver = channel[i64](2)?
    with spawn(total, receiver).drop_job()? as consumer:
        with sender as outbox, spawn(produce, outbox.share()).drop_job()?, spawn(produce, outbox.share()).drop_job()?:
            pass
        print(consumer.join()?)?
    Ok(())

60

Here the sender guard comes first: the producers send too, so only the consumer, joined later, needs the sender gone. The inner block joins both producers, then drops the last sender. The consumer sees disconnection and returns, so consumer.join() after the block cannot hang. If a producer fails to start, ? leaves through the same exits in the same order.

Failed starts keep their endpoints

A job that fails to start returns inside SpawnError[F], with every endpoint it captured. ? keeps that error until the function returns, which is after the block’s joins:

def report(receiver: Receiver[i64]) -> Result[(), IoError | OverflowError[i64]]:
    mut total = 0
    while True:
        match receiver.recv():
            case Ok(value):
                total = total + value
            case Err(_):
                break
    print(total)

def produce(sender: Sender[i64], count: i64) -> Result[(), SendError[i64] | OverflowError[i64]]:
    for n in range(count):
        sender.send(n * n)?
    Ok(())

type Errors = AllocError | SpawnError[()] | SpawnError[OnceClosure[[], Result[(), SendError[i64] | OverflowError[i64]]]]

def main() -> Result[(), Errors]:
    sender, receiver = channel[i64](2)?
    count = 5
    job = def once [sender, count]() -> Result[(), SendError[i64] | OverflowError[i64]]:
        produce(sender, count)
    with spawn(report, receiver).drop_job()?, spawn(job)?:
        pass
    Ok(())

leaving this `with` block joins a task whose worker holds `Receiver[i64]` while the error that `?` propagates still holds `Sender[i64]`
call `drop_job()` on the start

When you will not retry the job, call drop_job() on the start. It drops the job, and its sender, before ? leaves, and the error becomes SpawnError[()]:

def report(receiver: Receiver[i64]) -> Result[(), IoError | OverflowError[i64]]:
    mut total = 0
    while True:
        match receiver.recv():
            case Ok(value):
                total = total + value
            case Err(_):
                break
    print(total)

def produce(sender: Sender[i64], count: i64) -> Result[(), SendError[i64] | OverflowError[i64]]:
    for n in range(count):
        sender.send(n * n)?
    Ok(())

type Errors = AllocError | SpawnError[()]

def main() -> Result[(), Errors]:
    sender, receiver = channel[i64](2)?
    count = 5
    job = def once [sender, count]() -> Result[(), SendError[i64] | OverflowError[i64]]:
        produce(sender, count)
    with spawn(report, receiver).drop_job()?, spawn(job).drop_job()?:
        pass
    Ok(())

30

A named start, spawn(produce, sender, count), binds the same kind of job from its arguments, so it needs drop_job() here too.

The channel reference lists every way a function can hold an endpoint, and the programs the check rejects although they could not hang.