# 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() → RunStatistics

  Returns an object containing run-loop-level debugging information:

class trio.lowlevel.RunStatistics

  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 be `inf` if there are no pending deadlines.

  - `run_sync_soon_queue_size` (int): The number of unprocessed callbacks queued via `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.

### Checking for Trio

If you want to interact with an active Trio run – perhaps you need to know the `current_time()` or the `current_task()` – then Trio needs to have certain state available to it or else you will get a `RuntimeError("must be called from async context")`. This requires that you either be:

- indirectly inside (and on the same thread as) a call to `trio.run()`, for run-level information such as the `current_time()` or `current_clock()`; or

- indirectly inside a Trio task, for task-level information such as the `current_task()` or `current_effective_deadline()`.

Internally, this state is provided by thread-local variables tracking the current run and the current task. Sometimes, it’s useful to know in advance whether a call will fail or to have dynamic information for safeguards against running something inside or outside Trio. To do so, call `trio.lowlevel.in_trio_run()` or `trio.lowlevel.in_trio_task()`, which will provide answers according to the following table.

situation | `trio.lowlevel.in_trio_run()` | `trio.lowlevel.in_trio_task()`
--- | --- | ---
inside a Trio-flavored async function | `True` | `True`
in a thread without an active call to `trio.run()` | `False` | `False`
in a guest run’s host loop | `True` | `False`
inside an instrument call | `True` | depends
in a thread created by `trio.to_thread.run_sync()` | `False` | `False`
inside an abort function | `True` | `True`

trio.lowlevel.in_trio_run() → bool

  Check whether we are in a Trio run. This returns `True` if and only if `current_time()` will succeed.

  See also the discussion of differing ways of detecting Trio.

trio.lowlevel.in_trio_task() → bool

  Check whether we are in a Trio task. This returns `True` if and only if `current_task()` will succeed.

  See also the discussion of differing ways of detecting Trio.

### The current clock

trio.lowlevel.current_clock() → Clock

  Returns the current `Clock`.

### 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: Instrument) → None

  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: Instrument) → None

  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: float) → None

    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() → None

    Called just before `trio.run()` returns.

  after_task_step(task: Task) → None

    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: float) → None

    Called before blocking to wait for I/O readiness.

    Parameters:

      timeout (float) – The number of seconds we are willing to wait.

  before_run() → None

    Called at the beginning of `trio.run()`.

  before_task_step(task: Task) → None

    Called immediately before we resume running the given task.

    Parameters:

      task (trio.lowlevel.Task) – The task that is about to run.

  task_exited(task: Task) → None

    Called when the given task exits.

    Parameters:

      task (trio.lowlevel.Task) – The finished task.

  task_scheduled(task: Task) → None

    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: Task) → None

    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 process spawning

await trio.lowlevel.open_process(command: str | bytes | os.PathLike | Sequence[str | bytes | os.PathLike], *, stdin: int | HasFileno | None = None, stdout: int | HasFileno | None = None, stderr: int | HasFileno | None = None, **options: object) → Process

  Execute a child program in a new process.

  After construction, you can interact with the child process by writing data to its `stdin` stream (a `SendStream`), reading data from its `stdout` and/or `stderr` streams (both `ReceiveStream`s), sending it signals using `terminate`, `kill`, or `send_signal`, and waiting for it to exit using `wait`. See `trio.Process` for details.

  Each standard stream is only available if you specify that a pipe should be created for it. For example, if you pass `stdin=subprocess.PIPE`, you can write to the `stdin` stream, else `stdin` will be `None`.

  Unlike `trio.run_process`, this function doesn’t do any kind of automatic management of the child process. It’s up to you to implement whatever semantics you want.

  Parameters:

    - command – The command to run. Typically this is a sequence of strings or bytes such as `['ls', '-l', 'directory with spaces']`, where the first element names the executable to invoke and the other elements specify its arguments. With `shell=True` in the `**options`, or on Windows, `command` can be a string or bytes, which will be parsed following platform-dependent quoting rules. In all cases `command` can be a path or a sequence of paths.

    - stdin – Specifies what the child process’s standard input stream should connect to: output written by the parent (`subprocess.PIPE`), nothing (`subprocess.DEVNULL`), or an open file (pass a file descriptor or something whose `fileno` method returns one). If `stdin` is unspecified, the child process will have the same standard input stream as its parent.

    - stdout – Like `stdin`, but for the child process’s standard output stream.

    - stderr – Like `stdin`, but for the child process’s standard error stream. An additional value `subprocess.STDOUT` is supported, which causes the child’s standard output and standard error messages to be intermixed on a single standard output stream, attached to whatever the `stdout` option says to attach it to.

    - **options – Other general subprocess options are also accepted.

  Returns:

    A new `trio.Process` object.

  Raises:

    OSError – if the process spawning fails, for example because the specified command could not be found.

## 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 integer `SOCKET` handle, or else an object with a `.fileno()` method which returns an integer `SOCKET` handle. File descriptors aren’t supported, and neither are handles that refer to anything besides a `SOCKET`.

  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 of `obj`.

  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` or `wait_writable` calls on the given object to immediately wake up and raise `ClosedResourceError`.

  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:

  1. Explicitly mark the object as closed, so that any new attempts to use it will abort before they start.

  2. Call `notify_closing` to wake up any already-existing users.

  3. 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: `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 across `os.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 use `FdStream(os.dup(sys.stdin.fileno()))` 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 of `sys.stdin` until the stream returned by `FdStream` is closed. See issue #174 for a discussion of the challenges involved in relaxing this restriction.

  Warning

  one specific consequence of non-blocking mode applying to the entire open file description is that when your program is run with multiple standard streams connected to a TTY (as in a terminal emulator), all of the streams become non-blocking when you construct an `FdStream` for any of them. For example, if you construct an `FdStream` for standard input, you might observe Python loggers begin to fail with `BlockingIOError`.

  Parameters:

    fd (int) – The fd to be wrapped.

  Returns:

    A new `FdStream` object.

### Kqueue-specific API

TODO: these are implemented, but are currently more of a sketch than anything real. See #26.

trio.lowlevel.current_kqueue()

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26.

await trio.lowlevel.wait_kevent(ident, filter, abort_func)

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26.

with trio.lowlevel.monitor_kevent(ident, filter) as queue

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26.

### 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)

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

await trio.lowlevel.wait_overlapped(handle, lpOverlapped)

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

await trio.lowlevel.write_overlapped(handle, data)

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

await trio.lowlevel.readinto_overlapped(handle, data)

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

trio.lowlevel.current_iocp()

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

with trio.lowlevel.monitor_completion_key() as queue

  TODO: these are implemented, but are currently more of a sketch than anything real. See #26 and #52.

## Global state: system tasks and run-local variables

class trio.lowlevel.RunVar(name: str, default | type[~trio._core._local._NoValue] = ...)

  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 to `trio.run()` rather than a single task.

trio.lowlevel.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 `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 use `disable_ki_protection()` explicitly (and come up with some plan for what to do with a `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 `trio.run()`, after the main task and all system tasks have exited, the system nursery becomes closed. At this point, new calls to `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 `TrioToken.run_sync_soon()` at the right moment.

  Parameters:

    - async_fn – An async callable.

    - args – Positional arguments for `async_fn`. If you want to pass keyword arguments, use `functools.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 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:

    the newly spawned task

  Return type:

    Task

## 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:

  1. 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()` and `trio.from_thread` use to communicate with worker threads, that `trio.open_signal_receiver` uses to receive notifications about signals, and so forth.

  2. Each call to `trio.run()` has exactly one associated `TrioToken` object, so you can use it to identify a particular call.

  run_sync_soon(sync_fn: Callable[[Unpack[PosArgsT]], object], *args: Unpack[PosArgsT], idempotent: bool = False) → None

    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 want `sync_fn` to be interruptible by control-C, then you need to use `disable_ki_protection()` explicitly.

    - If `sync_fn` raises an exception, then it’s converted into a `TrioInternalError` and all tasks are cancelled. You should be careful that `sync_fn` doesn’t crash.

    All calls with `idempotent=False` are processed in strict first-in first-out order.

    If `idempotent=True`, then `sync_fn` and `args` 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` and `idempotent=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 before `trio.run()` exits.)

trio.lowlevel.current_trio_token() → TrioToken

  Retrieve the `TrioToken` for the current call to `trio.run()`.

## Spawning threads

trio.lowlevel.start_thread_soon(fn: Callable[[], RetT], deliver: Callable[[outcome.Outcome[RetT]], object], name: str | None = None) → None

  Runs `deliver(outcome.capture(fn))` in a worker thread.

  Generally `fn` does some blocking work, and `deliver` delivers the result back to whoever is interested.

  This is a low-level, no-frills interface, very similar to using `threading.Thread` to spawn a thread directly. The main difference is that this function tries to reuse threads when possible, so it can be a bit faster than `threading.Thread`.

  Worker threads have the `daemon` flag set, which means that if your main thread exits, worker threads will automatically be killed. If you want to make sure that your `fn` runs to completion, then you should make sure that the main thread remains alive until `deliver` is called.

  It is safe to call this function simultaneously from multiple threads.

  Parameters:

    - fn (sync function) – Performs arbitrary blocking work.

    - deliver (sync function) – Takes the `outcome.Outcome` of `fn`, and delivers it. Must not block.

  Because worker threads are cached and reused for multiple calls, neither function should mutate thread-level state, like `threading.local` objects – or if they do, they should be careful to revert their changes before returning.

  Note

  The split between `fn` and `deliver` serves two purposes. First, it’s convenient, since most callers need something like this anyway.

  Second, it avoids a small race condition that could cause too many threads to be spawned. Consider a program that wants to run several jobs sequentially on a thread, so the main thread submits a job, waits for it to finish, submits another job, etc. In theory, this program should only need one worker thread. But what could happen is:

  1. Worker thread: First job finishes, and calls `deliver`.

  2. Main thread: receives notification that the job finished, and calls `start_thread_soon`.

  3. Main thread: sees that no worker threads are marked idle, so spawns a second worker thread.

  4. Original worker thread: marks itself as idle.

  To avoid this, threads mark themselves as idle before calling `deliver`.

  Is this potential extra thread a major problem? Maybe not, but it’s easy enough to avoid, and we figure that if the user is trying to limit how many threads they’re using then it’s polite to respect that.

## 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:

- The top-level function in regular user tasks is unprotected.

- The top-level function in system tasks is protected.

- If a function doesn’t specify otherwise, then it inherits the protection state of its caller.

This 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 by `KeyboardInterrupt` 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 uses `TrioToken.run_sync_soon()` to get into the Trio thread. `run_sync_soon()` callbacks are run with `KeyboardInterrupt` protection enabled, and `trio.from_thread.run()` takes advantage of this to safely set up the machinery for sending a response back to the original thread, but then uses `disable_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 by `KeyboardInterrupt`. (Though if it contains any checkpoints, then it can still receive `KeyboardInterrupt` 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 a `Lock`, where a poorly-timed `KeyboardInterrupt` could leave the lock in an inconsistent state and cause a deadlock.

  Since KeyboardInterrupt protection is tracked per code object, any attempt to conditionally protect the same block of code in different ways is unlikely to behave how you expect. If you try to conditionally protect a closure, it will be unconditionally protected instead:

      def example(protect: bool) -> bool:
          def inner() -> bool:
              return trio.lowlevel.currently_ki_protected()
          if protect:
              inner = trio.lowlevel.enable_ki_protection(inner)
          return inner()

      async def amain():
          assert example(False) == False
          assert example(True) == True  # once protected ...
          assert example(False) == True  # ... always protected

      trio.run(amain)

  If you really need conditional protection, you can achieve it by giving each KI-protected instance of the closure its own code object:

      def example(protect: bool) -> bool:
          def inner() -> bool:
              return trio.lowlevel.currently_ki_protected()
          if protect:
              inner.__code__ = inner.__code__.replace()
              inner = trio.lowlevel.enable_ki_protection(inner)
          return inner()

      async def amain():
          assert example(False) == False
          assert example(True) == True
          assert example(False) == False

      trio.run(amain)

  (This isn’t done by default because it carries some memory overhead and reduces the potential for specializing optimizations in recent versions of CPython.)

trio.lowlevel.currently_ki_protected() → bool

  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 for `assert`s and unit tests.

  Returns:

    True if protection is enabled, and False otherwise.

  Return type:

    bool

## Sleeping and waking

### Wait queue abstraction

class trio.lowlevel.ParkingLot

  A fair wait queue with cancellation and requeuing.

  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, and `if parking_lot: ...` to check whether there are any parked tasks.

  broken_by: list[Task]
  break_lot(task: Task | None = None) → None

    Break this lot, with `task` noted as the task that broke it.

    This causes all parked tasks to raise an error, and any future tasks attempting to park to error. Unpark & repark become no-ops as the parking lot is empty.

    The error raised contains a reference to the task sent as a parameter. The task is also saved in the parking lot in the `broken_by` attribute.

  await park() → None

    Park the current task until woken by a call to `unpark()` or `unpark_all()`.

    Raises:

      BrokenResourceError – if attempting to park in a broken lot, or the lot breaks before we get to unpark.

  repark(new_lot: ParkingLot, *, count: int | float = 1) → None

    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|math.inf) – the number of tasks to move.

  repark_all(new_lot: ParkingLot) → None

    Move all parked tasks from one `ParkingLot` object to another.

    See `repark()` for details.

  statistics() → ParkingLotStatistics

    Return an object containing debugging information.

    Currently the following fields are defined:

    - `tasks_waiting`: The number of tasks blocked on this lot’s `park()` method.

  unpark(*, count: int | float = 1) → list[Task]

    Unpark one or more tasks.

    This wakes up `count` tasks that are blocked in `park()`. If there are fewer than `count` tasks parked, then wakes as many tasks are available and then returns successfully.

    Parameters:

      count (int | math.inf) – the number of tasks to unpark.

  unpark_all() → list[Task]

    Unpark all parked tasks.

class trio.lowlevel.ParkingLotStatistics(tasks_waiting: int)

  An object containing debugging information for a ParkingLot.

  Currently, the following fields are defined:

  - `tasks_waiting` (int): The number of tasks blocked on this lot’s `trio.lowlevel.ParkingLot.park()` method.

trio.lowlevel.add_parking_lot_breaker(task: Task, lot: ParkingLot) → None

  Register a task as a breaker for a lot. See `ParkingLot.break_lot()`.

  Raises:

    trio.BrokenResourceError – if the task has already exited.

trio.lowlevel.remove_parking_lot_breaker(task: Task, lot: ParkingLot) → None

  Deregister a task as a breaker for a lot. See `ParkingLot.break_lot()`

### Low-level checkpoint functions

await trio.lowlevel.checkpoint() → None

  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 calling `checkpoint()`.)

The next two functions are used together to make up a checkpoint:

await trio.lowlevel.checkpoint_if_cancelled() → None

  Issue a checkpoint if the calling context has been cancelled.

  Equivalent to (but potentially more efficient than):

      if trio.current_effective_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() → None

  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. Using `cancel_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 by `attempt_operation` or `wait_for_operation_to_be_ready`, by keeping the `while True:` loop outside of the `except 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: Callable[[Callable[[], NoReturn]], Abort]) → Any

  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 in `abort_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:

  1. “Someone” calls `reschedule()` on the current task, and `wait_task_rescheduled()` returns or raises whatever value or error was passed to `reschedule()`.

  2. 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 return `Abort.SUCCEEDED`, in which case the task will automatically be rescheduled with an appropriate `Cancelled` 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 calls `reschedule()`.

     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 the `abort_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 the `raise_cancel` callback passed to `abort_func` to raise a `Cancelled` (or possibly `KeyboardInterrupt`) 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 to `wait_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 than `Abort.SUCCEEDED` or `Abort.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 calling `reschedule()` 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(*values)

  `enum.Enum` used as the return value from abort functions.

  See `wait_task_rescheduled()` for details.

  SUCCEEDED
  FAILED

trio.lowlevel.reschedule(task: Task, next_send: outcome.Outcome[object] = <object object>) → None

  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 to `wait_task_rescheduled()`. (And when counting, keep in mind that returning `Abort.SUCCEEDED` from an abort callback is equivalent to calling `reschedule()` 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):
            # We might have to try several times to acquire the lock.
            while self._held:
                # Someone else has the lock, so we have to wait.
                task = trio.lowlevel.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)
                # At this point the lock was released -- but someone else
                # might have swooped in and taken it again before we
                # woke up. So we loop around to check the 'while' condition
                # again.
            # if we reach this point, it means that the 'while' condition
            # has just failed, so we know no-one is holding the lock, and
            # we can take it.
            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 called `current_task()`.

  Return type:

    Task

class trio.lowlevel.Task

  A `Task` object represents a concurrent “thread” of execution. It has no public constructor; Trio internally creates a `Task` object for each call to `nursery.start(...)` or `nursery.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 this `Task` is running, but can be overridden by passing `name=` to `start` or `start_soon`.

  coro

    This task’s coroutine object.

  for ... in iter_await_frames() → Iterator[tuple[types.FrameType, int]]

    Iterates recursively over the coroutine-like objects this task is waiting on, yielding the frame and line number at each frame.

    This is similar to `traceback.walk_stack` in a synchronous context. Note that `traceback.walk_stack` returns frames from the bottom of the call stack to the top, while this function starts from `Task.coro` and works it way down.

    Example usage: extracting a stack trace:

        import traceback

        def print_stack_for_task(task):
            ss = traceback.StackSummary.extract(task.iter_await_frames())
            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.

  eventual_parent_nursery

    The nursery this task will be inside after it calls `task_status.started()`.

    If this task has already called `started()`, or if it was not spawned using `nursery.start()`, then its `eventual_parent_nursery` is `None`.

  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. (See `wait_task_rescheduled()` for details.)

## Using “guest mode” to run Trio on top of other event loops

### What is “guest mode”?

An event loop acts as a central coordinator to manage all the IO happening in your program. Normally, that means that your application has to pick one event loop, and use it for everything. But what if you like Trio, but also need to use a framework like Qt or PyGame that has its own event loop? Then you need some way to run both event loops at once.

It is possible to combine event loops, but the standard approaches all have significant downsides:

- Polling: this is where you use a busy-loop to manually check for IO on both event loops many times per second. This adds latency, and wastes CPU time and electricity.

- Pluggable IO backends: this is where you reimplement one of the event loop APIs on top of the other, so you effectively end up with just one event loop. This requires a significant amount of work for each pair of event loops you want to integrate, and different backends inevitably end up with inconsistent behavior, forcing users to program against the least-common-denominator. And if the two event loops expose different feature sets, it may not even be possible to implement one in terms of the other.

- Running the two event loops in separate threads: This works, but most event loop APIs aren’t thread-safe, so in this approach you need to keep careful track of which code runs on which event loop, and remember to use explicit inter-thread messaging whenever you interact with the other loop – or else risk obscure race conditions and data corruption.

That’s why Trio offers a fourth option: guest mode. Guest mode lets you execute `trio.run` on top of some other “host” event loop, like Qt. Its advantages are:

- Efficiency: guest mode is event-driven instead of using a busy-loop, so it has low latency and doesn’t waste electricity.

- No need to think about threads: your Trio code runs in the same thread as the host event loop, so you can freely call sync Trio APIs from the host, and call sync host APIs from Trio. For example, if you’re making a GUI app with Qt as the host loop, then making a cancel button and connecting it to a `trio.CancelScope` is as easy as writing:

      # Trio code can create Qt objects without any special ceremony...
      my_cancel_button = QPushButton("Cancel")
      # ...and Qt can call back to Trio just as easily
      my_cancel_button.clicked.connect(my_cancel_scope.cancel)

  (For async APIs, it’s not that simple, but you can use sync APIs to build explicit bridges between the two worlds, e.g. by passing async functions and their results back and forth through queues.)

- Consistent behavior: guest mode uses the same code as regular Trio: the same scheduler, same IO code, same everything. So you get the full feature set and everything acts the way you expect.

- Simple integration and broad compatibility: pretty much every event loop offers some threadsafe “schedule a callback” operation, and that’s all you need to use it as a host loop.

### Really? How is that possible?

Note

You can use guest mode without reading this section. It’s included for those who enjoy understanding how things work.

All event loops have the same basic structure. They loop through two operations, over and over:

1. Wait for the operating system to notify them that something interesting has happened, like data arriving on a socket or a timeout passing. They do this by invoking a platform-specific `sleep_until_something_happens()` system call – `select`, `epoll`, `kqueue`, `GetQueuedCompletionEvents`, etc.

2. Run all the user tasks that care about whatever happened, then go back to step 1.

The problem here is step 1. Two different event loops on the same thread can take turns running user tasks in step 2, but when they’re idle and nothing is happening, they can’t both invoke their own `sleep_until_something_happens()` function at the same time.

The “polling” and “pluggable backend” strategies solve this by hacking the loops so both step 1s can run at the same time in the same thread. Keeping everything in one thread is great for step 2, but the step 1 hacks create problems.

The “separate threads” strategy solves this by moving both steps into separate threads. This makes step 1 work, but the downside is that now the user tasks in step 2 are running separate threads as well, so users are forced to deal with inter-thread coordination.

The idea behind guest mode is to combine the best parts of each approach: we move Trio’s step 1 into a separate worker thread, while keeping Trio’s step 2 in the main host thread. This way, when the application is idle, both event loops do their `sleep_until_something_happens()` at the same time in their own threads. But when the app wakes up and your code is actually running, it all happens in a single thread. The threading trickiness is all handled transparently inside Trio.

Concretely, we unroll Trio’s internal event loop into a chain of callbacks, and as each callback finishes, it schedules the next callback onto the host loop or a worker thread as appropriate. So the only thing the host loop has to provide is a way to schedule a callback onto the main thread from a worker thread.

Coordinating between Trio and the host loop does add some overhead. The main cost is switching in and out of the background thread, since this requires cross-thread messaging. This is cheap (on the order of a few microseconds, assuming your host loop is implemented efficiently), but it’s not free.

But, there’s a nice optimization we can make: we only need the thread when our `sleep_until_something_happens()` call actually sleeps, that is, when the Trio part of your program is idle and has nothing to do. So before we switch into the worker thread, we double-check whether we’re idle, and if not, then we skip the worker thread and jump directly to step 2. This means that your app only pays the extra thread-switching penalty at moments when it would otherwise be sleeping, so it should have minimal effect on your app’s overall performance.

The total overhead will depend on your host loop, your platform, your application, etc. But we expect that in most cases, apps running in guest mode should only be 5-10% slower than the same code using `trio.run`. If you find that’s not true for your app, then please let us know and we’ll see if we can fix it!

### Implementing guest mode for your favorite event loop

Let’s walk through what you need to do to integrate Trio’s guest mode with your favorite event loop. Treat this section like a checklist.

Getting started: The first step is to get something basic working. Here’s a minimal example of running Trio on top of asyncio, that you can use as a model:

    import asyncio
    import trio

    # A tiny Trio program
    async def trio_main():
        for _ in range(5):
            print("Hello from Trio!")
            # This is inside Trio, so we have to use Trio APIs
            await trio.sleep(1)
        return "trio done!"

    # The code to run it as a guest inside asyncio
    async def asyncio_main():
        asyncio_loop = asyncio.get_running_loop()

        def run_sync_soon_threadsafe(fn):
            asyncio_loop.call_soon_threadsafe(fn)

        def done_callback(trio_main_outcome):
            print(f"Trio program ended with: {trio_main_outcome}")

        # This is where the magic happens:
        trio.lowlevel.start_guest_run(
            trio_main,
            run_sync_soon_threadsafe=run_sync_soon_threadsafe,
            done_callback=done_callback,
        )

        # Let the host loop run for a while to give trio_main time to
        # finish. (WARNING: This is a hack. See below for better
        # approaches.)
        #
        # This function is in asyncio, so we have to use asyncio APIs.
        await asyncio.sleep(10)

    asyncio.run(asyncio_main())

You can see we’re using asyncio-specific APIs to start up a loop, and then we call `trio.lowlevel.start_guest_run`. This function is very similar to `trio.run`, and takes all the same arguments. But it has two differences:

First, instead of blocking until `trio_main` has finished, it schedules `trio_main` to start running on top of the host loop, and then returns immediately. So `trio_main` is running in the background – that’s why we have to sleep and give it time to finish.

And second, it requires two extra keyword arguments: `run_sync_soon_threadsafe`, and `done_callback`.

For `run_sync_soon_threadsafe`, we need a function that takes a synchronous callback, and schedules it to run on your host loop. And this function needs to be “threadsafe” in the sense that you can safely call it from any thread. So you need to figure out how to write a function that does that using your host loop’s API. For asyncio, this is easy because `call_soon_threadsafe` does exactly what we need; for your loop, it might be more or less complicated.

For `done_callback`, you pass in a function that Trio will automatically invoke when the Trio run finishes, so you know it’s done and what happened. For this basic starting version, we just print the result; in the next section we’ll discuss better alternatives.

At this stage you should be able to run a simple Trio program inside your host loop. Now we’ll turn that prototype into something solid.

Loop lifetimes: One of the trickiest things in most event loops is shutting down correctly. And having two event loops makes this even harder!

If you can, we recommend following this pattern:

- Start up your host loop

- Immediately call `start_guest_run` to start Trio

- When Trio finishes and your `done_callback` is invoked, shut down the host loop

- Make sure that nothing else shuts down your host loop

This way, your two event loops have the same lifetime, and your program automatically exits when your Trio function finishes.

Here’s how we’d extend our asyncio example to implement this pattern:

    # Improved version, that shuts down properly after Trio finishes
    async def asyncio_main():
        asyncio_loop = asyncio.get_running_loop()

        def run_sync_soon_threadsafe(fn):
            asyncio_loop.call_soon_threadsafe(fn)

        # Revised 'done' callback: set a Future
        done_fut = asyncio_loop.create_future()
        def done_callback(trio_main_outcome):
            done_fut.set_result(trio_main_outcome)

        trio.lowlevel.start_guest_run(
            trio_main,
            run_sync_soon_threadsafe=run_sync_soon_threadsafe,
            done_callback=done_callback,
        )

        # Wait for the guest run to finish
        trio_main_outcome = await done_fut
        # Pass through the return value or exception from the guest run
        return trio_main_outcome.unwrap()

And then you can encapsulate all this machinery in a utility function that exposes a `trio.run`-like API, but runs both loops together:

    def trio_run_with_asyncio(trio_main, *args, **trio_run_kwargs):
        async def asyncio_main():
            # same as above
            ...

        return asyncio.run(asyncio_main())

Technically, it is possible to use other patterns. But there are some important limitations you have to respect:

- You must let the Trio program run to completion. Many event loops let you stop the event loop at any point, and any pending callbacks/tasks/etc. just… don’t run. Trio follows a more structured system, where you can cancel things, but the code always runs to completion, so `finally` blocks run, resources are cleaned up, etc. If you stop your host loop early, before the `done_callback` is invoked, then that cuts off the Trio run in the middle without a chance to clean up. This can leave your code in an inconsistent state, and will definitely leave Trio’s internals in an inconsistent state, which will cause errors if you try to use Trio again in that thread.

  Some programs need to be able to quit at any time, for example in response to a GUI window being closed or a user selecting a “Quit” from a menu. In these cases, we recommend wrapping your whole program in a `trio.CancelScope`, and cancelling it when you want to quit.

- Each host loop can only have one `start_guest_run` at a time. If you try to start a second one, you’ll get an error. If you need to run multiple Trio functions at the same time, then start up a single Trio run, open a nursery, and then start your functions as child tasks in that nursery.

- Unless you or your host loop register a handler for `signal.SIGINT` before starting Trio (this is not common), then Trio will take over delivery of `KeyboardInterrupt`s. And since Trio can’t tell which host code is safe to interrupt, it will only deliver `KeyboardInterrupt` into the Trio part of your code. This is fine if your program is set up to exit when the Trio part exits, because the `KeyboardInterrupt` will propagate out of Trio and then trigger the shutdown of your host loop, which is just what you want.

Given these constraints, we think the simplest approach is to always start and stop the two loops together.

Signal management: “Signals” are a low-level inter-process communication primitive. When you hit control-C to kill a program, that uses a signal. Signal handling in Python has a lot of moving parts. One of those parts is `signal.set_wakeup_fd`, which event loops use to make sure that they wake up when a signal arrives so they can respond to it. (If you’ve ever had an event loop ignore you when you hit control-C, it was probably because they weren’t using `signal.set_wakeup_fd` correctly.)

But, only one event loop can use `signal.set_wakeup_fd` at a time. And in guest mode that can cause problems: Trio and the host loop might start fighting over who’s using `signal.set_wakeup_fd`.

Some event loops, like asyncio, won’t work correctly unless they win this fight. Fortunately, Trio is a little less picky: as long as someone makes sure that the program wakes up when a signal arrives, it should work correctly. So if your host loop wants `signal.set_wakeup_fd`, then you should disable Trio’s `signal.set_wakeup_fd` support, and then both loops will work correctly.

On the other hand, if your host loop doesn’t use `signal.set_wakeup_fd`, then the only way to make everything work correctly is to enable Trio’s `signal.set_wakeup_fd` support.

By default, Trio assumes that your host loop doesn’t use `signal.set_wakeup_fd`. It does try to detect when this creates a conflict with the host loop, and print a warning – but unfortunately, by the time it detects it, the damage has already been done. So if you’re getting this warning, then you should disable Trio’s `signal.set_wakeup_fd` support by passing `host_uses_signal_set_wakeup_fd=True` to `start_guest_run`.

If you aren’t seeing any warnings with your initial prototype, you’re probably fine. But the only way to be certain is to check your host loop’s source. For example, asyncio may or may not use `signal.set_wakeup_fd` depending on the Python version and operating system.

A small optimization: Finally, consider a small optimization. Some event loops offer two versions of their “call this function soon” API: one that can be used from any thread, and one that can only be used from the event loop thread, with the latter being cheaper. For example, asyncio has both `call_soon_threadsafe` and `call_soon`.

If you have a loop like this, then you can also pass a `run_sync_soon_not_threadsafe=...` kwarg to `start_guest_run`, and Trio will automatically use it when appropriate.

If your loop doesn’t have a split like this, then don’t worry about it; `run_sync_soon_not_threadsafe=` is optional. (If it’s not passed, then Trio will just use your threadsafe version in all cases.)

That’s it! If you’ve followed all these steps, you should now have a cleanly-integrated hybrid event loop. Go make some cool GUIs/games/whatever!

### Limitations

In general, almost all Trio features should work in guest mode. The exception is features which rely on Trio having a complete picture of everything that your program is doing, since obviously, it can’t control the host loop or see what it’s doing.

Custom clocks can be used in guest mode, but they only affect Trio timeouts, not host loop timeouts. And the autojump clock and related `trio.testing.wait_all_tasks_blocked` can technically be used in guest mode, but they’ll only take Trio tasks into account when decided whether to jump the clock or whether all tasks are blocked.

### Reference

trio.lowlevel.start_guest_run(async_fn: Callable[..., Awaitable[RetT]], *args: object, run_sync_soon_threadsafe: Callable[[Callable[[], object]], object], done_callback: Callable[[outcome.Outcome[RetT]], object], run_sync_soon_not_threadsafe: Callable[[Callable[[], object]], object] | None = None, host_uses_signal_set_wakeup_fd: bool = False, clock: Clock | None = None, instruments: Sequence[Instrument] = (), restrict_keyboard_interrupt_to_checkpoints: bool = False, strict_exception_groups: bool = True) → None

  Start a “guest” run of Trio on top of some other “host” event loop.

  Each host loop can only have one guest run at a time.

  You should always let the Trio run finish before stopping the host loop; if not, it may leave Trio’s internal data structures in an inconsistent state. You might be able to get away with it if you immediately exit the program, but it’s safest not to go there in the first place.

  Generally, the best way to do this is wrap this in a function that starts the host loop and then immediately starts the guest run, and then shuts down the host when the guest run completes.

  Once `start_guest_run()` returns successfully, the guest run has been set up enough that you can invoke sync-colored Trio functions such as `current_time()`, `spawn_system_task()`, and `current_trio_token()`. If a `TrioInternalError` occurs during this early setup of the guest run, it will be raised out of `start_guest_run()`. All other errors, including all errors raised by the async_fn, will be delivered to your done_callback at some point after `start_guest_run()` returns successfully.

  Parameters:

    - run_sync_soon_threadsafe –

      An arbitrary callable, which will be passed a function as its sole argument:

          def my_run_sync_soon_threadsafe(fn):
              ...

      This callable should schedule `fn()` to be run by the host on its next pass through its loop. Must support being called from arbitrary threads.

    - done_callback –

      An arbitrary callable:

          def my_done_callback(run_outcome):
              ...

      When the Trio run has finished, Trio will invoke this callback to let you know. The argument is an `outcome.Outcome`, reporting what would have been returned or raised by `trio.run`. This function can do anything you want, but commonly you’ll want it to shut down the host loop, unwrap the outcome, etc.

    - run_sync_soon_not_threadsafe – Like `run_sync_soon_threadsafe`, but will only be called from inside the host loop’s main thread. Optional, but if your host loop allows you to implement this more efficiently than `run_sync_soon_threadsafe` then passing it will make things a bit faster.

    - host_uses_signal_set_wakeup_fd (bool) – Pass `True` if your host loop uses `signal.set_wakeup_fd`, and `False` otherwise. For more details, see Implementing guest mode for your favorite event loop.

  For the meaning of other arguments, see `trio.run`.

## 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: Outcome[object]) → object

  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: Callable[[Callable[[], NoReturn]], Abort]) → object

  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 use `reattach_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 with `current_task()`. You can also use this `Task` object to retrieve the coroutine object – see `Task.coro`.

  Parameters:

    abort_func – Same as for `wait_task_rescheduled()`, except that it must return `Abort.FAILED`. (If it returned `Abort.SUCCEEDED`, then Trio would attempt to reschedule the detached task directly without going through `reattach_detached_coroutine_object()`, which would be bad.) Your `abort_func` should still arrange for whatever the coroutine object is doing to be cancelled, and then reattach to Trio and call the `raise_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: Task, yield_value: object) → None

  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.

  Parameters:

    - task (Task) – The Trio task object that the current coroutine was detached from.

    - yield_value (object) – The object to yield to the current coroutine runner.