GitHub

Concurrency and interruption

PreviousNext

Fibers, structured concurrency, races and timeouts.

Every effect runs on a fiber, a lightweight thread on pyfect's scheduler. The same program runs with run_sync (no event loop) or on asyncio with run_async.

Running effects concurrently

fx.for_each and fx.all take a concurrency limit (a number or "unbounded"; sequential by default). The first failure interrupts the others.

for_each.py
import pyfect as fx


def fetch(n: int) -> fx.Effect[int]:
    return fx.sleep(0.05).as_(n * 10)


# Up to 3 fetches at a time; results keep the input order.
program = fx.for_each(range(6), fetch, concurrency=3)

seconds, result = fx.run_sync(program.timed())
assert result == [0, 10, 20, 30, 40, 50]
assert seconds < 0.25  # two rounds of three, not six sleeps in a row

# all runs a fixed set of effects and keeps each one's type: tuple[int, str].
pair = fx.all(fetch(1), fx.succeed("two"), concurrency="unbounded")
assert fx.run_sync(pair) == (10, "two")

Fibers

eff.fork() starts a child fiber. join waits for its result, interrupt stops it and waits for its finalizers. A fiber's children are interrupted when it finishes, so forked work never outlives the code that started it (fork_daemon opts out).

fork()fork()fork()interruptinterruptinterruptprogramchild fiberchild fibergrandchild fiber
fork()fork()fork()interruptinterruptinterruptprogramchild fiberchild fibergrandchild fiber

Interruption is not an exception you can catch: handlers like catch_all don't see it, and on_interrupt, ensuring and scope finalizers always run. Code that must not be interrupted halfway goes in fx.uninterruptible_mask.

fibers.py
import pyfect as fx

log: list[str] = []


@fx.gen
def worker():
    yield from fx.sleep(10)
    log.append("finished")  # never reached: the worker is interrupted


@fx.gen
def program():
    # fork starts a child fiber and returns at once.
    cleanup = fx.sync(lambda: log.append("interrupted"))
    fiber = yield from worker.on_interrupt(cleanup).fork()
    yield from fx.sleep(0.01)
    yield from fiber.interrupt()  # returns once the child has finished cleaning up
    task = yield from fx.succeed(21).map(lambda n: n * 2).fork()
    return (yield from task.join())


assert fx.run_sync(program) == 42
assert log == ["interrupted"]

Races and timeouts

race, race_all and timeout always wait for the losers to finish cleaning up before returning.

Coordinating fibers: Ref, Deferred, Queue, PubSub, Semaphore, Latch, FiberSet.

race.py
import pyfect as fx

released: list[str] = []


def replica(name: str, delay: float) -> fx.Effect[str]:
    return fx.sleep(delay).as_(name).on_interrupt(fx.sync(lambda: released.append(name)))


# The first to succeed wins; the loser is interrupted, and race waits for it.
fastest = replica("eu", 0.01).race(replica("us", 1))
winner = fx.run_sync(fastest)
print(f"winner: {winner}, interrupted: {released}")
assert winner == "eu" and released == ["us"]

# timeout fails with TimeoutException, also interrupting the slow effect.
slow = replica("ap", 1).timeout(0.01)
exit = fx.run_sync_exit(slow)
print(f"timeout: {type(exit).__name__}, interrupted: {released}")
assert isinstance(exit, fx.Failure)

Reference: Fiber, pyfect.queue, for_each.