diff --git a/CHANGELOG.md b/CHANGELOG.md index f440f47..b83127d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,10 @@ Pre-releases (`b*`, `rc*`) are not listed. - Annotations throughout `cuvis` restated in the forms Python 3.10 provides: `Union[A, B]` and `Optional[A]` became `A | B` and `A | None`, and `Tuple`, `FrozenSet`, `Sequence`, `Callable` and `Awaitable` now come from `builtins` and `collections.abc` rather than `typing`. Every signature denotes what it denoted before; `cuvis.cube_utils.ImageData.__getitem__` keeps `Union`, because its member list contains a forward reference and `|` cannot join a type to a string at runtime. - `cuvis.AcquisitionContext.capture` - parameter `to_interal` renamed to `to_internal`. +- `cuvis.Async.AsyncMesu.__await__`, `cuvis.Async.Async.__await__`, `cuvis.Worker.get_next_result_async` - await the SDK's own blocking wait in a worker thread instead of polling it every 10 ms or 100 ms, so they resume when the SDK finishes rather than at the next poll. + Each outstanding wait occupies one pooled thread, and it cannot be cancelled, because a thread blocked in C is not interruptible. +- `cuvis.Worker.register_worker_callback` - awaits each result instead of re-checking every 1 ms, so an idle registered callback no longer occupies a core. + `cuvis.Worker.reset_worker_callback` therefore takes effect only once the current one-second wait expires. ### Removed @@ -69,6 +73,7 @@ Pre-releases (`b*`, `rc*`) are not listed. - `cuvis.AcquisitionContext.register_ready_callback` - parameter `callback` was annotated `Callable[None, Awaitable[None]]`, which is not a valid `Callable` form; it is now `Callable[[], Awaitable[None]]`, matching the no-argument call the implementation makes. - `cuvis.AcquisitionContext.capture` - `to_internal=True` raised `TypeError` instead of queueing the measurement, because it passed a Python `0` where SWIG requires a null pointer. - `cuvis.Calibration`, `cuvis.AcquisitionContext`, `cuvis.ProcessingContext`, `cuvis.SessionFile`, `cuvis.Measurement`, `cuvis.Viewer`, `cuvis.Worker`, `cuvis.CubeExporter`, `cuvis.EnviExporter`, `cuvis.TiffExporter`, `cuvis.ViewExporter` - `__del__` raised `TypeError` after a failed construction, because it freed a handle that was still `None`. +- `cuvis.Worker.get_next_result_async` - a timeout returned a `cuvis.Worker.WorkerResult` built from handles the SDK never filled in; it now raises `cuvis.cuvis_aux.SDKException`. ## [3.5.3.2] - 2026-08-19 diff --git a/cuvis/Async.py b/cuvis/Async.py index e9ad752..95ca8e7 100644 --- a/cuvis/Async.py +++ b/cuvis/Async.py @@ -17,6 +17,25 @@ def _to_ms(value: int | timedelta) -> int: raise SDKException("Unknown type for converting to ms") +# The SDK's own waits take a timeout in ms and treat 0 as "wait for ever". +_WAIT_FOREVER = 0 + + +async def _wait_off_the_loop(blocking_get): + """Run one of the SDK's blocking waits in a worker thread. + + The wrappers release the GIL for the duration of the call, so the thread parks in the + SDK and the event loop keeps running. This is what makes the awaitables below real: + they resume when the SDK is done, not when the next poll happens to come round. + + The cost is one pooled thread per outstanding wait, and a wait that never completes + cannot be cancelled, because a thread blocked in C is not interruptible. Both go away + only if the SDK hands out something the event loop can watch directly. + """ + loop = a.get_running_loop() + return await loop.run_in_executor(None, blocking_get, _WAIT_FOREVER) + + class AsyncMesu(object): def __init__(self, handle): self._handle = handle @@ -46,17 +65,8 @@ def get( def __await__(self) -> Measurement | None: async def _wait_for_return(): - _status_ptr = cuvis_il.new_p_cuvis_status_t() - while True: - if cuvis_il.status_ok != cuvis_il.cuvis_async_capture_status( - self._handle, _status_ptr - ): - raise SDKException() - status = cuvis_il.p_cuvis_status_t_value(_status_ptr) - if status == cuvis_il.status_ok: - return self.get(0)[0] - else: - await a.sleep(10.0 / 1000) + mesu, _ = await _wait_off_the_loop(self.get) + return mesu return _wait_for_return().__await__() @@ -101,20 +111,7 @@ def get(self, timeout_ms: int | timedelta) -> AsyncResult: # Python Magic Methods def __await__(self) -> AsyncResult: - async def _wait_for_return(): - _status_ptr = cuvis_il.new_p_cuvis_status_t() - while True: - if cuvis_il.status_ok != cuvis_il.cuvis_async_call_status( - self._handle, _status_ptr - ): - raise SDKException() - status = cuvis_il.p_cuvis_status_t_value(_status_ptr) - if status == cuvis_il.status_ok: - return self.get(0) - else: - await a.sleep(10.0 / 1000) - - return _wait_for_return().__await__() + return _wait_off_the_loop(self.get).__await__() def __del__(self): if self._handle is None: diff --git a/cuvis/Worker.py b/cuvis/Worker.py index 12e982e..05b0068 100644 --- a/cuvis/Worker.py +++ b/cuvis/Worker.py @@ -14,6 +14,11 @@ from dataclasses import dataclass from collections.abc import Callable, Awaitable +# How long each wait for a worker result blocks before looping round. Long enough that a +# quiet worker costs nothing, short enough that reset_worker_callback is not held up by +# more than one window. +_RESULT_WAIT_MS = 1000 + @dataclass class WorkerResult: @@ -152,29 +157,40 @@ def get_next_result(self, timeout) -> WorkerResult: view = None return WorkerResult(mesu, view) - async def get_next_result_async(self, timeout: int) -> WorkerResult: - poll_intervall = 100 + def _wait_for_result(self, timeout_ms: int): + """The SDK's own blocking wait, returning None instead of raising on a timeout. + + `get_next_result` reports "nothing arrived" as an error, and `SDKException` logs + every instance it is constructed with. A wait that expects to come back empty + cannot go through it without filling the log with tracebacks, so the status is + read here directly. + """ ptr_mesu = cuvis_il.new_p_int() ptr_view = cuvis_il.new_p_int() + if cuvis_il.status_ok != cuvis_il.cuvis_worker_get_next_result( + self._handle, ptr_mesu, ptr_view, timeout_ms + ): + return None + view = ( + Viewer._create_view_data(None, cuvis_il.p_int_value(ptr_view)) + if self._viewer_set + else None + ) + return WorkerResult(Measurement(cuvis_il.p_int_value(ptr_mesu)), view) - tries = 0 - while tries * poll_intervall < timeout: - if self.has_next_result(): - await a.sleep(0) - if cuvis_il.status_ok != cuvis_il.cuvis_worker_get_next_result( - self._handle, ptr_mesu, ptr_view, 100 - ): - raise SDKException() - break - else: - tries += 1 - await a.sleep(poll_intervall / 1000) - mesu = Measurement(cuvis_il.p_int_value(ptr_mesu)) - if self._viewer_set: - view = Viewer._create_view_data(None, cuvis_il.p_int_value(ptr_view)) - else: - view = None - return WorkerResult(mesu, view) + async def get_next_result_async(self, timeout: int) -> WorkerResult: + """Wait for the next result without polling for it. + + The SDK's own wait is blocking and releases the GIL, so it runs in a worker + thread while the event loop carries on. Unlike the polling version this raises + `SDKException` when the timeout passes rather than returning a result built from + handles the SDK never filled in. + """ + loop = a.get_running_loop() + result = await loop.run_in_executor(None, self._wait_for_result, timeout) + if result is None: + raise SDKException("Worker produced no result within {} ms".format(timeout)) + return result @property @copydoc(cuvis_il.cuvis_worker_get_input_queue_limit) @@ -311,17 +327,18 @@ def register_worker_callback( self, callback: Callable[[WorkerResult], Awaitable[None]] ) -> None: self.reset_worker_callback() - poll_time = 0.001 async def _internal_worker_loop(): + loop = a.get_running_loop() while True: - if self.has_next_result(): - workerContainer = await self.get_next_result_async(1000) - a.create_task(callback(workerContainer)) - - # TODO limit number of created task objects like in the cpp wrapper - else: - await a.sleep(poll_time) + result = await loop.run_in_executor( + None, self._wait_for_result, _RESULT_WAIT_MS + ) + if result is None: + continue # nothing arrived inside the window; wait again + + a.create_task(callback(result)) + # TODO limit number of created task objects like in the cpp wrapper self._worker_poll_task = a.create_task(_internal_worker_loop())