diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd
index fd562ad75b68626c31dfa4044ff5f170bf5c3e2c..1ed5d4b229981eda35b12c6d2a0bbaa2edeb4473 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd
@@ -37,3 +37,5 @@ cdef class CompletionQueue:
cdef bint is_polling
cdef bint is_shutting_down
cdef bint is_shutdown
+
+ cdef _interpret_event(self, grpc.grpc_event event)
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx
index 2cf49707b48dbb83eb69da6194e331c1879f6af9..635a38fe28e04af1ecd81b3150ce79323535d65a 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx
@@ -46,35 +46,13 @@ cdef class CompletionQueue:
self.poll_condition = threading.Condition()
self.is_polling = False
- def poll(self, records.Timespec deadline=None):
- # We name this 'poll' to avoid problems with CPython's expectations for
- # 'special' methods (like next and __next__).
- cdef grpc.gpr_timespec c_deadline = grpc.gpr_inf_future(
- grpc.GPR_CLOCK_REALTIME)
+ cdef _interpret_event(self, grpc.grpc_event event):
cdef records.OperationTag tag = None
cdef object user_tag = None
cdef call.Call operation_call = None
cdef records.CallDetails request_call_details = None
cdef records.Metadata request_metadata = None
cdef records.Operations batch_operations = None
- if deadline is not None:
- c_deadline = deadline.c_time
- cdef grpc.grpc_event event
-
- # Poll within a critical section
- # TODO consider making queue polling contention a hard error to enable
- # easier bug discovery
- with self.poll_condition:
- while self.is_polling:
- self.poll_condition.wait(float(deadline) - time.time())
- self.is_polling = True
- with nogil:
- event = grpc.grpc_completion_queue_next(
- self.c_completion_queue, c_deadline, NULL)
- with self.poll_condition:
- self.is_polling = False
- self.poll_condition.notify()
-
if event.type == grpc.GRPC_QUEUE_TIMEOUT:
return records.Event(
event.type, False, None, None, None, None, False, None)
@@ -104,6 +82,54 @@ cdef class CompletionQueue:
request_call_details, request_metadata, tag.is_new_request,
batch_operations)
+ def poll(self, records.Timespec deadline=None):
+ # We name this 'poll' to avoid problems with CPython's expectations for
+ # 'special' methods (like next and __next__).
+ cdef grpc.gpr_timespec c_deadline = grpc.gpr_inf_future(
+ grpc.GPR_CLOCK_REALTIME)
+ if deadline is not None:
+ c_deadline = deadline.c_time
+ cdef grpc.grpc_event event
+
+ # Poll within a critical section
+ # TODO(atash) consider making queue polling contention a hard error to
+ # enable easier bug discovery
+ with self.poll_condition:
+ while self.is_polling:
+ self.poll_condition.wait(float(deadline) - time.time())
+ self.is_polling = True
+ with nogil:
+ event = grpc.grpc_completion_queue_next(
+ self.c_completion_queue, c_deadline, NULL)
+ with self.poll_condition:
+ self.is_polling = False
+ self.poll_condition.notify()
+ return self._interpret_event(event)
+
+ def pluck(self, records.OperationTag tag, records.Timespec deadline=None):
+ # Plucking a 'None' tag is equivalent to passing control to GRPC core until
+ # the deadline.
+ cdef grpc.gpr_timespec c_deadline = grpc.gpr_inf_future(
+ grpc.GPR_CLOCK_REALTIME)
+ if deadline is not None:
+ c_deadline = deadline.c_time
+ cdef grpc.grpc_event event
+
+ # Poll within a critical section
+ # TODO(atash) consider making queue polling contention a hard error to
+ # enable easier bug discovery
+ with self.poll_condition:
+ while self.is_polling:
+ self.poll_condition.wait(float(deadline) - time.time())
+ self.is_polling = True
+ with nogil:
+ event = grpc.grpc_completion_queue_pluck(
+ self.c_completion_queue, <cpython.PyObject *>tag, c_deadline, NULL)
+ with self.poll_condition:
+ self.is_polling = False
+ self.poll_condition.notify()
+ return self._interpret_event(event)
+
def shutdown(self):
grpc.grpc_completion_queue_shutdown(self.c_completion_queue)
self.is_shutting_down = True
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxd b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxd
index 643cdc9e3d17d96a8cdec224e14831df7fde0e35..2b8c19d69c55e1c54c01826cc3b7aec61a40e28a 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxd
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxd
@@ -292,6 +292,9 @@ cdef extern from "grpc/grpc.h":
grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,
gpr_timespec deadline,
void *reserved) nogil
+ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag,
+ gpr_timespec deadline,
+ void *reserved) nogil
void grpc_completion_queue_shutdown(grpc_completion_queue *cq)
void grpc_completion_queue_destroy(grpc_completion_queue *cq)
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx
index 46df8bf77f8d1228f842b67fdbe68d8fbd12ab4c..b0bafbc1f49158e415e2ab7854c3c75dfca4eeeb 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx
@@ -89,6 +89,8 @@ cdef class Server:
self.register_completion_queue(self.backup_shutdown_queue)
self.is_started = True
grpc.grpc_server_start(self.c_server)
+ # Ensure the core has gotten a chance to do the start-up work
+ self.backup_shutdown_queue.pluck(None, records.Timespec(None))
def add_http2_port(self, address,
credentials.ServerCredentials server_credentials=None):