Introspecting and extending Trio with trio.lowlevel
¶
trio.lowlevel
contains low-level APIs for introspecting and
extending Trio. If you’re writing ordinary, everyday code, then you
can ignore this module completely. But sometimes you need something a
bit lower level. Here are some examples of situations where you should
reach for trio.lowlevel
:
You want to implement a new synchronization primitive that Trio doesn’t (yet) provide, like a reader-writer lock.
You want to extract low-level metrics to monitor the health of your application.
You want to use a low-level operating system interface that Trio doesn’t (yet) provide its own wrappers for, like watching a filesystem directory for changes.
You want to implement an interface for calling between Trio and another event loop within the same process.
You’re writing a debugger and want to visualize Trio’s task tree.
You need to interoperate with a C library whose API exposes raw file descriptors.
You don’t need to be scared of trio.lowlevel
, as long as you
take proper precautions. These are real public APIs, with strictly
defined and carefully documented semantics. They’re the same tools we
use to implement all the nice high-level APIs in the trio
namespace. But, be careful. Some of those strict semantics have nasty
big pointy teeth. If you make a
mistake, Trio may not be able to handle it gracefully; conventions and
guarantees that are followed strictly in the rest of Trio do not
always apply. When you use this module, it’s your job to think about
how you’re going to handle the tricky cases so you can expose a
friendly Trio-style API to your users.
Debugging and instrumentation¶
Trio tries hard to provide useful hooks for debugging and
instrumentation. Some are documented above (the nursery introspection
attributes, trio.Lock.statistics()
, etc.). Here are some more.
Global statistics¶
-
trio.lowlevel.
current_statistics
()¶ Returns an object containing 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 beinf
if there are no pending deadlines.run_sync_soon_queue_size
(int): The number of unprocessed callbacks queued viatrio.lowlevel.TrioToken.run_sync_soon()
.io_statistics
(object): Some statistics from Trio’s I/O backend. This always has an attributebackend
which is a string naming which operating-system-specific I/O backend is in use; the other attributes vary between backends.
Instrument API¶
The instrument API provides a standard way to add custom instrumentation to the run loop. Want to make a histogram of scheduling latencies, log a stack trace of any task that blocks the run loop for >50 ms, or measure what percentage of your process’s running time is spent waiting for I/O? This is the place.
The general idea is that at any given moment, trio.run()
maintains a set of “instruments”, which are objects that implement the
trio.abc.Instrument
interface. When an interesting event
happens, it loops over these instruments and notifies them by calling
an appropriate method. The tutorial has a simple example of
using this for tracing.
Since this hooks into Trio at a rather low level, you do have to be
careful. The callbacks are run synchronously, and in many cases if
they error out then there isn’t any plausible way to propagate this
exception (for instance, we might be deep in the guts of the exception
propagation machinery…). Therefore our current strategy for handling
exceptions raised by instruments is to (a) log an exception to the
"trio.abc.Instrument"
logger, which by default prints a stack
trace to standard error and (b) disable the offending instrument.
You can register an initial list of instruments by passing them to
trio.run()
. add_instrument()
and
remove_instrument()
let you add and remove instruments at
runtime.
-
trio.lowlevel.
add_instrument
(instrument)¶ Start instrumenting the current run loop with the given instrument.
- Parameters
instrument (trio.abc.Instrument) – The instrument to activate.
If
instrument
is already active, does nothing.
-
trio.lowlevel.
remove_instrument
(instrument)¶ Stop instrumenting the current run loop with the given instrument.
- Parameters
instrument (trio.abc.Instrument) – The instrument to de-activate.
- Raises
KeyError – if the instrument is not currently active. This could occur either because you never added it, or because you added it and then it raised an unhandled exception and was automatically deactivated.
And here’s the interface to implement if you want to build your own
Instrument
:
-
class
trio.abc.
Instrument
¶ The interface for run loop instrumentation.
Instruments don’t have to inherit from this abstract base class, and all of these methods are optional. This class serves mostly as documentation.
-
after_io_wait
(timeout)¶ Called after handling pending I/O.
- Parameters
timeout (float) – The number of seconds we were willing to wait. This much time may or may not have elapsed, depending on whether any I/O was ready.
-
after_run
()¶ Called just before
trio.run()
returns.
-
after_task_step
(task)¶ Called when we return to the main run loop after a task has yielded.
- Parameters
task (trio.lowlevel.Task) – The task that just ran.
-
before_io_wait
(timeout)¶ Called before blocking to wait for I/O readiness.
- Parameters
timeout (float) – The number of seconds we are willing to wait.
-
before_run
()¶ Called at the beginning of
trio.run()
.
-
before_task_step
(task)¶ Called immediately before we resume running the given task.
- Parameters
task (trio.lowlevel.Task) – The task that is about to run.
-
task_exited
(task)¶ Called when the given task exits.
- Parameters
task (trio.lowlevel.Task) – The finished task.
-
task_scheduled
(task)¶ Called when the given task becomes runnable.
It may still be some time before it actually runs, if there are other runnable tasks ahead of it.
- Parameters
task (trio.lowlevel.Task) – The task that became runnable.
-
task_spawned
(task)¶ Called when the given task is created.
- Parameters
task (trio.lowlevel.Task) – The new task.
-
The tutorial has a fully-worked example of defining a custom instrument to log Trio’s internal scheduling decisions.
Low-level I/O primitives¶
Different environments expose different low-level APIs for performing
async I/O. trio.lowlevel
exposes these APIs in a relatively
direct way, so as to allow maximum power and flexibility for higher
level code. However, this means that the exact API provided may vary
depending on what system Trio is running on.
Universally available API¶
All environments provide the following functions:
-
await
trio.lowlevel.
wait_readable
(obj)¶ Block until the kernel reports that the given object is readable.
On Unix systems,
obj
must either be an integer file descriptor, or else an object with a.fileno()
method which returns an integer file descriptor. Any kind of file descriptor can be passed, though the exact semantics will depend on your kernel. For example, this probably won’t do anything useful for on-disk files.On Windows systems,
obj
must either be an integerSOCKET
handle, or else an object with a.fileno()
method which returns an integerSOCKET
handle. File descriptors aren’t supported, and neither are handles that refer to anything besides aSOCKET
.- Raises
trio.BusyResourceError – if another task is already waiting for the given socket to become readable.
trio.ClosedResourceError – if another task calls
notify_closing()
while this function is still working.
-
await
trio.lowlevel.
wait_writable
(obj)¶ Block until the kernel reports that the given object is writable.
See
wait_readable
for the definition ofobj
.- Raises
trio.BusyResourceError – if another task is already waiting for the given socket to become writable.
trio.ClosedResourceError – if another task calls
notify_closing()
while this function is still working.
-
trio.lowlevel.
notify_closing
(obj)¶ Call this before closing a file descriptor (on Unix) or socket (on Windows). This will cause any
wait_readable
orwait_writable
calls on the given object to immediately wake up and raiseClosedResourceError
.This doesn’t actually close the object – you still have to do that yourself afterwards. Also, you want to be careful to make sure no new tasks start waiting on the object in between when you call this and when it’s actually closed. So to close something properly, you usually want to do these steps in order:
Explicitly mark the object as closed, so that any new attempts to use it will abort before they start.
Call
notify_closing
to wake up any already-existing users.Actually close the object.
It’s also possible to do them in a different order if that’s more convenient, but only if you make sure not to have any checkpoints in between the steps. This way they all happen in a single atomic step, so other tasks won’t be able to tell what order they happened in anyway.
Unix-specific API¶
FdStream
supports wrapping Unix files (such as a pipe or TTY) as
a stream.
If you have two different file descriptors for sending and receiving,
and want to bundle them together into a single bidirectional
Stream
, then use trio.StapledStream
:
bidirectional_stream = trio.StapledStream(
trio.lowlevel.FdStream(write_fd),
trio.lowlevel.FdStream(read_fd)
)
-
class
trio.lowlevel.
FdStream
(fd: int)¶ Bases:
trio.abc.Stream
Represents a stream given the file descriptor to a pipe, TTY, etc.
fd must refer to a file that is open for reading and/or writing and supports non-blocking I/O (pipes and TTYs will work, on-disk files probably not). The returned stream takes ownership of the fd, so closing the stream will close the fd too. As with
os.fdopen
, you should not directly use an fd after you have wrapped it in a stream using this function.To be used as a Trio stream, an open file must be placed in non-blocking mode. Unfortunately, this impacts all I/O that goes through the underlying open file, including I/O that uses a different file descriptor than the one that was passed to Trio. If other threads or processes are using file descriptors that are related through
os.dup
or inheritance acrossos.fork
to the one that Trio is using, they are unlikely to be prepared to have non-blocking I/O semantics suddenly thrust upon them. For example, you can useFdStream(os.dup(0))
to obtain a stream for reading from standard input, but it is only safe to do so with heavy caveats: your stdin must not be shared by any other processes and you must not make any calls to synchronous methods ofsys.stdin
until the stream returned byFdStream
is closed. See issue #174 for a discussion of the challenges involved in relaxing this restriction.
Kqueue-specific API¶
TODO: these are implemented, but are currently more of a sketch than anything real. See #26.
-
trio.lowlevel.
current_kqueue
()¶
-
await
trio.lowlevel.
wait_kevent
(ident, filter, abort_func)¶
-
with
trio.lowlevel.
monitor_kevent
(ident, filter) as queue¶
Windows-specific API¶
-
await
trio.lowlevel.
WaitForSingleObject
(handle)¶ Async and cancellable variant of WaitForSingleObject. Windows only.
- Parameters
handle – A Win32 object handle, as a Python integer.
- Raises
OSError – If the handle is invalid, e.g. when it is already closed.
TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.
-
trio.lowlevel.
register_with_iocp
(handle)¶
-
await
trio.lowlevel.
wait_overlapped
(handle, lpOverlapped)¶
-
trio.lowlevel.
current_iocp
()¶
-
with
trio.lowlevel.
monitor_completion_key
() as queue¶
Global state: system tasks and run-local variables¶
-
class
trio.lowlevel.
RunVar
(name, default=<object object>)¶ The run-local variant of a context variable.
RunVar
objects are similar to context variable objects, except that they are shared across a single call totrio.run()
rather than a single task.
-
trio.lowlevel.
spawn_system_task
(async_fn, *args, name=None)¶ 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
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
KeyboardInterrupt
protection enabled. If you want your task to be interruptible by control-C, then you need to usedisable_ki_protection()
explicitly (and come up with some plan for what to do with aKeyboardInterrupt
, given that system tasks aren’t allowed to raise exceptions).System tasks do not inherit context variables from their creator.
- Parameters
async_fn – An async callable.
args – Positional arguments for
async_fn
. If you want to pass keyword arguments, usefunctools.partial()
.name – The name for this task. Only used for debugging/introspection (e.g.
repr(task_obj)
). If this isn’t a string,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 thename=
to make debugging easier.
- Returns
the newly spawned task
- Return type
Trio tokens¶
-
class
trio.lowlevel.
TrioToken
¶ An opaque object representing a single call to
trio.run()
.It has no public constructor; instead, see
current_trio_token()
.This object has two uses:
It lets you re-enter the Trio run loop from external threads or signal handlers. This is the low-level primitive that
trio.to_thread()
andtrio.from_thread
use to communicate with worker threads, thattrio.open_signal_receiver
uses to receive notifications about signals, and so forth.Each call to
trio.run()
has exactly one associatedTrioToken
object, so you can use it to identify a particular call.
-
run_sync_soon
(sync_fn, *args, idempotent=False)¶ Schedule a call to
sync_fn(*args)
to occur in the context of a Trio task.This is safe to call from the main thread, from other threads, and from signal handlers. This is the fundamental primitive used to re-enter the Trio run loop from outside of it.
The call will happen “soon”, but there’s no guarantee about exactly when, and no mechanism provided for finding out when it’s happened. If you need this, you’ll have to build your own.
The call is effectively run as part of a system task (see
spawn_system_task()
). In particular this means that:KeyboardInterrupt
protection is enabled by default; if you wantsync_fn
to be interruptible by control-C, then you need to usedisable_ki_protection()
explicitly.If
sync_fn
raises an exception, then it’s converted into aTrioInternalError
and all tasks are cancelled. You should be careful thatsync_fn
doesn’t crash.
All calls with
idempotent=False
are processed in strict first-in first-out order.If
idempotent=True
, thensync_fn
andargs
must be hashable, and Trio will make a best-effort attempt to discard any call submission which is equal to an already-pending call. Trio will process these in first-in first-out order.Any ordering guarantees apply separately to
idempotent=False
andidempotent=True
calls; there’s no rule for how calls in the different categories are ordered with respect to each other.- Raises
trio.RunFinishedError – if the associated call to
trio.run()
has already exited. (Any call that doesn’t raise this error is guaranteed to be fully processed beforetrio.run()
exits.)
-
trio.lowlevel.
current_trio_token
()¶ Retrieve the
TrioToken
for the current call totrio.run()
.
Safer KeyboardInterrupt handling¶
Trio’s handling of control-C is designed to balance usability and
safety. On the one hand, there are sensitive regions (like the core
scheduling loop) where it’s simply impossible to handle arbitrary
KeyboardInterrupt
exceptions while maintaining our core
correctness invariants. On the other, if the user accidentally writes
an infinite loop, we do want to be able to break out of that. Our
solution is to install a default signal handler which checks whether
it’s safe to raise KeyboardInterrupt
at the place where the
signal is received. If so, then we do; otherwise, we schedule a
KeyboardInterrupt
to be delivered to the main task at the next
available opportunity (similar to how Cancelled
is
delivered).
So that’s great, but – how do we know whether we’re in one of the sensitive parts of the program or not?
This is determined on a function-by-function basis. By default, a function is protected if its caller is, and not if its caller isn’t; this is helpful because it means you only need to override the defaults at places where you transition from protected code to unprotected code or vice-versa.
These transitions are accomplished using two function decorators:
-
@
trio.lowlevel.
disable_ki_protection
¶ Decorator that marks the given regular function, generator function, async function, or async generator function as unprotected against
KeyboardInterrupt
, i.e., the code inside this function can be rudely interrupted byKeyboardInterrupt
at any moment.If you have multiple decorators on the same function, then this should be at the bottom of the stack (closest to the actual function).
An example of where you’d use this is in implementing something like
trio.from_thread.run()
, which usesTrioToken.run_sync_soon()
to get into the Trio thread.run_sync_soon()
callbacks are run withKeyboardInterrupt
protection enabled, andtrio.from_thread.run()
takes advantage of this to safely set up the machinery for sending a response back to the original thread, but then usesdisable_ki_protection()
when entering the user-provided function.
-
@
trio.lowlevel.
enable_ki_protection
¶ Decorator that marks the given regular function, generator function, async function, or async generator function as protected against
KeyboardInterrupt
, i.e., the code inside this function won’t be rudely interrupted byKeyboardInterrupt
. (Though if it contains any checkpoints, then it can still receiveKeyboardInterrupt
at those. This is considered a polite interruption.)Warning
Be very careful to only use this decorator on functions that you know will either exit in bounded time, or else pass through a checkpoint regularly. (Of course all of your functions should have this property, but if you mess it up here then you won’t even be able to use control-C to escape!)
If you have multiple decorators on the same function, then this should be at the bottom of the stack (closest to the actual function).
An example of where you’d use this is on the
__exit__
implementation for something like aLock
, where a poorly-timedKeyboardInterrupt
could leave the lock in an inconsistent state and cause a deadlock.
-
trio.lowlevel.
currently_ki_protected
()¶ Check whether the calling code has
KeyboardInterrupt
protection enabled.It’s surprisingly easy to think that one’s
KeyboardInterrupt
protection is enabled when it isn’t, or vice-versa. This function tells you what Trio thinks of the matter, which makes it useful forassert
s and unit tests.- Returns
True if protection is enabled, and False otherwise.
- Return type
Sleeping and waking¶
Wait queue abstraction¶
-
class
trio.lowlevel.
ParkingLot
¶ A fair wait queue with cancellation and requeueing.
This class encapsulates the tricky parts of implementing a wait queue. It’s useful for implementing higher-level synchronization primitives like queues and locks.
In addition to the methods below, you can use
len(parking_lot)
to get the number of parked tasks, andif parking_lot: ...
to check whether there are any parked tasks.-
await
park
()¶ Park the current task until woken by a call to
unpark()
orunpark_all()
.
-
repark
(new_lot, *, count=1)¶ Move parked tasks from one
ParkingLot
object to another.This dequeues
count
tasks from one lot, and requeues them on another, preserving order. For example:async def parker(lot): print("sleeping") await lot.park() print("woken") async def main(): lot1 = trio.lowlevel.ParkingLot() lot2 = trio.lowlevel.ParkingLot() async with trio.open_nursery() as nursery: nursery.start_soon(parker, lot1) await trio.testing.wait_all_tasks_blocked() assert len(lot1) == 1 assert len(lot2) == 0 lot1.repark(lot2) assert len(lot1) == 0 assert len(lot2) == 1 # This wakes up the task that was originally parked in lot1 lot2.unpark()
If there are fewer than
count
tasks parked, then reparks as many tasks as are available and then returns successfully.- Parameters
new_lot (ParkingLot) – the parking lot to move tasks to.
count (int) – the number of tasks to move.
-
repark_all
(new_lot)¶ Move all parked tasks from one
ParkingLot
object to another.See
repark()
for details.
-
statistics
()¶ Return an object containing debugging information.
Currently the following fields are defined:
tasks_waiting
: The number of tasks blocked on this lot’spark()
method.
-
unpark
(*, count=1)¶ Unpark one or more tasks.
This wakes up
count
tasks that are blocked inpark()
. If there are fewer thancount
tasks parked, then wakes as many tasks are available and then returns successfully.- Parameters
count (int) – the number of tasks to unpark.
-
unpark_all
()¶ Unpark all parked tasks.
-
await
Low-level checkpoint functions¶
-
await
trio.lowlevel.
checkpoint
()¶ A pure checkpoint.
This checks for cancellation and allows other tasks to be scheduled, without otherwise blocking.
Note that the scheduler has the option of ignoring this and continuing to run the current task if it decides this is appropriate (e.g. for increased efficiency).
Equivalent to
await trio.sleep(0)
(which is implemented by callingcheckpoint()
.)
The next two functions are used together to make up a checkpoint:
-
await
trio.lowlevel.
checkpoint_if_cancelled
()¶ Issue a checkpoint if the calling context has been cancelled.
Equivalent to (but potentially more efficient than):
if trio.current_deadline() == -inf: await trio.lowlevel.checkpoint()
This is either a no-op, or else it allow other tasks to be scheduled and then raises
trio.Cancelled
.Typically used together with
cancel_shielded_checkpoint()
.
-
await
trio.lowlevel.
cancel_shielded_checkpoint
()¶ Introduce a schedule point, but not a cancel point.
This is not a checkpoint, but it is half of a checkpoint, and when combined with
checkpoint_if_cancelled()
it can make a full checkpoint.Equivalent to (but potentially more efficient than):
with trio.CancelScope(shield=True): await trio.lowlevel.checkpoint()
These are commonly used in cases where you have an operation that might-or-might-not block, and you want to implement Trio’s standard checkpoint semantics. Example:
async def operation_that_maybe_blocks():
await checkpoint_if_cancelled()
try:
ret = attempt_operation()
except BlockingIOError:
# need to block and then retry, which we do below
pass
else:
# operation succeeded, finish the checkpoint then return
await cancel_shielded_checkpoint()
return ret
while True:
await wait_for_operation_to_be_ready()
try:
return attempt_operation()
except BlockingIOError:
pass
This logic is a bit convoluted, but accomplishes all of the following:
Every successful execution path passes through a checkpoint (assuming that
wait_for_operation_to_be_ready
is an unconditional checkpoint)Our cancellation semantics say that
Cancelled
should only be raised if the operation didn’t happen. Usingcancel_shielded_checkpoint()
on the early-exit branch accomplishes this.On the path where we do end up blocking, we don’t pass through any schedule points before that, which avoids some unnecessary work.
Avoids implicitly chaining the
BlockingIOError
with any errors raised byattempt_operation
orwait_for_operation_to_be_ready
, by keeping thewhile True:
loop outside of theexcept BlockingIOError:
block.
These functions can also be useful in other situations. For example,
when trio.to_thread.run_sync()
schedules some work to run in a
worker thread, it blocks until the work is finished (so it’s a
schedule point), but by default it doesn’t allow cancellation. So to
make sure that the call always acts as a checkpoint, it calls
checkpoint_if_cancelled()
before starting the thread.
Low-level blocking¶
-
await
trio.lowlevel.
wait_task_rescheduled
(abort_func)¶ Put the current task to sleep, with cancellation support.
This is the lowest-level API for blocking in Trio. Every time a
Task
blocks, it does so by calling this function (usually indirectly via some higher-level API).This is a tricky interface with no guard rails. If you can use
ParkingLot
or the built-in I/O wait functions instead, then you should.Generally the way it works is that before calling this function, you make arrangements for “someone” to call
reschedule()
on the current task at some later point.Then you call
wait_task_rescheduled()
, passing inabort_func
, an “abort callback”.(Terminology: in Trio, “aborting” is the process of attempting to interrupt a blocked task to deliver a cancellation.)
There are two possibilities for what happens next:
“Someone” calls
reschedule()
on the current task, andwait_task_rescheduled()
returns or raises whatever value or error was passed toreschedule()
.The call’s context transitions to a cancelled state (e.g. due to a timeout expiring). When this happens, the
abort_func
is called. Its interface looks like:def abort_func(raise_cancel): ... return trio.lowlevel.Abort.SUCCEEDED # or FAILED
It should attempt to clean up any state associated with this call, and in particular, arrange that
reschedule()
will not be called later. If (and only if!) it is successful, then it should returnAbort.SUCCEEDED
, in which case the task will automatically be rescheduled with an appropriateCancelled
error.Otherwise, it should return
Abort.FAILED
. This means that the task can’t be cancelled at this time, and still has to make sure that “someone” eventually callsreschedule()
.At that point there are again two possibilities. You can simply ignore the cancellation altogether: wait for the operation to complete and then reschedule and continue as normal. (For example, this is what
trio.to_thread.run_sync()
does if cancellation is disabled.) The other possibility is that theabort_func
does succeed in cancelling the operation, but for some reason isn’t able to report that right away. (Example: on Windows, it’s possible to request that an async (“overlapped”) I/O operation be cancelled, but this request is also asynchronous – you don’t find out until later whether the operation was actually cancelled or not.) To report a delayed cancellation, then you should reschedule the task yourself, and call theraise_cancel
callback passed toabort_func
to raise aCancelled
(or possiblyKeyboardInterrupt
) exception into this task. Either of the approaches sketched below can work:# Option 1: # Catch the exception from raise_cancel and inject it into the task. # (This is what Trio does automatically for you if you return # Abort.SUCCEEDED.) trio.lowlevel.reschedule(task, outcome.capture(raise_cancel)) # Option 2: # wait to be woken by "someone", and then decide whether to raise # the error from inside the task. outer_raise_cancel = None def abort(inner_raise_cancel): nonlocal outer_raise_cancel outer_raise_cancel = inner_raise_cancel TRY_TO_CANCEL_OPERATION() return trio.lowlevel.Abort.FAILED await wait_task_rescheduled(abort) if OPERATION_WAS_SUCCESSFULLY_CANCELLED: # raises the error outer_raise_cancel()
In any case it’s guaranteed that we only call the
abort_func
at most once per call towait_task_rescheduled()
.
Sometimes, it’s useful to be able to share some mutable sleep-related data between the sleeping task, the abort function, and the waking task. You can use the sleeping task’s
custom_sleep_data
attribute to store this data, and Trio won’t touch it, except to make sure that it gets cleared when the task is rescheduled.Warning
If your
abort_func
raises an error, or returns any value other thanAbort.SUCCEEDED
orAbort.FAILED
, then Trio will crash violently. Be careful! Similarly, it is entirely possible to deadlock a Trio program by failing to reschedule a blocked task, or cause havoc by callingreschedule()
too many times. Remember what we said up above about how you should use a higher-level API if at all possible?
-
class
trio.lowlevel.
Abort
¶ enum.Enum
used as the return value from abort functions.See
wait_task_rescheduled()
for details.
-
trio.lowlevel.
reschedule
(task, next_send=<object object>)¶ Reschedule the given task with the given
outcome.Outcome
.See
wait_task_rescheduled()
for the gory details.There must be exactly one call to
reschedule()
for every call towait_task_rescheduled()
. (And when counting, keep in mind that returningAbort.SUCCEEDED
from an abort callback is equivalent to callingreschedule()
once.)- Parameters
task (trio.lowlevel.Task) – the task to be rescheduled. Must be blocked in a call to
wait_task_rescheduled()
.next_send (outcome.Outcome) – the value (or error) to return (or raise) from
wait_task_rescheduled()
.
Here’s an example lock class implemented using
wait_task_rescheduled()
directly. This implementation has a
number of flaws, including lack of fairness, O(n) cancellation,
missing error checking, failure to insert a checkpoint on the
non-blocking path, etc. If you really want to implement your own lock,
then you should study the implementation of trio.Lock
and use
ParkingLot
, which handles some of these issues for you. But
this does serve to illustrate the basic structure of the
wait_task_rescheduled()
API:
class NotVeryGoodLock:
def __init__(self):
self._blocked_tasks = collections.deque()
self._held = False
async def acquire(self):
while self._held:
task = trio.current_task()
self._blocked_tasks.append(task)
def abort_fn(_):
self._blocked_tasks.remove(task)
return trio.lowlevel.Abort.SUCCEEDED
await trio.lowlevel.wait_task_rescheduled(abort_fn)
self._held = True
def release(self):
self._held = False
if self._blocked_tasks:
woken_task = self._blocked_tasks.popleft()
trio.lowlevel.reschedule(woken_task)
Task API¶
-
trio.lowlevel.
current_root_task
()¶ Returns the current root
Task
.This is the task that is the ultimate parent of all other tasks.
-
trio.lowlevel.
current_task
()¶ Return the
Task
object representing the current task.- Returns
the
Task
that calledcurrent_task()
.- Return type
-
class
trio.lowlevel.
Task
¶ A
Task
object represents a concurrent “thread” of execution. It has no public constructor; Trio internally creates aTask
object for each call tonursery.start(...)
ornursery.start_soon(...)
.Its public members are mostly useful for introspection and debugging:
-
name
¶ String containing this
Task
’s name. Usually the name of the function thisTask
is running, but can be overridden by passingname=
tostart
orstart_soon
.
-
coro
¶ This task’s coroutine object. Example usage: extracting a stack trace:
import traceback def walk_coro_stack(coro): while coro is not None: if hasattr(coro, "cr_frame"): # A real coroutine yield coro.cr_frame, coro.cr_frame.f_lineno coro = coro.cr_await else: # A generator decorated with @types.coroutine yield coro.gi_frame, coro.gi_frame.f_lineno coro = coro.gi_yieldfrom def print_stack_for_task(task): ss = traceback.StackSummary.extract(walk_coro_stack(task.coro)) print("".join(ss.format()))
-
context
¶ This task’s
contextvars.Context
object.
-
parent_nursery
¶ The nursery this task is inside (or None if this is the “init” task).
Example use case: drawing a visualization of the task tree in a debugger.
-
child_nurseries
¶ The nurseries this task contains.
This is a list, with outer nurseries before inner nurseries.
-
custom_sleep_data
¶ Trio doesn’t assign this variable any meaning, except that it sets it to
None
whenever a task is rescheduled. It can be used to share data between the different tasks involved in putting a task to sleep and then waking it up again. (Seewait_task_rescheduled()
for details.)
-
Handing off live coroutine objects between coroutine runners¶
Internally, Python’s async/await syntax is built around the idea of
“coroutine objects” and “coroutine runners”. A coroutine object
represents the state of an async callstack. But by itself, this is
just a static object that sits there. If you want it to do anything,
you need a coroutine runner to push it forward. Every Trio task has an
associated coroutine object (see Task.coro
), and the Trio
scheduler acts as their coroutine runner.
But of course, Trio isn’t the only coroutine runner in Python –
asyncio
has one, other event loops have them, you can even
define your own.
And in some very, very unusual circumstances, it even makes sense to transfer a single coroutine object back and forth between different coroutine runners. That’s what this section is about. This is an extremely exotic use case, and assumes a lot of expertise in how Python async/await works internally. For motivating examples, see trio-asyncio issue #42, and trio issue #649. For more details on how coroutines work, we recommend André Caron’s A tale of event loops, or going straight to PEP 492 for the full details.
-
await
trio.lowlevel.
permanently_detach_coroutine_object
(final_outcome)¶ Permanently detach the current task from the Trio scheduler.
Normally, a Trio task doesn’t exit until its coroutine object exits. When you call this function, Trio acts like the coroutine object just exited and the task terminates with the given outcome. This is useful if you want to permanently switch the coroutine object over to a different coroutine runner.
When the calling coroutine enters this function it’s running under Trio, and when the function returns it’s running under the foreign coroutine runner.
You should make sure that the coroutine object has released any Trio-specific resources it has acquired (e.g. nurseries).
- Parameters
final_outcome (outcome.Outcome) – Trio acts as if the current task exited with the given return value or exception.
Returns or raises whatever value or exception the new coroutine runner uses to resume the coroutine.
-
await
trio.lowlevel.
temporarily_detach_coroutine_object
(abort_func)¶ Temporarily detach the current coroutine object from the Trio scheduler.
When the calling coroutine enters this function it’s running under Trio, and when the function returns it’s running under the foreign coroutine runner.
The Trio
Task
will continue to exist, but will be suspended until you usereattach_detached_coroutine_object()
to resume it. In the mean time, you can use another coroutine runner to schedule the coroutine object. In fact, you have to – the function doesn’t return until the coroutine is advanced from outside.Note that you’ll need to save the current
Task
object to later resume; you can retrieve it withcurrent_task()
. You can also use thisTask
object to retrieve the coroutine object – seeTask.coro
.- Parameters
abort_func – Same as for
wait_task_rescheduled()
, except that it must returnAbort.FAILED
. (If it returnedAbort.SUCCEEDED
, then Trio would attempt to reschedule the detached task directly without going throughreattach_detached_coroutine_object()
, which would be bad.) Yourabort_func
should still arrange for whatever the coroutine object is doing to be cancelled, and then reattach to Trio and call theraise_cancel
callback, if possible.
Returns or raises whatever value or exception the new coroutine runner uses to resume the coroutine.
-
await
trio.lowlevel.
reattach_detached_coroutine_object
(task, yield_value)¶ Reattach a coroutine object that was detached using
temporarily_detach_coroutine_object()
.When the calling coroutine enters this function it’s running under the foreign coroutine runner, and when the function returns it’s running under Trio.
This must be called from inside the coroutine being resumed, and yields whatever value you pass in. (Presumably you’ll pass a value that will cause the current coroutine runner to stop scheduling this task.) Then the coroutine is resumed by the Trio scheduler at the next opportunity.