Repository: florimondmanca/asgi-lifespan

Issue:
Remove dependency on anyio?
`anyio` was introduced to quickly get something running with support for both asyncio, trio and curio.

Curio isn't that well supported in the ASGI ecosystem anyway, so we may as well just drop it *for now*.

This leaves us with asyncio and trio.

We only use `anyio` in `LifespanManager`, and we use the following primitives:

- Async generator finalization.
- Events.
- Queues.
- Cancellation exception class.
- Task groups (one task only).
- Timeouts ("fail after…").

Now, it's definitely a lot. But apart from task groups, all primitives are already present in both asyncio and trio, and we can easily build an abstraction layer on top of them, e.g. `ConcurrencyBackend`, like what HTTPX does.

Asyncio doesn't have task groups, but the good news is we don't need to reimplement the whole thing. We only use a task group in `LifespanManager` to run a coroutine in the background while we do stuff in the foreground. This can *almost* always be rewritten to use `asyncio.gather()`. The subtlety here is that we `yield` in the foreground. But there might be a way to rewrite the whole thing not to have the `yield` in the way (and break the context-managed usage internally). I've got some insights on this based on https://github.com/encode/httpx/pull/526.

So, hmm, we might look into this at some point? It's much easier because we don't actually perform any I/O (no pun intended), only coroutine-level concurrency.

Discussion:
codecov-io: # [Codecov](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=h1) Report
> Merging [#18](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=desc) into [master](https://codecov.io/gh/florimondmanca/asgi-lifespan/commit/5075ec129ae132b2d194fd6ae39813e0a16d77b4?src=pr&el=desc) will **decrease** coverage by `0.25%`.
> The diff coverage is `99.3%`.

[![Impacted file tree graph](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/graphs/tree.svg?width=650&token=Kd4Ce4dVb6&height=150&src=pr)](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master      #18      +/-   ##
==========================================
- Coverage     100%   99.74%   -0.26%     
==========================================
  Files          11       15       +4     
  Lines         289      394     +105     
==========================================
+ Hits          289      393     +104     
- Misses          0        1       +1
```


| [Impacted Files](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [tests/test\_manager.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-dGVzdHMvdGVzdF9tYW5hZ2VyLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [asgi\_lifespan/concurrency/base.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-YXNnaV9saWZlc3Bhbi9jb25jdXJyZW5jeS9iYXNlLnB5) | `100% <100%> (ø)` | |
| [asgi\_lifespan/manager.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-YXNnaV9saWZlc3Bhbi9tYW5hZ2VyLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [tests/test\_app.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-dGVzdHMvdGVzdF9hcHAucHk=) | `100% <100%> (ø)` | :arrow_up: |
| [asgi\_lifespan/concurrency/asyncio.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-YXNnaV9saWZlc3Bhbi9jb25jdXJyZW5jeS9hc3luY2lvLnB5) | `100% <100%> (ø)` | |
| [tests/test\_middleware.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-dGVzdHMvdGVzdF9taWRkbGV3YXJlLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [tests/conftest.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-dGVzdHMvY29uZnRlc3QucHk=) | `100% <100%> (ø)` | |
| [tests/concurrency.py](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree#diff-dGVzdHMvY29uY3VycmVuY3kucHk=) | `94.73% <94.73%> (ø)` | |
| ... and [2 more](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=footer). Last update [5075ec1...67608ec](https://codecov.io/gh/florimondmanca/asgi-lifespan/pull/18?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/README.md b/README.md
index ed7aa23..7cc2df5 100644
--- a/README.md
+++ b/README.md
@@ -27,14 +27,12 @@ Modular components for adding [lifespan protocol](https://asgi.readthedocs.io/en
 - Create a lifespan-capable ASGI app with event handler registration support using `Lifespan`.
 - Add lifespan support to an ASGI app using `LifespanMiddleware`.
 - Send lifespan events to an ASGI app (e.g. for testing) using `LifespanManager`.
-- Support for [asyncio], [trio] and [curio] (provided by [anyio]).
+- Support for [asyncio] and [trio].
 - Fully type-annotated.
 - 100% test coverage.
 
 [asyncio]: https://docs.python.org/3/library/asyncio
-[trio]: https://anyio.readthedocs.io/en/latest/
-[curio]: https://anyio.readthedocs.io/en/latest/
-[anyio]: https://anyio.readthedocs.io
+[trio]: https://trio.readthedocs.io
 
 ## Installation
 
@@ -173,9 +171,6 @@ asyncio.run(main())
 
 import trio
 trio.run(main)
-
-import curio
-curio.run(main)
 ```
 
 Run `$ python main.py` in your terminal, and you should get the following output:
diff --git a/asgi_lifespan/_concurrency/__init__.py b/asgi_lifespan/_concurrency/__init__.py
new file mode 100644
index 0000000..4ee0c4e
--- /dev/null
+++ b/asgi_lifespan/_concurrency/__init__.py
@@ -0,0 +1,22 @@
+from .base import ConcurrencyBackend
+
+
+def detect_concurrency_backend() -> ConcurrencyBackend:
+    from .asyncio import AsyncioBackend
+
+    try:
+        import sniffio
+    except ImportError:  # pragma: no cover
+        # sniffio would have been installed with trio,
+        # so we must be running on asyncio.
+        return AsyncioBackend()
+
+    library = sniffio.current_async_library()
+    assert library in ("asyncio", "trio")
+
+    if library == "asyncio":
+        return AsyncioBackend()
+
+    from .trio import TrioBackend
+
+    return TrioBackend()
diff --git a/asgi_lifespan/_concurrency/asyncio.py b/asgi_lifespan/_concurrency/asyncio.py
new file mode 100644
index 0000000..2096303
--- /dev/null
+++ b/asgi_lifespan/_concurrency/asyncio.py
@@ -0,0 +1,83 @@
+import asyncio
+import contextlib
+import types
+import typing
+
+from .base import BaseEvent, BaseQueue, ConcurrencyBackend
+
+
+class AsyncioEvent(BaseEvent):
+    def __init__(self) -> None:
+        self._event = asyncio.Event()
+
+    def set(self) -> None:
+        self._event.set()
+
+    async def wait(self) -> None:
+        await self._event.wait()
+
+
+class AsyncioQueue(BaseQueue):
+    def __init__(self, capacity: int) -> None:
+        self._queue: asyncio.Queue[typing.Any] = asyncio.Queue(maxsize=capacity)
+
+    async def get(self) -> typing.Any:
+        return await self._queue.get()
+
+    async def put(self, value: typing.Any) -> None:
+        await self._queue.put(value)
+
+
+class AsyncioBackend(ConcurrencyBackend):
+    def create_event(self) -> BaseEvent:
+        return AsyncioEvent()
+
+    def create_queue(self, capacity: int) -> BaseQueue:
+        return AsyncioQueue(capacity=capacity)
+
+    async def run_and_fail_after(
+        self,
+        seconds: typing.Optional[float],
+        coroutine: typing.Callable[[], typing.Awaitable[None]],
+    ) -> None:
+        try:
+            await asyncio.wait_for(coroutine(), timeout=seconds)
+        except asyncio.TimeoutError:
+            raise TimeoutError
+
+    def run_in_background(
+        self, coroutine: typing.Callable[[], typing.Awaitable[None]]
+    ) -> typing.AsyncContextManager:
+        return Background(coroutine)
+
+
+class Background:
+    def __init__(self, coroutine: typing.Callable[[], typing.Awaitable[None]]) -> None:
+        self.coroutine = coroutine
+        self.task: typing.Optional[asyncio.Task] = None
+        self._task_exception: typing.Optional[BaseException] = None
+
+    async def __aenter__(self) -> None:
+        async def run_and_silence_cancelled() -> None:
+            with contextlib.suppress(asyncio.CancelledError):
+                await self.coroutine()
+
+        loop = asyncio.get_event_loop()
+        self.task = loop.create_task(run_and_silence_cancelled())
+
+    async def __aexit__(
+        self,
+        exc_type: typing.Optional[typing.Type[BaseException]] = None,
+        exc_value: BaseException = None,
+        traceback: types.TracebackType = None,
+    ) -> None:
+        assert self.task is not None
+
+        _, pending = await asyncio.wait({self.task}, timeout=0)
+        if pending:
+            self.task.cancel()
+
+        await self.task
+
+        if exc_type is None:
+            self.task.result()
diff --git a/asgi_lifespan/_concurrency/base.py b/asgi_lifespan/_concurrency/base.py
new file mode 100644
index 0000000..da7c456
--- /dev/null
+++ b/asgi_lifespan/_concurrency/base.py
@@ -0,0 +1,53 @@
+import typing
+
+
+class BaseEvent:
+    def set(self) -> None:
+        raise NotImplementedError  # pragma: no cover
+
+    async def wait(self) -> None:
+        raise NotImplementedError  # pragma: no cover
+
+
+class BaseQueue:
+    async def get(self) -> typing.Any:
+        raise NotImplementedError  # pragma: no cover
+
+    async def put(self, value: typing.Any) -> None:
+        raise NotImplementedError  # pragma: no cover
+
+
+class ConcurrencyBackend:
+    def create_event(self) -> BaseEvent:
+        raise NotImplementedError  # pragma: no cover
+
+    def create_queue(self, capacity: int) -> BaseQueue:
+        raise NotImplementedError  # pragma: no cover
+
+    async def run_and_fail_after(
+        self,
+        seconds: typing.Optional[float],
+        coroutine: typing.Callable[[], typing.Awaitable[None]],
+    ) -> None:
+        raise NotImplementedError  # pragma: no cover
+
+    def finalize(
+        self, agen: typing.AsyncGenerator[None, None]
+    ) -> typing.AsyncContextManager:
+        return Finalize(agen)
+
+    def run_in_background(
+        self, coroutine: typing.Callable[[], typing.Awaitable[None]]
+    ) -> typing.AsyncContextManager:
+        raise NotImplementedError  # pragma: no cover
+
+
+class Finalize:
+    def __init__(self, agen: typing.AsyncGenerator):
+        self._agen = agen
+
+    async def __aenter__(self) -> typing.AsyncGenerator:
+        return self._agen
+
+    async def __aexit__(self, *args: typing.Any) -> None:
+        await self._agen.aclose()
diff --git a/asgi_lifespan/_concurrency/trio.py b/asgi_lifespan/_concurrency/trio.py
new file mode 100644
index 0000000..eb6826e
--- /dev/null
+++ b/asgi_lifespan/_concurrency/trio.py
@@ -0,0 +1,77 @@
+import types
+import typing
+
+import trio
+
+from ..compat import AsyncExitStack
+from .base import BaseEvent, BaseQueue, ConcurrencyBackend
+
+
+class TrioEvent(BaseEvent):
+    def __init__(self) -> None:
+        self._event = trio.Event()
+
+    def set(self) -> None:
+        self._event.set()
+
+    async def wait(self) -> None:
+        await self._event.wait()
+
+
+class TrioQueue(BaseQueue):
+    def __init__(self, capacity: int) -> None:
+        self._send_channel, self._receive_channel = trio.open_memory_channel(
+            max_buffer_size=capacity
+        )
+
+    async def get(self) -> typing.Any:
+        return await self._receive_channel.receive()
+
+    async def put(self, value: typing.Any) -> None:
+        await self._send_channel.send(value)
+
+
+class TrioBackend(ConcurrencyBackend):
+    def create_event(self) -> BaseEvent:
+        return TrioEvent()
+
+    def create_queue(self, capacity: int) -> BaseQueue:
+        return TrioQueue(capacity=capacity)
+
+    async def run_and_fail_after(
+        self,
+        seconds: typing.Optional[float],
+        coroutine: typing.Callable[[], typing.Awaitable[None]],
+    ) -> None:
+        with trio.move_on_after(seconds if seconds is not None else float("inf")):
+            await coroutine()
+            return
+        raise TimeoutError
+
+    def run_in_background(
+        self, coroutine: typing.Callable[[], typing.Awaitable[None]]
+    ) -> typing.AsyncContextManager:
+        return Background(coroutine)
+
+
+class Background:
+    def __init__(self, coroutine: typing.Callable[[], typing.Awaitable[None]]) -> None:
+        self.coroutine = coroutine
+        self._exit_stack = AsyncExitStack()
+
+    async def __aenter__(self) -> None:
+        nursery = await self._exit_stack.enter_async_context(trio.open_nursery())
+        nursery.start_soon(self.coroutine)
+
+    async def __aexit__(
+        self,
+        exc_type: typing.Optional[typing.Type[BaseException]] = None,
+        exc_value: BaseException = None,
+        traceback: types.TracebackType = None,
+    ) -> None:
+        try:
+            await self._exit_stack.__aexit__(exc_type, exc_value, traceback)
+        except trio.MultiError as exc:
+            for exception in exc.exceptions:
+                if not isinstance(exception, (trio.Cancelled, GeneratorExit)):
+                    raise exception
diff --git a/asgi_lifespan/manager.py b/asgi_lifespan/manager.py
index ac2c674..e8dd62d 100644
--- a/asgi_lifespan/manager.py
+++ b/asgi_lifespan/manager.py
@@ -2,8 +2,7 @@ import inspect
 import typing
 from types import TracebackType
 
-import anyio
-
+from ._concurrency import ConcurrencyBackend, detect_concurrency_backend
 from .compat import AsyncExitStack, asynccontextmanager
 from .exceptions import LifespanNotSupported
 
@@ -16,6 +15,7 @@ class LifespanManager:
         shutdown_timeout: typing.Optional[float] = 5,
     ) -> None:
         self.app = app
+        self.concurrency_backend = detect_concurrency_backend()
         self.startup_timeout = startup_timeout
         self.shutdown_timeout = shutdown_timeout
         self._exit_stack = AsyncExitStack()
@@ -27,12 +27,14 @@ class LifespanManager:
             self.app,
             startup_timeout=self.startup_timeout,
             shutdown_timeout=self.shutdown_timeout,
+            concurrency_backend=self.concurrency_backend,
         )
 
         # Ensure the async generator used by @asynccontextmanager is properly
-        # closed on exit by wrapping it in anyio.finalize().
-        # This prevents a warning raised by curio in particular.
-        agen = await self._exit_stack.enter_async_context(anyio.finalize(agen))
+        # closed on exit, even in case of an exception.
+        agen = await self._exit_stack.enter_async_context(
+            self.concurrency_backend.finalize(agen)
+        )
         assert inspect.isasyncgen(agen)
 
         @asynccontextmanager
@@ -58,12 +60,13 @@ async def _lifespan_manager_generator(
     app: typing.Callable,
     startup_timeout: typing.Optional[float],
     shutdown_timeout: typing.Optional[float],
-) -> typing.AsyncIterator[None]:
-    startup_complete = anyio.create_event()
-    shutdown_complete = anyio.create_event()
+    concurrency_backend: ConcurrencyBackend,
+) -> typing.AsyncGenerator[None, None]:
+    startup_complete = concurrency_backend.create_event()
+    shutdown_complete = concurrency_backend.create_event()
 
     scope = {"type": "lifespan"}
-    receive_queue = anyio.create_queue(capacity=1)
+    receive_queue = concurrency_backend.create_queue(capacity=2)
     receive_called = False
 
     async def receive() -> dict:
@@ -79,17 +82,19 @@ async def _lifespan_manager_generator(
             )
 
         if message["type"] == "lifespan.startup.complete":
-            await startup_complete.set()
+            startup_complete.set()
         elif message["type"] == "lifespan.shutdown.complete":
-            await shutdown_complete.set()
+            shutdown_complete.set()
 
     async def run_app() -> None:
         try:
             await app(scope, receive, send)
         except Exception as exc:
-            if isinstance(exc, anyio.get_cancelled_exc_class()):
-                # Stay out of the way of task cancellation.
-                raise
+            # If the app crashed, we should abort startup and shutdown as soon
+            # as possible, instead of waiting until timeout.
+            startup_complete.set()
+            shutdown_complete.set()
+
             if not receive_called:
                 raise LifespanNotSupported(
                     "Application failed before making its first call to 'receive()'. "
@@ -98,17 +103,16 @@ async def _lifespan_manager_generator(
                     "If that is not the case, then this crash is unexpected and "
                     "there is probably more debug output in the cause traceback."
                 ) from exc
-            raise
 
-    async with anyio.create_task_group() as group:
-        await group.spawn(run_app)
+            raise
 
+    async with concurrency_backend.run_in_background(run_app):
         await receive_queue.put({"type": "lifespan.startup"})
-        async with anyio.fail_after(startup_timeout):
-            await startup_complete.wait()
-
+        await concurrency_backend.run_and_fail_after(
+            startup_timeout, startup_complete.wait
+        )
         yield
-
         await receive_queue.put({"type": "lifespan.shutdown"})
-        async with anyio.fail_after(shutdown_timeout):
-            await shutdown_complete.wait()
+        await concurrency_backend.run_and_fail_after(
+            shutdown_timeout, shutdown_complete.wait
+        )
diff --git a/requirements.txt b/requirements.txt
index 63caa30..e0aec61 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,6 @@
 -e .
 
 black
-curio
 flake8
 flake8-bugbear
 flake8-comprehensions
@@ -9,5 +8,7 @@ isort
 mypy
 nox
 pytest
+pytest-asyncio
 pytest-cov
+pytest-trio
 trio
diff --git a/setup.cfg b/setup.cfg
index 193d197..40cafec 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -11,7 +11,7 @@ combine_as_imports = True
 force_grid_wrap = 0
 include_trailing_comma = True
 known_first_party = asgi_lifespan,tests
-known_third_party = anyio,nox,pytest,setuptools
+known_third_party = nox,pytest,setuptools,trio
 line_length = 88
 multi_line_output = 3
 
@@ -21,4 +21,3 @@ addopts =
   --cov=asgi_lifespan
   --cov=tests
   --cov-report=term-missing
-  --anyio-backends=asyncio,trio,curio
diff --git a/setup.py b/setup.py
index 5d5501d..17f6509 100644
--- a/setup.py
+++ b/setup.py
@@ -38,7 +38,6 @@ setup(
     author_email="florimond.manca@gmail.com",
     packages=get_packages("asgi_lifespan"),
     install_requires=[
-        "anyio",
         "async_generator; python_version < '3.7'",
         "async_exit_stack; python_version < '3.7'",
     ],


Test patch:
diff --git a/tests/concurrency.py b/tests/concurrency.py
new file mode 100644
index 0000000..fd0170a
--- /dev/null
+++ b/tests/concurrency.py
@@ -0,0 +1,64 @@
+"""
+This module contains concurrency utilities that are only used in tests, thus not
+required as part of the ConcurrencyBackend API.
+"""
+
+import asyncio
+import functools
+import typing
+
+import trio
+
+from asgi_lifespan._concurrency.asyncio import AsyncioBackend
+from asgi_lifespan._concurrency.base import ConcurrencyBackend
+from asgi_lifespan._concurrency.trio import TrioBackend
+
+
+@functools.singledispatch
+async def sleep(concurrency_backend: ConcurrencyBackend, seconds: int) -> None:
+    raise NotImplementedError  # pragma: no cover
+
+
+@sleep.register(AsyncioBackend)
+async def _sleep_asyncio(concurrency_backend: ConcurrencyBackend, seconds: int) -> None:
+    await asyncio.sleep(seconds)
+
+
+@sleep.register(TrioBackend)
+async def _sleep_trio(concurrency_backend: ConcurrencyBackend, seconds: int) -> None:
+    await trio.sleep(seconds)
+
+
+@functools.singledispatch
+async def run_and_move_on_after(
+    concurrency_backend: ConcurrencyBackend,
+    seconds: typing.Optional[float],
+    coroutine: typing.Callable[[], typing.Awaitable[None]],
+) -> bool:
+    raise NotImplementedError  # pragma: no cover
+
+
+@run_and_move_on_after.register(AsyncioBackend)
+async def _run_and_move_on_after_asyncio(
+    concurrency_backend: ConcurrencyBackend,
+    seconds: typing.Optional[float],
+    coroutine: typing.Callable[[], typing.Awaitable[None]],
+) -> bool:
+    try:
+        await asyncio.wait_for(coroutine(), timeout=seconds)
+    except asyncio.TimeoutError:
+        return True
+    else:
+        raise NotImplementedError  # pragma: no cover
+
+
+@run_and_move_on_after.register(TrioBackend)
+async def _run_and_move_on_after_trio(
+    concurrency_backend: ConcurrencyBackend,
+    seconds: typing.Optional[float],
+    coroutine: typing.Callable[[], typing.Awaitable[None]],
+) -> bool:
+    with trio.move_on_after(seconds if seconds is not None else float("inf")):
+        await coroutine()
+        raise NotImplementedError  # pragma: no cover
+    return True
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..cb5bdf3
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,11 @@
+import pytest
+
+
+@pytest.fixture(
+    params=[
+        pytest.param(None, marks=pytest.mark.asyncio),
+        pytest.param(None, marks=pytest.mark.trio),
+    ]
+)
+def concurrency() -> None:
+    return
diff --git a/tests/test_app.py b/tests/test_app.py
index 5ace5be..0a45e97 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -56,9 +56,9 @@ def build_asgi_arguments(
     return scope, receive, send
 
 
-@pytest.mark.anyio
+@pytest.mark.usefixtures("concurrency")
 @pytest.mark.parametrize("startup_failed", (False, True))
-async def test_lifespan_app(startup_failed: bool,) -> None:
+async def test_lifespan_app(startup_failed: bool) -> None:
     log: Log = []
     messages = [{"type": "lifespan.shutdown"}, {"type": "lifespan.startup"}]
 
diff --git a/tests/test_manager.py b/tests/test_manager.py
index 03bd035..6ee36d2 100644
--- a/tests/test_manager.py
+++ b/tests/test_manager.py
@@ -1,13 +1,15 @@
 import contextlib
 import typing
 
-import anyio
 import pytest
 
 from asgi_lifespan import Lifespan, LifespanManager, LifespanNotSupported
+from asgi_lifespan._concurrency import detect_concurrency_backend
 
+from . import concurrency
 
-@pytest.mark.anyio
+
+@pytest.mark.usefixtures("concurrency")
 @pytest.mark.parametrize("startup_exception", (None, ValueError))
 @pytest.mark.parametrize("body_exception", (None, ValueError))
 @pytest.mark.parametrize("shutdown_exception", (None, ValueError))
@@ -94,44 +96,46 @@ async def test_lifespan_manager(
 async def slow_startup(
     scope: dict, receive: typing.Callable, send: typing.Callable
 ) -> None:
+    concurrency_backend = detect_concurrency_backend()
     message = await receive()
     assert message["type"] == "lifespan.startup"
-    await anyio.sleep(0.05)
+    await concurrency.sleep(concurrency_backend, 0.05)
     # ...
 
 
 async def slow_shutdown(
     scope: dict, receive: typing.Callable, send: typing.Callable
 ) -> None:
+    concurrency_backend = detect_concurrency_backend()
+
     message = await receive()
     assert message["type"] == "lifespan.startup"
     await send({"type": "lifespan.startup.complete"})
 
     message = await receive()
     assert message["type"] == "lifespan.shutdown"
-    await anyio.sleep(0.05)
+    await concurrency.sleep(concurrency_backend, 0.05)
     # ...
 
 
-@pytest.mark.anyio
+@pytest.mark.usefixtures("concurrency")
 @pytest.mark.parametrize("app", [slow_startup, slow_shutdown])
 async def test_lifespan_timeout(app: typing.Callable) -> None:
-    async with anyio.move_on_after(0.02) as cancel_scope:
-        with pytest.raises(TimeoutError):
-            async with LifespanManager(
-                app, startup_timeout=0.01, shutdown_timeout=0.01
-            ):
-                pass
-    assert not cancel_scope.cancel_called
+    with pytest.raises(TimeoutError):
+        async with LifespanManager(app, startup_timeout=0.01, shutdown_timeout=0.01):
+            pass
 
 
-@pytest.mark.anyio
+@pytest.mark.usefixtures("concurrency")
 @pytest.mark.parametrize("app", [slow_startup, slow_shutdown])
 async def test_lifespan_no_timeout(app: typing.Callable) -> None:
-    async with anyio.move_on_after(0.02) as cancel_scope:
+    async def main() -> None:
         async with LifespanManager(app, startup_timeout=None, shutdown_timeout=None):
             pass
-    assert cancel_scope.cancel_called
+
+    concurrency_backend = detect_concurrency_backend()
+    timed_out = await concurrency.run_and_move_on_after(concurrency_backend, 0.02, main)
+    assert timed_out
 
 
 async def http_only(
@@ -162,7 +166,7 @@ async def http_no_assert_receive_request(
     # ...
 
 
-@pytest.mark.anyio
+@pytest.mark.usefixtures("concurrency")
 @pytest.mark.parametrize(
     "app",
     [
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 71ac3c9..368cbb8 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -5,7 +5,7 @@ import pytest
 from asgi_lifespan import LifespanMiddleware
 
 
-@pytest.mark.anyio
+@pytest.mark.usefixtures("concurrency")
 async def test_lifespan_middleware() -> None:
     scopes: typing.List[dict] = []