Coverage for src/lilbee/runtime/daemon_call.py: 100%

1"""A call run on a daemon thread that the caller can wait on, and that exit never joins. 

2 

3concurrent.futures joins every executor worker at interpreter exit whatever its 

4daemon flag, so a worker stuck in a network read holds the process open. 

5""" 

6 

7from __future__ import annotations 

8 

9import threading 

10from collections.abc import Callable 

11from typing import Generic, TypeVar, cast 

12 

13T = TypeVar("T") 

14 

15 

16class DaemonCall(Generic[T]): 

17 """Run *fn* on a started daemon thread; ``result()`` returns its value or re-raises.""" 

18 

19 def __init__(self, fn: Callable[[], T], *, name: str) -> None: 

20 self._fn = fn 

21 self._value: T | None = None 

22 self._error: BaseException | None = None 

23 self._thread = threading.Thread(target=self._run, name=name, daemon=True) 

24 self._thread.start() 

25 

26 def _run(self) -> None: 

27 try: 

28 self._value = self._fn() 

29 except BaseException as exc: # re-raised to the waiting caller by result() 

30 self._error = exc 

31 

32 def wait(self, timeout: float | None = None) -> bool: 

33 """Block until the call finishes or *timeout* passes; return whether it finished.""" 

34 self._thread.join(timeout) 

35 return not self._thread.is_alive() 

36 

37 def result(self) -> T: 

38 """Wait for the call, then return its value or re-raise its exception.""" 

39 self.wait() 

40 if self._error is not None: 

41 raise self._error 

42 return cast(T, self._value)