forked from python-trio/trio
-
Notifications
You must be signed in to change notification settings - Fork 0
/
_generated_run.py
269 lines (215 loc) · 10.2 KB
/
_generated_run.py
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
# ***********************************************************
# ******* WARNING: AUTOGENERATED! ALL EDITS WILL BE LOST ******
# *************************************************************
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from ._ki import LOCALS_KEY_KI_PROTECTION_ENABLED
from ._run import _NO_SEND, GLOBAL_RUN_CONTEXT, RunStatistics, Task
if TYPE_CHECKING:
import contextvars
from collections.abc import Awaitable, Callable
from outcome import Outcome
from typing_extensions import Unpack
from .._abc import Clock
from ._entry_queue import TrioToken
from ._run import PosArgT
__all__ = [
"current_clock",
"current_root_task",
"current_statistics",
"current_time",
"current_trio_token",
"reschedule",
"spawn_system_task",
"wait_all_tasks_blocked",
]
def current_statistics() -> RunStatistics:
"""Returns ``RunStatistics``, which contains run-loop-level debugging information.
Currently, the following fields are defined:
* ``tasks_living`` (int): The number of tasks that have been spawned
and not yet exited.
* ``tasks_runnable`` (int): The number of tasks that are currently
queued on the run queue (as opposed to blocked waiting for something
to happen).
* ``seconds_to_next_deadline`` (float): The time until the next
pending cancel scope deadline. May be negative if the deadline has
expired but we haven't yet processed cancellations. May be
:data:`~math.inf` if there are no pending deadlines.
* ``run_sync_soon_queue_size`` (int): The number of
unprocessed callbacks queued via
:meth:`trio.lowlevel.TrioToken.run_sync_soon`.
* ``io_statistics`` (object): Some statistics from Trio's I/O
backend. This always has an attribute ``backend`` which is a string
naming which operating-system-specific I/O backend is in use; the
other attributes vary between backends.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.current_statistics()
except AttributeError:
raise RuntimeError("must be called from async context") from None
def current_time() -> float:
"""Returns the current time according to Trio's internal clock.
Returns:
float: The current time.
Raises:
RuntimeError: if not inside a call to :func:`trio.run`.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.current_time()
except AttributeError:
raise RuntimeError("must be called from async context") from None
def current_clock() -> Clock:
"""Returns the current :class:`~trio.abc.Clock`."""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.current_clock()
except AttributeError:
raise RuntimeError("must be called from async context") from None
def current_root_task() -> Task | None:
"""Returns the current root :class:`Task`.
This is the task that is the ultimate parent of all other tasks.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.current_root_task()
except AttributeError:
raise RuntimeError("must be called from async context") from None
def reschedule(task: Task, next_send: Outcome[Any] = _NO_SEND) -> None:
"""Reschedule the given task with the given
:class:`outcome.Outcome`.
See :func:`wait_task_rescheduled` for the gory details.
There must be exactly one call to :func:`reschedule` for every call to
:func:`wait_task_rescheduled`. (And when counting, keep in mind that
returning :data:`Abort.SUCCEEDED` from an abort callback is equivalent
to calling :func:`reschedule` once.)
Args:
task (trio.lowlevel.Task): the task to be rescheduled. Must be blocked
in a call to :func:`wait_task_rescheduled`.
next_send (outcome.Outcome): the value (or error) to return (or
raise) from :func:`wait_task_rescheduled`.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.reschedule(task, next_send)
except AttributeError:
raise RuntimeError("must be called from async context") from None
def spawn_system_task(
async_fn: Callable[[Unpack[PosArgT]], Awaitable[object]],
*args: Unpack[PosArgT],
name: object = None,
context: (contextvars.Context | None) = None,
) -> Task:
"""Spawn a "system" task.
System tasks have a few differences from regular tasks:
* They don't need an explicit nursery; instead they go into the
internal "system nursery".
* If a system task raises an exception, then it's converted into a
:exc:`~trio.TrioInternalError` and *all* tasks are cancelled. If you
write a system task, you should be careful to make sure it doesn't
crash.
* System tasks are automatically cancelled when the main task exits.
* By default, system tasks have :exc:`KeyboardInterrupt` protection
*enabled*. If you want your task to be interruptible by control-C,
then you need to use :func:`disable_ki_protection` explicitly (and
come up with some plan for what to do with a
:exc:`KeyboardInterrupt`, given that system tasks aren't allowed to
raise exceptions).
* System tasks do not inherit context variables from their creator.
Towards the end of a call to :meth:`trio.run`, after the main
task and all system tasks have exited, the system nursery
becomes closed. At this point, new calls to
:func:`spawn_system_task` will raise ``RuntimeError("Nursery
is closed to new arrivals")`` instead of creating a system
task. It's possible to encounter this state either in
a ``finally`` block in an async generator, or in a callback
passed to :meth:`TrioToken.run_sync_soon` at the right moment.
Args:
async_fn: An async callable.
args: Positional arguments for ``async_fn``. If you want to pass
keyword arguments, use :func:`functools.partial`.
name: The name for this task. Only used for debugging/introspection
(e.g. ``repr(task_obj)``). If this isn't a string,
:func:`spawn_system_task` will try to make it one. A common use
case is if you're wrapping a function before spawning a new
task, you might pass the original function as the ``name=`` to
make debugging easier.
context: An optional ``contextvars.Context`` object with context variables
to use for this task. You would normally get a copy of the current
context with ``context = contextvars.copy_context()`` and then you would
pass that ``context`` object here.
Returns:
Task: the newly spawned task
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.spawn_system_task(
async_fn, *args, name=name, context=context
)
except AttributeError:
raise RuntimeError("must be called from async context") from None
def current_trio_token() -> TrioToken:
"""Retrieve the :class:`TrioToken` for the current call to
:func:`trio.run`.
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return GLOBAL_RUN_CONTEXT.runner.current_trio_token()
except AttributeError:
raise RuntimeError("must be called from async context") from None
async def wait_all_tasks_blocked(cushion: float = 0.0) -> None:
"""Block until there are no runnable tasks.
This is useful in testing code when you want to give other tasks a
chance to "settle down". The calling task is blocked, and doesn't wake
up until all other tasks are also blocked for at least ``cushion``
seconds. (Setting a non-zero ``cushion`` is intended to handle cases
like two tasks talking to each other over a local socket, where we
want to ignore the potential brief moment between a send and receive
when all tasks are blocked.)
Note that ``cushion`` is measured in *real* time, not the Trio clock
time.
If there are multiple tasks blocked in :func:`wait_all_tasks_blocked`,
then the one with the shortest ``cushion`` is the one woken (and
this task becoming unblocked resets the timers for the remaining
tasks). If there are multiple tasks that have exactly the same
``cushion``, then all are woken.
You should also consider :class:`trio.testing.Sequencer`, which
provides a more explicit way to control execution ordering within a
test, and will often produce more readable tests.
Example:
Here's an example of one way to test that Trio's locks are fair: we
take the lock in the parent, start a child, wait for the child to be
blocked waiting for the lock (!), and then check that we can't
release and immediately re-acquire the lock::
async def lock_taker(lock):
await lock.acquire()
lock.release()
async def test_lock_fairness():
lock = trio.Lock()
await lock.acquire()
async with trio.open_nursery() as nursery:
nursery.start_soon(lock_taker, lock)
# child hasn't run yet, we have the lock
assert lock.locked()
assert lock._owner is trio.lowlevel.current_task()
await trio.testing.wait_all_tasks_blocked()
# now the child has run and is blocked on lock.acquire(), we
# still have the lock
assert lock.locked()
assert lock._owner is trio.lowlevel.current_task()
lock.release()
try:
# The child has a prior claim, so we can't have it
lock.acquire_nowait()
except trio.WouldBlock:
assert lock._owner is not trio.lowlevel.current_task()
print("PASS")
else:
print("FAIL")
"""
locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
try:
return await GLOBAL_RUN_CONTEXT.runner.wait_all_tasks_blocked(cushion)
except AttributeError:
raise RuntimeError("must be called from async context") from None