Concurrency

Introduction

Sometimes you need several independent, slow things done and none of them depends on the others. almasix.concurrency runs them at the same time and gives you the results in the shape you asked for.

from almasix.concurrency import Concurrency

user_count, order_count = Concurrency.run([
    lambda: User.query().count(),
    lambda: Order.query().count(),
])

The ConcurrencyServiceProvider (registered with the foundation) binds a manager built from config/concurrency.py.

File map

Piece	Path
Façade	src/almasix/concurrency/facade.py — Concurrency
Manager	src/almasix/concurrency/manager.py — ConcurrencyManager
Drivers	src/almasix/concurrency/drivers/ — thread, fork, process, sync
Task shapes	src/almasix/concurrency/tasks.py — TaskSet
Deferral	src/almasix/concurrency/deferred.py — DeferredTasks
Provider	src/almasix/concurrency/provider.py
Exceptions	src/almasix/concurrency/exceptions.py

How it works

Concurrency.run() takes one callable, a list of them, or a mapping, and returns results in the same shape:

Concurrency.run(lambda: heavy())              # ["result"]

Concurrency.run([first, second])              # ["a", "b"] — in order

Concurrency.run({                             # {"users": 10, "orders": 4}
    "users": lambda: User.query().count(),
    "orders": lambda: Order.query().count(),
})

Results come back in the order the tasks were given, not the order they finished.

Failures

If a task raises, run() re-raises it — but only once every other task has settled, so nothing is left running in the background:

try:
    Concurrency.run([safe, risky])
except ValueError:
    ...

Drivers

Configure the default in config/concurrency.py, or pick one per call:

Concurrency.run(tasks, "fork")
Concurrency.driver("sync").run(tasks)
Concurrency.set_default_driver("fork")

Driver	Parallelism	Accepts	Notes
thread (default)	I/O-bound only	any callable	A thread pool; max_workers defaults to 16
fork	true	any callable	Unix only. Children inherit memory, so closures work; results must be picklable
process	true	picklable callables	Spawns a fresh interpreter, so tasks must be importable — a module-level function, not a lambda
sync	none	any callable	Runs in order on the current thread, for debugging

Almasix defaults to thread where Laravel defaults to process, because Python has real threads and PHP does not. Threads accept any closure and cover the work this page is actually for — queries, HTTP calls, file reads — all of which release the GIL. Reach for fork when the work is CPU-bound.

fork is unsafe to mix with threads in the parent process, which is a constraint of fork(2) rather than of Almasix.

Custom drivers

Concurrency.extend("my-driver", lambda app, config, name: MyDriver(config))

A driver subclasses Driver and implements execute(task_set), returning results in the task order.

Deferring tasks

When you want the work done but do not need the results:

Concurrency.defer([
    lambda: metrics.record(request),
    lambda: audit.log(request),
])

Laravel runs deferred tasks after the HTTP response is sent. Almasix has no post-response hook yet, so they run on a background thread — and defer() returns a DeferredTasks handle so tests can wait() for them:

deferred = Concurrency.defer([task])
deferred.finished()
deferred.wait(timeout=5)

Concurrency under ASGI

Almasix runs on ASGI, where the natural way to overlap work is the event loop. arun() awaits coroutines (or plain callables) concurrently and has no Laravel counterpart:

async def index():
    users, posts = await Concurrency.arun([
        lambda: fetch_users(),
        lambda: fetch_posts(),
    ])

Use arun() inside async controllers and run() everywhere else. Calling the blocking run() from inside a running event loop would block it.

Configuration

config/concurrency.py

config = {
    "default": env("CONCURRENCY_DRIVER", "thread"),
    "drivers": {
        "thread": {"driver": "thread", "max_workers": 16},
        "fork": {"driver": "fork"},
        "process": {"driver": "process"},
        "sync": {"driver": "sync"},
    },
}

A named entry can point at any driver, so "fast": {"driver": "sync"} gives you an alias.

Deliberate deviations from Laravel

• thread is the default driver, and is an addition; see above.
• arun() is an addition for the ASGI path.
• defer() runs on a background thread and returns a waitable handle, because Almasix has no post-response deferral hook yet. The same deviation applies to Batch.defer() in the HTTP client.
• The process driver rejects unpicklable tasks with a clear error instead of serializing closures. Laravel leans on SerializableClosure, which has no dependency-free Python equivalent.