import pytest
|
import asyncio
|
import contextvars
|
import sys
|
import traceback
|
import queue
|
from functools import partial
|
from math import inf
|
import signal
|
import socket
|
import threading
|
import time
|
import warnings
|
|
import trio
|
import trio.testing
|
from .tutil import gc_collect_harder, buggy_pypy_asyncgens, restore_unraisablehook
|
from ..._util import signal_raise
|
|
# The simplest possible "host" loop.
|
# Nice features:
|
# - we can run code "outside" of trio using the schedule function passed to
|
# our main
|
# - final result is returned
|
# - any unhandled exceptions cause an immediate crash
|
def trivial_guest_run(trio_fn, **start_guest_run_kwargs):
|
todo = queue.Queue()
|
|
host_thread = threading.current_thread()
|
|
def run_sync_soon_threadsafe(fn):
|
if host_thread is threading.current_thread(): # pragma: no cover
|
crash = partial(
|
pytest.fail, "run_sync_soon_threadsafe called from host thread"
|
)
|
todo.put(("run", crash))
|
todo.put(("run", fn))
|
|
def run_sync_soon_not_threadsafe(fn):
|
if host_thread is not threading.current_thread(): # pragma: no cover
|
crash = partial(
|
pytest.fail, "run_sync_soon_not_threadsafe called from worker thread"
|
)
|
todo.put(("run", crash))
|
todo.put(("run", fn))
|
|
def done_callback(outcome):
|
todo.put(("unwrap", outcome))
|
|
trio.lowlevel.start_guest_run(
|
trio_fn,
|
run_sync_soon_not_threadsafe,
|
run_sync_soon_threadsafe=run_sync_soon_threadsafe,
|
run_sync_soon_not_threadsafe=run_sync_soon_not_threadsafe,
|
done_callback=done_callback,
|
**start_guest_run_kwargs,
|
)
|
|
try:
|
while True:
|
op, obj = todo.get()
|
if op == "run":
|
obj()
|
elif op == "unwrap":
|
return obj.unwrap()
|
else: # pragma: no cover
|
assert False
|
finally:
|
# Make sure that exceptions raised here don't capture these, so that
|
# if an exception does cause us to abandon a run then the Trio state
|
# has a chance to be GC'ed and warn about it.
|
del todo, run_sync_soon_threadsafe, done_callback
|
|
|
def test_guest_trivial():
|
async def trio_return(in_host):
|
await trio.sleep(0)
|
return "ok"
|
|
assert trivial_guest_run(trio_return) == "ok"
|
|
async def trio_fail(in_host):
|
raise KeyError("whoopsiedaisy")
|
|
with pytest.raises(KeyError, match="whoopsiedaisy"):
|
trivial_guest_run(trio_fail)
|
|
|
def test_guest_can_do_io():
|
async def trio_main(in_host):
|
record = []
|
a, b = trio.socket.socketpair()
|
with a, b:
|
async with trio.open_nursery() as nursery:
|
|
async def do_receive():
|
record.append(await a.recv(1))
|
|
nursery.start_soon(do_receive)
|
await trio.testing.wait_all_tasks_blocked()
|
|
await b.send(b"x")
|
|
assert record == [b"x"]
|
|
trivial_guest_run(trio_main)
|
|
|
def test_host_can_directly_wake_trio_task():
|
async def trio_main(in_host):
|
ev = trio.Event()
|
in_host(ev.set)
|
await ev.wait()
|
return "ok"
|
|
assert trivial_guest_run(trio_main) == "ok"
|
|
|
def test_host_altering_deadlines_wakes_trio_up():
|
def set_deadline(cscope, new_deadline):
|
cscope.deadline = new_deadline
|
|
async def trio_main(in_host):
|
with trio.CancelScope() as cscope:
|
in_host(lambda: set_deadline(cscope, -inf))
|
await trio.sleep_forever()
|
assert cscope.cancelled_caught
|
|
with trio.CancelScope() as cscope:
|
# also do a change that doesn't affect the next deadline, just to
|
# exercise that path
|
in_host(lambda: set_deadline(cscope, 1e6))
|
in_host(lambda: set_deadline(cscope, -inf))
|
await trio.sleep(999)
|
assert cscope.cancelled_caught
|
|
return "ok"
|
|
assert trivial_guest_run(trio_main) == "ok"
|
|
|
def test_warn_set_wakeup_fd_overwrite():
|
assert signal.set_wakeup_fd(-1) == -1
|
|
async def trio_main(in_host):
|
return "ok"
|
|
a, b = socket.socketpair()
|
with a, b:
|
a.setblocking(False)
|
|
# Warn if there's already a wakeup fd
|
signal.set_wakeup_fd(a.fileno())
|
try:
|
with pytest.warns(RuntimeWarning, match="signal handling code.*collided"):
|
assert trivial_guest_run(trio_main) == "ok"
|
finally:
|
assert signal.set_wakeup_fd(-1) == a.fileno()
|
|
signal.set_wakeup_fd(a.fileno())
|
try:
|
with pytest.warns(RuntimeWarning, match="signal handling code.*collided"):
|
assert (
|
trivial_guest_run(trio_main, host_uses_signal_set_wakeup_fd=False)
|
== "ok"
|
)
|
finally:
|
assert signal.set_wakeup_fd(-1) == a.fileno()
|
|
# Don't warn if there isn't already a wakeup fd
|
with warnings.catch_warnings():
|
warnings.simplefilter("error")
|
assert trivial_guest_run(trio_main) == "ok"
|
|
with warnings.catch_warnings():
|
warnings.simplefilter("error")
|
assert (
|
trivial_guest_run(trio_main, host_uses_signal_set_wakeup_fd=True)
|
== "ok"
|
)
|
|
# If there's already a wakeup fd, but we've been told to trust it,
|
# then it's left alone and there's no warning
|
signal.set_wakeup_fd(a.fileno())
|
try:
|
|
async def trio_check_wakeup_fd_unaltered(in_host):
|
fd = signal.set_wakeup_fd(-1)
|
assert fd == a.fileno()
|
signal.set_wakeup_fd(fd)
|
return "ok"
|
|
with warnings.catch_warnings():
|
warnings.simplefilter("error")
|
assert (
|
trivial_guest_run(
|
trio_check_wakeup_fd_unaltered,
|
host_uses_signal_set_wakeup_fd=True,
|
)
|
== "ok"
|
)
|
finally:
|
assert signal.set_wakeup_fd(-1) == a.fileno()
|
|
|
def test_host_wakeup_doesnt_trigger_wait_all_tasks_blocked():
|
# This is designed to hit the branch in unrolled_run where:
|
# idle_primed=True
|
# runner.runq is empty
|
# events is Truth-y
|
# ...and confirm that in this case, wait_all_tasks_blocked does not get
|
# triggered.
|
def set_deadline(cscope, new_deadline):
|
print(f"setting deadline {new_deadline}")
|
cscope.deadline = new_deadline
|
|
async def trio_main(in_host):
|
async def sit_in_wait_all_tasks_blocked(watb_cscope):
|
with watb_cscope:
|
# Overall point of this test is that this
|
# wait_all_tasks_blocked should *not* return normally, but
|
# only by cancellation.
|
await trio.testing.wait_all_tasks_blocked(cushion=9999)
|
assert False # pragma: no cover
|
assert watb_cscope.cancelled_caught
|
|
async def get_woken_by_host_deadline(watb_cscope):
|
with trio.CancelScope() as cscope:
|
print("scheduling stuff to happen")
|
# Altering the deadline from the host, to something in the
|
# future, will cause the run loop to wake up, but then
|
# discover that there is nothing to do and go back to sleep.
|
# This should *not* trigger wait_all_tasks_blocked.
|
#
|
# So the 'before_io_wait' here will wait until we're blocking
|
# with the wait_all_tasks_blocked primed, and then schedule a
|
# deadline change. The critical test is that this should *not*
|
# wake up 'sit_in_wait_all_tasks_blocked'.
|
#
|
# The after we've had a chance to wake up
|
# 'sit_in_wait_all_tasks_blocked', we want the test to
|
# actually end. So in after_io_wait we schedule a second host
|
# call to tear things down.
|
class InstrumentHelper:
|
def __init__(self):
|
self.primed = False
|
|
def before_io_wait(self, timeout):
|
print(f"before_io_wait({timeout})")
|
if timeout == 9999: # pragma: no branch
|
assert not self.primed
|
in_host(lambda: set_deadline(cscope, 1e9))
|
self.primed = True
|
|
def after_io_wait(self, timeout):
|
if self.primed: # pragma: no branch
|
print("instrument triggered")
|
in_host(lambda: cscope.cancel())
|
trio.lowlevel.remove_instrument(self)
|
|
trio.lowlevel.add_instrument(InstrumentHelper())
|
await trio.sleep_forever()
|
assert cscope.cancelled_caught
|
watb_cscope.cancel()
|
|
async with trio.open_nursery() as nursery:
|
watb_cscope = trio.CancelScope()
|
nursery.start_soon(sit_in_wait_all_tasks_blocked, watb_cscope)
|
await trio.testing.wait_all_tasks_blocked()
|
nursery.start_soon(get_woken_by_host_deadline, watb_cscope)
|
|
return "ok"
|
|
assert trivial_guest_run(trio_main) == "ok"
|
|
|
@restore_unraisablehook()
|
def test_guest_warns_if_abandoned():
|
# This warning is emitted from the garbage collector. So we have to make
|
# sure that our abandoned run is garbage. The easiest way to do this is to
|
# put it into a function, so that we're sure all the local state,
|
# traceback frames, etc. are garbage once it returns.
|
def do_abandoned_guest_run():
|
async def abandoned_main(in_host):
|
in_host(lambda: 1 / 0)
|
while True:
|
await trio.sleep(0)
|
|
with pytest.raises(ZeroDivisionError):
|
trivial_guest_run(abandoned_main)
|
|
with pytest.warns(RuntimeWarning, match="Trio guest run got abandoned"):
|
do_abandoned_guest_run()
|
gc_collect_harder()
|
|
# If you have problems some day figuring out what's holding onto a
|
# reference to the unrolled_run generator and making this test fail,
|
# then this might be useful to help track it down. (It assumes you
|
# also hack start_guest_run so that it does 'global W; W =
|
# weakref(unrolled_run_gen)'.)
|
#
|
# import gc
|
# print(trio._core._run.W)
|
# targets = [trio._core._run.W()]
|
# for i in range(15):
|
# new_targets = []
|
# for target in targets:
|
# new_targets += gc.get_referrers(target)
|
# new_targets.remove(targets)
|
# print("#####################")
|
# print(f"depth {i}: {len(new_targets)}")
|
# print(new_targets)
|
# targets = new_targets
|
|
with pytest.raises(RuntimeError):
|
trio.current_time()
|
|
|
def aiotrio_run(trio_fn, *, pass_not_threadsafe=True, **start_guest_run_kwargs):
|
loop = asyncio.new_event_loop()
|
|
async def aio_main():
|
trio_done_fut = loop.create_future()
|
|
def trio_done_callback(main_outcome):
|
print(f"trio_fn finished: {main_outcome!r}")
|
trio_done_fut.set_result(main_outcome)
|
|
if pass_not_threadsafe:
|
start_guest_run_kwargs["run_sync_soon_not_threadsafe"] = loop.call_soon
|
|
trio.lowlevel.start_guest_run(
|
trio_fn,
|
run_sync_soon_threadsafe=loop.call_soon_threadsafe,
|
done_callback=trio_done_callback,
|
**start_guest_run_kwargs,
|
)
|
|
return (await trio_done_fut).unwrap()
|
|
try:
|
return loop.run_until_complete(aio_main())
|
finally:
|
loop.close()
|
|
|
def test_guest_mode_on_asyncio():
|
async def trio_main():
|
print("trio_main!")
|
|
to_trio, from_aio = trio.open_memory_channel(float("inf"))
|
from_trio = asyncio.Queue()
|
|
aio_task = asyncio.ensure_future(aio_pingpong(from_trio, to_trio))
|
|
# Make sure we have at least one tick where we don't need to go into
|
# the thread
|
await trio.sleep(0)
|
|
from_trio.put_nowait(0)
|
|
async for n in from_aio:
|
print(f"trio got: {n}")
|
from_trio.put_nowait(n + 1)
|
if n >= 10:
|
aio_task.cancel()
|
return "trio-main-done"
|
|
async def aio_pingpong(from_trio, to_trio):
|
print("aio_pingpong!")
|
|
try:
|
while True:
|
n = await from_trio.get()
|
print(f"aio got: {n}")
|
to_trio.send_nowait(n + 1)
|
except asyncio.CancelledError:
|
raise
|
except: # pragma: no cover
|
traceback.print_exc()
|
raise
|
|
assert (
|
aiotrio_run(
|
trio_main,
|
# Not all versions of asyncio we test on can actually be trusted,
|
# but this test doesn't care about signal handling, and it's
|
# easier to just avoid the warnings.
|
host_uses_signal_set_wakeup_fd=True,
|
)
|
== "trio-main-done"
|
)
|
|
assert (
|
aiotrio_run(
|
trio_main,
|
# Also check that passing only call_soon_threadsafe works, via the
|
# fallback path where we use it for everything.
|
pass_not_threadsafe=False,
|
host_uses_signal_set_wakeup_fd=True,
|
)
|
== "trio-main-done"
|
)
|
|
|
def test_guest_mode_internal_errors(monkeypatch, recwarn):
|
with monkeypatch.context() as m:
|
|
async def crash_in_run_loop(in_host):
|
m.setattr("trio._core._run.GLOBAL_RUN_CONTEXT.runner.runq", "HI")
|
await trio.sleep(1)
|
|
with pytest.raises(trio.TrioInternalError):
|
trivial_guest_run(crash_in_run_loop)
|
|
with monkeypatch.context() as m:
|
|
async def crash_in_io(in_host):
|
m.setattr("trio._core._run.TheIOManager.get_events", None)
|
await trio.sleep(0)
|
|
with pytest.raises(trio.TrioInternalError):
|
trivial_guest_run(crash_in_io)
|
|
with monkeypatch.context() as m:
|
|
async def crash_in_worker_thread_io(in_host):
|
t = threading.current_thread()
|
old_get_events = trio._core._run.TheIOManager.get_events
|
|
def bad_get_events(*args):
|
if threading.current_thread() is not t:
|
raise ValueError("oh no!")
|
else:
|
return old_get_events(*args)
|
|
m.setattr("trio._core._run.TheIOManager.get_events", bad_get_events)
|
|
await trio.sleep(1)
|
|
with pytest.raises(trio.TrioInternalError):
|
trivial_guest_run(crash_in_worker_thread_io)
|
|
gc_collect_harder()
|
|
|
def test_guest_mode_ki():
|
assert signal.getsignal(signal.SIGINT) is signal.default_int_handler
|
|
# Check SIGINT in Trio func and in host func
|
async def trio_main(in_host):
|
with pytest.raises(KeyboardInterrupt):
|
signal_raise(signal.SIGINT)
|
|
# Host SIGINT should get injected into Trio
|
in_host(partial(signal_raise, signal.SIGINT))
|
await trio.sleep(10)
|
|
with pytest.raises(KeyboardInterrupt) as excinfo:
|
trivial_guest_run(trio_main)
|
assert excinfo.value.__context__ is None
|
# Signal handler should be restored properly on exit
|
assert signal.getsignal(signal.SIGINT) is signal.default_int_handler
|
|
# Also check chaining in the case where KI is injected after main exits
|
final_exc = KeyError("whoa")
|
|
async def trio_main_raising(in_host):
|
in_host(partial(signal_raise, signal.SIGINT))
|
raise final_exc
|
|
with pytest.raises(KeyboardInterrupt) as excinfo:
|
trivial_guest_run(trio_main_raising)
|
assert excinfo.value.__context__ is final_exc
|
|
assert signal.getsignal(signal.SIGINT) is signal.default_int_handler
|
|
|
def test_guest_mode_autojump_clock_threshold_changing():
|
# This is super obscure and probably no-one will ever notice, but
|
# technically mutating the MockClock.autojump_threshold from the host
|
# should wake up the guest, so let's test it.
|
|
clock = trio.testing.MockClock()
|
|
DURATION = 120
|
|
async def trio_main(in_host):
|
assert trio.current_time() == 0
|
in_host(lambda: setattr(clock, "autojump_threshold", 0))
|
await trio.sleep(DURATION)
|
assert trio.current_time() == DURATION
|
|
start = time.monotonic()
|
trivial_guest_run(trio_main, clock=clock)
|
end = time.monotonic()
|
# Should be basically instantaneous, but we'll leave a generous buffer to
|
# account for any CI weirdness
|
assert end - start < DURATION / 2
|
|
|
@pytest.mark.skipif(buggy_pypy_asyncgens, reason="PyPy 7.2 is buggy")
|
@pytest.mark.xfail(
|
sys.implementation.name == "pypy",
|
reason="async generator issue under investigation",
|
)
|
@restore_unraisablehook()
|
def test_guest_mode_asyncgens():
|
import sniffio
|
|
record = set()
|
|
async def agen(label):
|
assert sniffio.current_async_library() == label
|
try:
|
yield 1
|
finally:
|
library = sniffio.current_async_library()
|
try:
|
await sys.modules[library].sleep(0)
|
except trio.Cancelled:
|
pass
|
record.add((label, library))
|
|
async def iterate_in_aio():
|
# "trio" gets inherited from our Trio caller if we don't set this
|
sniffio.current_async_library_cvar.set("asyncio")
|
await agen("asyncio").asend(None)
|
|
async def trio_main():
|
task = asyncio.ensure_future(iterate_in_aio())
|
done_evt = trio.Event()
|
task.add_done_callback(lambda _: done_evt.set())
|
with trio.fail_after(1):
|
await done_evt.wait()
|
|
await agen("trio").asend(None)
|
|
gc_collect_harder()
|
|
# Ensure we don't pollute the thread-level context if run under
|
# an asyncio without contextvars support (3.6)
|
context = contextvars.copy_context()
|
context.run(aiotrio_run, trio_main, host_uses_signal_set_wakeup_fd=True)
|
|
assert record == {("asyncio", "asyncio"), ("trio", "trio")}
|
