Mailman 3 - Async-sig

Additive merging of async generators
by James Stidard 20 Jul '18

20 Jul '18

Hi, I was wondering if anyone has had experience implementing a similar pattern to this or has alternative suggestions of how to achieve it. I have a bunch of async generators which I’d like to be able to merge into a single async generator I can iterate over. I found Vincent’s aiostream library which gives me this without too much effort: from asyncio import sleep, run from aiostream.stream import merge async def go(): yield 0 await sleep(1) yield 50 await sleep(1) yield 100 async def main(): tasks = merge(go(), go(), go()) async for v in tasks: print(v) if __name__ == '__main__': run(main()) However, I also would like to be able to add additional tasks into the list once I’ve started iterating the list so something more akin to: from asyncio import sleep, run from aiostream.stream import merge async def go(): yield 0 await sleep(1) yield 50 await sleep(1) yield 100 async def main(): tasks = merge(go(), go(), go()) async for v in tasks: If v == 50: tasks.merge(go()) print(v) if __name__ == '__main__': run(main()) Has anyone been able to achieve something like this? p.s. I know appending to a list you’re iterating is bad practice, I assume the same would be true modifying this stream object, but think the example illustrates what I’m trying to achieve. Thanks, James

2 3

Re: [Async-sig] Additive merging of async generators
by twisteroid ambassador 19 Jul '18

19 Jul '18

If I were to implement something like this from scratch, I would write a class that basically wraps an asyncio.Queue. For each input async generator to be merged, simply iterate through it and put each element in the queue, and write an output async generator that gets and yields elements from the queue. Maintain a counter of source generators that haven't been exhausted yet, and when that falls to 0, stop the output generator. Should be pretty straightforward.

1 0

Tips and tricks to track memory leaks in realtime inside an AsyncIO daemon ?
by Ludovic Gasc 18 Jun '18

18 Jun '18

Hi, We have now an AsyncIO daemon with memory leaks. To track this, I'm thinking to use this: https://pythonhosted.org/Pympler/muppy.html and objgraph. But, because it's a live daemon and not a script, instead to implement an HTTP endpoint to launch the memory snapshot locally, I'm thinking to use epdb: https://github.com/sassoftware/epdb To have directly a Python console to explore interactively inside the process to have a more flexible way to debug. I did a quick'n'dirty lab', it seems to work more or less to run epdb inside a thread executor, but before to continue on this idea, I'm interested in how you track your memory leaks inside your AsyncIO daemons ? Thanks for your feedbacks. -- Ludovic Gasc (GMLudo)

2 2

async file IO library
by Dima Tisnek 09 Jun '18

09 Jun '18

Hi group, It seems that Python docs don't make a recommendation about which library to use for asynchronous access to files when using asyncio. Is there a canonical recommendation? Is it a good idea? There's at least one 3rd party library popular enough to deserve own stackoverflow tag, however AFAIU it uses thread pool, while most OS'es provide some or other API, making native implementation possible, at least in theory. What are your thoughts? Thanks, Dima

2 1

Who will be present at EuroPython 2018 ?
by Ludovic Gasc 20 May '18

20 May '18

Hi, You certainly know that EuroPython is 23-29 July: https://ep2018.europython.eu/en/ I have seen Yury's tweet about it: https://twitter.com/1st1/status/997910868573720579 Who has planned to be present ? It might be the opportunity to do a sprint code around AsyncIO to improve it or increase the documentation like: https://asyncio.readthedocs.io/en/latest/ Have a nice week-end. -- Ludovic Gasc (GMLudo)

2 1

Re: [Async-sig] [python-tulip] Re: asyncio.Lock equivalent for multiple processes
by Ludovic Gasc 20 May '18

20 May '18

Hi Antoine & Chris, Thanks a lot for the advisory lock, I didn't know this feature in PostgreSQL. Indeed, it seems to fit my problem. The small latest problem I have is that we have string names for locks, but advisory locks accept only integers. Nevertheless, it isn't a problem, I will do a mapping between names and integers. Yours. -- Ludovic Gasc (GMLudo) 2018-04-17 13:41 GMT+02:00 Antoine Pitrou <solipsis(a)pitrou.net>: > On Tue, 17 Apr 2018 13:34:47 +0200 > Ludovic Gasc <gmludo(a)gmail.com> wrote: > > Hi Nickolai, > > > > Thanks for your suggestions, especially for the file system lock: We > don't > > have often locks, but we must be sure it's locked. > > > > For 1) and 4) suggestions, in fact we have several systems to sync and > also > > a PostgreSQL transaction, the request must be treated by the same worker > > from beginning to end and the other systems aren't idempotent at all, > it's > > "old-school" proprietary systems, good luck to change that ;-) > > If you already have a PostgreSQL connection, can't you use a PostgreSQL > lock? e.g. an "advisory lock" as described in > https://www.postgresql.org/docs/9.1/static/explicit-locking.html > > Regards > > Antoine. > > >

3 5

async_stagger: Happy Eyeballs implementation in pure asyncio
by twisteroid ambassador 15 May '18

15 May '18

Repo: https://github.com/twisteroidambassador/async_stagger Docs: http://async-stagger.readthedocs.io/en/latest/ Provides near drop-in replacements for open_connection() and create_connection() using Happy Eyeballs. Also exposes the underlying scheduling logic where you can plug in your own coroutines to run. I basically ported trio's implementation to asyncio, and it turned out to be not too difficult. Cheers, twistero

5 6

ANN: miniasync, a small library build on top of asyncio for simple use cases
by Alice Heaton 04 May '18

04 May '18

Hello, I have been working on miniasync, a small library build on top of asyncio to facilitate the occasional use of async code in otherwise synchronous applications. A typical use case is an otherwise synchronous application, which at some point wants to send notifications to multiple online services. miniasync makes it easier to implement this asynchronously without having to write asyncio boilerplate. http://miniasync.readthedocs.io asyncio is vast, powerful and flexible. But there is a learning curve - you need to understand the event loop api (other async languages like Javascript don't expose that), how to create it and run tasks on it, which means you need to understand about Tasks and Futures, etc. The flexibility is great, and gives Python developers a lot of scope for implementing advanced applications, but it's a put off for simple use cases. miniasync exposes a single function, "miniasync.run", which takes a list of coroutine objects, creates a local event loop, runs all the co-routines until they're complete, and returns their result in the order they were defined: import aiofiles import miniasync async def get_file_content(filename): async with aiofiles.open(filename, mode='r') as f: return await f.read() results = miniasync.run( get_file_content('file1.txt'), get_file_content('file2.txt'), ) assert results == [ '<the content of file1.txt>', '<the content of file2.txt>' ] This is similar to using a combination of gather and run_until_complete, but as a single step and without explicit reference to the loop. This also differs from gather in that: - Instead of saying you either want exceptions raised or returned, you need to list explicitly exceptions you want returned, all other exceptions are raised (explicit is better than implicit); - Unhandled exceptions cause all other tasks to be cancelled. miniasync.run always creates a new loop (so you can nest invocations of miniasync.run, and rely on each invocation only executing it's parameters). For the cases where you need the loop before running (eg. for creating a asyncio.Queue object to be shared amongst your co-routines), miniasync also exposes a context manager that creates a loop and lets you run co-routines on it: import asyncio import miniasync async def coro1(q): q.put_nowait('world') return 'hello' async def coro2(q): return await q.get() with miniasync.loop() as loop: q = asyncio.Queue() results = loop.run( coro1(q), coro2(q) ) assert results == ['hello', 'world'] You can pip install miniasync to try out, or read the docs on readthedocs (as above). For now miniasync covers what was my main issue with running simple async code. Other things I have in mind for the future is a simple interface for running multiple http requests (based on top of aiohttp). I'm keen to hear about other issues that could be simplified for simple use cases. :) Alice

3 3

"Coroutines" sometimes run without being scheduled on an event loop
by twisteroid ambassador 04 May '18

04 May '18

Hi, tl;dr: coroutine functions and regular functions returning Futures behave differently: the latter may start running immediately without being scheduled on a loop, or even with no loop running. This might be bad since the two are sometimes advertised to be interchangeable. I find that sometimes I want to construct a coroutine object, store it for some time, and run it later. Most times it works like one would expect: I call a coroutine function which gives me a coroutine object, I hold on to the coroutine object, I later await it or use loop.create_task(), asyncio.gather(), etc. on it, and only then it starts to run. However, I have found some cases where the "coroutine" starts running immediately. The first example is loop.run_in_executor(). I guess this is somewhat unsurprising since the passed function don't actually run in the event loop. Demonstrated below with strace and the interactive console: $ strace -e connect -f python3 Python 3.6.5 (default, Apr 4 2018, 15:01:18) [GCC 7.3.1 20180303 (Red Hat 7.3.1-5)] on linux Type "help", "copyright", "credits" or "license" for more information. >>> import asyncio >>> import socket >>> s = socket.socket() >>> loop = asyncio.get_event_loop() >>> coro = loop.sock_connect(s, ('127.0.0.1', 80)) >>> loop.run_until_complete(asyncio.sleep(1)) >>> task = loop.create_task(coro) >>> loop.run_until_complete(asyncio.sleep(1)) connect(3, {sa_family=AF_INET, sin_port=htons(80), sin_addr=inet_addr("127.0.0.1")}, 16) = -1 ECONNREFUSED (Connection refused) >>> s.close() >>> s = socket.socket() >>> coro2 = loop.run_in_executor(None, s.connect, ('127.0.0.1', 80)) strace: Process 13739 attached >>> [pid 13739] connect(3, {sa_family=AF_INET, sin_port=htons(80), sin_addr=inet_addr("127.0.0.1")}, 16) = -1 ECONNREFUSED (Connection refused) >>> coro2 <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib64/python3.6/asyncio/futures.py:403]> >>> loop.run_until_complete(asyncio.sleep(1)) >>> coro2 <Future finished exception=ConnectionRefusedError(111, 'Connection refused')> >>> Note that with loop.sock_connect(), the connect syscall is only run after loop.create_task() is called on the coroutine AND the loop is running. On the other hand, as soon as loop.run_in_executor() is called on socket.connect, the connect syscall gets called, without the event loop running at all. Another such case is with Python 3.4.2, where even loop.sock_connect() will run immediately: $ strace -e connect -f python3 Python 3.4.2 (default, Oct 8 2014, 10:45:20) [GCC 4.9.1] on linux Type "help", "copyright", "credits" or "license" for more information. >>> import socket >>> import asyncio >>> loop = asyncio.get_event_loop() >>> s = socket.socket() >>> c = loop.sock_connect(s, ('127.0.0.1', 82)) connect(7, {sa_family=AF_INET, sin_port=htons(82), sin_addr=inet_addr("127.0.0.1")}, 16) = -1ECONNREFUSED (Connection refused) >>> c <Future finished exception=ConnectionRefusedError(111, 'Connection refused')> >>> In both these cases, the misbehaving "coroutine" aren't actually defined as coroutine functions, but regular functions returning a Future, which is probably why they don't act like coroutines. However, coroutine functions and regular functions returning Futures are often used interchangeably: Python docs Section 18.5.3.1 even says: > Note: In this documentation, some methods are documented as coroutines, even if they are plain Python functions returning a Future. This is intentional to have a freedom of tweaking the implementation of these functions in the future. In particular, both run_in_executor() and sock_connect() are documented as coroutines. If an asyncio API may change from a function returning Future to a coroutine function and vice versa any time, then one cannot rely on the behavior of creating the "coroutine object" not running the coroutine immediately. This seems like an important Gotcha waiting to bite someone. Back to the scenario in the beginning. If I want to write a function that takes coroutine objects and schedule them to run later, and some coroutine objects turn out to be misbehaving like above, then they will run too early. To avoid this, I could either 1. pass the coroutine functions and their arguments separately "callback style", 2. use functools.partial or lambdas, or 3. always pass in real coroutine objects returned from coroutine functions defined with "async def". Does this sound right? Thanks, twistero

6 9

New blog post: Notes on structured concurrency, or: Go statement considered harmful
by Nathaniel Smith 27 Apr '18

27 Apr '18

Hi all, I just posted another essay on concurrent API design: https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-con… This is the one that finally gets at the core reasons why Trio exists; I've been trying to figure out how to write it for at least a year now. I hope you like it. (Guido: this is the one you should read :-). Or if it's too much, you can jump to the conclusion [1], and I'm happy to come find you somewhere with a whiteboard, if that'd be helpful!) -n [1] https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-con… -- Nathaniel J. Smith -- https://vorpus.org

5 8