agronholm / anyio Goto Github PK

High level asynchronous concurrency and networking framework that works on top of either trio or asyncio

License: MIT License

Python 100.00%

anyio's Introduction

AnyIO is an asynchronous networking and concurrency library that works on top of either asyncio or trio. It implements trio-like structured concurrency (SC) on top of asyncio and works in harmony with the native SC of trio itself.

Applications and libraries written against AnyIO's API will run unmodified on either asyncio or trio. AnyIO can also be adopted into a library or application incrementally – bit by bit, no full refactoring necessary. It will blend in with the native libraries of your chosen backend.

Documentation

View full documentation at: https://anyio.readthedocs.io/

Features

AnyIO offers the following functionality:

Task groups (nurseries in trio terminology)
High-level networking (TCP, UDP and UNIX sockets)
- Happy eyeballs algorithm for TCP connections (more robust than that of asyncio on Python 3.8)
- async/await style UDP sockets (unlike asyncio where you still have to use Transports and Protocols)
A versatile API for byte streams and object streams
Inter-task synchronization and communication (locks, conditions, events, semaphores, object streams)
Worker threads
Subprocesses
Asynchronous file I/O (using worker threads)
Signal handling

AnyIO also comes with its own pytest plugin which also supports asynchronous fixtures. It even works with the popular Hypothesis library.

anyio's People

Contributors

Stargazers

Watchers

Forkers

smurfix theendsofinvention malinoff standy66 rnovatorov thedrow yilinxiong ledmonster tjstum ermakov-oleg pquentin fantix mjwestcott gitter-badger sobolevn shalevy1 graingert daa bohblue2 ioistired sbtinstruments d4n weaver-viii makkus cjmartian johnzeringue byronformwalt lordmauve uspike belm0 siloam generalcommission jonathanplasse alexdelorenzo flyte kiplangatbryan jellezijlstra mayst0731 wireboy5 davidbrochart mgorny xiaoconstantine richardsheridan bruingineer bellyfat pntone spikingneurons vbkbmqj jonathanslenders tiangolo jpmallarino emg110 dlax cclauss didimelli kludex bluenix2 adriangb valerianweiss salva hroncok hauntsaninja pqtrung onlyone0001 fairhopeweb maartenbreddels hassanpydev hsuyuming danielhasan1 jhominal lalinsky amos402 conorstevenson arungahlot-woolies gabrielbusta djun bo-at-pleno imlzg gschaffner e7dal william3johnson inferno-inc simon-oram fivepapertigers nihiyama helderco haobibo wangzhaoyong marcinbarczynski kovan smithed kclamalex diez-canseco-ramirez gwk saorisakura tapetersen sappjw ggopinathan-athene discdiver nikicat

anyio's Issues

Expose curio-like tcp_server_socket / unix_server_socket API

Curio separates socket construction logic into separate *_server_socket synchronous functions. This gives ability to start listening in synchronous context, then do something with the socket, e.g. get actual bound port number when port=0 was passed to auto select, and only after that start accepting connections in asynchronous context.

This is the use case that may be beneficial in unit tests, for example, consider the following scenario:

Create a socket without any specific port, bind it and listen (sync)
Get bound port number
Spin up client thread / process with the port number obtained in part 2 (sync)
Start accepting incoming connections (async)

I think it would be a great addition and this is really easy to implement.

Ability to specify backend with `pytest.mark.anyio`

I came here from pytest-dev/pytest-asyncio#61 and primarily pytest-dev/pytest-asyncio#124

From what I can tell the pytest.mark.anyio marker is solving my problem but I think it's mostly incidental. If I use the pytest.mark.anyio on only the tests I want to be asyncio tests then everything works fine, but really what I need is the ability to specify test-by-test which backend it should use. Not sure if that's a use case that you would like to support.

SocketStream.receive_until() doesn't work with TLS

The current implementation relies on the ability to peek into the kernel buffer and this does not work when TLS is active. The only recourse is to employ buffering inside the instance.

Add Hypothesis support to pytest plugin

Hypothesis requires some explicit support to work properly with async tests. This would also make @Zac-HD happy :)

Aiohttp fails, claiming it's not inside a task

code:

    async def do_thing():
        ...
        async with fail_after(some_timeout):
            async with ClientSession() as session:
                async with session.get(url) as resp:
                    data = await resp.json()

error:

Traceback (most recent call last):
  File "my_file.py", line 138, in do_thing
    async with session.get(url) as resp:
  File "/usr/lib/python3.7/site-packages/aiohttp/client.py", line 855, in __aenter__
    self._resp = await self._coro
  File "/usr/lib/python3.7/site-packages/aiohttp/client.py", line 321, in _request
    with timer:
  File "/usr/lib/python3.7/site-packages/aiohttp/helpers.py", line 658, in __enter__
    raise RuntimeError('Timeout context manager should be used '
RuntimeError: Timeout context manager should be used inside a task

Expose setsockopt in public API

This would be useful for setting up TCP keepalive and TCP_NODELAY flags. Right now the only option is to assume every SocketStream is anyio._networking.SocketStream and obtain raw socket via _socket._raw_socket

Add a way to use Locks with values

Asyncio has

fut = loop.create_future()
# spawn a task that will set fut
await fut
return fut.result()

AnyIO does not offer a way to set values on a Lock or similar, so having something like that would be really useful.

Add default socket options

Apparently both asyncio and trio set TCP_NODELAY on stream sockets by default.
Likewise, SO_REUSEADDR should be set for servers, except on Windows.
On Windows, SO_EXCLUSIVEADDRUSE should possible be set for servers.

Pre-release status makes it hard to install the package

I'm trying to install asks with pipenv and getting an error:

[pipenv.exceptions.ResolutionFailure]: Warning: Your dependencies could not be resolved. You likely have a mismatch in your sub-dependencies.
  First try clearing your dependency cache with $ pipenv lock --clear, then try the original command again.
 Alternatively, you can use $ pipenv install --skip-lock to bypass this mechanism, then run $ pipenv graph to inspect the situation.
  Hint: try $ pipenv lock --pre if it is a pre-release dependency.
ERROR: ERROR: Could not find a version that matches anyio
Skipped pre-versions: 1.0.0a1, 1.0.0a1, 1.0.0a2, 1.0.0a2, 1.0.0b1, 1.0.0b1, 1.0.0b2, 1.0.0b2
There are incompatible versions in the resolved dependencies.

timeout discovery

Trio has a current_effective_deadline() method, which is very useful for interfacing with libraries that require explicit timeout parameters. AnyIO should provide that.

Add subprocess support

There should be some support for subprocesses, at least on the level that all the backends provide.

Two kinds of use cases can be seen for subprocesses:

Running external commands
Running CPU-bound functions

The first case would probably be easier to implement, but eventually we need both.

Difference in cancellation behavior: asyncio vs. trio

import anyio
async def one():
    async with anyio.open_cancel_scope() as sc:
        await anyio.sleep(0.1)
    raise RuntimeError("This should not be printed")

async def bar(tg):
    try:
        await anyio.sleep(1)
    finally:
        await tg.spawn(one)
    pass
async def foo():
    async with anyio.create_task_group() as tg:
        await tg.spawn(bar,tg)
        await tg.cancel_scope.cancel()
anyio.run(foo, backend="trio") # OK
anyio.run(foo, backend="asyncio")  # fails

Cancellation semantics on asyncio and curio differ slightly from trio's

The following script produces different results on the trio backend vs the asyncio and curio ones:

import anyio

scope = None


async def scoped():
    global scope
    async with anyio.open_cancel_scope() as scope:
        async with anyio.open_cancel_scope() as sc2:
            try:
                await scope.cancel()
                # await anyio.sleep(2)
            except BaseException as exc:
                await sc2.cancel()
                raise

        print("You should not see this.")


async def check():
    async with anyio.create_task_group() as tg:
        await tg.spawn(scoped)
        await anyio.sleep(1)
        await scope.cancel()
        await anyio.sleep(1)

anyio.run(scoped, backend='trio')

To fix this, we would need some way to track which cancellation exception was triggered by which cancel scope but currently we do not have the means for this.

Implement introspection of tasks

To help debug applications, there should be a way to get information on the running tasks in the current event loop. There is already code to that end, but it's incomplete.

Add documentation

Write manual for how to use the various APIs in practice.

Cancel scopes stopped working with trio 0.11

import anyio
import trio

async def main():
    async with anyio.open_cancel_scope():
        pass

trio.run(main)

/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/anyio/_backends/trio.py:34: TrioDeprecationWarning: trio.open_cancel_scope is deprecated since Trio 0.10.0; use trio.CancelScope instead (https://github.com/python-trio/trio/issues/607)
  with trio.open_cancel_scope(shield=shield) as cancel_scope:
Traceback (most recent call last):
  File "test.py", line 8, in <module>
    trio.run(main)
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/trio/_core/_run.py", line 1444, in run
    raise runner.main_task_outcome.error
  File "test.py", line 5, in main
    async with anyio.open_cancel_scope():
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/async_generator/_util.py", line 34, in __aenter__
    return await self._agen.asend(None)
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/async_generator/_impl.py", line 366, in step
    return await ANextIter(self._it, start_fn, *args)
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/async_generator/_impl.py", line 197, in __next__
    return self._invoke(first_fn, *first_args)
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/async_generator/_impl.py", line 209, in _invoke
    result = fn(*args)
  File "/Users/astepanov/.virtualenvs/work/lib/python3.6/site-packages/anyio/_backends/trio.py", line 35, in open_cancel_scope
    cancel_scope.cancel = wrap_as_awaitable(cancel_scope.cancel)
AttributeError: 'CancelScope' object attribute 'cancel' is read-only

UNIX server socket files are not automatically removed when closed

UNIX server socket files should be automatically deleted when the socket is closed, but this is not currently being done.

Cancellation issue with asyncio and curio

The following test fails with the latest master (5a2b4c9):

import pytest
import anyio


@pytest.mark.anyio
async def test():
    with pytest.raises(TimeoutError):
        async with anyio.fail_after(1):
            async with anyio.create_task_group() as group:
                await group.spawn(anyio.sleep, 2)
                await anyio.sleep(2)

Output:

± pytest -v --anyio-backends=asyncio,trio,curio
============================ test session starts =============================
platform linux -- Python 3.7.2, pytest-3.10.1, py-1.7.0, pluggy-0.8.1 --
plugins: trio-0.5.1, asyncio-0.10.0, anyio-1.0.0rc1.post5

test.py::test[asyncio] FAILED                        [ 33%]
test.py::test[trio] PASSED                           [ 66%]
test.py::test[curio] FAILED                          [100%]

The corresponding asyncio exception:

E           anyio._backends.asyncio.AsyncioExceptionGroup: 2 exceptions were raised in the task group:
E           ----------------------------
E           Traceback (most recent call last):
E           
E             File "/home/vinmic/anyio/anyio/_backends/asyncio.py", line 384, in _run_wrapped_task
E               await func(*args)
E           
E             File "/home/vinmic/anyio/anyio/_backends/asyncio.py", line 157, in sleep
E               await asyncio.sleep(delay)
E           
E             File "/home/vinmic/miniconda/lib/python3.7/asyncio/tasks.py", line 568, in sleep
E               return await future
E           
E           concurrent.futures._base.CancelledError
E           ----------------------------
E           Traceback (most recent call last):
E           
E             File "/home/vinmic/anyio/test.py", line 12, in test
E               await anyio.sleep(2)
E           
E             File "/home/vinmic/anyio/anyio/_backends/asyncio.py", line 157, in sleep
E               await asyncio.sleep(delay)
E           
E             File "/home/vinmic/miniconda/lib/python3.7/asyncio/tasks.py", line 568, in sleep
E               return await future
E           
E           concurrent.futures._base.CancelledError

../anyio/anyio/_backends/asyncio.py:375: AsyncioExceptionGroup

Cancelling a timeout in a nested scope fails w/ asyncio

Another one … you probably need to check parent scopes recursively, not just the current one.

Fails with asyncio. Works with trio.

@pytest.mark.anyio
async def test_nexted_fail_after():
    async def killer(scope):
        await sleep(0.1)
        await scope.cancel()

    async with create_task_group() as tg:
        async with open_cancel_scope() as scope:
            async with open_cancel_scope():
                await tg.spawn(killer, scope)
                async with fail_after(1):
                    await sleep(2)    
                    raise RuntimeError("Do not go here")
                raise RuntimeError("Do not go here either")
            raise RuntimeError("Do not go here squared")   
                                                           
    assert scope.cancel_called

TLS seemingly broken with Trio backend

Traceback (most recent call last):
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/anyio/_networking.py", line 150, in start_tls
    self._raw_socket.do_handshake()
  File "/usr/lib64/python3.6/ssl.py", line 1068, in do_handshake
    self._sslobj.do_handshake()
  File "/usr/lib64/python3.6/ssl.py", line 689, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLWantReadError: The operation did not complete (read) (_ssl.c:841)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/laura/dev/libs/asyncws/aa.py", line 17, in <module>
    anyio.run(test, backend="trio")
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/anyio/__init__.py", line 112, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_run.py", line 1397, in run
    return result.unwrap()
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_result.py", line 119, in unwrap
    raise self.error
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_run.py", line 1502, in run_impl
    msg = task.coro.send(next_send)
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_run.py", line 1085, in init
    return system_nursery._reap_and_unwrap(task)
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_run.py", line 447, in _reap_and_unwrap
    return task._result.unwrap()
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_result.py", line 119, in unwrap
    raise self.error
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/_run.py", line 1502, in run_impl
    msg = task.coro.send(next_send)
  File "/home/laura/dev/libs/asyncws/aa.py", line 10, in test
    async with open_websocket("wss://echo.websocket.org") as ws:
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/async_generator/_util.py", line 34, in __aenter__
    return await self._agen.asend(None)
  File "/home/laura/dev/libs/asyncws/asyncwebsockets/client.py", line 25, in open_websocket
    await ws.__ainit__(addr=addr, ssl=ssl, path=url.path)
  File "/home/laura/dev/libs/asyncws/asyncwebsockets/_specific/anyio.py", line 15, in __ainit__
    self._sock = await anyio.connect_tcp(*addr, tls=True)
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/anyio/__init__.py", line 301, in connect_tcp
    await stream.start_tls()
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/anyio/_networking.py", line 216, in start_tls
    await self._socket.start_tls(ssl_context, self._server_hostname)
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/anyio/_networking.py", line 153, in start_tls
    await self._wait_readable()
  File "/home/laura/.virtualenvs/asyncws-Sz4NY5bC/lib/python3.6/site-packages/trio/_core/__init__.py", line 47, in wait_socket_readable
    raise TypeError("need a socket")
TypeError: need a socket

Provide a context manager for fail_after() and move_on_after()

At least for move_on_after(), it is important to let the user know if the context block was exited normally or due to a timeout. Trio also provides additional features like the ability to add time to the timeout, plus others which we may consider supporting.

Crashed task does not cancel siblings if host task has finished

If a host task is waiting for task in a task group to finish and one of them raises an exception, this does not cancel the other subtasks even though it should. This works correctly on trio, but not on the other backends.

AsyncFile difference

You use Trio's "native" async file, which uses await s.aclose() to async-close a file (Trio convention).
However, on asyncio there is await s.close().

[Trio] Task groups are marked inactive early

Code:

import anyio


async def second():
    print("This never happens")


async def first(tg):
    await anyio.sleep(0)
    await tg.spawn(second)


async def main():
    async with anyio.create_task_group() as tg:
        await tg.spawn(first, tg)


anyio.run(main, backend="trio")

Error:

Traceback (most recent call last):
  File "/home/laura/dev/discord/curious/aaa.py", line 18, in <module>
    anyio.run(main, backend="trio")
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/__init__.py", line 112, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/trio/_core/_run.py", line 1337, in run
    raise runner.main_task_outcome.error
  File "/home/laura/dev/discord/curious/aaa.py", line 15, in main
    await tg.spawn(first, tg)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/async_generator/_util.py", line 42, in __aexit__
    await self._agen.asend(None)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/async_generator/_impl.py", line 366, in step
    return await ANextIter(self._it, start_fn, *args)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/async_generator/_impl.py", line 202, in send
    return self._invoke(self._it.send, value)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/async_generator/_impl.py", line 209, in _invoke
    result = fn(*args)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/_backends/trio.py", line 90, in create_task_group
    tg._active = False
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/trio/_core/_run.py", line 397, in __aexit__
    raise combined_error_from_nursery
  File "/home/laura/dev/discord/curious/aaa.py", line 10, in first
    await tg.spawn(second)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/_backends/trio.py", line 78, in spawn
    raise RuntimeError('This task group is not active; no new tasks can be spawned.')
RuntimeError: This task group is not active; no new tasks can be spawned.

Both curio and asyncio backends allow this to work.

curio backend spews extra error on main function erroring

Traceback (most recent call last):
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/_backends/curio.py", line 21, in run
    return kernel.run(func, *args)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 180, in run
    raise ret_exc
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 746, in _run_coro
    trap = current._send(current.next_value)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/task.py", line 173, in _task_runner
    self.next_value = await coro
  File "/home/laura/dev/discord/curious/repro.py", line 23, in main
    await err()
  File "/home/laura/dev/discord/curious/repro.py", line 19, in err
    raise AttributeError("fewsfwsefwf")
AttributeError: fewsfwsefwf

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/laura/dev/discord/curious/repro.py", line 26, in <module>
    anyio.run(main, backend="curio")
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/__init__.py", line 112, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/anyio/_backends/curio.py", line 24, in run
    kernel.run(shutdown=True)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 146, in run
    self._runner.send(None)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 605, in _run_coro
    _init_loopback()
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 277, in _init_loopback
    kernel._call_at_shutdown(kernel._notify_sock.close)
  File "/home/laura/.local/share/virtualenvs/curious-yiBjxk2Q/lib/python3.7/site-packages/curio/kernel.py", line 133, in _call_at_shutdown
    self._shutdown_funcs.append(func)
AttributeError: 'NoneType' object has no attribute 'append'

Seems like you try and shut the kernel down when it's already done it internally.

`accept_connections` doesn't

    async for conn in server.accept_connections():
  File "/usr/lib/python3/dist-packages/async_generator/_impl.py", line 271, in __anext__
    return await self._do_it(self._it.__next__)
  File "/usr/lib/python3/dist-packages/async_generator/_impl.py", line 290, in _do_it
    return await ANextIter(self._it, start_fn, *args)
  File "/usr/lib/python3/dist-packages/async_generator/_impl.py", line 197, in __next__
    return self._invoke(first_fn, *first_args)
  File "/usr/lib/python3/dist-packages/async_generator/_impl.py", line 209, in _invoke
    result = fn(*args)
  File "/src/anyio/anyio/_networking.py", line 384, in accept_connections
    while not self._socket.closed:
  File "/src/anyio/anyio/_networking.py", line 23, in __getattr__
    return getattr(self._raw_socket, item)
AttributeError: 'socket' object has no attribute 'closed'

Yes, this requires a testcase …

diff --git a/anyio/_networking.py b/anyio/_networking.py
index 71f6d6f..05f329b 100644
--- a/anyio/_networking.py
+++ b/anyio/_networking.py
@@ -381,7 +381,7 @@ class SocketStreamServer(abc.SocketStreamServer):
 
     @async_generator
     async def accept_connections(self):
-        while not self._socket.closed:
+        while not self._socket._closed:
             await yield_(await self.accept())

Socket listeners do not get close notifications

If one task is waiting for a socket to become ready for reading or writing and another task closes it, the former task is left waiting forever. Trio has a native mechanism to handle this gracefully, and for the other two we could whip up a solution fairly easily.

Use uvloop as event loop if available

Link to docs?

Someone pointed out this library to me, and it looks like it may be really helpful. I see that you have docs, and it looks like they should be on Read the Docs, but my Googling didn't turn anything up: (YMMV) https://www.google.com/search?q=site%3Areadthedocs.org+python+anyio&rlz=1C1CHBF_enUS720US720&oq=site%3Areadthedocs.org+python+anyio&aqs=chrome..69i57j69i58.6112j0j7&sourceid=chrome&ie=UTF-8

If this library has published docs, it would be great if you could link to them in the README.

Cancellation doesn't survive its scope

This should work, but doesn't. I have no good idea why because AFAIU cancelling an asyncio task is a one-stop thing which raises a CancelledError and clears the task's must_cancel flag.

import asyncio
import anyio

async def main():
    try:
        async with anyio.fail_after(0.1):
            await anyio.sleep(1)
    except TimeoutError:
        print("Timed out.")
    await anyio.sleep(0.2)
    print("Coninued OK.")

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Add shielding support

Add support for deferring the cancellation of a task until a critical block of code is complete.
The best way to do this would probably be using a context manager.

Nested cancellation scopes fail to eat "their" exceptions

This code does not print "Done". It should.

import anyio

scope = None
async def scoped():
    async with anyio.open_cancel_scope() as sc:
        async with anyio.open_cancel_scope() as sc2:
            global scope
            scope = sc
            await anyio.sleep(2)

async def check():
    async with anyio.create_task_group() as tg:
        await tg.spawn(scoped)
        await anyio.sleep(1)
        await scope.cancel()
        await anyio.sleep(1)
        print("Done.")

anyio.run(check)

This patch fixes the immediate problem, but it's fragile. A "real" solution needs to store the cancel scope that's responsible for the cancellation in the exception, like Trio does.

diff --git a/anyio/_backends/asyncio.py b/anyio/_backends/asyncio.py
index 5f470d0..34a364d 100644
--- a/anyio/_backends/asyncio.py
+++ b/anyio/_backends/asyncio.py
@@ -221,7 +221,7 @@ async def open_cancel_scope(deadline: float = float('inf'), shield: bool = False
     except asyncio.CancelledError as exc:
         if timeout_expired:
             raise TimeoutError().with_traceback(exc.__traceback__) from None
-        else:
+        elif not scope._cancel_called:
             raise
     finally:
         if timeout_task:

pytest plugin doesn't interoperate with trio

I'm not versed in pytest's internals to get a handle on this one.

smurf@dev:~/src/trio-asyncio$ pytest-3 -sxv tests/
========================================== test session starts ==========================================
platform linux -- Python 3.6.7, pytest-3.6.4, py-1.5.4, pluggy-0.6.0 -- /usr/bin/python3
cachedir: .pytest_cache
rootdir: /src/trio-asyncio, inifile: setup.cfg
plugins: trio-0.4.2, cov-2.5.1, anyio-0.0.post49
collected 1112 items / 2 skipped                                                                        

tests/test_aio_subprocess.py::test_subprocess_exec ERROR

================================================ ERRORS =================================================
________________________________ ERROR at setup of test_subprocess_exec _________________________________
file /src/trio-asyncio/tests/test_aio_subprocess.py, line 74
  @pytest.mark.trio
  async def test_subprocess_exec(loop):
      await run_subprocess_exec(loop)
file /usr/lib/python3/dist-packages/anyio/pytest_plugin.py, line 17
      def wrapper(*args, **kwargs):
E       fixture 'anyio_backend' not found
>       available fixtures: autojump_clock, cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, loop, mock_clock, monkeypatch, nursery, pytestconfig, record_property, record_xml_attribute, record_xml_property, recwarn, tmpdir, tmpdir_factory
>       use 'pytest --fixtures [testpath]' for help on them.

/usr/lib/python3/dist-packages/anyio/pytest_plugin.py:17
=========================================== warnings summary ============================================
tests/python/test_selector_events.py::TestBaseSelectorEventLoop
  cannot collect test class 'TestBaseSelectorEventLoop' because it has a __init__ constructor

tests/python/test_subprocess.py::TestSubprocessTransport
  cannot collect test class 'TestSubprocessTransport' because it has a __init__ constructor

-- Docs: http://doc.pytest.org/en/latest/warnings.html
============================ 2 skipped, 2 warnings, 1 error in 0.36 seconds =============================

Properly shut down TLS connections

TLS connections should be closed with the proper shutdown handshake. If the socket is closed before this has been done, an exception should be raised. This behavior should be optional, however, since HTTPS (among others) does commonly skip this step.

please hide internals

When I'm importing trio in my IDE, I'd rather not see anyio._backends.trio as an option.

Need a finalise helper

Curio complains about async generators:

RuntimeError: Async generator with async finalization must be wrapped by
async with curio.meta.finalize(agen) as agen:
    async for n in agen:
         ...

Unfortunately, this can't be done without hard-depending on curio.

Guard against concurrent read/write from/to streams

Right now there is nothing blocking multiple tasks from reading from or writing to streams. Each stream should know if there's already a task performing either of those operations and either raise an exception or block the operation. Looking at trio should yield the answer on which action to take then.

Issue with nested task groups and cancellation

As far as I can tell, this issue is different from #34.

When an outer task group cancels a task containing an inner task group, the inner task group does not re-raise the CancelledError:

import trio
import anyio
import pytest


@pytest.mark.anyio
async def test_anyio():

    async def g():
        async with anyio.create_task_group() as group:
            await anyio.sleep(1)
        assert False

    async def f():
        async with anyio.create_task_group() as group:
            await group.spawn(g)
            await anyio.sleep(0)
            await group.cancel_scope.cancel()

    await f()


@pytest.mark.trio
async def test_trio():

    async def g():
        async with trio.open_nursery() as nursery:
            await trio.sleep(1)
        assert False

    async def f():
        async with trio.open_nursery() as nursery:
            nursery.start_soon(g)
            await trio.sleep(0)
            nursery.cancel_scope.cancel()

    await f()

Reports:

test_anyio.py::test_anyio[asyncio] FAILED
test_anyio.py::test_anyio[curio] FAILED
test_anyio.py::test_anyio[trio] FAILED                  
test_anyio.py::test_trio PASSED

As a side note, test_anyio[trio] does not fail for the same reason as the other test_anyio: it raises an ExceptionGroup containing two trio.Cancelled exceptions.

Allow use of mixed asyncio/curio/trio

Our project aims to allow the end user to write extensions in any of the async libraries provided by anyio, but it seems this is either undocumented or not yet possible. Please implement a way to make this work.

code used:

import anyio, curio, trio

async def curio_coro():
    await curio.sleep(1)
    print("Hello from curio!")

async def trio_coro():
    await trio.sleep(1)
    print("Hello from trio!")

async def run_both():
    async with anyio.create_task_group() as tg:
        await tg.spawn(curio_coro)
        await tg.spawn(trio_coro)

anyio.run(run_both)

In this context, curio_coro and trio_coro would be functions written by our end users.

Provide text streams

All the streams in AnyIO are currently bytes based. Providing support for automatic en/decoding of unicode strings given an encoding (and possibly a fallback encoding) might theoretically make it much easier to implement text based protocols.

receive_some may ignore max_bytes

When data is buffered, max_bytes is ignored. It should not be.

Reproduce by sending b"abc" to a socket, call receive_until(b'a'), then receive_some(1).

diff --git a/anyio/_networking.py b/anyio/_networking.py
index 71f6d6f..32e7732 100644
--- a/anyio/_networking.py
+++ b/anyio/_networking.py
@@ -222,7 +222,10 @@ class SocketStream(abc.SocketStream):
 
     async def receive_some(self, max_bytes: Optional[int]) -> bytes:
         if self._buffer:
-            data, self._buffer = self._buffer, b''
+            if max_bytes is not None and len(self._buffer) > max_bytes:
+                data, self._buffer = self._buffer[:max_bytes], self_buffer[max_bytes:]
+            else:
+                data, self._buffer = self._buffer, b''
             return data
 
         return await self._socket.recv(max_bytes)

anyio pytest marker doesn't work with strict

laura:users@antiskill ~/d/l/rio-redis master …6 ./rw
(rio-redis-wQCA5cC-) $ pytest
============================== test session starts ===============================
platform linux -- Python 3.7.1, pytest-3.9.2, py-1.7.0, pluggy-0.8.0
rootdir: /home/laura/dev/libs/rio-redis, inifile:
plugins: cov-2.6.0, anyio-1.0.0a2
collected 36 items                                                               

rioredis/tests/test_commands.py ...........                                [ 30%]
rioredis/tests/test_pool.py .                                              [ 33%]
rioredis/tests/test_commands.py ...........                                [ 63%]
rioredis/tests/test_pool.py .                                              [ 66%]
rioredis/tests/test_commands.py ...........                                [ 97%]
rioredis/tests/test_pool.py .                                              [100%]

=========================== 36 passed in 0.17 seconds ============================
laura:users@antiskill ~/d/l/rio-redis master …6 ./rw
(rio-redis-wQCA5cC-) $ pytest --strict
============================== test session starts ===============================
platform linux -- Python 3.7.1, pytest-3.9.2, py-1.7.0, pluggy-0.8.0
rootdir: /home/laura/dev/libs/rio-redis, inifile:
plugins: cov-2.6.0, anyio-1.0.0a2
collected 0 items / 2 errors                                                     

===================================== ERRORS =====================================
________________ ERROR collecting rioredis/tests/test_commands.py ________________
'anyio' not a registered marker
__________________ ERROR collecting rioredis/tests/test_pool.py __________________
'anyio' not a registered marker
!!!!!!!!!!!!!!!!!!!! Interrupted: 2 errors during collection !!!!!!!!!!!!!!!!!!!!!
============================ 2 error in 0.08 seconds =============================

https://circleci.com/gh/Fuyukai/rio-redis/44

Integrate with the exceptiongroup library

The current implementation of exception groups in anyio is quite poor. This has been left alone because of the emerging exceptiongroup library which should provide a much better implementation. Once it's been released, we should add a dependency on it and remove our built-in implementation.

Port?

You seem to be ignoring the portargument here:

https://github.com/agronholm/hyperio/blob/2f68daa8f8f73b0d5983411995619a43557ac68a/hyperio/backends/asyncio.py#L435

Add capacity limiters

Add support for trio-like CapacityLimiters. These are most commonly used for limiting the maximum number of running threads, but can be used in other roles as well.

Provide access to the SSL object for TLS enabled streams

Currently there is no way to get the peer certificate, negotiated ALPN protocol or any other SSL related information after the handshake. The SocketStream class should provide a way to access this information.

asyncio: Multiple consecutive cancel scopes cause a crash

The problem is that asyncio does not run callbacks immediately when a Future is done, so CancelScope._task_done is queued twice for the same task, which fails the second time around.

import asyncio
import anyio

async def main():
    async with anyio.open_cancel_scope() as scope:
        pass
    async with anyio.open_cancel_scope() as scope:
        pass
    await asyncio.sleep(0.1)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

curio task groups eat errors

import anyio


async def err():
    print("This definitely runs!")
    raise AttributeError("fewsfwsefwf")


async def main():
    async with anyio.create_task_group() as tg:
        await tg.spawn(err)


anyio.run(main, backend="curio")

Output in my PyCharm:

This definitely runs!

Process finished with exit code 0

socket.sendall broken if OS doesn't accept whole buffer in one go

This comment notes that anyio's sendall corrupts data: python-trio/purerpc#10 (comment)

Looking at the implementation, it seems obvious that we need to adjust the buffer we pass to send based on how much data was already sent.

Add server side support for opportunistic TLS

Currently, if an SSLContext has been provided to a stream server, it will always try to do a handshake at the beginning. For services like IMAP which have the STARTTLS command, there should be a way to disable the automatic handshake but leave it as an option to do it later, using the provided SSLContext.