diff options
author | Masood Malekghassemi <soltanmm@users.noreply.github.com> | 2015-12-17 17:48:28 -0800 |
---|---|---|
committer | Masood Malekghassemi <soltanmm@users.noreply.github.com> | 2015-12-21 13:07:39 -0800 |
commit | f872c6aa9fe3980df3a6d1425f83ac8bd373da7f (patch) | |
tree | 7254e389cac34813d226557a95c3bc403e2a5b25 /src | |
parent | afd3fdc31a2282786ee6c3cffb7a68b566734860 (diff) |
Fix Python test timeouts
We've been assuming that the server upon invoking start was started.
This has been false; invoking start means that the server will
eventually be started, and to make this reliably observable requires
passing control to core.
Diffstat (limited to 'src')
4 files changed, 56 insertions, 23 deletions
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pxd index fd562ad75b..1ed5d4b229 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 2cf49707b4..635a38fe28 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 643cdc9e3d..2b8c19d69c 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 46df8bf77f..b0bafbc1f4 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): |