Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add asyncio.streams Comm #2165

Closed
wants to merge 7 commits into from
Closed

Add asyncio.streams Comm #2165

wants to merge 7 commits into from

Conversation

mrocklin
Copy link
Member

@mrocklin mrocklin commented Aug 7, 2018

This only works with Python 3.7. I had to pull out and modify some asyncio code. It also makes no attempt at TLS.

infos = [socket.getaddrinfo(host, port, family=family, flags=flags) for
host in hosts]
infos = set(itertools.chain.from_iterable(infos))
infos = [info for info in infos if info[1] == socket.SocketKind.SOCK_STREAM]
Copy link
Member Author

@mrocklin mrocklin Aug 7, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I changed the first two of these lines to use the socket blocking API rather than asyncio's non-blocking API.

The third line is a hack. I added it to get past errors like the following:

    def _start_serving(self):
        if self._serving:
            return
        self._serving = True
        for sock in self._sockets:
>           sock.listen(self._backlog)
E           OSError: [Errno 95] Operation not supported

I suspect that there is a way to handle this by passing in the right arguments to start_server, but I haven't found it yet.

@mrocklin
Copy link
Member Author

mrocklin commented Aug 8, 2018

This does get us down to about a 600us roundtrip time, down from 800us, which is a bit of a win.

In [1]: from dask.distributed import Client
   ...: client = Client()
   ...: client
   ...: 
   ...: 
Out[1]: <Client: scheduler='tcp://127.0.0.1:35289' processes=4 cores=4>

In [2]: async def f():
   ...:     for i in range(10000):
   ...:         await client.scheduler.identity()
   ...: %time client.sync(f)
   ...: 
   ...: 
CPU times: user 5.8 s, sys: 467 ms, total: 6.27 s
Wall time: 6.31 s

@mrocklin
Copy link
Member Author

mrocklin commented Aug 8, 2018

Oooh, I can get this down to 470us if

  1. I bundle all of the writes into one for small messages (this is definitely doable)
  2. I engage Asyncio equivalent of call_soon tornadoweb/tornado#2463

At this point about 70% of scheduler time is spent in socket.send (see https://stackoverflow.com/questions/51731690/why-does-asyncio-spend-time-in-socket-senddata)

@mrocklin mrocklin changed the title WIP prototype of an Asyncio streams comm Add asyncio.streams Comm Aug 8, 2018
@mrocklin
Copy link
Member Author

mrocklin commented Aug 8, 2018

OK, this could use review by someone who knows this stack better than I do.

This currently implements a functional Comm using asyncio streams. This seems to perform a bit better when doing many small fast messages. I haven't yet checked with larger messages (though that should be interesting).

There are a few challenges:

  1. asyncio's start_server function is asynchronous, which means that Listener.start would need to be asynchronous, which means that all of the Dask Server.listen methods would need to become asynchronous (they are currently not asynchronous) which would be a bit of a pain.

    Currently my solution to this is to copy over and patch a couple functions in asyncio (see distributed/comm/_asyncio_utils.py), but this is fragile and only works for specific versions (the Server class changed a bit between Python 3.6 and 3.7).

  2. I haven't figured out how to do IPV6. I suspect that this is a simple matter of passing a different flag when creating a connections or starting a server.

  3. Currently we fail distributed/comm/tests/test_comms.py::test_tcp_comm_closed_implicit. When the server severs our connection we don't feel it on the client side.

@pitrou if you have time to look things over here I would appreciate it.

@mrocklin
Copy link
Member Author

mrocklin commented Aug 8, 2018

Current tests on the CI systems aren't that meaningful. This only works in Python 3.7 due to copying over and modifying asyncio code.

@mrocklin
Copy link
Member Author

mrocklin commented Aug 8, 2018

Oh, and I'm also somewhat blocked on this problem: https://stackoverflow.com/questions/51731690/why-does-asyncio-spend-time-in-socket-senddata

A very simple benchmark that sends a small message back and forth currently gets up to spending about 70% of its time in socket.send.

@pitrou
Copy link
Member

pitrou commented Aug 9, 2018

I also suggest you run a benchmark with large messages, because asyncio doesn't have all the optimizations that we added to Tornado.

@pitrou
Copy link
Member

pitrou commented Aug 9, 2018

A very simple benchmark that sends a small message back and forth currently gets up to spending about 70% of its time in socket.send.

Just for the record, which profiler are you using? cProfile or a sampling profiler? Do you have other threads going on in the background?

And how much send calls is that per second or, another way to phrase it, what is the average duration of a send call?

@mrocklin
Copy link
Member Author

mrocklin commented Aug 9, 2018

Just for the record, which profiler are you using? cProfile or a sampling profiler? Do you have other threads going on in the background?

Sampling profiler. The same one that we use for worker threads. See #2144

I also suggest you run a benchmark with large messages, because asyncio doesn't have all the optimizations that we added to Tornado.

I agree. My short-term plan is to include both, but have tornado be default. I'm dealing with some workloads now that need relatively low-latency communication of many small messages. Using this and many other tricks I can get round-trip message latency down to about 450us (see #2156)

And how much send calls is that per second or, another way to phrase it, what is the average duration of a send call?

That's a good question. I'll find out.

@@ -0,0 +1,186 @@
"""
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This stuff is probably no longer necessary. Since doing this work we've made Listeners optionaly awaitable, so we should be able to handle things appropriately on the Dask side.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

3 participants