bluesky_tiled_plugins.writing._buffer

This module vendors BufferingWrapper from bluesky.callbacks.buffer, with the addition of the flush method, so that beamline profiles can use the read-after-write synchronization and avoids a bluesky dependency.

Once a bluesky release includes BufferingWrapper.flush, this copy can be replaced by a re-export (or deprecated). The implementation is self-contained (standard library only); keep it in sync with the upstream module.

Module Contents

Classes

BufferingWrapper

A wrapper for callbacks that processes documents in a separate thread.

Data

logger

API

bluesky_tiled_plugins.writing._buffer.logger

‘getLogger(…)’

class bluesky_tiled_plugins.writing._buffer.BufferingWrapper(target: collections.abc.Callable, queue_size: int = 1000000)

A wrapper for callbacks that processes documents in a separate thread.

This class allows a callback to be executed in a background thread, processing documents as they are received. This prevent the blocking of RunEngine on any slow I/O operations by the callback. It handles graceful shutdown on exit or signal termination, ensuring that no new documents are accepted after shutdown has been initiated.

The wrapped callback should be thread-safe and not subscribed to the RE directly. If it maintains shared mutable state, it must protect it using internal locking.

Because documents are processed asynchronously, a run may not be fully written by the time RE(plan) returns. Code that reads a run back immediately after it completes (e.g. db[-1] or db[uid]) must call :meth:flush first to wait for the buffer to drain; otherwise it may observe a missing or incomplete run, or silently pick up the previous one.

Parameters

target : callable
    The instance of a callback that will be called with the documents.
    It should accept two parameters: `name` and `doc`.
queue_size : int, optional
    The maximum size of the internal queue. Default is 1,000,000.

Usage

tw = TiledWriter(client)
buff_tw = BufferingWrapper(tw)
RE.subscribe(buff_tw)

flush(timeout: float | None = None) → bool

Block until all documents enqueued so far have been processed.

This provides a synchronization barrier for read-after-write patterns: a plan or script that reads a run back (e.g. db[uid]) immediately after RE(plan) returns must call flush() first, because the wrapped callback processes documents asynchronously and may not have written the run yet by the time the plan finishes.

Documents enqueued by another thread while flushing extend the wait: the method returns once the queue has no unfinished items.

Parameters

timeout : float, optional Maximum time to wait, in seconds. By default, wait indefinitely.

Returns

bool True if the buffer was fully drained; False if the timeout expired (or the worker thread is no longer running) with items remaining.

shutdown(wait: bool = True)