Elide cygrpc.Timespec

pull/13974/head
Nathaniel Manista 7 years ago
parent 2eb22fd67d
commit 31ddbff8cf
  1. 55
      src/python/grpcio/grpc/_channel.py
  2. 26
      src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi
  3. 9
      src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi
  4. 5
      src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi
  5. 72
      src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
  6. 2
      src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
  7. 19
      src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi
  8. 30
      src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi
  9. 1
      src/python/grpcio/grpc/_cython/cygrpc.pxd
  10. 1
      src/python/grpcio/grpc/_cython/cygrpc.pyx
  11. 3
      src/python/grpcio/grpc/_server.py
  12. 7
      src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py
  13. 6
      src/python/grpcio_tests/tests/unit/_cython/_channel_test.py
  14. 1
      src/python/grpcio_tests/tests/unit/_cython/_common.py
  15. 6
      src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py
  16. 6
      src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py
  17. 3
      src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py
  18. 38
      src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
  19. 2
      src/python/grpcio_tests/tests/unit/_cython/test_utilities.py

@ -27,7 +27,6 @@ from grpc.framework.foundation import callable_util
_USER_AGENT = 'grpc-python/{}'.format(_grpcio_metadata.__version__)
_EMPTY_FLAGS = 0
_INFINITE_FUTURE = cygrpc.Timespec(float('+inf'))
_UNARY_UNARY_INITIAL_DUE = (
cygrpc.OperationType.send_initial_metadata,
@ -61,11 +60,7 @@ _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = (
def _deadline(timeout):
if timeout is None:
return None, _INFINITE_FUTURE
else:
deadline = time.time() + timeout
return deadline, cygrpc.Timespec(deadline)
return None if timeout is None else time.time() + timeout
def _unknown_code_details(unknown_cygrpc_code, details):
@ -420,15 +415,15 @@ class _Rendezvous(grpc.RpcError, grpc.Future, grpc.Call):
def _start_unary_request(request, timeout, request_serializer):
deadline, deadline_timespec = _deadline(timeout)
deadline = _deadline(timeout)
serialized_request = _common.serialize(request, request_serializer)
if serialized_request is None:
state = _RPCState((), (), (), grpc.StatusCode.INTERNAL,
'Exception serializing request!')
rendezvous = _Rendezvous(state, None, None, deadline)
return deadline, deadline_timespec, None, rendezvous
return deadline, None, rendezvous
else:
return deadline, deadline_timespec, serialized_request, None
return deadline, serialized_request, None
def _end_unary_response_blocking(state, call, with_call, deadline):
@ -453,10 +448,10 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
self._response_deserializer = response_deserializer
def _prepare(self, request, timeout, metadata):
deadline, deadline_timespec, serialized_request, rendezvous = (
_start_unary_request(request, timeout, self._request_serializer))
deadline, serialized_request, rendezvous = (_start_unary_request(
request, timeout, self._request_serializer))
if serialized_request is None:
return None, None, None, None, rendezvous
return None, None, None, rendezvous
else:
state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None)
operations = (
@ -467,18 +462,17 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
)
return state, operations, deadline, deadline_timespec, None
return state, operations, deadline, None
def _blocking(self, request, timeout, metadata, credentials):
state, operations, deadline, deadline_timespec, rendezvous = self._prepare(
state, operations, deadline, rendezvous = self._prepare(
request, timeout, metadata)
if rendezvous:
raise rendezvous
else:
completion_queue = cygrpc.CompletionQueue()
call = self._channel.create_call(None, 0, completion_queue,
self._method, None,
deadline_timespec)
self._method, None, deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
call_error = call.start_client_batch(operations, None)
@ -498,13 +492,13 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
return _end_unary_response_blocking(state, call, True, deadline)
def future(self, request, timeout=None, metadata=None, credentials=None):
state, operations, deadline, deadline_timespec, rendezvous = self._prepare(
state, operations, deadline, rendezvous = self._prepare(
request, timeout, metadata)
if rendezvous:
return rendezvous
else:
call, drive_call = self._managed_call(None, 0, self._method, None,
deadline_timespec)
deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
event_handler = _event_handler(state, call,
@ -530,14 +524,14 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable):
self._response_deserializer = response_deserializer
def __call__(self, request, timeout=None, metadata=None, credentials=None):
deadline, deadline_timespec, serialized_request, rendezvous = (
_start_unary_request(request, timeout, self._request_serializer))
deadline, serialized_request, rendezvous = (_start_unary_request(
request, timeout, self._request_serializer))
if serialized_request is None:
raise rendezvous
else:
state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None)
call, drive_call = self._managed_call(None, 0, self._method, None,
deadline_timespec)
deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
event_handler = _event_handler(state, call,
@ -573,11 +567,11 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable):
self._response_deserializer = response_deserializer
def _blocking(self, request_iterator, timeout, metadata, credentials):
deadline, deadline_timespec = _deadline(timeout)
deadline = _deadline(timeout)
state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None)
completion_queue = cygrpc.CompletionQueue()
call = self._channel.create_call(None, 0, completion_queue,
self._method, None, deadline_timespec)
self._method, None, deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
with state.condition:
@ -624,10 +618,10 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable):
timeout=None,
metadata=None,
credentials=None):
deadline, deadline_timespec = _deadline(timeout)
deadline = _deadline(timeout)
state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None)
call, drive_call = self._managed_call(None, 0, self._method, None,
deadline_timespec)
deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
event_handler = _event_handler(state, call, self._response_deserializer)
@ -665,10 +659,10 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable):
timeout=None,
metadata=None,
credentials=None):
deadline, deadline_timespec = _deadline(timeout)
deadline = _deadline(timeout)
state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None)
call, drive_call = self._managed_call(None, 0, self._method, None,
deadline_timespec)
deadline)
if credentials is not None:
call.set_credentials(credentials._credentials)
event_handler = _event_handler(state, call, self._response_deserializer)
@ -737,7 +731,8 @@ def _channel_managed_call_management(state):
flags: An integer bitfield of call flags.
method: The RPC method.
host: A host string for the created call.
deadline: A cygrpc.Timespec to be the deadline of the created call.
deadline: A float to be the deadline of the created call or None if the
call is to have an infinite deadline.
Returns:
A cygrpc.Call with which to conduct an RPC and a function to call if
@ -827,8 +822,8 @@ def _poll_connectivity(state, channel, initial_try_to_connect):
completion_queue = cygrpc.CompletionQueue()
while True:
channel.watch_connectivity_state(connectivity,
cygrpc.Timespec(time.time() + 0.2),
completion_queue, None)
time.time() + 0.2, completion_queue,
None)
event = completion_queue.poll()
with state.lock:
if not state.callbacks_and_connectivities and not state.try_to_connect:

@ -42,7 +42,7 @@ cdef class Channel:
def create_call(self, Call parent, int flags,
CompletionQueue queue not None,
method, host, Timespec deadline not None):
method, host, object deadline):
if queue.is_shutting_down:
raise ValueError("queue must not be shutting down or shutdown")
cdef grpc_slice method_slice = _slice_from_bytes(method)
@ -56,14 +56,13 @@ cdef class Channel:
cdef grpc_call *parent_call = NULL
if parent is not None:
parent_call = parent.c_call
with nogil:
operation_call.c_call = grpc_channel_create_call(
self.c_channel, parent_call, flags,
queue.c_completion_queue, method_slice, host_slice_ptr,
deadline.c_time, NULL)
grpc_slice_unref(method_slice)
if host_slice_ptr:
grpc_slice_unref(host_slice)
operation_call.c_call = grpc_channel_create_call(
self.c_channel, parent_call, flags,
queue.c_completion_queue, method_slice, host_slice_ptr,
_timespec_from_time(deadline), NULL)
grpc_slice_unref(method_slice)
if host_slice_ptr:
grpc_slice_unref(host_slice)
return operation_call
def check_connectivity_state(self, bint try_to_connect):
@ -75,13 +74,12 @@ cdef class Channel:
def watch_connectivity_state(
self, grpc_connectivity_state last_observed_state,
Timespec deadline not None, CompletionQueue queue not None, tag):
object deadline, CompletionQueue queue not None, tag):
cdef _ConnectivityTag connectivity_tag = _ConnectivityTag(tag)
cpython.Py_INCREF(connectivity_tag)
with nogil:
grpc_channel_watch_connectivity_state(
self.c_channel, last_observed_state, deadline.c_time,
queue.c_completion_queue, <cpython.PyObject *>connectivity_tag)
grpc_channel_watch_connectivity_state(
self.c_channel, last_observed_state, _timespec_from_time(deadline),
queue.c_completion_queue, <cpython.PyObject *>connectivity_tag)
def target(self):
cdef char *target = NULL

@ -52,17 +52,18 @@ cdef class CompletionQueue:
cpython.Py_DECREF(tag)
return tag.event(event)
def poll(self, Timespec deadline=None):
def poll(self, deadline=None):
# We name this 'poll' to avoid problems with CPython's expectations for
# 'special' methods (like next and __next__).
cdef gpr_timespec c_increment
cdef gpr_timespec c_timeout
cdef gpr_timespec c_deadline
if deadline is None:
c_deadline = gpr_inf_future(GPR_CLOCK_REALTIME)
else:
c_deadline = _timespec_from_time(deadline)
with nogil:
c_increment = gpr_time_from_millis(_INTERRUPT_CHECK_PERIOD_MS, GPR_TIMESPAN)
c_deadline = gpr_inf_future(GPR_CLOCK_REALTIME)
if deadline is not None:
c_deadline = deadline.c_time
while True:
c_timeout = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), c_increment)

@ -18,11 +18,6 @@ cdef grpc_slice _copy_slice(grpc_slice slice) nogil
cdef grpc_slice _slice_from_bytes(bytes value) nogil
cdef class Timespec:
cdef gpr_timespec c_time
cdef class CallDetails:
cdef grpc_call_details c_details

@ -123,74 +123,6 @@ class CompressionLevel:
high = GRPC_COMPRESS_LEVEL_HIGH
cdef class Timespec:
def __cinit__(self, time):
if time is None:
with nogil:
self.c_time = gpr_now(GPR_CLOCK_REALTIME)
return
if isinstance(time, int):
time = float(time)
if isinstance(time, float):
if time == float("+inf"):
with nogil:
self.c_time = gpr_inf_future(GPR_CLOCK_REALTIME)
elif time == float("-inf"):
with nogil:
self.c_time = gpr_inf_past(GPR_CLOCK_REALTIME)
else:
self.c_time.seconds = time
self.c_time.nanoseconds = (time - float(self.c_time.seconds)) * 1e9
self.c_time.clock_type = GPR_CLOCK_REALTIME
elif isinstance(time, Timespec):
self.c_time = (<Timespec>time).c_time
else:
raise TypeError("expected time to be float, int, or Timespec, not {}"
.format(type(time)))
@property
def seconds(self):
# TODO(atash) ensure that everywhere a Timespec is created that it's
# converted to GPR_CLOCK_REALTIME then and not every time someone wants to
# read values off in Python.
cdef gpr_timespec real_time
with nogil:
real_time = (
gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME))
return real_time.seconds
@property
def nanoseconds(self):
cdef gpr_timespec real_time = (
gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME))
return real_time.nanoseconds
def __float__(self):
cdef gpr_timespec real_time = (
gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME))
return <double>real_time.seconds + <double>real_time.nanoseconds / 1e9
def __richcmp__(Timespec self not None, Timespec other not None, int op):
cdef gpr_timespec self_c_time = self.c_time
cdef gpr_timespec other_c_time = other.c_time
cdef int result = gpr_time_cmp(self_c_time, other_c_time)
if op == 0: # <
return result < 0
elif op == 2: # ==
return result == 0
elif op == 4: # >
return result > 0
elif op == 1: # <=
return result <= 0
elif op == 3: # !=
return result != 0
elif op == 5: # >=
return result >= 0
else:
raise ValueError('__richcmp__ `op` contract violated')
cdef class CallDetails:
def __cinit__(self):
@ -213,9 +145,7 @@ cdef class CallDetails:
@property
def deadline(self):
timespec = Timespec(float("-inf"))
timespec.c_time = self.c_details.deadline
return timespec
return _time_from_timespec(self.c_details.deadline)
cdef class SslPemKeyCertPair:

@ -106,7 +106,7 @@ cdef class Server:
with nogil:
grpc_server_start(self.c_server)
# Ensure the core has gotten a chance to do the start-up work
self.backup_shutdown_queue.poll(Timespec(None))
self.backup_shutdown_queue.poll(deadline=time.time())
def add_http2_port(self, bytes address,
ServerCredentials server_credentials=None):

@ -0,0 +1,19 @@
# Copyright 2018 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cdef gpr_timespec _timespec_from_time(object time)
cdef double _time_from_timespec(gpr_timespec timespec)

@ -0,0 +1,30 @@
# Copyright 2018 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cdef gpr_timespec _timespec_from_time(object time):
cdef gpr_timespec timespec
if time is None:
return gpr_inf_future(GPR_CLOCK_REALTIME)
else:
timespec.seconds = time
timespec.nanoseconds = (time - float(timespec.seconds)) * 1e9
timespec.clock_type = GPR_CLOCK_REALTIME
return timespec
cdef double _time_from_timespec(gpr_timespec timespec):
cdef gpr_timespec real_timespec = gpr_convert_clock_type(
timespec, GPR_CLOCK_REALTIME)
return <double>real_timespec.seconds + <double>real_timespec.nanoseconds / 1e9

@ -25,3 +25,4 @@ include "_cygrpc/records.pxd.pxi"
include "_cygrpc/security.pxd.pxi"
include "_cygrpc/server.pxd.pxi"
include "_cygrpc/tag.pxd.pxi"
include "_cygrpc/time.pxd.pxi"

@ -32,6 +32,7 @@ include "_cygrpc/records.pyx.pxi"
include "_cygrpc/security.pyx.pxi"
include "_cygrpc/server.pyx.pxi"
include "_cygrpc/tag.pyx.pxi"
include "_cygrpc/time.pyx.pxi"
#
# initialize gRPC

@ -220,8 +220,7 @@ class _Context(grpc.ServicerContext):
return self._state.client is not _CANCELLED and not self._state.statused
def time_remaining(self):
return max(
float(self._rpc_event.call_details.deadline) - time.time(), 0)
return max(self._rpc_event.call_details.deadline - time.time(), 0)
def cancel(self):
self._rpc_event.call.cancel()

@ -20,7 +20,6 @@ from grpc._cython import cygrpc
from grpc.framework.foundation import logging_pool
from tests.unit.framework.common import test_constants
_INFINITE_FUTURE = cygrpc.Timespec(float('+inf'))
_EMPTY_FLAGS = 0
_EMPTY_METADATA = ()
@ -171,9 +170,9 @@ class CancelManyCallsTest(unittest.TestCase):
with client_condition:
client_calls = []
for index in range(test_constants.RPC_CONCURRENCY):
client_call = channel.create_call(
None, _EMPTY_FLAGS, client_completion_queue, b'/twinkies',
None, _INFINITE_FUTURE)
client_call = channel.create_call(None, _EMPTY_FLAGS,
client_completion_queue,
b'/twinkies', None, None)
operations = (
cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
_EMPTY_FLAGS),

@ -31,9 +31,9 @@ def _connectivity_loop(channel, completion_queue):
for _ in range(100):
connectivity = channel.check_connectivity_state(True)
channel.watch_connectivity_state(connectivity,
cygrpc.Timespec(time.time() + 0.2),
completion_queue, None)
completion_queue.poll(deadline=cygrpc.Timespec(float('+inf')))
time.time() + 0.2, completion_queue,
None)
completion_queue.poll()
def _create_loop_destroy():

@ -20,7 +20,6 @@ from grpc._cython import cygrpc
RPC_COUNT = 4000
INFINITE_FUTURE = cygrpc.Timespec(float('+inf'))
EMPTY_FLAGS = 0
INVOCATION_METADATA = (

@ -41,9 +41,9 @@ class Test(_common.RpcTest, unittest.TestCase):
server_request_call_tag,
})
client_call = self.channel.create_call(
None, _common.EMPTY_FLAGS, self.client_completion_queue,
b'/twinkies', None, _common.INFINITE_FUTURE)
client_call = self.channel.create_call(None, _common.EMPTY_FLAGS,
self.client_completion_queue,
b'/twinkies', None, None)
client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag'
client_complete_rpc_tag = 'client_complete_rpc_tag'
with self.client_condition:

@ -36,9 +36,9 @@ class Test(_common.RpcTest, unittest.TestCase):
server_request_call_tag,
})
client_call = self.channel.create_call(
None, _common.EMPTY_FLAGS, self.client_completion_queue,
b'/twinkies', None, _common.INFINITE_FUTURE)
client_call = self.channel.create_call(None, _common.EMPTY_FLAGS,
self.client_completion_queue,
b'/twinkies', None, None)
client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag'
client_complete_rpc_tag = 'client_complete_rpc_tag'
with self.client_condition:

@ -18,7 +18,6 @@ import unittest
from grpc._cython import cygrpc
_INFINITE_FUTURE = cygrpc.Timespec(float('+inf'))
_EMPTY_FLAGS = 0
_EMPTY_METADATA = ()
@ -156,7 +155,7 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase):
client_call = channel.create_call(None, _EMPTY_FLAGS,
client_completion_queue, b'/twinkies',
None, _INFINITE_FUTURE)
None, None)
client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag'
client_complete_rpc_tag = 'client_complete_rpc_tag'
with client_condition:

@ -37,21 +37,6 @@ def _metadata_plugin(context, callback):
class TypeSmokeTest(unittest.TestCase):
def testTimespec(self):
now = time.time()
now_timespec_a = cygrpc.Timespec(now)
now_timespec_b = cygrpc.Timespec(now)
self.assertAlmostEqual(now, float(now_timespec_a), places=8)
self.assertEqual(now_timespec_a, now_timespec_b)
self.assertLess(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
self.assertGreater(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
self.assertGreaterEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
self.assertGreaterEqual(cygrpc.Timespec(now), cygrpc.Timespec(now))
self.assertLessEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
self.assertLessEqual(cygrpc.Timespec(now), cygrpc.Timespec(now))
self.assertNotEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
self.assertNotEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
def testCompletionQueueUpDown(self):
completion_queue = cygrpc.CompletionQueue()
del completion_queue
@ -147,7 +132,7 @@ class ServerClientMixin(object):
try:
call_result = call.start_client_batch(operations, tag)
self.assertEqual(cygrpc.CallError.ok, call_result)
event = queue.poll(deadline)
event = queue.poll(deadline=deadline)
self.assertEqual(cygrpc.CompletionType.operation_complete,
event.completion_type)
self.assertTrue(event.success)
@ -176,8 +161,6 @@ class ServerClientMixin(object):
RESPONSE = b'his name is robert paulson'
METHOD = b'twinkies'
cygrpc_deadline = cygrpc.Timespec(DEADLINE)
server_request_tag = object()
request_call_result = self.server.request_call(
self.server_completion_queue, self.server_completion_queue,
@ -188,7 +171,7 @@ class ServerClientMixin(object):
client_call_tag = object()
client_call = self.client_channel.create_call(
None, 0, self.client_completion_queue, METHOD, self.host_argument,
cygrpc_deadline)
DEADLINE)
client_initial_metadata = (
(
CLIENT_METADATA_ASCII_KEY,
@ -210,9 +193,9 @@ class ServerClientMixin(object):
], client_call_tag)
self.assertEqual(cygrpc.CallError.ok, client_start_batch_result)
client_event_future = test_utilities.CompletionQueuePollFuture(
self.client_completion_queue, cygrpc_deadline)
self.client_completion_queue, DEADLINE)
request_event = self.server_completion_queue.poll(cygrpc_deadline)
request_event = self.server_completion_queue.poll(deadline=DEADLINE)
self.assertEqual(cygrpc.CompletionType.operation_complete,
request_event.completion_type)
self.assertIsInstance(request_event.call, cygrpc.Call)
@ -223,7 +206,7 @@ class ServerClientMixin(object):
self.assertEqual(METHOD, request_event.call_details.method)
self.assertEqual(self.expected_host, request_event.call_details.host)
self.assertLess(
abs(DEADLINE - float(request_event.call_details.deadline)),
abs(DEADLINE - request_event.call_details.deadline),
DEADLINE_TOLERANCE)
server_call_tag = object()
@ -248,7 +231,7 @@ class ServerClientMixin(object):
], server_call_tag)
self.assertEqual(cygrpc.CallError.ok, server_start_batch_result)
server_event = self.server_completion_queue.poll(cygrpc_deadline)
server_event = self.server_completion_queue.poll(deadline=DEADLINE)
client_event = client_event_future.result()
self.assertEqual(6, len(client_event.batch_operations))
@ -310,7 +293,6 @@ class ServerClientMixin(object):
DEADLINE_TOLERANCE = 0.25
METHOD = b'twinkies'
cygrpc_deadline = cygrpc.Timespec(DEADLINE)
empty_metadata = ()
server_request_tag = object()
@ -319,26 +301,26 @@ class ServerClientMixin(object):
server_request_tag)
client_call = self.client_channel.create_call(
None, 0, self.client_completion_queue, METHOD, self.host_argument,
cygrpc_deadline)
DEADLINE)
# Prologue
def perform_client_operations(operations, description):
return self._perform_operations(operations, client_call,
self.client_completion_queue,
cygrpc_deadline, description)
DEADLINE, description)
client_event_future = perform_client_operations([
cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS),
cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
], "Client prologue")
request_event = self.server_completion_queue.poll(cygrpc_deadline)
request_event = self.server_completion_queue.poll(deadline=DEADLINE)
server_call = request_event.call
def perform_server_operations(operations, description):
return self._perform_operations(operations, server_call,
self.server_completion_queue,
cygrpc_deadline, description)
DEADLINE, description)
server_event_future = perform_server_operations([
cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS),

@ -49,4 +49,4 @@ class CompletionQueuePollFuture(SimpleFuture):
def __init__(self, completion_queue, deadline):
super(CompletionQueuePollFuture,
self).__init__(lambda: completion_queue.poll(deadline))
self).__init__(lambda: completion_queue.poll(deadline=deadline))

Loading…
Cancel
Save