Make the fail back mode triggering more robust

pull/23783/head
Lidi Zheng 4 years ago
parent c0632acaff
commit 03813bf6fc
  1. 1
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pxd.pxi
  2. 43
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi

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

@ -17,12 +17,22 @@ import socket
cdef gpr_timespec _GPR_INF_FUTURE = gpr_inf_future(GPR_CLOCK_REALTIME) cdef gpr_timespec _GPR_INF_FUTURE = gpr_inf_future(GPR_CLOCK_REALTIME)
cdef float _POLL_AWAKE_INTERVAL_S = 0.2 cdef float _POLL_AWAKE_INTERVAL_S = 0.2
cdef bint _no_fd_monitoring = False # 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": IF UNAME_SYSNAME == "Windows":
cdef void _unified_socket_write(int fd) nogil: cdef void _unified_socket_write(int fd) nogil:
win_socket_send(<WIN_SOCKET>fd, b"1", 1, 0) win_socket_send(<WIN_SOCKET>fd, b"1", 1, 0)
# If the event loop policy is Proactor, then immediately turn on fall back
# mode.
if asyncio.get_event_loop_policy() == asyncio.WindowsProactorEventLoopPolicy:
_has_fd_monitoring = False
elif asyncio.get_event_loop_policy() == asyncio.DefaultEventLoopPolicy:
if sys.version_info >= (3, 8, 0):
_has_fd_monitoring = False
ELSE: ELSE:
from posix cimport unistd from posix cimport unistd
@ -43,6 +53,7 @@ cdef class BaseCompletionQueue:
cdef class _BoundEventLoop: cdef class _BoundEventLoop:
def __cinit__(self, object loop, object read_socket, object handler): def __cinit__(self, object loop, object read_socket, object handler):
global _has_fd_monitoring
self.loop = loop self.loop = loop
self.read_socket = read_socket self.read_socket = read_socket
reader_function = functools.partial( reader_function = functools.partial(
@ -53,14 +64,18 @@ cdef class _BoundEventLoop:
# support is available or not. Checking the event loop policy is not # support is available or not. Checking the event loop policy is not
# good enough. The application can has its own loop implementation, or # good enough. The application can has its own loop implementation, or
# uses different types of event loops (e.g., 1 Proactor, 3 Selectors). # uses different types of event loops (e.g., 1 Proactor, 3 Selectors).
try: if _has_fd_monitoring:
self.loop.add_reader(self.read_socket, reader_function) try:
except NotImplementedError: self.loop.add_reader(self.read_socket, reader_function)
_no_fd_monitoring = True self._has_reader = True
except NotImplementedError:
_has_fd_monitoring = False
self._has_reader = False
def close(self): def close(self):
if self.loop: 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): cdef class PollerCompletionQueue(BaseCompletionQueue):
@ -106,15 +121,14 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
self._queue_mutex.lock() self._queue_mutex.lock()
self._queue.push(event) self._queue.push(event)
self._queue_mutex.unlock() self._queue_mutex.unlock()
if not _no_fd_monitoring: if _has_fd_monitoring:
_unified_socket_write(self._write_fd) _unified_socket_write(self._write_fd)
else: else:
with gil: with gil:
# Event loops can be paused or killed at any time. The # Event loops can be paused or killed at any time. So,
# most robust way to make sure someone is polling is # instead of deligate to any thread, the polling thread
# awaking all loops up. # should handle the distribution of the event.
for loop in self._loops: self._handle_events(None)
self._handle_events(loop)
def _poll_wrapper(self): def _poll_wrapper(self):
with nogil: with nogil:
@ -136,7 +150,10 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
self._write_socket.close() self._write_socket.close()
def _handle_events(self, object context_loop): 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 grpc_event event
cdef CallbackContext *context cdef CallbackContext *context

Loading…
Cancel
Save