Mailman 3 - Async-sig

no current event loop in thread 'MainThread'
by Pradip Caulagi July 7, 2017

July 7, 2017

I am trying to write a test for a function that uses async. Can I mix using unittest.TestCase and asyncio.test_utils.TestCase? When I do that, the loop in test_utils.TestCase seems to affect the other code I have. This is a simplified example - $ cat foo.py import asyncio import unittest from asyncio import test_utils async def foo(): return True class Foo1Test(test_utils.TestCase): def setUp(self): super().setUp() self.loop = self.new_test_loop() self.… [View More]

2 1

async documentation methods
by Dima Tisnek July 7, 2017

July 7, 2017

Hi all, I'm working to improve async docs, and I wonder if/how async methods ought to be marked in the documentation, for example library/async-sync.rst: """ ... It [lock] has two basic methods, `acquire()` and `release()`. ... """ In fact, these methods are not symmetric, the earlier is asynchronous and the latter synchronous: Definitions are `async def acquire()` and `def release()`. Likewise user is expected to call `await .acquire()` and `.release()`. This is user-facing documentation, … [View More]

7 15

Go-style generators in asyncio
by Vincent Michel July 6, 2017

July 6, 2017

Hi all, I've recently been looking into the go concurrency model to see how it compares to asyncio and an interesting concept caught my attention: go generators. It's quite similar to asynchronous generator, with a bit of extra concurrency. You can find a small write-up comparing the two concepts [1] and a possible implementation of go-style generators using asyncio [2]. [1]: https://gist.github.com/vxgmichel/4ea46d3ae5c270260471d304a2c8e97b [2]: https://gist.github.com/vxgmichel/… [View More]

1 0

Read remaining and ready buffer once it has been closed by your peer
by Pau Freixes July 5, 2017

July 5, 2017

Hi guys, The current implementation of StreamReader does not allow to read the remaining buffer if has had an exception and there was no waiters before to read the data. IMHO this is not the best way to handle this situation, the developer should be able to access to the remaining and ready buffer in somehow. For example, this data might bring some important stuff related to the reason of the exception. Ive just opened a bug [1] and made a PR with a simple proposal [2] with a very basic … [View More]

1 0

async testing question
by Chris Jerdonek July 4, 2017

July 4, 2017

I have a question about testing async code. Say I have a coroutine: async def do_things(): await do_something() await do_more() await do_even_more() And future: task = ensure_future(do_things()) Is there a way to write a test case to check that task.cancel() would behave correctly if, say, do_things() is waiting at the line do_more()? In real life, this situation can happen if a function like the following is called, and an exception happens in one of the … [View More]

6 11

question re: asyncio.Condition lock acquisition order
by Chris Jerdonek June 28, 2017

June 28, 2017

I have a couple questions about asyncio's synchronization primitives. Say a coroutine acquires an asyncio Condition's underlying lock, calls notify() (or notify_all()), and then releases the lock. In terms of which coroutines will acquire the lock next, is any preference given between (1) coroutines waiting to acquire the underlying lock, and (2) coroutines waiting on the Condition object itself? The documentation doesn't seem to say anything about this. Also, more generally (and I'm sure … [View More]

3 5

"read-write" synchronization
by Chris Jerdonek June 27, 2017

June 27, 2017

I'm relatively new to async programming in Python and am thinking through possibilities for doing "read-write" synchronization. I'm using asyncio, and the synchronization primitives that asyncio exposes are relatively simple [1]. Have options for async read-write synchronization already been discussed in any detail? I'm interested in designs where "readers" don't need to acquire a lock -- only writers. It seems like one way to deal with the main race condition I see that comes up would be to … [View More]

6 21

async generator confusion or bug?
by Dima Tisnek June 27, 2017

June 27, 2017

Hi group, I'm trying to cross-use an sync generator across several async functions. Is it allowed or a completely bad idea? (if so, why?) Here's MRE: import asyncio async def generator(): while True: x = yield print("received", x) await asyncio.sleep(0.1) async def user(name, g): print("sending", name) await g.asend(name) async def helper(): g = generator() await g.asend(None) await asyncio.gather(*[user(f"user-{x}", g) for x in range(3)]… [View More]

5 8

Cancelling SSL connection
by Mark E. Haase June 23, 2017

June 23, 2017

(I'm not sure if this is a newbie question or a bug report or something in between. I apologize in advance if its off-topic. Let me know if I should post this somewhere else.) If a task is cancelled while SSL is being negotiated, then an SSLError is raised, but there's no way (as far as I can tell) for the caller to catch it. (The example below is pretty contrived, but in an application I'm working on, the user can cancel downloads at any time.) Here's an example: import asyncio, random, … [View More]ssl async def download(host): ssl_context = ssl.create_default_context() reader, writer = await asyncio.open_connection(host, 443, ssl=ssl_context) request = f'HEAD / HTTP/1.1\r\nHost: {host}\r\n\r\n' writer.write(request.encode('ascii')) lines = list() while True: newdata = await reader.readline() if newdata == b'\r\n': break else: lines.append(newdata.decode('utf8').rstrip('\r\n')) return lines[0] async def main(): while True: task = asyncio.Task(download('www.python.org')) await asyncio.sleep(random.uniform(0.0, 0.5)) task.cancel() try: response = await task print(response) except asyncio.CancelledError: print('request cancelled!') except ssl.SSLError: print('caught SSL error') await asyncio.sleep(1) loop = asyncio.get_event_loop() loop.run_until_complete(main()) loop.close() Running this script yields the following output: HTTP/1.1 200 OK request cancelled! HTTP/1.1 200 OK HTTP/1.1 200 OK <asyncio.sslproto.SSLProtocol object at 0x7fe7c00e5a20>: SSL handshake failed Traceback (most recent call last): File "/usr/lib/python3.6/asyncio/base_events.py", line 803, in _create_connection_transport yield from waiter File "/usr/lib/python3.6/asyncio/tasks.py", line 304, in _wakeup future.result() concurrent.futures._base.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/lib/python3.6/asyncio/sslproto.py", line 577, in _on_handshake_complete raise handshake_exc File "/usr/lib/python3.6/asyncio/sslproto.py", line 638, in _process_write_backlog ssldata = self._sslpipe.shutdown(self._finalize) File "/usr/lib/python3.6/asyncio/sslproto.py", line 155, in shutdown ssldata, appdata = self.feed_ssldata(b'') File "/usr/lib/python3.6/asyncio/sslproto.py", line 219, in feed_ssldata self._sslobj.unwrap() File "/usr/lib/python3.6/ssl.py", line 692, in unwrap return self._sslobj.shutdown() ssl.SSLError: [SSL] shutdown while in init (_ssl.c:2299) Is this a bug that I should file, or is there some reason that it's intended to work this way? I can work around it with asyncio.shield(), but I think I would prefer for the asyncio/sslproto.py to catch the SSLError and ignore it. Maybe I'm being short sighted. Thanks, Mark [View Less]

3 3

Asyncio, implementing a new fair scheduling reactor
by Pau Freixes June 17, 2017

June 17, 2017

Hi guys, recently I've been trying to implement a POC of the default event loop implemented by asyncio but using a fair scheduling reactor. At the moment is just a POC [1], something to test the rationale and pending to be evolved in someone more mature, but before of that I would prefer to share my thoughts and get all of the comments from you. The current implementation is based on a FIFO queue that is filled with all of the callbacks that have to be executed, these callbacks can stand for: … [View More]1) Run tasks, either to be started or resumed 2) Run future callbacks. 3) Run scheduled callbacks. 4) Run file descriptors callbacks. Worth mentioning that most of them internally are chained in somehow, perhaps a future callback can wake up a resumed task. Also, have in mind that the API published by asyncio to schedule callbacks can be used by asyncio itself or by the user, callsoon for example. The usual flow the reactor is the following one: 1) Check the file descriptors with events, and stack the handlers into the reactor. 2) Pop all outdated scheduled callbacks and push them into the reactor. 3) Iterate for N first elements at the queue, where N stands for the number of the handles stacked at that moment. Future handles stacked during that iteration won't be handled, they must wait until next whole iteration 4) Go to the point 1. As you can observe here, the IO is only made once per loop and should wait until all handles that are in a specific moment are executed. This implements in somehow a natural backpressure, the read and also the accept the new connections will rely on the buffers run by the operating system. That implementation can be seen as simple, but it stands on a solid strategy and follows KISS design that helps to scare the bugs. Why fair scheduling? Not all code that is written in the same module, in terms of loop sharing, has the same requirements. Some part might need N and other parts M. When this implementation cant be decoupled, and it means that the cost of placing them into a separated pieces inside of your architecture are too expensive, in that scenario the developer cant express this difference to make the underlying implementation aware of that. For example, an API with a regular endpoint accessed by the user and another one with the health-check of the system, which has completely different requirements in terms of IO. Not only due to the nature of the resources accessed, also because of the frequency of use. Meanwhile, the healthcheck is accessed to a known a frequency at X seconds, the other endpoint has a variable frequency of use. Do you believe that asyncio will be able to preserve the health-check frequency at any moment? Absolutely not. Therefore, the idea of implementing a fair scheduling reactor is based on the needed of address these kind of situations, giving to the developer an interface to isolate different resources. Basic principles The basic principles of the implementation are: - The cost of the scheduling has to be the same of the current implementation, no overhead - The design has to follow the current one, having the implicit backpressure that was commented. I will focus in the second principle, taking into account that the first one is a matter of implementation. To achieve the same behavior, the new implementation only split the resources - handles, schedules, file descriptors - in isolated partitions to then implement for each partition the same algorithm than the current one. The developer can create a new partition using a new function called `spawn`, this function takes as an argument a coroutine, the task wrapped to that coroutine and all of the resources created inside this coroutine will belong to that partition. For example: >>> async def background_task(): >>> task = [ fetch() for i in range(1000)] >>> return (await asyncio.gather(*t)) >>> >>> async def foo(): >>> return (await asyncio.spawn(background_task())) All resources created inside the scope of the `background_tasks` are isolated to one partition. The 1000 sockets will schedule callbacks that will be stacked in the same queue. The partition is by default identified with the hash of the task that warps the `background_task`, but the user can pass an alternative value. >>> async def foo(): >>> return (await asyncio.spawn(healtheck(), partition='healthcheck')) Internally the implementation has a default ROOT partition that is used for all of these resources that are not executed inside of the scope of a spawn function. As you can guess, if you don't use the spawn method the reactor will run exactly as the current implementation. Having all the resources in the same queue. Round robin between partitions. The differents partitions that exist at some moment share the CPU resource using a round robin strategy. It gives the same chance to all partitions to run the same amount of handles, but with one particularity. Each time that a partition runs out of handles, the loop is restarted again to handle the file descriptors and the delayed calls but only for that specific partition that runs out of handles. The side effect is clear, have the same backpressure mechanism. But, per partition. The black hole of the current implementation. There is always a but, at least I've found a situation where this strategy can perform in the same way as the current one, without applying any fair scheduling. Although the code uses the spawn method. Have a look at the following snippet: >>> async def healtheck(request): >>> await request.resp() >>> >>> async def view(request): >>> return (await asyncio.spawn(healthcheck(request))) The task that wraps the healtcheck coroutine that is being isolated in a partition, won't be scheduled until the data from the file descriptor that is read by a callback that is in fact executed inside of the ROOT partition. Therefore, in the worst case scenario, the fair scheduling will become a simple FIFO scheduling. IMHO there is not an easy way to solve that issue, or at least without changing the current picture. And try to solve it, might end up having a messy implementation and a buggy code. Although, I believe that this still worth it, having in mind the benefits that it will bring us for all of those cases where the user needs to isolate resources. Thoughts, comments, and others will be welcomed. [1] https://github.com/pfreixes/qloop -- --pau [View Less]

3 3