| # Copyright 2016 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. |
| """Service-side implementation of gRPC Python.""" |
| |
| import collections |
| import enum |
| import logging |
| import threading |
| import time |
| |
| from concurrent import futures |
| import six |
| |
| import grpc |
| from grpc import _common |
| from grpc import _compression |
| from grpc import _interceptor |
| from grpc._cython import cygrpc |
| |
| _LOGGER = logging.getLogger(__name__) |
| |
| _SHUTDOWN_TAG = 'shutdown' |
| _REQUEST_CALL_TAG = 'request_call' |
| |
| _RECEIVE_CLOSE_ON_SERVER_TOKEN = 'receive_close_on_server' |
| _SEND_INITIAL_METADATA_TOKEN = 'send_initial_metadata' |
| _RECEIVE_MESSAGE_TOKEN = 'receive_message' |
| _SEND_MESSAGE_TOKEN = 'send_message' |
| _SEND_INITIAL_METADATA_AND_SEND_MESSAGE_TOKEN = ( |
| 'send_initial_metadata * send_message') |
| _SEND_STATUS_FROM_SERVER_TOKEN = 'send_status_from_server' |
| _SEND_INITIAL_METADATA_AND_SEND_STATUS_FROM_SERVER_TOKEN = ( |
| 'send_initial_metadata * send_status_from_server') |
| |
| _OPEN = 'open' |
| _CLOSED = 'closed' |
| _CANCELLED = 'cancelled' |
| |
| _EMPTY_FLAGS = 0 |
| |
| _DEALLOCATED_SERVER_CHECK_PERIOD_S = 1.0 |
| |
| |
| def _serialized_request(request_event): |
| return request_event.batch_operations[0].message() |
| |
| |
| def _application_code(code): |
| cygrpc_code = _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE.get(code) |
| return cygrpc.StatusCode.unknown if cygrpc_code is None else cygrpc_code |
| |
| |
| def _completion_code(state): |
| if state.code is None: |
| return cygrpc.StatusCode.ok |
| else: |
| return _application_code(state.code) |
| |
| |
| def _abortion_code(state, code): |
| if state.code is None: |
| return code |
| else: |
| return _application_code(state.code) |
| |
| |
| def _details(state): |
| return b'' if state.details is None else state.details |
| |
| |
| class _HandlerCallDetails( |
| collections.namedtuple('_HandlerCallDetails', ( |
| 'method', |
| 'invocation_metadata', |
| )), grpc.HandlerCallDetails): |
| pass |
| |
| |
| class _RPCState(object): |
| |
| def __init__(self): |
| self.condition = threading.Condition() |
| self.due = set() |
| self.request = None |
| self.client = _OPEN |
| self.initial_metadata_allowed = True |
| self.compression_algorithm = None |
| self.disable_next_compression = False |
| self.trailing_metadata = None |
| self.code = None |
| self.details = None |
| self.statused = False |
| self.rpc_errors = [] |
| self.callbacks = [] |
| self.aborted = False |
| |
| |
| def _raise_rpc_error(state): |
| rpc_error = grpc.RpcError() |
| state.rpc_errors.append(rpc_error) |
| raise rpc_error |
| |
| |
| def _possibly_finish_call(state, token): |
| state.due.remove(token) |
| if not _is_rpc_state_active(state) and not state.due: |
| callbacks = state.callbacks |
| state.callbacks = None |
| return state, callbacks |
| else: |
| return None, () |
| |
| |
| def _send_status_from_server(state, token): |
| |
| def send_status_from_server(unused_send_status_from_server_event): |
| with state.condition: |
| return _possibly_finish_call(state, token) |
| |
| return send_status_from_server |
| |
| |
| def _get_initial_metadata(state, metadata): |
| with state.condition: |
| if state.compression_algorithm: |
| compression_metadata = ( |
| _compression.compression_algorithm_to_metadata( |
| state.compression_algorithm),) |
| if metadata is None: |
| return compression_metadata |
| else: |
| return compression_metadata + tuple(metadata) |
| else: |
| return metadata |
| |
| |
| def _get_initial_metadata_operation(state, metadata): |
| operation = cygrpc.SendInitialMetadataOperation( |
| _get_initial_metadata(state, metadata), _EMPTY_FLAGS) |
| return operation |
| |
| |
| def _abort(state, call, code, details): |
| if state.client is not _CANCELLED: |
| effective_code = _abortion_code(state, code) |
| effective_details = details if state.details is None else state.details |
| if state.initial_metadata_allowed: |
| operations = ( |
| _get_initial_metadata_operation(state, None), |
| cygrpc.SendStatusFromServerOperation( |
| state.trailing_metadata, effective_code, effective_details, |
| _EMPTY_FLAGS), |
| ) |
| token = _SEND_INITIAL_METADATA_AND_SEND_STATUS_FROM_SERVER_TOKEN |
| else: |
| operations = (cygrpc.SendStatusFromServerOperation( |
| state.trailing_metadata, effective_code, effective_details, |
| _EMPTY_FLAGS),) |
| token = _SEND_STATUS_FROM_SERVER_TOKEN |
| call.start_server_batch(operations, |
| _send_status_from_server(state, token)) |
| state.statused = True |
| state.due.add(token) |
| |
| |
| def _receive_close_on_server(state): |
| |
| def receive_close_on_server(receive_close_on_server_event): |
| with state.condition: |
| if receive_close_on_server_event.batch_operations[0].cancelled(): |
| state.client = _CANCELLED |
| elif state.client is _OPEN: |
| state.client = _CLOSED |
| state.condition.notify_all() |
| return _possibly_finish_call(state, _RECEIVE_CLOSE_ON_SERVER_TOKEN) |
| |
| return receive_close_on_server |
| |
| |
| def _receive_message(state, call, request_deserializer): |
| |
| def receive_message(receive_message_event): |
| serialized_request = _serialized_request(receive_message_event) |
| if serialized_request is None: |
| with state.condition: |
| if state.client is _OPEN: |
| state.client = _CLOSED |
| state.condition.notify_all() |
| return _possibly_finish_call(state, _RECEIVE_MESSAGE_TOKEN) |
| else: |
| request = _common.deserialize(serialized_request, |
| request_deserializer) |
| with state.condition: |
| if request is None: |
| _abort(state, call, cygrpc.StatusCode.internal, |
| b'Exception deserializing request!') |
| else: |
| state.request = request |
| state.condition.notify_all() |
| return _possibly_finish_call(state, _RECEIVE_MESSAGE_TOKEN) |
| |
| return receive_message |
| |
| |
| def _send_initial_metadata(state): |
| |
| def send_initial_metadata(unused_send_initial_metadata_event): |
| with state.condition: |
| return _possibly_finish_call(state, _SEND_INITIAL_METADATA_TOKEN) |
| |
| return send_initial_metadata |
| |
| |
| def _send_message(state, token): |
| |
| def send_message(unused_send_message_event): |
| with state.condition: |
| state.condition.notify_all() |
| return _possibly_finish_call(state, token) |
| |
| return send_message |
| |
| |
| class _Context(grpc.ServicerContext): |
| |
| def __init__(self, rpc_event, state, request_deserializer): |
| self._rpc_event = rpc_event |
| self._state = state |
| self._request_deserializer = request_deserializer |
| |
| def is_active(self): |
| with self._state.condition: |
| return _is_rpc_state_active(self._state) |
| |
| def time_remaining(self): |
| return max(self._rpc_event.call_details.deadline - time.time(), 0) |
| |
| def cancel(self): |
| self._rpc_event.call.cancel() |
| |
| def add_callback(self, callback): |
| with self._state.condition: |
| if self._state.callbacks is None: |
| return False |
| else: |
| self._state.callbacks.append(callback) |
| return True |
| |
| def disable_next_message_compression(self): |
| with self._state.condition: |
| self._state.disable_next_compression = True |
| |
| def invocation_metadata(self): |
| return self._rpc_event.invocation_metadata |
| |
| def peer(self): |
| return _common.decode(self._rpc_event.call.peer()) |
| |
| def peer_identities(self): |
| return cygrpc.peer_identities(self._rpc_event.call) |
| |
| def peer_identity_key(self): |
| id_key = cygrpc.peer_identity_key(self._rpc_event.call) |
| return id_key if id_key is None else _common.decode(id_key) |
| |
| def auth_context(self): |
| return { |
| _common.decode(key): value |
| for key, value in six.iteritems( |
| cygrpc.auth_context(self._rpc_event.call)) |
| } |
| |
| def set_compression(self, compression): |
| with self._state.condition: |
| self._state.compression_algorithm = compression |
| |
| def send_initial_metadata(self, initial_metadata): |
| with self._state.condition: |
| if self._state.client is _CANCELLED: |
| _raise_rpc_error(self._state) |
| else: |
| if self._state.initial_metadata_allowed: |
| operation = _get_initial_metadata_operation( |
| self._state, initial_metadata) |
| self._rpc_event.call.start_server_batch( |
| (operation,), _send_initial_metadata(self._state)) |
| self._state.initial_metadata_allowed = False |
| self._state.due.add(_SEND_INITIAL_METADATA_TOKEN) |
| else: |
| raise ValueError('Initial metadata no longer allowed!') |
| |
| def set_trailing_metadata(self, trailing_metadata): |
| with self._state.condition: |
| self._state.trailing_metadata = trailing_metadata |
| |
| def abort(self, code, details): |
| # treat OK like other invalid arguments: fail the RPC |
| if code == grpc.StatusCode.OK: |
| _LOGGER.error( |
| 'abort() called with StatusCode.OK; returning UNKNOWN') |
| code = grpc.StatusCode.UNKNOWN |
| details = '' |
| with self._state.condition: |
| self._state.code = code |
| self._state.details = _common.encode(details) |
| self._state.aborted = True |
| raise Exception() |
| |
| def abort_with_status(self, status): |
| self._state.trailing_metadata = status.trailing_metadata |
| self.abort(status.code, status.details) |
| |
| def set_code(self, code): |
| with self._state.condition: |
| self._state.code = code |
| |
| def set_details(self, details): |
| with self._state.condition: |
| self._state.details = _common.encode(details) |
| |
| def _finalize_state(self): |
| pass |
| |
| |
| class _RequestIterator(object): |
| |
| def __init__(self, state, call, request_deserializer): |
| self._state = state |
| self._call = call |
| self._request_deserializer = request_deserializer |
| |
| def _raise_or_start_receive_message(self): |
| if self._state.client is _CANCELLED: |
| _raise_rpc_error(self._state) |
| elif not _is_rpc_state_active(self._state): |
| raise StopIteration() |
| else: |
| self._call.start_server_batch( |
| (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), |
| _receive_message(self._state, self._call, |
| self._request_deserializer)) |
| self._state.due.add(_RECEIVE_MESSAGE_TOKEN) |
| |
| def _look_for_request(self): |
| if self._state.client is _CANCELLED: |
| _raise_rpc_error(self._state) |
| elif (self._state.request is None and |
| _RECEIVE_MESSAGE_TOKEN not in self._state.due): |
| raise StopIteration() |
| else: |
| request = self._state.request |
| self._state.request = None |
| return request |
| |
| raise AssertionError() # should never run |
| |
| def _next(self): |
| with self._state.condition: |
| self._raise_or_start_receive_message() |
| while True: |
| self._state.condition.wait() |
| request = self._look_for_request() |
| if request is not None: |
| return request |
| |
| def __iter__(self): |
| return self |
| |
| def __next__(self): |
| return self._next() |
| |
| def next(self): |
| return self._next() |
| |
| |
| def _unary_request(rpc_event, state, request_deserializer): |
| |
| def unary_request(): |
| with state.condition: |
| if not _is_rpc_state_active(state): |
| return None |
| else: |
| rpc_event.call.start_server_batch( |
| (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), |
| _receive_message(state, rpc_event.call, |
| request_deserializer)) |
| state.due.add(_RECEIVE_MESSAGE_TOKEN) |
| while True: |
| state.condition.wait() |
| if state.request is None: |
| if state.client is _CLOSED: |
| details = '"{}" requires exactly one request message.'.format( |
| rpc_event.call_details.method) |
| _abort(state, rpc_event.call, |
| cygrpc.StatusCode.unimplemented, |
| _common.encode(details)) |
| return None |
| elif state.client is _CANCELLED: |
| return None |
| else: |
| request = state.request |
| state.request = None |
| return request |
| |
| return unary_request |
| |
| |
| def _call_behavior(rpc_event, |
| state, |
| behavior, |
| argument, |
| request_deserializer, |
| send_response_callback=None): |
| from grpc import _create_servicer_context |
| with _create_servicer_context(rpc_event, state, |
| request_deserializer) as context: |
| try: |
| response_or_iterator = None |
| if send_response_callback is not None: |
| response_or_iterator = behavior(argument, context, |
| send_response_callback) |
| else: |
| response_or_iterator = behavior(argument, context) |
| return response_or_iterator, True |
| except Exception as exception: # pylint: disable=broad-except |
| with state.condition: |
| if state.aborted: |
| _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, |
| b'RPC Aborted') |
| elif exception not in state.rpc_errors: |
| details = 'Exception calling application: {}'.format( |
| exception) |
| _LOGGER.exception(details) |
| _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, |
| _common.encode(details)) |
| return None, False |
| |
| |
| def _take_response_from_response_iterator(rpc_event, state, response_iterator): |
| try: |
| return next(response_iterator), True |
| except StopIteration: |
| return None, True |
| except Exception as exception: # pylint: disable=broad-except |
| with state.condition: |
| if state.aborted: |
| _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, |
| b'RPC Aborted') |
| elif exception not in state.rpc_errors: |
| details = 'Exception iterating responses: {}'.format(exception) |
| _LOGGER.exception(details) |
| _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, |
| _common.encode(details)) |
| return None, False |
| |
| |
| def _serialize_response(rpc_event, state, response, response_serializer): |
| serialized_response = _common.serialize(response, response_serializer) |
| if serialized_response is None: |
| with state.condition: |
| _abort(state, rpc_event.call, cygrpc.StatusCode.internal, |
| b'Failed to serialize response!') |
| return None |
| else: |
| return serialized_response |
| |
| |
| def _get_send_message_op_flags_from_state(state): |
| if state.disable_next_compression: |
| return cygrpc.WriteFlag.no_compress |
| else: |
| return _EMPTY_FLAGS |
| |
| |
| def _reset_per_message_state(state): |
| with state.condition: |
| state.disable_next_compression = False |
| |
| |
| def _send_response(rpc_event, state, serialized_response): |
| with state.condition: |
| if not _is_rpc_state_active(state): |
| return False |
| else: |
| if state.initial_metadata_allowed: |
| operations = ( |
| _get_initial_metadata_operation(state, None), |
| cygrpc.SendMessageOperation( |
| serialized_response, |
| _get_send_message_op_flags_from_state(state)), |
| ) |
| state.initial_metadata_allowed = False |
| token = _SEND_INITIAL_METADATA_AND_SEND_MESSAGE_TOKEN |
| else: |
| operations = (cygrpc.SendMessageOperation( |
| serialized_response, |
| _get_send_message_op_flags_from_state(state)),) |
| token = _SEND_MESSAGE_TOKEN |
| rpc_event.call.start_server_batch(operations, |
| _send_message(state, token)) |
| state.due.add(token) |
| _reset_per_message_state(state) |
| while True: |
| state.condition.wait() |
| if token not in state.due: |
| return _is_rpc_state_active(state) |
| |
| |
| def _status(rpc_event, state, serialized_response): |
| with state.condition: |
| if state.client is not _CANCELLED: |
| code = _completion_code(state) |
| details = _details(state) |
| operations = [ |
| cygrpc.SendStatusFromServerOperation( |
| state.trailing_metadata, code, details, _EMPTY_FLAGS), |
| ] |
| if state.initial_metadata_allowed: |
| operations.append(_get_initial_metadata_operation(state, None)) |
| if serialized_response is not None: |
| operations.append( |
| cygrpc.SendMessageOperation( |
| serialized_response, |
| _get_send_message_op_flags_from_state(state))) |
| rpc_event.call.start_server_batch( |
| operations, |
| _send_status_from_server(state, _SEND_STATUS_FROM_SERVER_TOKEN)) |
| state.statused = True |
| _reset_per_message_state(state) |
| state.due.add(_SEND_STATUS_FROM_SERVER_TOKEN) |
| |
| |
| def _unary_response_in_pool(rpc_event, state, behavior, argument_thunk, |
| request_deserializer, response_serializer): |
| cygrpc.install_context_from_request_call_event(rpc_event) |
| try: |
| argument = argument_thunk() |
| if argument is not None: |
| response, proceed = _call_behavior(rpc_event, state, behavior, |
| argument, request_deserializer) |
| if proceed: |
| serialized_response = _serialize_response( |
| rpc_event, state, response, response_serializer) |
| if serialized_response is not None: |
| _status(rpc_event, state, serialized_response) |
| finally: |
| cygrpc.uninstall_context() |
| |
| |
| def _stream_response_in_pool(rpc_event, state, behavior, argument_thunk, |
| request_deserializer, response_serializer): |
| cygrpc.install_context_from_request_call_event(rpc_event) |
| |
| def send_response(response): |
| if response is None: |
| _status(rpc_event, state, None) |
| else: |
| serialized_response = _serialize_response( |
| rpc_event, state, response, response_serializer) |
| if serialized_response is not None: |
| _send_response(rpc_event, state, serialized_response) |
| |
| try: |
| argument = argument_thunk() |
| if argument is not None: |
| if hasattr(behavior, 'experimental_non_blocking' |
| ) and behavior.experimental_non_blocking: |
| _call_behavior( |
| rpc_event, |
| state, |
| behavior, |
| argument, |
| request_deserializer, |
| send_response_callback=send_response) |
| else: |
| response_iterator, proceed = _call_behavior( |
| rpc_event, state, behavior, argument, request_deserializer) |
| if proceed: |
| _send_message_callback_to_blocking_iterator_adapter( |
| rpc_event, state, send_response, response_iterator) |
| finally: |
| cygrpc.uninstall_context() |
| |
| |
| def _is_rpc_state_active(state): |
| return state.client is not _CANCELLED and not state.statused |
| |
| |
| def _send_message_callback_to_blocking_iterator_adapter( |
| rpc_event, state, send_response_callback, response_iterator): |
| while True: |
| response, proceed = _take_response_from_response_iterator( |
| rpc_event, state, response_iterator) |
| if proceed: |
| send_response_callback(response) |
| if not _is_rpc_state_active(state): |
| break |
| else: |
| break |
| |
| |
| def _select_thread_pool_for_behavior(behavior, default_thread_pool): |
| if hasattr(behavior, 'experimental_thread_pool') and isinstance( |
| behavior.experimental_thread_pool, futures.ThreadPoolExecutor): |
| return behavior.experimental_thread_pool |
| else: |
| return default_thread_pool |
| |
| |
| def _handle_unary_unary(rpc_event, state, method_handler, default_thread_pool): |
| unary_request = _unary_request(rpc_event, state, |
| method_handler.request_deserializer) |
| thread_pool = _select_thread_pool_for_behavior(method_handler.unary_unary, |
| default_thread_pool) |
| return thread_pool.submit(_unary_response_in_pool, rpc_event, state, |
| method_handler.unary_unary, unary_request, |
| method_handler.request_deserializer, |
| method_handler.response_serializer) |
| |
| |
| def _handle_unary_stream(rpc_event, state, method_handler, default_thread_pool): |
| unary_request = _unary_request(rpc_event, state, |
| method_handler.request_deserializer) |
| thread_pool = _select_thread_pool_for_behavior(method_handler.unary_stream, |
| default_thread_pool) |
| return thread_pool.submit(_stream_response_in_pool, rpc_event, state, |
| method_handler.unary_stream, unary_request, |
| method_handler.request_deserializer, |
| method_handler.response_serializer) |
| |
| |
| def _handle_stream_unary(rpc_event, state, method_handler, default_thread_pool): |
| request_iterator = _RequestIterator(state, rpc_event.call, |
| method_handler.request_deserializer) |
| thread_pool = _select_thread_pool_for_behavior(method_handler.stream_unary, |
| default_thread_pool) |
| return thread_pool.submit( |
| _unary_response_in_pool, rpc_event, state, method_handler.stream_unary, |
| lambda: request_iterator, method_handler.request_deserializer, |
| method_handler.response_serializer) |
| |
| |
| def _handle_stream_stream(rpc_event, state, method_handler, |
| default_thread_pool): |
| request_iterator = _RequestIterator(state, rpc_event.call, |
| method_handler.request_deserializer) |
| thread_pool = _select_thread_pool_for_behavior(method_handler.stream_stream, |
| default_thread_pool) |
| return thread_pool.submit( |
| _stream_response_in_pool, rpc_event, state, |
| method_handler.stream_stream, lambda: request_iterator, |
| method_handler.request_deserializer, method_handler.response_serializer) |
| |
| |
| def _find_method_handler(rpc_event, generic_handlers, interceptor_pipeline): |
| |
| def query_handlers(handler_call_details): |
| for generic_handler in generic_handlers: |
| method_handler = generic_handler.service(handler_call_details) |
| if method_handler is not None: |
| return method_handler |
| return None |
| |
| handler_call_details = _HandlerCallDetails( |
| _common.decode(rpc_event.call_details.method), |
| rpc_event.invocation_metadata) |
| |
| if interceptor_pipeline is not None: |
| return interceptor_pipeline.execute(query_handlers, |
| handler_call_details) |
| else: |
| return query_handlers(handler_call_details) |
| |
| |
| def _reject_rpc(rpc_event, status, details): |
| rpc_state = _RPCState() |
| operations = ( |
| _get_initial_metadata_operation(rpc_state, None), |
| cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS), |
| cygrpc.SendStatusFromServerOperation(None, status, details, |
| _EMPTY_FLAGS), |
| ) |
| rpc_event.call.start_server_batch(operations, |
| lambda ignored_event: (rpc_state, (),)) |
| return rpc_state |
| |
| |
| def _handle_with_method_handler(rpc_event, method_handler, thread_pool): |
| state = _RPCState() |
| with state.condition: |
| rpc_event.call.start_server_batch( |
| (cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),), |
| _receive_close_on_server(state)) |
| state.due.add(_RECEIVE_CLOSE_ON_SERVER_TOKEN) |
| if method_handler.request_streaming: |
| if method_handler.response_streaming: |
| return state, _handle_stream_stream(rpc_event, state, |
| method_handler, thread_pool) |
| else: |
| return state, _handle_stream_unary(rpc_event, state, |
| method_handler, thread_pool) |
| else: |
| if method_handler.response_streaming: |
| return state, _handle_unary_stream(rpc_event, state, |
| method_handler, thread_pool) |
| else: |
| return state, _handle_unary_unary(rpc_event, state, |
| method_handler, thread_pool) |
| |
| |
| def _handle_call(rpc_event, generic_handlers, interceptor_pipeline, thread_pool, |
| concurrency_exceeded): |
| if not rpc_event.success: |
| return None, None |
| if rpc_event.call_details.method is not None: |
| try: |
| method_handler = _find_method_handler(rpc_event, generic_handlers, |
| interceptor_pipeline) |
| except Exception as exception: # pylint: disable=broad-except |
| details = 'Exception servicing handler: {}'.format(exception) |
| _LOGGER.exception(details) |
| return _reject_rpc(rpc_event, cygrpc.StatusCode.unknown, |
| b'Error in service handler!'), None |
| if method_handler is None: |
| return _reject_rpc(rpc_event, cygrpc.StatusCode.unimplemented, |
| b'Method not found!'), None |
| elif concurrency_exceeded: |
| return _reject_rpc(rpc_event, cygrpc.StatusCode.resource_exhausted, |
| b'Concurrent RPC limit exceeded!'), None |
| else: |
| return _handle_with_method_handler(rpc_event, method_handler, |
| thread_pool) |
| else: |
| return None, None |
| |
| |
| @enum.unique |
| class _ServerStage(enum.Enum): |
| STOPPED = 'stopped' |
| STARTED = 'started' |
| GRACE = 'grace' |
| |
| |
| class _ServerState(object): |
| |
| # pylint: disable=too-many-arguments |
| def __init__(self, completion_queue, server, generic_handlers, |
| interceptor_pipeline, thread_pool, maximum_concurrent_rpcs): |
| self.lock = threading.RLock() |
| self.completion_queue = completion_queue |
| self.server = server |
| self.generic_handlers = list(generic_handlers) |
| self.interceptor_pipeline = interceptor_pipeline |
| self.thread_pool = thread_pool |
| self.stage = _ServerStage.STOPPED |
| self.shutdown_events = None |
| self.maximum_concurrent_rpcs = maximum_concurrent_rpcs |
| self.active_rpc_count = 0 |
| |
| # TODO(https://github.com/grpc/grpc/issues/6597): eliminate these fields. |
| self.rpc_states = set() |
| self.due = set() |
| |
| # A "volatile" flag to interrupt the daemon serving thread |
| self.server_deallocated = False |
| |
| |
| def _add_generic_handlers(state, generic_handlers): |
| with state.lock: |
| state.generic_handlers.extend(generic_handlers) |
| |
| |
| def _add_insecure_port(state, address): |
| with state.lock: |
| return state.server.add_http2_port(address) |
| |
| |
| def _add_secure_port(state, address, server_credentials): |
| with state.lock: |
| return state.server.add_http2_port(address, |
| server_credentials._credentials) |
| |
| |
| def _request_call(state): |
| state.server.request_call(state.completion_queue, state.completion_queue, |
| _REQUEST_CALL_TAG) |
| state.due.add(_REQUEST_CALL_TAG) |
| |
| |
| # TODO(https://github.com/grpc/grpc/issues/6597): delete this function. |
| def _stop_serving(state): |
| if not state.rpc_states and not state.due: |
| state.server.destroy() |
| for shutdown_event in state.shutdown_events: |
| shutdown_event.set() |
| state.stage = _ServerStage.STOPPED |
| return True |
| else: |
| return False |
| |
| |
| def _on_call_completed(state): |
| with state.lock: |
| state.active_rpc_count -= 1 |
| |
| |
| def _process_event_and_continue(state, event): |
| should_continue = True |
| if event.tag is _SHUTDOWN_TAG: |
| with state.lock: |
| state.due.remove(_SHUTDOWN_TAG) |
| if _stop_serving(state): |
| should_continue = False |
| elif event.tag is _REQUEST_CALL_TAG: |
| with state.lock: |
| state.due.remove(_REQUEST_CALL_TAG) |
| concurrency_exceeded = ( |
| state.maximum_concurrent_rpcs is not None and |
| state.active_rpc_count >= state.maximum_concurrent_rpcs) |
| rpc_state, rpc_future = _handle_call( |
| event, state.generic_handlers, state.interceptor_pipeline, |
| state.thread_pool, concurrency_exceeded) |
| if rpc_state is not None: |
| state.rpc_states.add(rpc_state) |
| if rpc_future is not None: |
| state.active_rpc_count += 1 |
| rpc_future.add_done_callback( |
| lambda unused_future: _on_call_completed(state)) |
| if state.stage is _ServerStage.STARTED: |
| _request_call(state) |
| elif _stop_serving(state): |
| should_continue = False |
| else: |
| rpc_state, callbacks = event.tag(event) |
| for callback in callbacks: |
| try: |
| callback() |
| except Exception: # pylint: disable=broad-except |
| _LOGGER.exception('Exception calling callback!') |
| if rpc_state is not None: |
| with state.lock: |
| state.rpc_states.remove(rpc_state) |
| if _stop_serving(state): |
| should_continue = False |
| return should_continue |
| |
| |
| def _serve(state): |
| while True: |
| timeout = time.time() + _DEALLOCATED_SERVER_CHECK_PERIOD_S |
| event = state.completion_queue.poll(timeout) |
| if state.server_deallocated: |
| _begin_shutdown_once(state) |
| if event.completion_type != cygrpc.CompletionType.queue_timeout: |
| if not _process_event_and_continue(state, event): |
| return |
| # We want to force the deletion of the previous event |
| # ~before~ we poll again; if the event has a reference |
| # to a shutdown Call object, this can induce spinlock. |
| event = None |
| |
| |
| def _begin_shutdown_once(state): |
| with state.lock: |
| if state.stage is _ServerStage.STARTED: |
| state.server.shutdown(state.completion_queue, _SHUTDOWN_TAG) |
| state.stage = _ServerStage.GRACE |
| state.shutdown_events = [] |
| state.due.add(_SHUTDOWN_TAG) |
| |
| |
| def _stop(state, grace): |
| with state.lock: |
| if state.stage is _ServerStage.STOPPED: |
| shutdown_event = threading.Event() |
| shutdown_event.set() |
| return shutdown_event |
| else: |
| _begin_shutdown_once(state) |
| shutdown_event = threading.Event() |
| state.shutdown_events.append(shutdown_event) |
| if grace is None: |
| state.server.cancel_all_calls() |
| else: |
| |
| def cancel_all_calls_after_grace(): |
| shutdown_event.wait(timeout=grace) |
| with state.lock: |
| state.server.cancel_all_calls() |
| |
| thread = threading.Thread(target=cancel_all_calls_after_grace) |
| thread.start() |
| return shutdown_event |
| shutdown_event.wait() |
| return shutdown_event |
| |
| |
| def _start(state): |
| with state.lock: |
| if state.stage is not _ServerStage.STOPPED: |
| raise ValueError('Cannot start already-started server!') |
| state.server.start() |
| state.stage = _ServerStage.STARTED |
| _request_call(state) |
| |
| thread = threading.Thread(target=_serve, args=(state,)) |
| thread.daemon = True |
| thread.start() |
| |
| |
| def _validate_generic_rpc_handlers(generic_rpc_handlers): |
| for generic_rpc_handler in generic_rpc_handlers: |
| service_attribute = getattr(generic_rpc_handler, 'service', None) |
| if service_attribute is None: |
| raise AttributeError( |
| '"{}" must conform to grpc.GenericRpcHandler type but does ' |
| 'not have "service" method!'.format(generic_rpc_handler)) |
| |
| |
| def _augment_options(base_options, compression): |
| compression_option = _compression.create_channel_option(compression) |
| return tuple(base_options) + compression_option |
| |
| |
| class _Server(grpc.Server): |
| |
| # pylint: disable=too-many-arguments |
| def __init__(self, thread_pool, generic_handlers, interceptors, options, |
| maximum_concurrent_rpcs, compression): |
| completion_queue = cygrpc.CompletionQueue() |
| server = cygrpc.Server(_augment_options(options, compression)) |
| server.register_completion_queue(completion_queue) |
| self._state = _ServerState(completion_queue, server, generic_handlers, |
| _interceptor.service_pipeline(interceptors), |
| thread_pool, maximum_concurrent_rpcs) |
| |
| def add_generic_rpc_handlers(self, generic_rpc_handlers): |
| _validate_generic_rpc_handlers(generic_rpc_handlers) |
| _add_generic_handlers(self._state, generic_rpc_handlers) |
| |
| def add_insecure_port(self, address): |
| return _add_insecure_port(self._state, _common.encode(address)) |
| |
| def add_secure_port(self, address, server_credentials): |
| return _add_secure_port(self._state, _common.encode(address), |
| server_credentials) |
| |
| def start(self): |
| _start(self._state) |
| |
| def stop(self, grace): |
| return _stop(self._state, grace) |
| |
| def __del__(self): |
| if hasattr(self, '_state'): |
| # We can not grab a lock in __del__(), so set a flag to signal the |
| # serving daemon thread (if it exists) to initiate shutdown. |
| self._state.server_deallocated = True |
| |
| |
| def create_server(thread_pool, generic_rpc_handlers, interceptors, options, |
| maximum_concurrent_rpcs, compression): |
| _validate_generic_rpc_handlers(generic_rpc_handlers) |
| return _Server(thread_pool, generic_rpc_handlers, interceptors, options, |
| maximum_concurrent_rpcs, compression) |
