1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
"""
Background work that outlives the call which started it.
**asyncio keeps only a weak reference to a task.** A coroutine started with
`asyncio.ensure_future(...)` whose result is thrown away can therefore be
collected while it is still running: the loop logs "Task was destroyed but it is
pending!" and the work simply does not happen. There is no error at the caller,
nothing fails, and what was lost is whatever that coroutine was in the middle of
— a notification never written, a file never indexed, a reload half applied.
`spawn()` is the one door for that shape. The reference in `_running` is what
keeps the task alive; the done callback is what stops the set growing.
Two things it deliberately does **not** replace:
- A task whose handle the caller keeps — the daemon's service tasks, the
indexer's reconcile loop — is already held, and being held is the whole point.
- A task scoped to something that can end, where the owner must be able to
**cancel** it: a peer session's spawns die with the session
(`WebRTCPeerSession._spawn`), and an enrichment run is tracked by its own
module. Those keep their own sets, because a module-level one can only ever
grow and forget, never shut down.
It also logs what a discarded task swallowed. An exception in a task nobody
awaits is reported by asyncio at collection time, out of context and often not
at all; here it is logged where it happened, naming the work.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Coroutine
log = logging.getLogger(__name__)
# Strong references to everything in flight. Module-level on purpose: these are
# tasks with no owner to hold them, which is exactly why they need this.
_running: set[asyncio.Task] = set()
def spawn(coro: Coroutine[Any, Any, Any], *, what: str = "") -> asyncio.Task:
"""Run a coroutine in the background and hold on to it until it finishes.
The work names itself in the log: `what` defaults to the coroutine's own
qualified name, so no call site has to carry a label for the sake of a line
nobody reads until something breaks. Pass one only where the coroutine's
name is not the useful answer.
"""
if not what:
what = getattr(coro, "__qualname__", "") or repr(coro)
task = asyncio.ensure_future(coro)
_running.add(task)
task.add_done_callback(lambda t: _finished(t, what))
return task
def _finished(task: asyncio.Task, what: str) -> None:
_running.discard(task)
if task.cancelled():
return
exc = task.exception()
if exc is not None:
log.error("Background task %s failed: %s", what or "(unnamed)", exc,
exc_info=exc)
def pending() -> int:
"""How many background tasks are in flight. For diagnostics and tests."""
return len(_running)
|