Actually.... this doesn't need help from Trio, if we're clever. It's not the most elegant thing, but we can start a timer thread running, and then if the timeout expires it can call back into Trio. Of course we'll also want to be able to kill it early, so something like, a thread blocked in select would work.

def timeout_thread(wake_sock, timeout, trio_token, cancel_scope):
    readable, _, _ = select.select([wake_sock], [], [], timeout)
    if not readable:
        # we were woken by the timeout expiring
        try:
            trio_token.run_sync_soon(cancel_scope.cancel)
        except trio.RunFinishedError:
            pass

async with trio.open_nursery() as nursery:
    a, b = socket.socketpair()  # blocking sockets
    trio_token = trio.hazmat.current_trio_token()
    try:
        with trio.open_cancel_scope() as timeout_cancel_scope:
            nursery.start_soon(trio.run_sync_in_worker_thread, timeout_thread, b, timeout, trio_token, timeout_cancel_scope, limiter=UNLIMITED)
            # ... do actual test here ...
    finally:
        a.send(b"x")
        if timeout_cancel_scope.cancelled_caught:
            raise ...

This does require that the Trio scheduler be functioning properly, but so would a more intrusive version baked into the Trio scheduler. This also has the advantage that by being in a third-party library we have a lot of flexibility to adjust the response to the timeout however we want – e.g. instead of just cancelling the test, we could have it capture and print a snapshot of the task tree with stacktraces. Or try cancelling after X seconds, and then if the test is still running after another Y seconds (i.e. it's ignoring the cancellation), use harsher methods to disable pytest's output capturing, dump debugging information to the screen, and then call os._exit() to crash the process.

Uh oh!

Add test timeout support #53

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions