Merge pull request #23783 from lidizheng/aio-win-poll

[Aio] Add a fail-back polling mode for Windows+3.8+
reviewable/pr23871/r1
Lidi Zheng 5 years ago committed by GitHub
commit cd336404e9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 1
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pxd.pxi
  2. 33
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi
  3. 13
      tools/run_tests/run_tests.py

@ -53,6 +53,7 @@ cdef class BaseCompletionQueue:
cdef class _BoundEventLoop:
cdef readonly object loop
cdef readonly object read_socket # socket.socket
cdef bint _has_reader
cdef class PollerCompletionQueue(BaseCompletionQueue):

@ -17,6 +17,9 @@ import socket
cdef gpr_timespec _GPR_INF_FUTURE = gpr_inf_future(GPR_CLOCK_REALTIME)
cdef float _POLL_AWAKE_INTERVAL_S = 0.2
# This bool indicates if the event loop impl can monitor a given fd, or has
# loop.add_reader method.
cdef bint _has_fd_monitoring = True
IF UNAME_SYSNAME == "Windows":
cdef void _unified_socket_write(int fd) nogil:
@ -41,17 +44,29 @@ cdef class BaseCompletionQueue:
cdef class _BoundEventLoop:
def __cinit__(self, object loop, object read_socket, object handler):
global _has_fd_monitoring
self.loop = loop
self.read_socket = read_socket
reader_function = functools.partial(
handler,
loop
)
self.loop.add_reader(self.read_socket, reader_function)
# NOTE(lidiz) There isn't a way to cleanly pre-check if fd monitoring
# support is available or not. Checking the event loop policy is not
# good enough. The application can has its own loop implementation, or
# uses different types of event loops (e.g., 1 Proactor, 3 Selectors).
if _has_fd_monitoring:
try:
self.loop.add_reader(self.read_socket, reader_function)
self._has_reader = True
except NotImplementedError:
_has_fd_monitoring = False
self._has_reader = False
def close(self):
if self.loop:
self.loop.remove_reader(self.read_socket)
if self._has_reader:
self.loop.remove_reader(self.read_socket)
cdef class PollerCompletionQueue(BaseCompletionQueue):
@ -97,7 +112,14 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
self._queue_mutex.lock()
self._queue.push(event)
self._queue_mutex.unlock()
_unified_socket_write(self._write_fd)
if _has_fd_monitoring:
_unified_socket_write(self._write_fd)
else:
with gil:
# Event loops can be paused or killed at any time. So,
# instead of deligate to any thread, the polling thread
# should handle the distribution of the event.
self._handle_events(None)
def _poll_wrapper(self):
with nogil:
@ -119,7 +141,10 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
self._write_socket.close()
def _handle_events(self, object context_loop):
cdef bytes data = self._read_socket.recv(1)
cdef bytes data
if _has_fd_monitoring:
# If fd monitoring is working, clean the socket without blocking.
data = self._read_socket.recv(1)
cdef grpc_event event
cdef CallbackContext *context

@ -819,25 +819,30 @@ class PythonLanguage(object):
if args.compiler == 'default':
if os.name == 'nt':
return (python36_config,)
if args.iomgr_platform == 'gevent':
# TODO(https://github.com/grpc/grpc/issues/23784) allow
# gevent to run on later version once issue solved.
return (python36_config,)
else:
return (python38_config,)
else:
if args.iomgr_platform == 'asyncio':
return (python36_config,)
return (python36_config, python38_config)
elif os.uname()[0] == 'Darwin':
# NOTE(rbellevi): Testing takes significantly longer on
# MacOS, so we restrict the number of interpreter versions
# tested.
return (
python27_config,
python36_config,
python37_config,
python38_config,
)
else:
return (
python27_config,
python35_config,
python36_config,
python37_config,
python38_config,
)
elif args.compiler == 'python2.7':
return (python27_config,)

Loading…
Cancel
Save