Concurrency

Concurrency utilities for bridging sync and async code.

This module provides three primary helpers:

1. asyncify: Wrap a synchronous function so it can be awaited in async code by offloading it to a worker thread using AnyIO.
2. syncify: Wrap an asynchronous function so it can be called from synchronous code. By default, this uses a shared AnyIO BlockingPortal running in a background thread.
3. parallel_gather: Run independent coroutine objects concurrently, returning ordered results and cleaning up unfinished sibling tasks on failure.

Usage

from gllm_core.concurrency import asyncify, syncify, parallel_gather

# Asyncify a sync function
async_op = asyncify(blocking_fn)
result = await async_op(arg1, arg2)

# Syncify an async function
sync_op = syncify(async_fn)
result = sync_op(arg1, arg2)

# Parallel gather coroutines with ordered results and cancellation cleanup
results = await parallel_gather(*(worker(item) for item in items))

Notes: 1. For asyncify: Cancelling an await of an asyncified sync function cancels the awaiter, but the underlying thread cannot be forcibly interrupted. The function continues to run until it returns. 2. For syncify: A shared default BlockingPortal is lazily created on first use and shut down at process exit. 3. For parallel_gather: The helper owns the tasks it creates. If any coroutine raises, or if the awaiting task is cancelled, unfinished sibling tasks are cancelled and drained before the original exception is re-raised.

asyncify(func, *, cancellable=False, limiter=None)

Wrap a sync function into an awaitable callable using a worker thread.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	Synchronous function to wrap.	required
cancellable	bool	If True, allow cancellation of the awaiter while running in a worker thread. Defaults to False.	False
limiter	CapacityLimiter | None	Capacity limiter to throttle concurrent thread usage. Defaults to None.	None

Returns:

Type	Description
Callable[P, Awaitable[R]]	Callable[P, Awaitable[R]]: An async function that when awaited will execute func in a
Callable[P, Awaitable[R]]	worker thread and return its result.

Usage

async def handler() -> int:
    wrapped = asyncify(blocking_func)
    return await wrapped(1, 2)

get_default_portal()

Return the shared default BlockingPortal.

Returns:

Name	Type	Description
BlockingPortal	BlockingPortal	A process-wide portal running on a background thread.

parallel_gather(*coroutines, max_concurrency=None) async

Run independent coroutine objects concurrently and return ordered results.

The helper owns the tasks it creates. If any coroutine raises, or if the awaiting task is cancelled, unfinished sibling tasks are cancelled and drained before the original exception is re-raised.

Parameters:

Name	Type	Description	Default
*coroutines	Coroutine[Any, Any, T]	Independent coroutine objects to run concurrently.	()
max_concurrency	int | None	Optional cap on how many coroutines may execute at once.	None

Returns:

Type	Description
list[T]	list[T]: A list of results in the same order as the input coroutines, or an empty list when called with no arguments.

Raises:

Type	Description
TypeError	If any argument is not a coroutine object.
ValueError	If max_concurrency is less than or equal to 0 when provided.
Exception	The first exception raised by any coroutine.
CancelledError	If the awaiting task is cancelled.

Examples:

Basic usage — results are returned in input order, regardless of completion order:

async def fetch(id: int) -> str:
    await asyncio.sleep(random.random())
    return f"result-{id}"

results = await parallel_gather(fetch(1), fetch(2), fetch(3))
# -> ["result-1", "result-2", "result-3"]

Generator expression — building tasks dynamically:

async def process(item: int) -> int:
    return item * 2

items = [1, 2, 3]
results = await parallel_gather(*(process(i) for i in items))
# -> [2, 4, 6]

syncify(async_func, *, portal=None)

Wrap an async function to be callable from synchronous code.

Lifecycle and portals: 1. This helper uses an already running AnyIO BlockingPortal to execute the coroutine. 2. If portal is not provided, a process-wide shared portal is used. Its lifecycle is managed internally: it is created lazily on first use and shut down automatically at process exit. 3. If you provide a portal, you are expected to manage its lifecycle, typically with a context manager. This is recommended when making many calls in a bounded scope since it avoids per-call startup costs while allowing deterministic teardown.

Parameters:

Name	Type	Description	Default
async_func	Callable[P, Awaitable[R]]	Asynchronous function to wrap.	required
portal	BlockingPortal | None	Portal to use for calling the async function from sync code. Defaults to None, in which case a shared default portal is used.	None

Returns:

Type	Description
Callable[P, R]	Callable[P, R]: A synchronous function that runs the coroutine and returns its result.

Usage

# Use the default shared portal (most convenient)
def do_work(x: int) -> int:
    sync_call = syncify(async_func)
    return sync_call(x)

# Reuse a scoped portal for multiple calls (deterministic lifecycle)
from anyio.from_thread import start_blocking_portal

with start_blocking_portal() as portal:
    sync_call = syncify(async_func, portal=portal)
    a = sync_call(1)
    b = sync_call(2)

Notes

Creating a brand-new portal per call is discouraged due to the overhead of spinning up and tearing down a background event loop/thread. Prefer the shared portal or a scoped portal reused for a batch of calls.