Don't graceful shutdown cq to avoid the deadlock

pull/22311/head
Lidi Zheng 5 years ago
parent 7622a2de38
commit 5d1651ff65
  1. 2
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pxd.pxi
  2. 9
      src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi
  3. 4
      src/python/grpcio/grpc/_cython/_cygrpc/aio/grpc_aio.pyx.pxi
  4. 12
      src/python/grpcio_tests/tests_aio/unit/compatibility_test.py

@ -19,11 +19,11 @@ cdef class BaseCompletionQueue:
cdef class PollerCompletionQueue(BaseCompletionQueue):
cdef bint _shutdown
cdef object _shutdown_completed
cdef object _poller_thread
cdef object _loop
cdef void _poll(self) except *
cdef void shutdown(self) nogil
cdef class CallbackCompletionQueue(BaseCompletionQueue):

@ -12,6 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from libc.stdio cimport printf
cdef gpr_timespec _GPR_INF_FUTURE = gpr_inf_future(GPR_CLOCK_REALTIME)
@ -30,7 +32,6 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
def __cinit__(self):
self._cq = grpc_completion_queue_create_for_next(NULL)
self._shutdown = False
self._shutdown_completed = threading.Event()
self._poller_thread = threading.Thread(target=self._poll_wrapper, daemon=True)
self._poller_thread.start()
@ -48,7 +49,6 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
raise AssertionError("Core should not return GRPC_QUEUE_TIMEOUT!")
elif event.type == GRPC_QUEUE_SHUTDOWN:
self._shutdown = True
self._shutdown_completed.set()
else:
context = <CallbackContext *>event.tag
loop = <object>context.loop
@ -60,11 +60,10 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
def _poll_wrapper(self):
self._poll()
def shutdown(self):
cdef void shutdown(self) nogil:
# TODO(https://github.com/grpc/grpc/issues/22365) perform graceful shutdown
grpc_completion_queue_shutdown(self._cq)
self._shutdown_completed.wait()
grpc_completion_queue_destroy(self._cq)
self._poller_thread.join()
cdef class CallbackCompletionQueue(BaseCompletionQueue):

@ -103,12 +103,12 @@ def _grpc_shutdown_wrapper(_):
cdef _actual_aio_shutdown():
if _global_aio_state.engine is AsyncIOEngine.CUSTOM_IO_MANAGER:
future = schedule_coro_threadsafe(
_global_aio_state.cq.shutdown,
_global_aio_state.cq.shutdown(),
(<CallbackCompletionQueue>_global_aio_state.cq)._loop
)
future.add_done_callback(_grpc_shutdown_wrapper)
elif _global_aio_state.engine is AsyncIOEngine.POLLER:
_global_aio_state.cq.shutdown()
(<PollerCompletionQueue>_global_aio_state.cq).shutdown()
grpc_shutdown_blocking()
else:
raise ValueError('Unsupported engine type [%s]' % _global_aio_state.engine)

@ -16,23 +16,19 @@
import asyncio
import logging
import os
import unittest
import random
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor
import time
import random
from typing import Callable, Sequence, Tuple
import grpc
from grpc.experimental import aio
from grpc._cython import cygrpc
from tests_aio.unit._test_base import AioTestBase
from tests.unit.framework.common import test_constants
from tests.unit.framework.common import get_socket
from src.proto.grpc.testing import messages_pb2, test_pb2_grpc
from tests.unit.framework.common import test_constants
from tests_aio.unit._test_base import AioTestBase
from tests_aio.unit._test_server import start_test_server
from tests_aio.unit import _common
_NUM_STREAM_RESPONSES = 5
_REQUEST_PAYLOAD_SIZE = 7

Loading…
Cancel
Save