Async Set Many Pattern

Asynchronous `set_many()` pattern

This page documents a pattern for the set_many() implementation in Ralitsa's HLA script. It is a reference implementation for that script; it is not an API or a change to acsys-python.

Overview and problem statement

A batch setter usually has three independently completing operations:

start one operation for each setting;
receive one completion or failure for each operation; and
return the results, or report all failures after the batch has finished.

A tempting translation of synchronous code is to protect a shared pending set with threading.Lock and wait on a threading.Event:

with lock:
    while pending:
        event.wait()

That is unsafe in an asynchronous function. The event-loop thread blocks inside the synchronous wait and cannot run the completion callback that would remove an item from pending and release the lock. This is a deadlock, not merely slow scheduling. threading.Lock.acquire() has the same problem when called synchronously from an async def.

The implementation below uses an asyncio.Queue as the completion channel. A callback from a worker thread never touches the queue directly. It uses loop.call_soon_threadsafe() to schedule queue.put_nowait() on the event-loop thread, where the queue is owned. This also avoids an event/lock lost-wakeup race: every completion is an individual queue item.

Fit with `acsys-python`

The checked-out acsys-python source does not contain a set_many() function. Its setting API is DPM.apply_settings(input_array), which submits one ApplySettings_request for the whole input list and returns None; it does not expose a per-setting callback. The proposed callback adapter is therefore for the HLA's own submission layer, not a direct wrapper around DPM.apply_settings(). Its relevant async patterns are:

concurrent coroutines with asyncio.gather() in acsys/__init__.py;
concurrent DPM operations with tasks and asyncio.as_completed() in acsys/dpm/__init__.py;
an asyncio.Queue for streamed replies in Connection.request_stream(); and
an asyncio.Semaphore for DPM state ownership.

The library's protocol callbacks are loop-owned: they resolve futures or enqueue replies from the asyncio protocol callback. No worker-thread callback contract was found. Consequently, code that directly calls acsys.Connection methods should normally use await, asyncio.gather(), or asyncio.as_completed() rather than introduce this adapter. DPM.apply_settings() also holds its DPM state semaphore for the entire request and reports request-level completion, so calling it concurrently once per individual setting is not equivalent to this callback pattern. Use call_soon_threadsafe() only when the HLA submission layer really can invoke its completion callback from another thread.

This guide deliberately does not copy two unrelated pre-existing implementation details in the library: the synchronous socket connection setup and the run_until_complete() call in the request-stream cleanup path. Neither belongs in an async set_many() implementation.

Contract for the submission adapter

set_many() expects an async submit_one(setting, on_complete) callable. It must:

arrange for exactly one callback, normally as (result, None) or (None, exception);
invoke the callback either on the event-loop thread or from a worker thread;
propagate cancellation instead of converting asyncio.CancelledError into a setting failure; and
stop or unregister its worker callback when its coroutine is cancelled, if the underlying API supports that operation.

If submit_one() returns without invoking the callback, set_many() cannot know whether the setting succeeded, so the batch waits until its timeout. The implementation treats a second callback as a duplicate and ignores it. This protects the result from APIs that report a completion and then also raise a submission error, but it does not repair an incorrectly implemented adapter.

Reference implementation

The state is local to one invocation. Do not share a mutable event, lock, queue, pending set, or result dictionary between concurrent calls to set_many().

import asyncio
from collections.abc import Iterable
from typing import Any, Awaitable, Callable, Dict, Optional, Tuple


CompletionCallback = Callable[[Any, Optional[BaseException]], None]
SubmitOne = Callable[[Any, CompletionCallback], Awaitable[None]]
Completion = Tuple[int, Any, Optional[BaseException]]


class SetManyError(Exception):
    """One or more settings failed.

    ``errors`` maps the zero-based input index to the original exception.
    ``results`` contains successes from the same partially completed batch.
    """

    def __init__(
        self,
        errors: Dict[int, BaseException],
        results: Optional[Dict[int, Any]] = None,
    ) -> None:
        self.errors = errors
        self.results = results or {}
        super().__init__(f"{len(errors)} setting(s) failed")


async def set_many(
    settings: Iterable[Any],
    *,
    submit_one: SubmitOne,
    timeout: Optional[float] = None,
) -> Dict[int, Any]:
    """Submit all settings concurrently and wait for every completion.

    The returned dictionary is keyed by the zero-based input index. A timeout
    is the maximum total duration of the batch, not a timeout between
    individual completions. ``asyncio.TimeoutError`` is raised if the batch
    does not complete in time. ``SetManyError`` contains all reported setting
    failures when completion is otherwise reached.
    """
    if timeout is not None and timeout < 0:
        raise ValueError("timeout must be non-negative or None")

    loop = asyncio.get_running_loop()
    values = list(settings)
    if not values:
        return {}

    # Only the event-loop thread calls put_nowait(). Worker threads use
    # call_soon_threadsafe() below to marshal the operation to this loop.
    completions: asyncio.Queue[Completion] = asyncio.Queue()
    accepting_callbacks = True

    def notify(
        index: int,
        result: Any = None,
        error: Optional[BaseException] = None,
    ) -> None:
        """Deliver a completion safely from either kind of callback thread."""
        completion = (index, result, error)

        def enqueue() -> None:
            # This function always runs on the event-loop thread. A callback
            # that was already in flight when the batch timed out is harmless.
            if accepting_callbacks:
                completions.put_nowait(completion)

        try:
            # This is safe even when the caller is already on this loop and is
            # required when the callback originates in a worker thread.
            loop.call_soon_threadsafe(enqueue)
        except RuntimeError:
            # The worker may report completion while application shutdown is
            # closing the loop. There is no waiter left to notify then.
            pass

    def notify_for(index: int) -> CompletionCallback:
        def callback(
            result: Any = None,
            error: Optional[BaseException] = None,
        ) -> None:
            notify(index, result, error)

        return callback

    async def submit(index: int, setting: Any) -> None:
        try:
            await submit_one(setting, notify_for(index))
        except asyncio.CancelledError:
            # Cancellation is cancellation of the batch, not a setting error.
            raise
        except Exception as error:
            # A failure before the callback is delivered through the same
            # channel, so it cannot leave the collector waiting forever.
            notify(index, error=error)

    submission_tasks = [
        asyncio.create_task(submit(index, setting))
        for index, setting in enumerate(values)
    ]

    results: Dict[int, Any] = {}
    errors: Dict[int, BaseException] = {}
    completed = set()

    async def collect() -> None:
        while len(completed) < len(values):
            index, result, error = await completions.get()

            # A buggy adapter may call its callback more than once. Only the
            # first callback determines the setting's terminal state.
            if index in completed:
                continue

            completed.add(index)
            if error is None:
                results[index] = result
            else:
                errors[index] = error

    try:
        if timeout is None:
            await collect()
        else:
            await asyncio.wait_for(collect(), timeout)

        if errors:
            raise SetManyError(dict(errors), dict(results))

        # Preserve input order in the returned mapping, even though
        # completions may arrive in any order.
        return {index: results[index] for index in range(len(values))}
    finally:
        # Stop submission coroutines on success, timeout, or caller
        # cancellation. gather() observes every task and avoids the
        # "Task was destroyed but it is pending" warning.
        for task in submission_tasks:
            if not task.done():
                task.cancel()
        await asyncio.gather(*submission_tasks, return_exceptions=True)

        # This assignment runs on the event-loop thread. Any later callback
        # is still allowed to call notify(), but its marshalled enqueue is a
        # no-op. The adapter should nevertheless cancel/unregister its own
        # worker when possible; this flag is only a final safety boundary.
        accepting_callbacks = False

Important lifetime detail

call_soon_threadsafe() only schedules the enqueue operation; it does not cancel the producer that owns the worker thread. On timeout or cancellation, set_many() cancels and awaits its submission tasks, but a third-party worker may still finish later. The accepting_callbacks guard prevents a late completion from being consumed as part of a subsequent batch. The HLA adapter should also expose cancellation or callback removal where possible.

No completion task is created for each callback. That is intentional: queue insertion is a single loop callback, and set_many() owns the one collector coroutine. This makes cleanup bounded and observable. If an implementation instead uses asyncio.Event and asyncio.Lock, it must track and await every task created to record a completion, and it must marshal all access to those asyncio primitives onto the owning loop.

Caller setup

The caller supplies an adapter for the HLA's actual operation. The following is a shape example; hla.submit() is a placeholder and is not provided by acsys-python:

async def run_batch(hla, settings):
    async def submit_one(setting, on_complete):
        try:
            # Replace with the HLA operation. If this call blocks the event
            # loop, run the blocking portion in asyncio.to_thread() or an
            # executor rather than calling it directly here.
            result = await hla.submit(setting)
        except Exception as error:
            on_complete(None, error)
        else:
            on_complete(result, None)

    return await set_many(
        settings,
        submit_one=submit_one,
        timeout=30.0,
    )

If the HLA API invokes a callback from a worker thread, the adapter can pass that callback through to set_many():

async def submit_one(setting, on_complete):
    # This is illustrative only. The HLA API must define how the worker is
    # started, stopped, and joined when this coroutine is cancelled.
    await hla.submit_from_worker(setting, on_complete)

Do not construct asyncio.Event, asyncio.Lock, asyncio.Queue, or futures in one event loop and use them directly from another. Do not call threading.Event.wait() or threading.Lock.acquire() from the event-loop thread. call_soon_threadsafe() is the correct direction of travel: worker thread -> event-loop callback -> asyncio-owned state. It does not make a blocking HLA call non-blocking by itself.

Tests

These tests exercise the reference implementation's important paths. They require pytest and pytest-asyncio; the library repository already uses asyncio_mode = "auto" in pyproject.toml.

import asyncio

import pytest


async def test_set_many_accepts_loop_thread_callbacks():
    async def submit_one(setting, on_complete):
        on_complete(setting * 2, None)

    assert await set_many([3, 1, 2], submit_one=submit_one) == {
        0: 6,
        1: 2,
        2: 4,
    }


async def test_set_many_marshals_worker_thread_callbacks():
    async def submit_one(setting, on_complete):
        await asyncio.to_thread(on_complete, setting + 1, None)

    assert await set_many([10], submit_one=submit_one) == {0: 11}


async def test_set_many_collects_submission_failures():
    async def submit_one(setting, on_complete):
        if setting == "bad":
            raise ValueError("rejected")
        on_complete(setting.upper(), None)

    with pytest.raises(SetManyError) as raised:
        await set_many(["ok", "bad"], submit_one=submit_one)

    assert isinstance(raised.value.errors[1], ValueError)
    assert str(raised.value.errors[1]) == "rejected"
    assert raised.value.results == {0: "OK"}


async def test_set_many_handles_empty_input():
    async def never_called(setting, on_complete):
        raise AssertionError("empty batches must not submit")

    assert await set_many([], submit_one=never_called) == {}


async def test_set_many_times_out_and_cleans_up_submission_tasks():
    cancelled = asyncio.Event()

    async def submit_one(setting, on_complete):
        try:
            await asyncio.Event().wait()
        except asyncio.CancelledError:
            cancelled.set()
            raise

    with pytest.raises(asyncio.TimeoutError):
        await set_many([1], submit_one=submit_one, timeout=0.01)

    assert cancelled.is_set()


async def test_set_many_does_not_block_unrelated_async_work():
    ticks = 0
    finished = False

    async def submit_one(setting, on_complete):
        await asyncio.sleep(0.02)
        on_complete(setting, None)

    async def ticker():
        nonlocal ticks
        while not finished:
            ticks += 1
            await asyncio.sleep(0)

    ticker_task = asyncio.create_task(ticker())
    try:
        assert await set_many([42], submit_one=submit_one) == {0: 42}
    finally:
        finished = True
        await ticker_task

    assert ticks > 0

For a real HLA adapter, add an integration test that invokes the actual worker-thread completion path and verifies that cancellation unregisters the callback or stops the worker. A unit test that only invokes callbacks on the loop thread cannot prove that thread marshalling is correct.

Checklist

settings is materialized once, and the empty batch returns immediately.
Every submission either invokes its callback once or raises an ordinary exception.
asyncio.CancelledError is not converted into a setting failure.
Worker callbacks call loop.call_soon_threadsafe() and do not touch asyncio objects directly.
The timeout semantics (total batch timeout here) are documented at the call site.
Underlying worker operations are cancelled or callbacks unregistered on timeout/cancellation.
Submission tasks are cancelled and awaited before returning or raising.
Tests prove loop responsiveness, worker-thread completion, failure propagation, timeout, and cleanup.

Navigation Menu

Uh oh!

Async Set Many Pattern

Asynchronous `set_many()` pattern

Overview and problem statement

Fit with `acsys-python`

Contract for the submission adapter

Reference implementation

Important lifetime detail

Caller setup

Tests

Checklist

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

Uh oh!

Async Set Many Pattern

Asynchronous set_many() pattern

Overview and problem statement

Fit with acsys-python

Contract for the submission adapter

Reference implementation

Important lifetime detail

Caller setup

Tests

Checklist

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

Asynchronous `set_many()` pattern

Fit with `acsys-python`