codekingpro/portable-devtools

1# Copyright 2016 gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Shared implementation."""
15
16import logging
17import time
18from typing import Any, AnyStr, Callable, Optional, Union
19
20import grpc
21from grpc._cython import cygrpc
22from grpc._typing import DeserializingFunction
23from grpc._typing import SerializingFunction
24
25_LOGGER = logging.getLogger(__name__)
26
27CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = {
28    cygrpc.ConnectivityState.idle: grpc.ChannelConnectivity.IDLE,
29    cygrpc.ConnectivityState.connecting: grpc.ChannelConnectivity.CONNECTING,
30    cygrpc.ConnectivityState.ready: grpc.ChannelConnectivity.READY,
31    cygrpc.ConnectivityState.transient_failure: grpc.ChannelConnectivity.TRANSIENT_FAILURE,
32    cygrpc.ConnectivityState.shutdown: grpc.ChannelConnectivity.SHUTDOWN,
33}
34
35CYGRPC_STATUS_CODE_TO_STATUS_CODE = {
36    cygrpc.StatusCode.ok: grpc.StatusCode.OK,
37    cygrpc.StatusCode.cancelled: grpc.StatusCode.CANCELLED,
38    cygrpc.StatusCode.unknown: grpc.StatusCode.UNKNOWN,
39    cygrpc.StatusCode.invalid_argument: grpc.StatusCode.INVALID_ARGUMENT,
40    cygrpc.StatusCode.deadline_exceeded: grpc.StatusCode.DEADLINE_EXCEEDED,
41    cygrpc.StatusCode.not_found: grpc.StatusCode.NOT_FOUND,
42    cygrpc.StatusCode.already_exists: grpc.StatusCode.ALREADY_EXISTS,
43    cygrpc.StatusCode.permission_denied: grpc.StatusCode.PERMISSION_DENIED,
44    cygrpc.StatusCode.unauthenticated: grpc.StatusCode.UNAUTHENTICATED,
45    cygrpc.StatusCode.resource_exhausted: grpc.StatusCode.RESOURCE_EXHAUSTED,
46    cygrpc.StatusCode.failed_precondition: grpc.StatusCode.FAILED_PRECONDITION,
47    cygrpc.StatusCode.aborted: grpc.StatusCode.ABORTED,
48    cygrpc.StatusCode.out_of_range: grpc.StatusCode.OUT_OF_RANGE,
49    cygrpc.StatusCode.unimplemented: grpc.StatusCode.UNIMPLEMENTED,
50    cygrpc.StatusCode.internal: grpc.StatusCode.INTERNAL,
51    cygrpc.StatusCode.unavailable: grpc.StatusCode.UNAVAILABLE,
52    cygrpc.StatusCode.data_loss: grpc.StatusCode.DATA_LOSS,
53}
54STATUS_CODE_TO_CYGRPC_STATUS_CODE = {
55    grpc_code: cygrpc_code
56    for cygrpc_code, grpc_code in CYGRPC_STATUS_CODE_TO_STATUS_CODE.items()
57}
58
59MAXIMUM_WAIT_TIMEOUT = 0.1
60
61_ERROR_MESSAGE_PORT_BINDING_FAILED = (
62    "Failed to bind to address %s; set "
63    "GRPC_VERBOSITY=debug environment variable to see detailed error message."
64)
65
66
67def encode(s: AnyStr) -> bytes:
68    if isinstance(s, bytes):
69        return s
70    return s.encode("utf8")
71
72
73def decode(b: AnyStr) -> str:
74    if isinstance(b, bytes):
75        return b.decode("utf-8", "replace")
76    return b
77
78
79def _transform(
80    message: Any,
81    transformer: Union[SerializingFunction, DeserializingFunction, None],
82    exception_message: str,
83) -> Any:
84    if transformer is None:
85        return message
86    try:
87        return transformer(message)
88    except Exception:  # pylint: disable=broad-except
89        _LOGGER.exception(exception_message)
90        return None
91
92
93def serialize(message: Any, serializer: Optional[SerializingFunction]) -> bytes:
94    return _transform(message, serializer, "Exception serializing message!")
95
96
97def deserialize(
98    serialized_message: bytes, deserializer: Optional[DeserializingFunction]
99) -> Any:
100    return _transform(
101        serialized_message, deserializer, "Exception deserializing message!"
102    )
103
104
105def fully_qualified_method(group: str, method: str) -> str:
106    return "/{}/{}".format(group, method)
107
108
109def _wait_once(
110    wait_fn: Callable[..., bool],
111    timeout: float,
112    spin_cb: Optional[Callable[[], None]],
113):
114    wait_fn(timeout=timeout)
115    if spin_cb is not None:
116        spin_cb()
117
118
119def wait(
120    wait_fn: Callable[..., bool],
121    wait_complete_fn: Callable[[], bool],
122    timeout: Optional[float] = None,
123    spin_cb: Optional[Callable[[], None]] = None,
124) -> bool:
125    """Blocks waiting for an event without blocking the thread indefinitely.
126
127    See https://github.com/grpc/grpc/issues/19464 for full context. CPython's
128    `threading.Event.wait` and `threading.Condition.wait` methods, if invoked
129    without a timeout kwarg, may block the calling thread indefinitely. If the
130    call is made from the main thread, this means that signal handlers may not
131    run for an arbitrarily long period of time.
132
133    This wrapper calls the supplied wait function with an arbitrary short
134    timeout to ensure that no signal handler has to wait longer than
135    MAXIMUM_WAIT_TIMEOUT before executing.
136
137    Args:
138      wait_fn: A callable acceptable a single float-valued kwarg named
139        `timeout`. This function is expected to be one of `threading.Event.wait`
140        or `threading.Condition.wait`.
141      wait_complete_fn: A callable taking no arguments and returning a bool.
142        When this function returns true, it indicates that waiting should cease.
143      timeout: An optional float-valued number of seconds after which the wait
144        should cease.
145      spin_cb: An optional Callable taking no arguments and returning nothing.
146        This callback will be called on each iteration of the spin. This may be
147        used for, e.g. work related to forking.
148
149    Returns:
150      True if a timeout was supplied and it was reached. False otherwise.
151    """
152    if timeout is None:
153        while not wait_complete_fn():
154            _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb)
155    else:
156        end = time.time() + timeout
157        while not wait_complete_fn():
158            remaining = min(end - time.time(), MAXIMUM_WAIT_TIMEOUT)
159            if remaining < 0:
160                return True
161            _wait_once(wait_fn, remaining, spin_cb)
162    return False
163
164
165def validate_port_binding_result(address: str, port: int) -> int:
166    """Validates if the port binding succeed.
167
168    If the port returned by Core is 0, the binding is failed. However, in that
169    case, the Core API doesn't return a detailed failing reason. The best we
170    can do is raising an exception to prevent further confusion.
171
172    Args:
173        address: The address string to be bound.
174        port: An int returned by core
175    """
176    if port == 0:
177        # The Core API doesn't return a failure message. The best we can do
178        # is raising an exception to prevent further confusion.
179        raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address)
180    return port
181