[AIO] Minimal AsyncIO Server #20598

lidizheng · 2019-10-14T22:28:50Z

Related issue: #19496

This PR:

Extends AsyncIO IO manager to support server-side operations;
Adds more logic to AsyncSocket class;
Implements an AsyncIO server that can serve unary-unary handlers;
Adds a server test with grpc.aio.Channel;
Support both Bazel / setup.py build.

* Extends AsyncIO IO manager to support server-side operations; * Adds more logic to AsyncSocket class; * Implements an AsyncIO server that can serve unary-unary handlers; * Adds a server test with grpc.aio.Channel; * Support both Bazel / setup.py build.

pfreixes

Thanks for the hard work Lidiz!

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pyx.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pyx.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pxd.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

src/python/grpcio/grpc/experimental/aio/_server.py

pfreixes · 2019-10-17T10:11:47Z

src/python/grpcio_tests/tests_aio/benchmark/server.py

+    aio.init_grpc_aio()
+    loop = asyncio.get_event_loop()
+    loop.create_task(_start_async_server())
+    loop.run_forever()


is this already used in someplace? where can we see the results?

This is not automated right now. I just have a bunch of manual scripts to generate the result.

The important part of the benchmark is the client-side, I had two version, one is multiprocessing Python client, and one is Golang client. They are not fitting in this directory, so I just checked in this one file.

will be Aio added as part of the current and official gRPC benchmark results?

Yes, I will add it later. To add it into our benchmark suite:

The Async API should support both unary and streaming RPC;

Can work with sync API in same process (the control plane is also in gRPC).

src/python/grpcio_tests/tests_aio/unit/server_test.py

lidizheng

Thank you for reviewing! All comments resolved or discussed. PTALA.
(Bazel seems still fail on Kokoro, it passes locally.)

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pyx.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pxd.pxi

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

src/python/grpcio/grpc/experimental/aio/_server.py

lidizheng · 2019-10-17T22:06:32Z

src/python/grpcio_tests/tests_aio/benchmark/server.py

+    aio.init_grpc_aio()
+    loop = asyncio.get_event_loop()
+    loop.create_task(_start_async_server())
+    loop.run_forever()


This is not automated right now. I just have a bunch of manual scripts to generate the result.

The important part of the benchmark is the client-side, I had two version, one is multiprocessing Python client, and one is Golang client. They are not fitting in this directory, so I just checked in this one file.

src/python/grpcio_tests/tests_aio/unit/server_test.py

gnossen

Awesome stuff! I'm looking forward to using an asyncio server.

In general, it would be great if we could get more inline comments and docstrings. One of the biggest problems with the old stack is that it's pretty much impenetrable for anyone but the original authors. Optimizing for future readers should be a priority in the new stack.

gnossen · 2019-10-18T18:38:41Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+
+    cdef grpc_resolved_address c_addr
+    hostname = str_to_bytes(peer[0])
+    grpc_string_to_sockaddr(&c_addr, hostname, peer[1])


This is just a nit (since this has apparently been a problem for a while).

This function's signature is really odd. Right now, it's

void grpc_string_to_sockaddr(grpc_resolved_address* out, char* addr, int port)

It should probably be

void grpc_string_to_sockaddr(const char* addr, int port, grpc_resolved_address* out)

outparams should come last (though there are apparently quite a few functions that break that in core and C++)

addr is only ever read, so it should be const

gnossen · 2019-10-18T18:45:39Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+    cdef grpc_resolved_address c_addr
+    hostname = str_to_bytes(peer[0])
+    grpc_string_to_sockaddr(&c_addr, hostname, peer[1])
+    string.memcpy(<void*>addr, <void*>c_addr.addr, c_addr.len)


It's unfortunate that we have to copy here. It seems like it's only necessary to fit two interfaces together. Since we're in control of both interfaces, is it possible to eliminate this?

I added a TODO. Changing the underlying C-Core interface can be risky. And this is in the control plane, so the impact is not that high.

gnossen · 2019-10-18T19:08:44Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

@@ -81,39 +82,82 @@ cdef grpc_error* asyncio_socket_getpeername(
        grpc_custom_socket* grpc_socket,
        const grpc_sockaddr* addr,
        int* length) with gil:


Where is the GIL being used in this function? Unless I'm missing something, everything appears to be cdef.

I see. socket._peername is a Python object as well. Would it make sense to make it a native struct so we don't have to hold the GIL in this function?

socket._peername is a Python tuple, so I think we still need GIL here.

Python is treating all the addresses in a weird way that they are tuples of (address, port).

gnossen · 2019-10-18T19:09:49Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+    grpc_string_to_sockaddr(&c_addr, hostname, peer[1])
+    string.memcpy(<void*>addr, <void*>c_addr.addr, c_addr.len)
+    length[0] = c_addr.len
+    return grpc_error_none()


 cdef grpc_error* asyncio_socket_getsockname(
        grpc_custom_socket* grpc_socket,
        const grpc_sockaddr* addr,
        int* length) with gil:


Likewise. Where is the GIL being used?

Any access to Python object requires GIL; since they needs to keep refcount accurate. #i_am_inevitable

Ah. I see socket.sockname() now. Can we reduce the scope during which we hold the GIL to just that portion of the code?

To solve this problem correctly, we need to modify the signature of IO manager which is shared between gevent and AsyncIO. This work is doable, but for single thread application the impact for performance won't be that significant. I would prefer delay it until we introduced the sync-handler-executes-in-thread-pool feature.

I created a TODO issue #20689.

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

gnossen · 2019-10-18T20:49:48Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

+
+
+async def _handle_unary_unary_rpc(object method_handler, RPCState rpc_state, object loop):
+    # Receives request message


I think the fact that these blocks are commented signals that they should be broken out into well-named functions. If performance is a concern, try cdef inline.

I prefer this way since it gives just enough information for reader to understand all the details. Also, await can not be added to cdef.

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

lidizheng

Thank you for the detailed review. I removed the _AioServerState, but kept the RPCState, because we need a representation to track the life cycle of an RPC, which lives across the boundary of functions.

lidizheng · 2019-10-18T23:59:47Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+    grpc_string_to_sockaddr(&c_addr, hostname, peer[1])
+    string.memcpy(<void*>addr, <void*>c_addr.addr, c_addr.len)
+    length[0] = c_addr.len
+    return grpc_error_none()


 cdef grpc_error* asyncio_socket_getsockname(
        grpc_custom_socket* grpc_socket,
        const grpc_sockaddr* addr,
        int* length) with gil:


Any access to Python object requires GIL; since they needs to keep refcount accurate. #i_am_inevitable

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

lidizheng · 2019-10-19T00:09:44Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+            socket = native_socket.socket(family=native_socket.AF_INET6)
+            _asyncio_apply_socket_options(socket)
+            socket.bind((host, port))
+        except native_socket.gaierror:


I modified to use ipaddress (Python 3.3+) to detect if it is IPv6. The only concern is I hope the grpc_sockaddr they passes in is IP instead of other hostnames.

lidizheng · 2019-10-19T00:14:51Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+    cdef grpc_resolved_address c_addr
+    hostname = str_to_bytes(peer[0])
+    grpc_string_to_sockaddr(&c_addr, hostname, peer[1])
+    string.memcpy(<void*>addr, <void*>c_addr.addr, c_addr.len)


I added a TODO. Changing the underlying C-Core interface can be risky. And this is in the control plane, so the impact is not that high.

lidizheng · 2019-10-19T00:18:57Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

+            _asyncio_apply_socket_options(socket)
+            socket.bind((host, port))
+    except IOError as io_error:
+        return socket_error("bind", str(io_error))


Seems str is better, it gives the context.

try: open('a', 'r') except Exception as e: print(repr(e)) # IOError(2, 'No such file or directory') print(str(e)) # [Errno 2] No such file or directory: 'a'

src/python/grpcio_tests/tests_aio/unit/BUILD.bazel

src/python/grpcio_tests/tests_aio/unit/server_test.py

lidizheng · 2019-10-19T00:59:45Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pxd.pxi

+
+
+cdef class AioServer:
+    cdef _AioServerState _state


_AioServerState removed. Since Cython cannot use C pointers in Python function signature, I added a _CallbackCompletionQueue class to wrap the C pointer to completion queue.

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

pfreixes · 2019-10-20T15:36:14Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

+        # await waiter
+
+        loop.create_task(_handle_rpc(server_state, rpc_state, loop))
+        await asyncio.sleep(0)


The coroutine will break the execution returning back the control to the loop here [1], no matter if gRPC already has something for being delivered - like a pending request, the future that is being awaited within the _server_call_request_call function will have always to yield the execution for at least one loop iteration which would give enough time for executing the already accepted request.,

The only reason that I would think that this could help will be an optimization for giving more priority to the current request. But TBH doing this, we will start polluting our code with await asyncio.sleep(0) when we do not have yet probes that this is really helping a lot.

[1] https://github.com/grpc/grpc/pull/20598/files#diff-b53c077f7911f53dc8ce7656e16d35cdR184

pfreixes · 2019-10-20T15:45:26Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

+    async def start(self):
+        loop = asyncio.get_event_loop()
+        loop.create_task(_server_start(self._state))
+        await asyncio.sleep(0)


if the idea is helping the unit test I would move the await asyncio.sleep(0) into the unit test. If the idea is having a kind of synchronization, so having the start coroutine waiting till the server is already accepting requests then maybe we should make it a bit more explicitly by using an Asyncio synchronization primitive, which will be easily testable.

pfreixes · 2019-10-20T15:51:42Z

src/python/grpcio_tests/tests_aio/benchmark/server.py

+    aio.init_grpc_aio()
+    loop = asyncio.get_event_loop()
+    loop.create_task(_start_async_server())
+    loop.run_forever()


will be Aio added as part of the current and official gRPC benchmark results?

gnossen · 2019-10-19T01:16:03Z

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

@@ -81,39 +82,82 @@ cdef grpc_error* asyncio_socket_getpeername(
        grpc_custom_socket* grpc_socket,
        const grpc_sockaddr* addr,
        int* length) with gil:


I see. socket._peername is a Python object as well. Would it make sense to make it a native struct so we don't have to hold the GIL in this function?

lidizheng added kind/enhancement lang/Python priority/P2 release notes: yes Indicates if PR needs to be in release notes labels Oct 14, 2019

lidizheng requested a review from gnossen October 14, 2019 22:28

lidizheng assigned gnossen Oct 14, 2019

lidizheng force-pushed the aio-server branch from c3de19c to 70c4e86 Compare October 14, 2019 22:31

Minimal AsyncIO Server for gRPC

c6ae98d

* Extends AsyncIO IO manager to support server-side operations; * Adds more logic to AsyncSocket class; * Implements an AsyncIO server that can serve unary-unary handlers; * Adds a server test with grpc.aio.Channel; * Support both Bazel / setup.py build.

lidizheng force-pushed the aio-server branch from 70c4e86 to c6ae98d Compare October 14, 2019 22:32

lidizheng added 4 commits October 15, 2019 17:22

Cosmetic changes to increase readability

175c8e4

Try to pin the python_version to PY3

995c4f4

Merge branch 'master' into aio-server

25be7c9

Make YAPF happy

3c7cd15

pfreixes requested changes Oct 17, 2019

View reviewed changes

Adopt reviewer's advices

49c7f1d

lidizheng commented Oct 17, 2019

View reviewed changes

lidizheng mentioned this pull request Oct 18, 2019

provide strongly-typed option 3 lidizheng/grpc-api-examples#1

Open

gnossen requested changes Oct 18, 2019

View reviewed changes

Adopt reviewer's suggestions

2ced359

lidizheng commented Oct 19, 2019

View reviewed changes

pfreixes requested changes Oct 20, 2019

View reviewed changes

Remove asyncio.sleep(0)

08bbbcd

gnossen approved these changes Oct 21, 2019

View reviewed changes

pfreixes approved these changes Oct 21, 2019

View reviewed changes

lidizheng added 2 commits October 21, 2019 16:00

Suppress the ImportError of ipaddress for Python 2

fbe1bc9

Remove the usage of f-string

a588b5f

lidizheng merged commit 3c62c8a into grpc:master Oct 22, 2019

pfreixes mentioned this pull request Oct 23, 2019

[Aio] Replace the test client end2end server fixture for an asynchronous one #19762

Closed

lock bot locked as resolved and limited conversation to collaborators Jan 24, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[AIO] Minimal AsyncIO Server #20598

[AIO] Minimal AsyncIO Server #20598

lidizheng commented Oct 14, 2019

pfreixes left a comment

pfreixes Oct 17, 2019

lidizheng Oct 17, 2019

pfreixes Oct 20, 2019

lidizheng Oct 21, 2019

lidizheng left a comment

lidizheng Oct 17, 2019

gnossen left a comment •

edited

Loading

gnossen Oct 18, 2019

gnossen Oct 18, 2019

lidizheng Oct 19, 2019

gnossen Oct 18, 2019

gnossen Oct 19, 2019

lidizheng Oct 21, 2019

gnossen Oct 18, 2019

lidizheng Oct 18, 2019

gnossen Oct 19, 2019 •

edited

Loading

lidizheng Oct 19, 2019

gnossen Oct 18, 2019

lidizheng Oct 19, 2019

lidizheng left a comment

lidizheng Oct 18, 2019

lidizheng Oct 19, 2019

lidizheng Oct 19, 2019

lidizheng Oct 19, 2019

lidizheng Oct 19, 2019

pfreixes Oct 20, 2019

pfreixes Oct 20, 2019

pfreixes Oct 20, 2019

gnossen Oct 19, 2019



		async def _handle_unary_unary_rpc(object method_handler, RPCState rpc_state, object loop):
		# Receives request message

[AIO] Minimal AsyncIO Server #20598

[AIO] Minimal AsyncIO Server #20598

Conversation

lidizheng commented Oct 14, 2019

pfreixes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lidizheng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gnossen left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gnossen Oct 19, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lidizheng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gnossen left a comment •

edited

Loading

gnossen Oct 19, 2019 •

edited

Loading