Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #23163 from gnossen/wait_for_ready_default

Browse Source 
Default wait_for_ready to True in simple stubs
reviewable/pr22991/r2
Richard Belleville 5 years ago committed by GitHub
parent 5f7babc746 4b497f7156
commit c483ece5dd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 145 additions and 18 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 53
      src/python/grpcio/grpc/_simple_stubs.py
  2. 6
      src/python/grpcio/grpc/experimental/__init__.py
  3. 104
      src/python/grpcio_tests/tests_py3_only/unit/_simple_stubs_test.py

53
src/python/grpcio/grpc/_simple_stubs.py
Unescape View File

@ -49,11 +49,18 @@ if _MAXIMUM_CHANNELS_KEY in os.environ:
else:
_MAXIMUM_CHANNELS = 2**8
_DEFAULT_TIMEOUT_KEY = "GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS"
if _DEFAULT_TIMEOUT_KEY in os.environ:
_DEFAULT_TIMEOUT = float(os.environ[_DEFAULT_TIMEOUT_KEY])
_LOGGER.debug("Setting default timeout seconds to %f", _DEFAULT_TIMEOUT)
else:
_DEFAULT_TIMEOUT = 60.0
def _create_channel(target: str, options: Sequence[Tuple[str, str]],
channel_credentials: Optional[grpc.ChannelCredentials],
compression: Optional[grpc.Compression]) -> grpc.Channel:
if channel_credentials._credentials is grpc.experimental._insecure_channel_credentials:
if channel_credentials is grpc.experimental.insecure_channel_credentials():
_LOGGER.debug(f"Creating insecure channel with options '{options}' " +
f"and compression '{compression}'")
return grpc.insecure_channel(target,
@ -178,7 +185,7 @@ def unary_unary(
call_credentials: Optional[grpc.CallCredentials] = None,
compression: Optional[grpc.Compression] = None,
wait_for_ready: Optional[bool] = None,
timeout: Optional[float] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
) -> ResponseType:
"""Invokes a unary-unary RPC without an explicitly specified channel.
@ -221,9 +228,13 @@ def unary_unary(
immediately if the connection is not ready at the time the RPC is
invoked, or if it should wait until the connection to the server
becomes ready. When using this option, the user will likely also want
to set a timeout. Defaults to False.
to set a timeout. Defaults to True.
timeout: An optional duration of time in seconds to allow for the RPC,
after which an exception will be raised.
after which an exception will be raised. If timeout is unspecified,
defaults to a timeout controlled by the
GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is
unset, defaults to 60 seconds. Supply a value of None to indicate that
no timeout should be enforced.
metadata: Optional metadata to send to the server.
Returns:
@ -234,6 +245,7 @@ def unary_unary(
compression)
multicallable = channel.unary_unary(method, request_serializer,
response_deserializer)
wait_for_ready = wait_for_ready if wait_for_ready is not None else True
return multicallable(request,
metadata=metadata,
wait_for_ready=wait_for_ready,
@ -254,7 +266,7 @@ def unary_stream(
call_credentials: Optional[grpc.CallCredentials] = None,
compression: Optional[grpc.Compression] = None,
wait_for_ready: Optional[bool] = None,
timeout: Optional[float] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
) -> Iterator[ResponseType]:
"""Invokes a unary-stream RPC without an explicitly specified channel.
@ -296,9 +308,13 @@ def unary_stream(
immediately if the connection is not ready at the time the RPC is
invoked, or if it should wait until the connection to the server
becomes ready. When using this option, the user will likely also want
to set a timeout. Defaults to False.
to set a timeout. Defaults to True.
timeout: An optional duration of time in seconds to allow for the RPC,
after which an exception will be raised.
after which an exception will be raised. If timeout is unspecified,
defaults to a timeout controlled by the
GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is
unset, defaults to 60 seconds. Supply a value of None to indicate that
no timeout should be enforced.
metadata: Optional metadata to send to the server.
Returns:
@ -309,6 +325,7 @@ def unary_stream(
compression)
multicallable = channel.unary_stream(method, request_serializer,
response_deserializer)
wait_for_ready = wait_for_ready if wait_for_ready is not None else True
return multicallable(request,
metadata=metadata,
wait_for_ready=wait_for_ready,
@ -329,7 +346,7 @@ def stream_unary(
call_credentials: Optional[grpc.CallCredentials] = None,
compression: Optional[grpc.Compression] = None,
wait_for_ready: Optional[bool] = None,
timeout: Optional[float] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
) -> ResponseType:
"""Invokes a stream-unary RPC without an explicitly specified channel.
@ -371,9 +388,13 @@ def stream_unary(
immediately if the connection is not ready at the time the RPC is
invoked, or if it should wait until the connection to the server
becomes ready. When using this option, the user will likely also want
to set a timeout. Defaults to False.
to set a timeout. Defaults to True.
timeout: An optional duration of time in seconds to allow for the RPC,
after which an exception will be raised.
after which an exception will be raised. If timeout is unspecified,
defaults to a timeout controlled by the
GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is
unset, defaults to 60 seconds. Supply a value of None to indicate that
no timeout should be enforced.
metadata: Optional metadata to send to the server.
Returns:
@ -384,6 +405,7 @@ def stream_unary(
compression)
multicallable = channel.stream_unary(method, request_serializer,
response_deserializer)
wait_for_ready = wait_for_ready if wait_for_ready is not None else True
return multicallable(request_iterator,
metadata=metadata,
wait_for_ready=wait_for_ready,
@ -404,7 +426,7 @@ def stream_stream(
call_credentials: Optional[grpc.CallCredentials] = None,
compression: Optional[grpc.Compression] = None,
wait_for_ready: Optional[bool] = None,
timeout: Optional[float] = None,
timeout: Optional[float] = _DEFAULT_TIMEOUT,
metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
) -> Iterator[ResponseType]:
"""Invokes a stream-stream RPC without an explicitly specified channel.
@ -446,9 +468,13 @@ def stream_stream(
immediately if the connection is not ready at the time the RPC is
invoked, or if it should wait until the connection to the server
becomes ready. When using this option, the user will likely also want
to set a timeout. Defaults to False.
to set a timeout. Defaults to True.
timeout: An optional duration of time in seconds to allow for the RPC,
after which an exception will be raised.
after which an exception will be raised. If timeout is unspecified,
defaults to a timeout controlled by the
GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is
unset, defaults to 60 seconds. Supply a value of None to indicate that
no timeout should be enforced.
metadata: Optional metadata to send to the server.
Returns:
@ -459,6 +485,7 @@ def stream_stream(
compression)
multicallable = channel.stream_stream(method, request_serializer,
response_deserializer)
wait_for_ready = wait_for_ready if wait_for_ready is not None else True
return multicallable(request_iterator,
metadata=metadata,
wait_for_ready=wait_for_ready,

6
src/python/grpcio/grpc/experimental/__init__.py
Unescape View File

@ -41,7 +41,9 @@ class UsageError(Exception):
"""Raised by the gRPC library to indicate usage not allowed by the API."""
_insecure_channel_credentials = object()
_insecure_channel_credentials_sentinel = object()
_insecure_channel_credentials = grpc.ChannelCredentials(
_insecure_channel_credentials_sentinel)
def insecure_channel_credentials():
@ -53,7 +55,7 @@ def insecure_channel_credentials():
used with grpc.unary_unary, grpc.unary_stream, grpc.stream_unary, or
grpc.stream_stream.
"""
return grpc.ChannelCredentials(_insecure_channel_credentials)
return _insecure_channel_credentials
class ExperimentalApiWarning(Warning):

104
src/python/grpcio_tests/tests_py3_only/unit/_simple_stubs_test.py
Unescape View File

@ -19,19 +19,24 @@ import os
_MAXIMUM_CHANNELS = 10
os.environ["GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS"] = "1"
_DEFAULT_TIMEOUT = 1.0
os.environ["GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS"] = "2"
os.environ["GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM"] = str(_MAXIMUM_CHANNELS)
os.environ["GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS"] = str(_DEFAULT_TIMEOUT)
import contextlib
import datetime
import inspect
import logging
import threading
import unittest
import sys
import time
from typing import Callable, Optional
from tests.unit import test_common
from tests.unit.framework.common import get_socket
from tests.unit import resources
import grpc
import grpc.experimental
@ -50,6 +55,7 @@ _UNARY_UNARY = "/test/UnaryUnary"
_UNARY_STREAM = "/test/UnaryStream"
_STREAM_UNARY = "/test/StreamUnary"
_STREAM_STREAM = "/test/StreamStream"
_BLACK_HOLE = "/test/BlackHole"
@contextlib.contextmanager
@ -80,6 +86,17 @@ def _stream_stream_handler(request_iterator, context):
yield request
def _black_hole_handler(request, context):
event = threading.Event()
def _on_done():
event.set()
context.add_callback(_on_done)
while not event.is_set():
time.sleep(0.1)
class _GenericHandler(grpc.GenericRpcHandler):
def service(self, handler_call_details):
@ -91,6 +108,8 @@ class _GenericHandler(grpc.GenericRpcHandler):
return grpc.stream_unary_rpc_method_handler(_stream_unary_handler)
elif handler_call_details.method == _STREAM_STREAM:
return grpc.stream_stream_rpc_method_handler(_stream_stream_handler)
elif handler_call_details.method == _BLACK_HOLE:
return grpc.unary_unary_rpc_method_handler(_black_hole_handler)
else:
raise NotImplementedError()
@ -169,7 +188,8 @@ class SimpleStubsTest(unittest.TestCase):
target,
_UNARY_UNARY,
channel_credentials=grpc.experimental.
insecure_channel_credentials())
insecure_channel_credentials(),
timeout=None)
self.assertEqual(_REQUEST, response)
def test_unary_unary_secure(self):
@ -179,7 +199,8 @@ class SimpleStubsTest(unittest.TestCase):
_REQUEST,
target,
_UNARY_UNARY,
channel_credentials=grpc.local_channel_credentials())
channel_credentials=grpc.local_channel_credentials(),
timeout=None)
self.assertEqual(_REQUEST, response)
def test_channels_cached(self):
@ -311,6 +332,83 @@ class SimpleStubsTest(unittest.TestCase):
insecure=True,
channel_credentials=grpc.local_channel_credentials())
def test_default_wait_for_ready(self):
addr, port, sock = get_socket()
sock.close()
target = f'{addr}:{port}'
channel = grpc._simple_stubs.ChannelCache.get().get_channel(
target, (), None, True, None)
rpc_finished_event = threading.Event()
rpc_failed_event = threading.Event()
server = None
def _on_connectivity_changed(connectivity):
nonlocal server
if connectivity is grpc.ChannelConnectivity.TRANSIENT_FAILURE:
self.assertFalse(rpc_finished_event.is_set())
self.assertFalse(rpc_failed_event.is_set())
server = test_common.test_server()
server.add_insecure_port(target)
server.add_generic_rpc_handlers((_GenericHandler(),))
server.start()
channel.unsubscribe(_on_connectivity_changed)
elif connectivity in (grpc.ChannelConnectivity.IDLE,
grpc.ChannelConnectivity.CONNECTING):
pass
else:
self.fail("Encountered unknown state.")
channel.subscribe(_on_connectivity_changed)
def _send_rpc():
try:
response = grpc.experimental.unary_unary(_REQUEST,
target,
_UNARY_UNARY,
timeout=None,
insecure=True)
rpc_finished_event.set()
except Exception as e:
rpc_failed_event.set()
t = threading.Thread(target=_send_rpc)
t.start()
t.join()
self.assertFalse(rpc_failed_event.is_set())
self.assertTrue(rpc_finished_event.is_set())
if server is not None:
server.stop(None)
def assert_times_out(self, invocation_args):
with _server(None) as port:
target = f'localhost:{port}'
with self.assertRaises(grpc.RpcError) as cm:
response = grpc.experimental.unary_unary(_REQUEST,
target,
_BLACK_HOLE,
insecure=True,
**invocation_args)
self.assertEqual(grpc.StatusCode.DEADLINE_EXCEEDED,
cm.exception.code())
def test_default_timeout(self):
not_present = object()
wait_for_ready_values = [True, not_present]
timeout_values = [0.5, not_present]
cases = []
for wait_for_ready in wait_for_ready_values:
for timeout in timeout_values:
case = {}
if timeout is not not_present:
case["timeout"] = timeout
if wait_for_ready is not not_present:
case["wait_for_ready"] = wait_for_ready
cases.append(case)
for case in cases:
with self.subTest(**case):
self.assert_times_out(case)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)

Write Preview
Loading…
Cancel
Save