1# Copyright 2019 The 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"""Server-side implementation of gRPC Asyncio Python.""" 15 16from concurrent.futures import Executor 17from typing import Any, Optional, Sequence 18 19import grpc 20from grpc import _common 21from grpc import _compression 22from grpc._cython import cygrpc 23 24from . import _base_server 25from ._interceptor import ServerInterceptor 26from ._typing import ChannelArgumentType 27 28 29def _augment_channel_arguments( 30 base_options: ChannelArgumentType, compression: Optional[grpc.Compression] 31): 32 compression_option = _compression.create_channel_option(compression) 33 return tuple(base_options) + compression_option 34 35 36class Server(_base_server.Server): 37 """Serves RPCs.""" 38 39 def __init__( 40 self, 41 thread_pool: Optional[Executor], 42 generic_handlers: Optional[Sequence[grpc.GenericRpcHandler]], 43 interceptors: Optional[Sequence[Any]], 44 options: ChannelArgumentType, 45 maximum_concurrent_rpcs: Optional[int], 46 compression: Optional[grpc.Compression], 47 ): 48 self._loop = cygrpc.get_working_loop() 49 if interceptors: 50 invalid_interceptors = [ 51 interceptor 52 for interceptor in interceptors 53 if not isinstance(interceptor, ServerInterceptor) 54 ] 55 if invalid_interceptors: 56 raise ValueError( 57 "Interceptor must be ServerInterceptor, the " 58 f"following are invalid: {invalid_interceptors}" 59 ) 60 self._server = cygrpc.AioServer( 61 self._loop, 62 thread_pool, 63 generic_handlers, 64 interceptors, 65 _augment_channel_arguments(options, compression), 66 maximum_concurrent_rpcs, 67 ) 68 69 def add_generic_rpc_handlers( 70 self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler] 71 ) -> None: 72 """Registers GenericRpcHandlers with this Server. 73 74 This method is only safe to call before the server is started. 75 76 Args: 77 generic_rpc_handlers: A sequence of GenericRpcHandlers that will be 78 used to service RPCs. 79 """ 80 self._server.add_generic_rpc_handlers(generic_rpc_handlers) 81 82 def add_insecure_port(self, address: str) -> int: 83 """Opens an insecure port for accepting RPCs. 84 85 This method may only be called before starting the server. 86 87 Args: 88 address: The address for which to open a port. If the port is 0, 89 or not specified in the address, then the gRPC runtime will choose a port. 90 91 Returns: 92 An integer port on which the server will accept RPC requests. 93 """ 94 return _common.validate_port_binding_result( 95 address, self._server.add_insecure_port(_common.encode(address)) 96 ) 97 98 def add_secure_port( 99 self, address: str, server_credentials: grpc.ServerCredentials 100 ) -> int: 101 """Opens a secure port for accepting RPCs. 102 103 This method may only be called before starting the server. 104 105 Args: 106 address: The address for which to open a port. 107 if the port is 0, or not specified in the address, then the gRPC 108 runtime will choose a port. 109 server_credentials: A ServerCredentials object. 110 111 Returns: 112 An integer port on which the server will accept RPC requests. 113 """ 114 return _common.validate_port_binding_result( 115 address, 116 self._server.add_secure_port( 117 _common.encode(address), server_credentials 118 ), 119 ) 120 121 async def start(self) -> None: 122 """Starts this Server. 123 124 This method may only be called once. (i.e. it is not idempotent). 125 """ 126 await self._server.start() 127 128 async def stop(self, grace: Optional[float]) -> None: 129 """Stops this Server. 130 131 This method immediately stops the server from servicing new RPCs in 132 all cases. 133 134 If a grace period is specified, this method waits until all active 135 RPCs are finished or until the grace period is reached. RPCs that haven't 136 been terminated within the grace period are aborted. 137 If a grace period is not specified (by passing None for grace), all 138 existing RPCs are aborted immediately and this method blocks until 139 the last RPC handler terminates. 140 141 This method is idempotent and may be called at any time. Passing a 142 smaller grace value in a subsequent call will have the effect of 143 stopping the Server sooner (passing None will have the effect of 144 stopping the server immediately). Passing a larger grace value in a 145 subsequent call will not have the effect of stopping the server later 146 (i.e. the most restrictive grace value is used). 147 148 Args: 149 grace: A duration of time in seconds or None. 150 """ 151 await self._server.shutdown(grace) 152 153 async def wait_for_termination( 154 self, timeout: Optional[float] = None 155 ) -> bool: 156 """Block current coroutine until the server stops. 157 158 This is an EXPERIMENTAL API. 159 160 The wait will not consume computational resources during blocking, and 161 it will block until one of the two following conditions are met: 162 163 1) The server is stopped or terminated; 164 2) A timeout occurs if timeout is not `None`. 165 166 The timeout argument works in the same way as `threading.Event.wait()`. 167 https://docs.python.org/3/library/threading.html#threading.Event.wait 168 169 Args: 170 timeout: A floating point number specifying a timeout for the 171 operation in seconds. 172 173 Returns: 174 A bool indicates if the operation times out. 175 """ 176 return await self._server.wait_for_termination(timeout) 177 178 def __del__(self): 179 """Schedules a graceful shutdown in current event loop. 180 181 The Cython AioServer doesn't hold a ref-count to this class. It should 182 be safe to slightly extend the underlying Cython object's life span. 183 """ 184 if hasattr(self, "_server"): 185 if self._server.is_running(): 186 cygrpc.schedule_coro_threadsafe( 187 self._server.shutdown(None), 188 self._loop, 189 ) 190 191 192def server( 193 migration_thread_pool: Optional[Executor] = None, 194 handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None, 195 interceptors: Optional[Sequence[Any]] = None, 196 options: Optional[ChannelArgumentType] = None, 197 maximum_concurrent_rpcs: Optional[int] = None, 198 compression: Optional[grpc.Compression] = None, 199): 200 """Creates a Server with which RPCs can be serviced. 201 202 Args: 203 migration_thread_pool: A futures.ThreadPoolExecutor to be used by the 204 Server to execute non-AsyncIO RPC handlers for migration purpose. 205 handlers: An optional list of GenericRpcHandlers used for executing RPCs. 206 More handlers may be added by calling add_generic_rpc_handlers any time 207 before the server is started. 208 interceptors: An optional list of ServerInterceptor objects that observe 209 and optionally manipulate the incoming RPCs before handing them over to 210 handlers. The interceptors are given control in the order they are 211 specified. This is an EXPERIMENTAL API. 212 options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime) 213 to configure the channel. 214 maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server 215 will service before returning RESOURCE_EXHAUSTED status, or None to 216 indicate no limit. 217 compression: An element of grpc.compression, e.g. 218 grpc.compression.Gzip. This compression algorithm will be used for the 219 lifetime of the server unless overridden by set_compression. 220 221 Returns: 222 A Server object. 223 """ 224 return Server( 225 migration_thread_pool, 226 () if handlers is None else handlers, 227 () if interceptors is None else interceptors, 228 () if options is None else options, 229 maximum_concurrent_rpcs, 230 compression, 231 ) 232