Function wait_all_threads_completed

src/trio/_threads.py:87–112

Wait until no threads are still running tasks. This is intended to be used when testing code with trio.to_thread to make sure no tasks are still making progress in a thread. See the following code for a usage example:: async def wait_all_settled(): while True:

()

async def wait_all_threads_completed() -> None:
    """Wait until no threads are still running tasks.

    This is intended to be used when testing code with trio.to_thread to
    make sure no tasks are still making progress in a thread. See the
    following code for a usage example::

        async def wait_all_settled():
            while True:
                await trio.testing.wait_all_threads_complete()
                await trio.testing.wait_all_tasks_blocked()
                if trio.testing.active_thread_count() == 0:
                    break
    """

    await checkpoint()

    try:
        active_threads_local = _active_threads_local.get()
    except LookupError:
        # If there would have been active threads, the
        # _active_threads_local would have been set
        return

    while active_threads_local.count != 0:
        await active_threads_local.event.wait()


def active_thread_count() -> int: