Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/_channel.py: 26%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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"""Invocation-side implementation of gRPC Python."""
16import copy
17import functools
18import logging
19import os
20import sys
21import threading
22import time
23import types
24from typing import (
25 Any,
26 Callable,
27 Dict,
28 Iterator,
29 List,
30 Optional,
31 Sequence,
32 Set,
33 Tuple,
34 Union,
35)
37import grpc
38from grpc import _common
39from grpc import _compression
40from grpc import _grpcio_metadata
41from grpc import _observability
42from grpc._cython import cygrpc
43from grpc._typing import ChannelArgumentType
44from grpc._typing import DeserializingFunction
45from grpc._typing import IntegratedCallFactory
46from grpc._typing import MetadataType
47from grpc._typing import NullaryCallbackType
48from grpc._typing import ResponseType
49from grpc._typing import SerializingFunction
50from grpc._typing import UserTag
51import grpc.experimental
53_LOGGER = logging.getLogger(__name__)
55_USER_AGENT = "grpc-python/{}".format(_grpcio_metadata.__version__)
57_EMPTY_FLAGS = 0
59# NOTE(rbellevi): No guarantees are given about the maintenance of this
60# environment variable.
61_DEFAULT_SINGLE_THREADED_UNARY_STREAM = (
62 os.getenv("GRPC_SINGLE_THREADED_UNARY_STREAM") is not None
63)
65_UNARY_UNARY_INITIAL_DUE = (
66 cygrpc.OperationType.send_initial_metadata,
67 cygrpc.OperationType.send_message,
68 cygrpc.OperationType.send_close_from_client,
69 cygrpc.OperationType.receive_initial_metadata,
70 cygrpc.OperationType.receive_message,
71 cygrpc.OperationType.receive_status_on_client,
72)
73_UNARY_STREAM_INITIAL_DUE = (
74 cygrpc.OperationType.send_initial_metadata,
75 cygrpc.OperationType.send_message,
76 cygrpc.OperationType.send_close_from_client,
77 cygrpc.OperationType.receive_initial_metadata,
78 cygrpc.OperationType.receive_status_on_client,
79)
80_STREAM_UNARY_INITIAL_DUE = (
81 cygrpc.OperationType.send_initial_metadata,
82 cygrpc.OperationType.receive_initial_metadata,
83 cygrpc.OperationType.receive_message,
84 cygrpc.OperationType.receive_status_on_client,
85)
86_STREAM_STREAM_INITIAL_DUE = (
87 cygrpc.OperationType.send_initial_metadata,
88 cygrpc.OperationType.receive_initial_metadata,
89 cygrpc.OperationType.receive_status_on_client,
90)
92_CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = (
93 "Exception calling channel subscription callback!"
94)
96_OK_RENDEZVOUS_REPR_FORMAT = (
97 '<{} of RPC that terminated with:\n\tstatus = {}\n\tdetails = "{}"\n>'
98)
100_NON_OK_RENDEZVOUS_REPR_FORMAT = (
101 "<{} of RPC that terminated with:\n"
102 "\tstatus = {}\n"
103 '\tdetails = "{}"\n'
104 '\tdebug_error_string = "{}"\n'
105 ">"
106)
109def _deadline(timeout: Optional[float]) -> Optional[float]:
110 return None if timeout is None else time.time() + timeout
113def _unknown_code_details(
114 unknown_cygrpc_code: Optional[grpc.StatusCode], details: Optional[str]
115) -> str:
116 return 'Server sent unknown code {} and details "{}"'.format(
117 unknown_cygrpc_code, details
118 )
121class _RPCState:
122 condition: threading.Condition
123 due: Set[cygrpc.OperationType]
124 initial_metadata: Optional[MetadataType]
125 response: Any
126 trailing_metadata: Optional[MetadataType]
127 code: Optional[grpc.StatusCode]
128 details: Optional[str]
129 debug_error_string: Optional[str]
130 cancelled: bool
131 callbacks: List[NullaryCallbackType]
132 fork_epoch: Optional[int]
133 rpc_start_time: Optional[float] # In relative seconds
134 rpc_end_time: Optional[float] # In relative seconds
135 method: Optional[str]
136 target: Optional[str]
138 def __init__(
139 self,
140 due: Sequence[cygrpc.OperationType],
141 initial_metadata: Optional[MetadataType],
142 trailing_metadata: Optional[MetadataType],
143 code: Optional[grpc.StatusCode],
144 details: Optional[str],
145 ):
146 # `condition` guards all members of _RPCState. `notify_all` is called on
147 # `condition` when the state of the RPC has changed.
148 self.condition = threading.Condition()
150 # The cygrpc.OperationType objects representing events due from the RPC's
151 # completion queue. If an operation is in `due`, it is guaranteed that
152 # `operate()` has been called on a corresponding operation. But the
153 # converse is not true. That is, in the case of failed `operate()`
154 # calls, there may briefly be events in `due` that do not correspond to
155 # operations submitted to Core.
156 self.due = set(due)
157 self.initial_metadata = initial_metadata
158 self.response = None
159 self.trailing_metadata = trailing_metadata
160 self.code = code
161 self.details = details
162 self.debug_error_string = None
163 # The following three fields are used for observability.
164 # Updates to those fields do not trigger self.condition.
165 self.rpc_start_time = None
166 self.rpc_end_time = None
167 self.method = None
168 self.target = None
170 # The semantics of grpc.Future.cancel and grpc.Future.cancelled are
171 # slightly wonky, so they have to be tracked separately from the rest of the
172 # result of the RPC. This field tracks whether cancellation was requested
173 # prior to termination of the RPC.
174 self.cancelled = False
175 self.callbacks = []
176 self.fork_epoch = cygrpc.get_fork_epoch()
178 def reset_postfork_child(self):
179 self.condition = threading.Condition()
182def _abort(state: _RPCState, code: grpc.StatusCode, details: str) -> None:
183 if state.code is None:
184 state.code = code
185 state.details = details
186 if state.initial_metadata is None:
187 state.initial_metadata = ()
188 state.trailing_metadata = ()
191def _handle_event(
192 event: cygrpc.BaseEvent,
193 state: _RPCState,
194 response_deserializer: Optional[DeserializingFunction],
195) -> List[NullaryCallbackType]:
196 callbacks = []
197 for batch_operation in event.batch_operations:
198 operation_type = batch_operation.type()
199 state.due.remove(operation_type)
200 if operation_type == cygrpc.OperationType.receive_initial_metadata:
201 state.initial_metadata = batch_operation.initial_metadata()
202 elif operation_type == cygrpc.OperationType.receive_message:
203 serialized_response = batch_operation.message()
204 if serialized_response is not None:
205 response = _common.deserialize(
206 serialized_response, response_deserializer
207 )
208 if response is None:
209 details = "Exception deserializing response!"
210 _abort(state, grpc.StatusCode.INTERNAL, details)
211 else:
212 state.response = response
213 elif operation_type == cygrpc.OperationType.receive_status_on_client:
214 state.trailing_metadata = batch_operation.trailing_metadata()
215 if state.code is None:
216 code = _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE.get(
217 batch_operation.code()
218 )
219 if code is None:
220 state.code = grpc.StatusCode.UNKNOWN
221 state.details = _unknown_code_details(
222 code, batch_operation.details()
223 )
224 else:
225 state.code = code
226 state.details = batch_operation.details()
227 state.debug_error_string = batch_operation.error_string()
228 state.rpc_end_time = time.perf_counter()
229 _observability.maybe_record_rpc_latency(state)
230 callbacks.extend(state.callbacks)
231 state.callbacks = None
232 return callbacks
235def _event_handler(
236 state: _RPCState, response_deserializer: Optional[DeserializingFunction]
237) -> UserTag:
238 def handle_event(event):
239 with state.condition:
240 callbacks = _handle_event(event, state, response_deserializer)
241 state.condition.notify_all()
242 done = not state.due
243 for callback in callbacks:
244 try:
245 callback()
246 except Exception as e: # pylint: disable=broad-except
247 # NOTE(rbellevi): We suppress but log errors here so as not to
248 # kill the channel spin thread.
249 _LOGGER.error(
250 "Exception in callback %s: %s", repr(callback.func), repr(e)
251 )
252 return done and state.fork_epoch >= cygrpc.get_fork_epoch()
254 return handle_event
257# TODO(xuanwn): Create a base class for IntegratedCall and SegregatedCall.
258# pylint: disable=too-many-statements
259def _consume_request_iterator(
260 request_iterator: Iterator,
261 state: _RPCState,
262 call: Union[cygrpc.IntegratedCall, cygrpc.SegregatedCall],
263 request_serializer: SerializingFunction,
264 event_handler: Optional[UserTag],
265) -> None:
266 """Consume a request supplied by the user."""
268 def consume_request_iterator(): # pylint: disable=too-many-branches
269 # Iterate over the request iterator until it is exhausted or an error
270 # condition is encountered.
271 while True:
272 return_from_user_request_generator_invoked = False
273 try:
274 # The thread may die in user-code. Do not block fork for this.
275 cygrpc.enter_user_request_generator()
276 request = next(request_iterator)
277 except StopIteration:
278 break
279 except Exception: # pylint: disable=broad-except
280 cygrpc.return_from_user_request_generator()
281 return_from_user_request_generator_invoked = True
282 code = grpc.StatusCode.UNKNOWN
283 details = "Exception iterating requests!"
284 _LOGGER.exception(details)
285 call.cancel(
286 _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code], details
287 )
288 _abort(state, code, details)
289 return
290 finally:
291 if not return_from_user_request_generator_invoked:
292 cygrpc.return_from_user_request_generator()
293 serialized_request = _common.serialize(request, request_serializer)
294 with state.condition:
295 if state.code is None and not state.cancelled:
296 if serialized_request is None:
297 code = grpc.StatusCode.INTERNAL
298 details = "Exception serializing request!"
299 call.cancel(
300 _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code],
301 details,
302 )
303 _abort(state, code, details)
304 return
305 state.due.add(cygrpc.OperationType.send_message)
306 operations = (
307 cygrpc.SendMessageOperation(
308 serialized_request, _EMPTY_FLAGS
309 ),
310 )
311 operating = call.operate(operations, event_handler)
312 if not operating:
313 state.due.remove(cygrpc.OperationType.send_message)
314 return
316 def _done():
317 return (
318 state.code is not None
319 or cygrpc.OperationType.send_message
320 not in state.due
321 )
323 _common.wait(
324 state.condition.wait,
325 _done,
326 spin_cb=functools.partial(
327 cygrpc.block_if_fork_in_progress, state
328 ),
329 )
330 if state.code is not None:
331 return
332 else:
333 return
334 with state.condition:
335 if state.code is None:
336 state.due.add(cygrpc.OperationType.send_close_from_client)
337 operations = (
338 cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
339 )
340 operating = call.operate(operations, event_handler)
341 if not operating:
342 state.due.remove(
343 cygrpc.OperationType.send_close_from_client
344 )
346 consumption_thread = cygrpc.ForkManagedThread(
347 target=consume_request_iterator
348 )
349 consumption_thread.setDaemon(True)
350 consumption_thread.start()
353def _rpc_state_string(class_name: str, rpc_state: _RPCState) -> str:
354 """Calculates error string for RPC."""
355 with rpc_state.condition:
356 if rpc_state.code is None:
357 return "<{} object>".format(class_name)
358 if rpc_state.code is grpc.StatusCode.OK:
359 return _OK_RENDEZVOUS_REPR_FORMAT.format(
360 class_name, rpc_state.code, rpc_state.details
361 )
362 return _NON_OK_RENDEZVOUS_REPR_FORMAT.format(
363 class_name,
364 rpc_state.code,
365 rpc_state.details,
366 rpc_state.debug_error_string,
367 )
370class _InactiveRpcError(grpc.RpcError, grpc.Call, grpc.Future):
371 """An RPC error not tied to the execution of a particular RPC.
373 The RPC represented by the state object must not be in-progress or
374 cancelled.
376 Attributes:
377 _state: An instance of _RPCState.
378 """
380 _state: _RPCState
382 def __init__(self, state: _RPCState):
383 if not isinstance(state, _RPCState):
384 # Handles exception wrapping edge-cases.
385 # See https://github.com/grpc/grpc/issues/38713 and
386 # https://github.com/pytorch/pytorch/issues/34130.
387 msg = "Inactive RPC Error: Invalid RPC state type."
388 if isinstance(state, str):
389 msg = f"{msg} Details: {state}"
390 state = _RPCState((), (), (), grpc.StatusCode.INTERNAL, msg)
392 with state.condition:
393 self._state = _RPCState(
394 (),
395 copy.deepcopy(state.initial_metadata),
396 copy.deepcopy(state.trailing_metadata),
397 state.code,
398 copy.deepcopy(state.details),
399 )
400 self._state.response = copy.copy(state.response)
401 self._state.debug_error_string = copy.copy(state.debug_error_string)
403 def initial_metadata(self) -> Optional[MetadataType]:
404 return self._state.initial_metadata
406 def trailing_metadata(self) -> Optional[MetadataType]:
407 return self._state.trailing_metadata
409 def code(self) -> Optional[grpc.StatusCode]:
410 return self._state.code
412 def details(self) -> Optional[str]:
413 return _common.decode(self._state.details)
415 def debug_error_string(self) -> Optional[str]:
416 return _common.decode(self._state.debug_error_string)
418 def _repr(self) -> str:
419 return _rpc_state_string(self.__class__.__name__, self._state)
421 def __repr__(self) -> str:
422 return self._repr()
424 def __str__(self) -> str:
425 return self._repr()
427 def cancel(self) -> bool:
428 """See grpc.Future.cancel."""
429 return False
431 def cancelled(self) -> bool:
432 """See grpc.Future.cancelled."""
433 return False
435 def running(self) -> bool:
436 """See grpc.Future.running."""
437 return False
439 def done(self) -> bool:
440 """See grpc.Future.done."""
441 return True
443 def result(
444 self, timeout: Optional[float] = None
445 ) -> Any: # pylint: disable=unused-argument
446 """See grpc.Future.result."""
447 raise self
449 def exception(
450 self, timeout: Optional[float] = None # pylint: disable=unused-argument
451 ) -> Optional[Exception]:
452 """See grpc.Future.exception."""
453 return self
455 def traceback(
456 self, timeout: Optional[float] = None # pylint: disable=unused-argument
457 ) -> Optional[types.TracebackType]:
458 """See grpc.Future.traceback."""
459 try:
460 raise self
461 except grpc.RpcError:
462 return sys.exc_info()[2]
464 def add_done_callback(
465 self,
466 fn: Callable[[grpc.Future], None],
467 timeout: Optional[float] = None, # pylint: disable=unused-argument
468 ) -> None:
469 """See grpc.Future.add_done_callback."""
470 fn(self)
473class _Rendezvous(grpc.RpcError, grpc.RpcContext):
474 """An RPC iterator.
476 Attributes:
477 _state: An instance of _RPCState.
478 _call: An instance of SegregatedCall or IntegratedCall.
479 In either case, the _call object is expected to have operate, cancel,
480 and next_event methods.
481 _response_deserializer: A callable taking bytes and return a Python
482 object.
483 _deadline: A float representing the deadline of the RPC in seconds. Or
484 possibly None, to represent an RPC with no deadline at all.
485 """
487 _state: _RPCState
488 _call: Union[cygrpc.SegregatedCall, cygrpc.IntegratedCall]
489 _response_deserializer: Optional[DeserializingFunction]
490 _deadline: Optional[float]
492 def __init__(
493 self,
494 state: _RPCState,
495 call: Union[cygrpc.SegregatedCall, cygrpc.IntegratedCall],
496 response_deserializer: Optional[DeserializingFunction],
497 deadline: Optional[float],
498 ):
499 super(_Rendezvous, self).__init__()
500 self._state = state
501 self._call = call
502 self._response_deserializer = response_deserializer
503 self._deadline = deadline
505 def is_active(self) -> bool:
506 """See grpc.RpcContext.is_active"""
507 with self._state.condition:
508 return self._state.code is None
510 def time_remaining(self) -> Optional[float]:
511 """See grpc.RpcContext.time_remaining"""
512 with self._state.condition:
513 if self._deadline is None:
514 return None
515 return max(self._deadline - time.time(), 0)
517 def cancel(self) -> bool:
518 """See grpc.RpcContext.cancel"""
519 with self._state.condition:
520 if self._state.code is None:
521 code = grpc.StatusCode.CANCELLED
522 details = "Locally cancelled by application!"
523 self._call.cancel(
524 _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code], details
525 )
526 self._state.cancelled = True
527 _abort(self._state, code, details)
528 self._state.condition.notify_all()
529 return True
530 return False
532 def add_callback(self, callback: NullaryCallbackType) -> bool:
533 """See grpc.RpcContext.add_callback"""
534 with self._state.condition:
535 if self._state.callbacks is None:
536 return False
537 self._state.callbacks.append(callback)
538 return True
540 def __iter__(self):
541 return self
543 def next(self):
544 return self._next()
546 def __next__(self):
547 return self._next()
549 def _next(self):
550 raise NotImplementedError()
552 def debug_error_string(self) -> Optional[str]:
553 raise NotImplementedError()
555 def _repr(self) -> str:
556 return _rpc_state_string(self.__class__.__name__, self._state)
558 def __repr__(self) -> str:
559 return self._repr()
561 def __str__(self) -> str:
562 return self._repr()
564 def __del__(self) -> None:
565 with self._state.condition:
566 if self._state.code is None:
567 self._state.code = grpc.StatusCode.CANCELLED
568 self._state.details = "Cancelled upon garbage collection!"
569 self._state.cancelled = True
570 self._call.cancel(
571 _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[self._state.code],
572 self._state.details,
573 )
574 self._state.condition.notify_all()
577class _SingleThreadedRendezvous(
578 _Rendezvous, grpc.Call, grpc.Future
579): # pylint: disable=too-many-ancestors
580 """An RPC iterator operating entirely on a single thread.
582 The __next__ method of _SingleThreadedRendezvous does not depend on the
583 existence of any other thread, including the "channel spin thread".
584 However, this means that its interface is entirely synchronous. So this
585 class cannot completely fulfill the grpc.Future interface. The result,
586 exception, and traceback methods will never block and will instead raise
587 an exception if calling the method would result in blocking.
589 This means that these methods are safe to call from add_done_callback
590 handlers.
591 """
593 _state: _RPCState
595 def _is_complete(self) -> bool:
596 return self._state.code is not None
598 def cancelled(self) -> bool:
599 with self._state.condition:
600 return self._state.cancelled
602 def running(self) -> bool:
603 with self._state.condition:
604 return self._state.code is None
606 def done(self) -> bool:
607 with self._state.condition:
608 return self._state.code is not None
610 def result(self, timeout: Optional[float] = None) -> Any:
611 """Returns the result of the computation or raises its exception.
613 This method will never block. Instead, it will raise an exception
614 if calling this method would otherwise result in blocking.
616 Since this method will never block, any `timeout` argument passed will
617 be ignored.
618 """
619 del timeout
620 with self._state.condition:
621 if not self._is_complete():
622 error_msg = (
623 "_SingleThreadedRendezvous only supports "
624 "result() when the RPC is complete."
625 )
626 raise grpc.experimental.UsageError(error_msg)
627 if self._state.code is grpc.StatusCode.OK:
628 return self._state.response
629 if self._state.cancelled:
630 raise grpc.FutureCancelledError()
631 raise self
633 def exception(self, timeout: Optional[float] = None) -> Optional[Exception]:
634 """Return the exception raised by the computation.
636 This method will never block. Instead, it will raise an exception
637 if calling this method would otherwise result in blocking.
639 Since this method will never block, any `timeout` argument passed will
640 be ignored.
641 """
642 del timeout
643 with self._state.condition:
644 if not self._is_complete():
645 error_msg = (
646 "_SingleThreadedRendezvous only supports "
647 "exception() when the RPC is complete."
648 )
649 raise grpc.experimental.UsageError(error_msg)
650 if self._state.code is grpc.StatusCode.OK:
651 return None
652 if self._state.cancelled:
653 raise grpc.FutureCancelledError()
654 return self
656 def traceback(
657 self, timeout: Optional[float] = None
658 ) -> Optional[types.TracebackType]:
659 """Access the traceback of the exception raised by the computation.
661 This method will never block. Instead, it will raise an exception
662 if calling this method would otherwise result in blocking.
664 Since this method will never block, any `timeout` argument passed will
665 be ignored.
666 """
667 del timeout
668 with self._state.condition:
669 if not self._is_complete():
670 msg = (
671 "_SingleThreadedRendezvous only supports "
672 "traceback() when the RPC is complete."
673 )
674 raise grpc.experimental.UsageError(msg)
675 if self._state.code is grpc.StatusCode.OK:
676 return None
677 if self._state.cancelled:
678 raise grpc.FutureCancelledError()
679 try:
680 raise self
681 except grpc.RpcError:
682 return sys.exc_info()[2]
684 def add_done_callback(self, fn: Callable[[grpc.Future], None]) -> None:
685 with self._state.condition:
686 if self._state.code is None:
687 self._state.callbacks.append(functools.partial(fn, self))
688 return
690 fn(self)
692 def initial_metadata(self) -> Optional[MetadataType]:
693 """See grpc.Call.initial_metadata"""
694 with self._state.condition:
695 # NOTE(gnossen): Based on our initial call batch, we are guaranteed
696 # to receive initial metadata before any messages.
697 while self._state.initial_metadata is None:
698 self._consume_next_event()
699 return self._state.initial_metadata
701 def trailing_metadata(self) -> Optional[MetadataType]:
702 """See grpc.Call.trailing_metadata"""
703 with self._state.condition:
704 if self._state.trailing_metadata is None:
705 error_msg = (
706 "Cannot get trailing metadata until RPC is completed."
707 )
708 raise grpc.experimental.UsageError(error_msg)
709 return self._state.trailing_metadata
711 def code(self) -> Optional[grpc.StatusCode]:
712 """See grpc.Call.code"""
713 with self._state.condition:
714 if self._state.code is None:
715 error_msg = "Cannot get code until RPC is completed."
716 raise grpc.experimental.UsageError(error_msg)
717 return self._state.code
719 def details(self) -> Optional[str]:
720 """See grpc.Call.details"""
721 with self._state.condition:
722 if self._state.details is None:
723 error_msg = "Cannot get details until RPC is completed."
724 raise grpc.experimental.UsageError(error_msg)
725 return _common.decode(self._state.details)
727 def _consume_next_event(self) -> Optional[cygrpc.BaseEvent]:
728 event = self._call.next_event()
729 with self._state.condition:
730 callbacks = _handle_event(
731 event, self._state, self._response_deserializer
732 )
733 for callback in callbacks:
734 # NOTE(gnossen): We intentionally allow exceptions to bubble up
735 # to the user when running on a single thread.
736 callback()
737 return event
739 def _next_response(self) -> Any:
740 while True:
741 self._consume_next_event()
742 with self._state.condition:
743 if self._state.response is not None:
744 response = self._state.response
745 self._state.response = None
746 return response
747 if cygrpc.OperationType.receive_message not in self._state.due:
748 if self._state.code is grpc.StatusCode.OK:
749 raise StopIteration()
750 if self._state.code is not None:
751 raise self
753 def _next(self) -> Any:
754 with self._state.condition:
755 if self._state.code is None:
756 # We tentatively add the operation as expected and remove
757 # it if the enqueue operation fails. This allows us to guarantee that
758 # if an event has been submitted to the core completion queue,
759 # it is in `due`. If we waited until after a successful
760 # enqueue operation then a signal could interrupt this
761 # thread between the enqueue operation and the addition of the
762 # operation to `due`. This would cause an exception on the
763 # channel spin thread when the operation completes and no
764 # corresponding operation would be present in state.due.
765 # Note that, since `condition` is held through this block, there is
766 # no data race on `due`.
767 self._state.due.add(cygrpc.OperationType.receive_message)
768 operating = self._call.operate(
769 (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), None
770 )
771 if not operating:
772 self._state.due.remove(cygrpc.OperationType.receive_message)
773 elif self._state.code is grpc.StatusCode.OK:
774 raise StopIteration()
775 else:
776 raise self
777 return self._next_response()
779 def debug_error_string(self) -> Optional[str]:
780 with self._state.condition:
781 if self._state.debug_error_string is None:
782 error_msg = (
783 "Cannot get debug error string until RPC is completed."
784 )
785 raise grpc.experimental.UsageError(error_msg)
786 return _common.decode(self._state.debug_error_string)
789class _MultiThreadedRendezvous(
790 _Rendezvous, grpc.Call, grpc.Future
791): # pylint: disable=too-many-ancestors
792 """An RPC iterator that depends on a channel spin thread.
794 This iterator relies upon a per-channel thread running in the background,
795 dequeueing events from the completion queue, and notifying threads waiting
796 on the threading.Condition object in the _RPCState object.
798 This extra thread allows _MultiThreadedRendezvous to fulfill the grpc.Future interface
799 and to mediate a bidirection streaming RPC.
800 """
802 _state: _RPCState
804 def initial_metadata(self) -> Optional[MetadataType]:
805 """See grpc.Call.initial_metadata"""
806 with self._state.condition:
808 def _done():
809 return self._state.initial_metadata is not None
811 _common.wait(self._state.condition.wait, _done)
812 return self._state.initial_metadata
814 def trailing_metadata(self) -> Optional[MetadataType]:
815 """See grpc.Call.trailing_metadata"""
816 with self._state.condition:
818 def _done():
819 return self._state.trailing_metadata is not None
821 _common.wait(self._state.condition.wait, _done)
822 return self._state.trailing_metadata
824 def code(self) -> Optional[grpc.StatusCode]:
825 """See grpc.Call.code"""
826 with self._state.condition:
828 def _done():
829 return self._state.code is not None
831 _common.wait(self._state.condition.wait, _done)
832 return self._state.code
834 def details(self) -> Optional[str]:
835 """See grpc.Call.details"""
836 with self._state.condition:
838 def _done():
839 return self._state.details is not None
841 _common.wait(self._state.condition.wait, _done)
842 return _common.decode(self._state.details)
844 def debug_error_string(self) -> Optional[str]:
845 with self._state.condition:
847 def _done():
848 return self._state.debug_error_string is not None
850 _common.wait(self._state.condition.wait, _done)
851 return _common.decode(self._state.debug_error_string)
853 def cancelled(self) -> bool:
854 with self._state.condition:
855 return self._state.cancelled
857 def running(self) -> bool:
858 with self._state.condition:
859 return self._state.code is None
861 def done(self) -> bool:
862 with self._state.condition:
863 return self._state.code is not None
865 def _is_complete(self) -> bool:
866 return self._state.code is not None
868 def result(self, timeout: Optional[float] = None) -> Any:
869 """Returns the result of the computation or raises its exception.
871 See grpc.Future.result for the full API contract.
872 """
873 with self._state.condition:
874 timed_out = _common.wait(
875 self._state.condition.wait, self._is_complete, timeout=timeout
876 )
877 if timed_out:
878 raise grpc.FutureTimeoutError()
879 if self._state.code is grpc.StatusCode.OK:
880 return self._state.response
881 if self._state.cancelled:
882 raise grpc.FutureCancelledError()
883 raise self
885 def exception(self, timeout: Optional[float] = None) -> Optional[Exception]:
886 """Return the exception raised by the computation.
888 See grpc.Future.exception for the full API contract.
889 """
890 with self._state.condition:
891 timed_out = _common.wait(
892 self._state.condition.wait, self._is_complete, timeout=timeout
893 )
894 if timed_out:
895 raise grpc.FutureTimeoutError()
896 if self._state.code is grpc.StatusCode.OK:
897 return None
898 if self._state.cancelled:
899 raise grpc.FutureCancelledError()
900 return self
902 def traceback(
903 self, timeout: Optional[float] = None
904 ) -> Optional[types.TracebackType]:
905 """Access the traceback of the exception raised by the computation.
907 See grpc.future.traceback for the full API contract.
908 """
909 with self._state.condition:
910 timed_out = _common.wait(
911 self._state.condition.wait, self._is_complete, timeout=timeout
912 )
913 if timed_out:
914 raise grpc.FutureTimeoutError()
915 if self._state.code is grpc.StatusCode.OK:
916 return None
917 if self._state.cancelled:
918 raise grpc.FutureCancelledError()
919 try:
920 raise self
921 except grpc.RpcError:
922 return sys.exc_info()[2]
924 def add_done_callback(self, fn: Callable[[grpc.Future], None]) -> None:
925 with self._state.condition:
926 if self._state.code is None:
927 self._state.callbacks.append(functools.partial(fn, self))
928 return
930 fn(self)
932 def _next(self) -> Any:
933 with self._state.condition:
934 if self._state.code is None:
935 event_handler = _event_handler(
936 self._state, self._response_deserializer
937 )
938 self._state.due.add(cygrpc.OperationType.receive_message)
939 operating = self._call.operate(
940 (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),),
941 event_handler,
942 )
943 if not operating:
944 self._state.due.remove(cygrpc.OperationType.receive_message)
945 elif self._state.code is grpc.StatusCode.OK:
946 raise StopIteration()
947 else:
948 raise self
950 def _response_ready():
951 return self._state.response is not None or (
952 cygrpc.OperationType.receive_message not in self._state.due
953 and self._state.code is not None
954 )
956 _common.wait(self._state.condition.wait, _response_ready)
957 if self._state.response is not None:
958 response = self._state.response
959 self._state.response = None
960 return response
961 if cygrpc.OperationType.receive_message not in self._state.due:
962 if self._state.code is grpc.StatusCode.OK:
963 raise StopIteration()
964 if self._state.code is not None:
965 raise self
968def _start_unary_request(
969 request: Any,
970 timeout: Optional[float],
971 request_serializer: Optional[SerializingFunction],
972) -> Tuple[Optional[float], Optional[bytes], Optional[grpc.RpcError]]:
973 deadline = _deadline(timeout)
974 serialized_request = _common.serialize(request, request_serializer)
975 if serialized_request is None:
976 state = _RPCState(
977 (),
978 (),
979 (),
980 grpc.StatusCode.INTERNAL,
981 "Exception serializing request!",
982 )
983 error = _InactiveRpcError(state)
984 return deadline, None, error
985 return deadline, serialized_request, None
988def _end_unary_response_blocking(
989 state: _RPCState,
990 call: cygrpc.SegregatedCall,
991 with_call: bool,
992 deadline: Optional[float],
993) -> Union[ResponseType, Tuple[ResponseType, grpc.Call]]:
994 if state.code is grpc.StatusCode.OK:
995 if with_call:
996 rendezvous = _MultiThreadedRendezvous(state, call, None, deadline)
997 return state.response, rendezvous
998 return state.response
999 raise _InactiveRpcError(state) # pytype: disable=not-instantiable
1002def _stream_unary_invocation_operations(
1003 metadata: Optional[MetadataType], initial_metadata_flags: int
1004) -> Sequence[Sequence[cygrpc.Operation]]:
1005 return (
1006 (
1007 cygrpc.SendInitialMetadataOperation(
1008 metadata, initial_metadata_flags
1009 ),
1010 cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
1011 cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
1012 ),
1013 (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),),
1014 )
1017def _stream_unary_invocation_operations_and_tags(
1018 metadata: Optional[MetadataType], initial_metadata_flags: int
1019) -> Sequence[Tuple[Sequence[cygrpc.Operation], Optional[UserTag]]]:
1020 return tuple(
1021 (
1022 operations,
1023 None,
1024 )
1025 for operations in _stream_unary_invocation_operations(
1026 metadata, initial_metadata_flags
1027 )
1028 )
1031def _determine_deadline(user_deadline: Optional[float]) -> Optional[float]:
1032 parent_deadline = cygrpc.get_deadline_from_context()
1033 if parent_deadline is None and user_deadline is None:
1034 return None
1035 if parent_deadline is not None and user_deadline is None:
1036 return parent_deadline
1037 if user_deadline is not None and parent_deadline is None:
1038 return user_deadline
1039 return min(parent_deadline, user_deadline)
1042class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
1043 _channel: cygrpc.Channel
1044 _managed_call: IntegratedCallFactory
1045 _method: bytes
1046 _target: bytes
1047 _request_serializer: Optional[SerializingFunction]
1048 _response_deserializer: Optional[DeserializingFunction]
1049 _context: Any
1050 _registered_call_handle: Optional[int]
1052 __slots__ = [
1053 "_channel",
1054 "_context",
1055 "_managed_call",
1056 "_method",
1057 "_request_serializer",
1058 "_response_deserializer",
1059 "_target",
1060 ]
1062 # pylint: disable=too-many-arguments
1063 def __init__(
1064 self,
1065 channel: cygrpc.Channel,
1066 managed_call: IntegratedCallFactory,
1067 method: bytes,
1068 target: bytes,
1069 request_serializer: Optional[SerializingFunction],
1070 response_deserializer: Optional[DeserializingFunction],
1071 _registered_call_handle: Optional[int],
1072 ):
1073 self._channel = channel
1074 self._managed_call = managed_call
1075 self._method = method
1076 self._target = target
1077 self._request_serializer = request_serializer
1078 self._response_deserializer = response_deserializer
1079 self._context = cygrpc.build_census_context()
1080 self._registered_call_handle = _registered_call_handle
1082 def _prepare(
1083 self,
1084 request: Any,
1085 timeout: Optional[float],
1086 metadata: Optional[MetadataType],
1087 wait_for_ready: Optional[bool],
1088 compression: Optional[grpc.Compression],
1089 ) -> Tuple[
1090 Optional[_RPCState],
1091 Optional[Sequence[cygrpc.Operation]],
1092 Optional[float],
1093 Optional[grpc.RpcError],
1094 ]:
1095 deadline, serialized_request, rendezvous = _start_unary_request(
1096 request, timeout, self._request_serializer
1097 )
1098 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1099 wait_for_ready
1100 )
1101 augmented_metadata = _compression.augment_metadata(
1102 metadata, compression
1103 )
1104 if serialized_request is None:
1105 return None, None, None, rendezvous
1106 state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None)
1107 operations = (
1108 cygrpc.SendInitialMetadataOperation(
1109 augmented_metadata, initial_metadata_flags
1110 ),
1111 cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS),
1112 cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
1113 cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
1114 cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
1115 cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
1116 )
1117 return state, operations, deadline, None
1119 def _blocking(
1120 self,
1121 request: Any,
1122 timeout: Optional[float] = None,
1123 metadata: Optional[MetadataType] = None,
1124 credentials: Optional[grpc.CallCredentials] = None,
1125 wait_for_ready: Optional[bool] = None,
1126 compression: Optional[grpc.Compression] = None,
1127 ) -> Tuple[_RPCState, cygrpc.SegregatedCall]:
1128 state, operations, deadline, rendezvous = self._prepare(
1129 request, timeout, metadata, wait_for_ready, compression
1130 )
1131 if state is None:
1132 raise rendezvous # pylint: disable-msg=raising-bad-type
1133 state.rpc_start_time = time.perf_counter()
1134 state.method = _common.decode(self._method)
1135 state.target = _common.decode(self._target)
1136 call = self._channel.segregated_call(
1137 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1138 self._method,
1139 None,
1140 _determine_deadline(deadline),
1141 metadata,
1142 None if credentials is None else credentials._credentials,
1143 (
1144 (
1145 operations,
1146 None,
1147 ),
1148 ),
1149 self._context,
1150 self._registered_call_handle,
1151 )
1152 event = call.next_event()
1153 _handle_event(event, state, self._response_deserializer)
1154 return state, call
1156 def __call__(
1157 self,
1158 request: Any,
1159 timeout: Optional[float] = None,
1160 metadata: Optional[MetadataType] = None,
1161 credentials: Optional[grpc.CallCredentials] = None,
1162 wait_for_ready: Optional[bool] = None,
1163 compression: Optional[grpc.Compression] = None,
1164 ) -> Any:
1165 state, call = self._blocking(
1166 request, timeout, metadata, credentials, wait_for_ready, compression
1167 )
1168 return _end_unary_response_blocking(state, call, False, None)
1170 def with_call(
1171 self,
1172 request: Any,
1173 timeout: Optional[float] = None,
1174 metadata: Optional[MetadataType] = None,
1175 credentials: Optional[grpc.CallCredentials] = None,
1176 wait_for_ready: Optional[bool] = None,
1177 compression: Optional[grpc.Compression] = None,
1178 ) -> Tuple[Any, grpc.Call]:
1179 state, call = self._blocking(
1180 request, timeout, metadata, credentials, wait_for_ready, compression
1181 )
1182 return _end_unary_response_blocking(state, call, True, None)
1184 def future(
1185 self,
1186 request: Any,
1187 timeout: Optional[float] = None,
1188 metadata: Optional[MetadataType] = None,
1189 credentials: Optional[grpc.CallCredentials] = None,
1190 wait_for_ready: Optional[bool] = None,
1191 compression: Optional[grpc.Compression] = None,
1192 ) -> _MultiThreadedRendezvous:
1193 state, operations, deadline, rendezvous = self._prepare(
1194 request, timeout, metadata, wait_for_ready, compression
1195 )
1196 if state is None:
1197 raise rendezvous # pylint: disable-msg=raising-bad-type
1198 event_handler = _event_handler(state, self._response_deserializer)
1199 state.rpc_start_time = time.perf_counter()
1200 state.method = _common.decode(self._method)
1201 state.target = _common.decode(self._target)
1202 call = self._managed_call(
1203 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1204 self._method,
1205 None,
1206 deadline,
1207 metadata,
1208 None if credentials is None else credentials._credentials,
1209 (operations,),
1210 event_handler,
1211 self._context,
1212 self._registered_call_handle,
1213 )
1214 return _MultiThreadedRendezvous(
1215 state, call, self._response_deserializer, deadline
1216 )
1219class _SingleThreadedUnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable):
1220 _channel: cygrpc.Channel
1221 _method: bytes
1222 _target: bytes
1223 _request_serializer: Optional[SerializingFunction]
1224 _response_deserializer: Optional[DeserializingFunction]
1225 _context: Any
1226 _registered_call_handle: Optional[int]
1228 __slots__ = [
1229 "_channel",
1230 "_context",
1231 "_method",
1232 "_request_serializer",
1233 "_response_deserializer",
1234 "_target",
1235 ]
1237 # pylint: disable=too-many-arguments
1238 def __init__(
1239 self,
1240 channel: cygrpc.Channel,
1241 method: bytes,
1242 target: bytes,
1243 request_serializer: SerializingFunction,
1244 response_deserializer: DeserializingFunction,
1245 _registered_call_handle: Optional[int],
1246 ):
1247 self._channel = channel
1248 self._method = method
1249 self._target = target
1250 self._request_serializer = request_serializer
1251 self._response_deserializer = response_deserializer
1252 self._context = cygrpc.build_census_context()
1253 self._registered_call_handle = _registered_call_handle
1255 def __call__( # pylint: disable=too-many-locals
1256 self,
1257 request: Any,
1258 timeout: Optional[float] = None,
1259 metadata: Optional[MetadataType] = None,
1260 credentials: Optional[grpc.CallCredentials] = None,
1261 wait_for_ready: Optional[bool] = None,
1262 compression: Optional[grpc.Compression] = None,
1263 ) -> _SingleThreadedRendezvous:
1264 deadline = _deadline(timeout)
1265 serialized_request = _common.serialize(
1266 request, self._request_serializer
1267 )
1268 if serialized_request is None:
1269 state = _RPCState(
1270 (),
1271 (),
1272 (),
1273 grpc.StatusCode.INTERNAL,
1274 "Exception serializing request!",
1275 )
1276 raise _InactiveRpcError(state)
1278 state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None)
1279 call_credentials = (
1280 None if credentials is None else credentials._credentials
1281 )
1282 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1283 wait_for_ready
1284 )
1285 augmented_metadata = _compression.augment_metadata(
1286 metadata, compression
1287 )
1288 operations = (
1289 (
1290 cygrpc.SendInitialMetadataOperation(
1291 augmented_metadata, initial_metadata_flags
1292 ),
1293 cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS),
1294 cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
1295 ),
1296 (cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),),
1297 (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),),
1298 )
1299 operations_and_tags = tuple((ops, None) for ops in operations)
1300 state.rpc_start_time = time.perf_counter()
1301 state.method = _common.decode(self._method)
1302 state.target = _common.decode(self._target)
1303 call = self._channel.segregated_call(
1304 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1305 self._method,
1306 None,
1307 _determine_deadline(deadline),
1308 metadata,
1309 call_credentials,
1310 operations_and_tags,
1311 self._context,
1312 self._registered_call_handle,
1313 )
1314 return _SingleThreadedRendezvous(
1315 state, call, self._response_deserializer, deadline
1316 )
1319class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable):
1320 _channel: cygrpc.Channel
1321 _managed_call: IntegratedCallFactory
1322 _method: bytes
1323 _target: bytes
1324 _request_serializer: Optional[SerializingFunction]
1325 _response_deserializer: Optional[DeserializingFunction]
1326 _context: Any
1327 _registered_call_handle: Optional[int]
1329 __slots__ = [
1330 "_channel",
1331 "_context",
1332 "_managed_call",
1333 "_method",
1334 "_request_serializer",
1335 "_response_deserializer",
1336 "_target",
1337 ]
1339 # pylint: disable=too-many-arguments
1340 def __init__(
1341 self,
1342 channel: cygrpc.Channel,
1343 managed_call: IntegratedCallFactory,
1344 method: bytes,
1345 target: bytes,
1346 request_serializer: SerializingFunction,
1347 response_deserializer: DeserializingFunction,
1348 _registered_call_handle: Optional[int],
1349 ):
1350 self._channel = channel
1351 self._managed_call = managed_call
1352 self._method = method
1353 self._target = target
1354 self._request_serializer = request_serializer
1355 self._response_deserializer = response_deserializer
1356 self._context = cygrpc.build_census_context()
1357 self._registered_call_handle = _registered_call_handle
1359 def __call__( # pylint: disable=too-many-locals
1360 self,
1361 request: Any,
1362 timeout: Optional[float] = None,
1363 metadata: Optional[MetadataType] = None,
1364 credentials: Optional[grpc.CallCredentials] = None,
1365 wait_for_ready: Optional[bool] = None,
1366 compression: Optional[grpc.Compression] = None,
1367 ) -> _MultiThreadedRendezvous:
1368 deadline, serialized_request, rendezvous = _start_unary_request(
1369 request, timeout, self._request_serializer
1370 )
1371 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1372 wait_for_ready
1373 )
1374 if serialized_request is None:
1375 raise rendezvous # pylint: disable-msg=raising-bad-type
1376 augmented_metadata = _compression.augment_metadata(
1377 metadata, compression
1378 )
1379 state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None)
1380 operations = (
1381 (
1382 cygrpc.SendInitialMetadataOperation(
1383 augmented_metadata, initial_metadata_flags
1384 ),
1385 cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS),
1386 cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
1387 cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
1388 ),
1389 (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),),
1390 )
1391 state.rpc_start_time = time.perf_counter()
1392 state.method = _common.decode(self._method)
1393 state.target = _common.decode(self._target)
1394 call = self._managed_call(
1395 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1396 self._method,
1397 None,
1398 _determine_deadline(deadline),
1399 metadata,
1400 None if credentials is None else credentials._credentials,
1401 operations,
1402 _event_handler(state, self._response_deserializer),
1403 self._context,
1404 self._registered_call_handle,
1405 )
1406 return _MultiThreadedRendezvous(
1407 state, call, self._response_deserializer, deadline
1408 )
1411class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable):
1412 _channel: cygrpc.Channel
1413 _managed_call: IntegratedCallFactory
1414 _method: bytes
1415 _target: bytes
1416 _request_serializer: Optional[SerializingFunction]
1417 _response_deserializer: Optional[DeserializingFunction]
1418 _context: Any
1419 _registered_call_handle: Optional[int]
1421 __slots__ = [
1422 "_channel",
1423 "_context",
1424 "_managed_call",
1425 "_method",
1426 "_request_serializer",
1427 "_response_deserializer",
1428 "_target",
1429 ]
1431 # pylint: disable=too-many-arguments
1432 def __init__(
1433 self,
1434 channel: cygrpc.Channel,
1435 managed_call: IntegratedCallFactory,
1436 method: bytes,
1437 target: bytes,
1438 request_serializer: Optional[SerializingFunction],
1439 response_deserializer: Optional[DeserializingFunction],
1440 _registered_call_handle: Optional[int],
1441 ):
1442 self._channel = channel
1443 self._managed_call = managed_call
1444 self._method = method
1445 self._target = target
1446 self._request_serializer = request_serializer
1447 self._response_deserializer = response_deserializer
1448 self._context = cygrpc.build_census_context()
1449 self._registered_call_handle = _registered_call_handle
1451 def _blocking(
1452 self,
1453 request_iterator: Iterator,
1454 timeout: Optional[float],
1455 metadata: Optional[MetadataType],
1456 credentials: Optional[grpc.CallCredentials],
1457 wait_for_ready: Optional[bool],
1458 compression: Optional[grpc.Compression],
1459 ) -> Tuple[_RPCState, cygrpc.SegregatedCall]:
1460 deadline = _deadline(timeout)
1461 state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None)
1462 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1463 wait_for_ready
1464 )
1465 augmented_metadata = _compression.augment_metadata(
1466 metadata, compression
1467 )
1468 state.rpc_start_time = time.perf_counter()
1469 state.method = _common.decode(self._method)
1470 state.target = _common.decode(self._target)
1471 call = self._channel.segregated_call(
1472 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1473 self._method,
1474 None,
1475 _determine_deadline(deadline),
1476 augmented_metadata,
1477 None if credentials is None else credentials._credentials,
1478 _stream_unary_invocation_operations_and_tags(
1479 augmented_metadata, initial_metadata_flags
1480 ),
1481 self._context,
1482 self._registered_call_handle,
1483 )
1484 _consume_request_iterator(
1485 request_iterator, state, call, self._request_serializer, None
1486 )
1487 while True:
1488 event = call.next_event()
1489 with state.condition:
1490 _handle_event(event, state, self._response_deserializer)
1491 state.condition.notify_all()
1492 if not state.due:
1493 break
1494 return state, call
1496 def __call__(
1497 self,
1498 request_iterator: Iterator,
1499 timeout: Optional[float] = None,
1500 metadata: Optional[MetadataType] = None,
1501 credentials: Optional[grpc.CallCredentials] = None,
1502 wait_for_ready: Optional[bool] = None,
1503 compression: Optional[grpc.Compression] = None,
1504 ) -> Any:
1505 state, call = self._blocking(
1506 request_iterator,
1507 timeout,
1508 metadata,
1509 credentials,
1510 wait_for_ready,
1511 compression,
1512 )
1513 return _end_unary_response_blocking(state, call, False, None)
1515 def with_call(
1516 self,
1517 request_iterator: Iterator,
1518 timeout: Optional[float] = None,
1519 metadata: Optional[MetadataType] = None,
1520 credentials: Optional[grpc.CallCredentials] = None,
1521 wait_for_ready: Optional[bool] = None,
1522 compression: Optional[grpc.Compression] = None,
1523 ) -> Tuple[Any, grpc.Call]:
1524 state, call = self._blocking(
1525 request_iterator,
1526 timeout,
1527 metadata,
1528 credentials,
1529 wait_for_ready,
1530 compression,
1531 )
1532 return _end_unary_response_blocking(state, call, True, None)
1534 def future(
1535 self,
1536 request_iterator: Iterator,
1537 timeout: Optional[float] = None,
1538 metadata: Optional[MetadataType] = None,
1539 credentials: Optional[grpc.CallCredentials] = None,
1540 wait_for_ready: Optional[bool] = None,
1541 compression: Optional[grpc.Compression] = None,
1542 ) -> _MultiThreadedRendezvous:
1543 deadline = _deadline(timeout)
1544 state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None)
1545 event_handler = _event_handler(state, self._response_deserializer)
1546 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1547 wait_for_ready
1548 )
1549 augmented_metadata = _compression.augment_metadata(
1550 metadata, compression
1551 )
1552 state.rpc_start_time = time.perf_counter()
1553 state.method = _common.decode(self._method)
1554 state.target = _common.decode(self._target)
1555 call = self._managed_call(
1556 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1557 self._method,
1558 None,
1559 deadline,
1560 augmented_metadata,
1561 None if credentials is None else credentials._credentials,
1562 _stream_unary_invocation_operations(
1563 metadata, initial_metadata_flags
1564 ),
1565 event_handler,
1566 self._context,
1567 self._registered_call_handle,
1568 )
1569 _consume_request_iterator(
1570 request_iterator,
1571 state,
1572 call,
1573 self._request_serializer,
1574 event_handler,
1575 )
1576 return _MultiThreadedRendezvous(
1577 state, call, self._response_deserializer, deadline
1578 )
1581class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable):
1582 _channel: cygrpc.Channel
1583 _managed_call: IntegratedCallFactory
1584 _method: bytes
1585 _target: bytes
1586 _request_serializer: Optional[SerializingFunction]
1587 _response_deserializer: Optional[DeserializingFunction]
1588 _context: Any
1589 _registered_call_handle: Optional[int]
1591 __slots__ = [
1592 "_channel",
1593 "_context",
1594 "_managed_call",
1595 "_method",
1596 "_request_serializer",
1597 "_response_deserializer",
1598 "_target",
1599 ]
1601 # pylint: disable=too-many-arguments
1602 def __init__(
1603 self,
1604 channel: cygrpc.Channel,
1605 managed_call: IntegratedCallFactory,
1606 method: bytes,
1607 target: bytes,
1608 request_serializer: Optional[SerializingFunction],
1609 response_deserializer: Optional[DeserializingFunction],
1610 _registered_call_handle: Optional[int],
1611 ):
1612 self._channel = channel
1613 self._managed_call = managed_call
1614 self._method = method
1615 self._target = target
1616 self._request_serializer = request_serializer
1617 self._response_deserializer = response_deserializer
1618 self._context = cygrpc.build_census_context()
1619 self._registered_call_handle = _registered_call_handle
1621 def __call__(
1622 self,
1623 request_iterator: Iterator,
1624 timeout: Optional[float] = None,
1625 metadata: Optional[MetadataType] = None,
1626 credentials: Optional[grpc.CallCredentials] = None,
1627 wait_for_ready: Optional[bool] = None,
1628 compression: Optional[grpc.Compression] = None,
1629 ) -> _MultiThreadedRendezvous:
1630 deadline = _deadline(timeout)
1631 state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None)
1632 initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready(
1633 wait_for_ready
1634 )
1635 augmented_metadata = _compression.augment_metadata(
1636 metadata, compression
1637 )
1638 operations = (
1639 (
1640 cygrpc.SendInitialMetadataOperation(
1641 augmented_metadata, initial_metadata_flags
1642 ),
1643 cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
1644 ),
1645 (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),),
1646 )
1647 event_handler = _event_handler(state, self._response_deserializer)
1648 state.rpc_start_time = time.perf_counter()
1649 state.method = _common.decode(self._method)
1650 state.target = _common.decode(self._target)
1651 call = self._managed_call(
1652 cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS,
1653 self._method,
1654 None,
1655 _determine_deadline(deadline),
1656 augmented_metadata,
1657 None if credentials is None else credentials._credentials,
1658 operations,
1659 event_handler,
1660 self._context,
1661 self._registered_call_handle,
1662 )
1663 _consume_request_iterator(
1664 request_iterator,
1665 state,
1666 call,
1667 self._request_serializer,
1668 event_handler,
1669 )
1670 return _MultiThreadedRendezvous(
1671 state, call, self._response_deserializer, deadline
1672 )
1675class _InitialMetadataFlags(int):
1676 """Stores immutable initial metadata flags"""
1678 def __new__(cls, value: int = _EMPTY_FLAGS):
1679 value &= cygrpc.InitialMetadataFlags.used_mask
1680 return super(_InitialMetadataFlags, cls).__new__(cls, value)
1682 def with_wait_for_ready(self, wait_for_ready: Optional[bool]) -> int:
1683 if wait_for_ready is not None:
1684 if wait_for_ready:
1685 return self.__class__(
1686 self
1687 | cygrpc.InitialMetadataFlags.wait_for_ready
1688 | cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set
1689 )
1690 if not wait_for_ready:
1691 return self.__class__(
1692 self & ~cygrpc.InitialMetadataFlags.wait_for_ready
1693 | cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set
1694 )
1695 return self
1698class _ChannelCallState:
1699 channel: cygrpc.Channel
1700 managed_calls: int
1701 threading: bool
1703 def __init__(self, channel: cygrpc.Channel):
1704 self.lock = threading.Lock()
1705 self.channel = channel
1706 self.managed_calls = 0
1707 self.threading = False
1709 def reset_postfork_child(self) -> None:
1710 self.managed_calls = 0
1712 def __del__(self):
1713 try:
1714 self.channel.close(
1715 cygrpc.StatusCode.cancelled, "Channel deallocated!"
1716 )
1717 except (TypeError, AttributeError):
1718 pass
1721def _run_channel_spin_thread(state: _ChannelCallState) -> None:
1722 def channel_spin():
1723 while True:
1724 cygrpc.block_if_fork_in_progress(state)
1725 event = state.channel.next_call_event()
1726 if event.completion_type == cygrpc.CompletionType.queue_timeout:
1727 continue
1728 call_completed = event.tag(event)
1729 if call_completed:
1730 with state.lock:
1731 state.managed_calls -= 1
1732 if state.managed_calls == 0:
1733 return
1735 channel_spin_thread = cygrpc.ForkManagedThread(target=channel_spin)
1736 channel_spin_thread.setDaemon(True)
1737 channel_spin_thread.start()
1740def _channel_managed_call_management(state: _ChannelCallState):
1741 # pylint: disable=too-many-arguments
1742 def create(
1743 flags: int,
1744 method: bytes,
1745 host: Optional[str],
1746 deadline: Optional[float],
1747 metadata: Optional[MetadataType],
1748 credentials: Optional[cygrpc.CallCredentials],
1749 operations: Sequence[Sequence[cygrpc.Operation]],
1750 event_handler: UserTag,
1751 context: Any,
1752 _registered_call_handle: Optional[int],
1753 ) -> cygrpc.IntegratedCall:
1754 """Creates a cygrpc.IntegratedCall.
1756 Args:
1757 flags: An integer bitfield of call flags.
1758 method: The RPC method.
1759 host: A host string for the created call.
1760 deadline: A float to be the deadline of the created call or None if
1761 the call is to have an infinite deadline.
1762 metadata: The metadata for the call or None.
1763 credentials: A cygrpc.CallCredentials or None.
1764 operations: A sequence of sequences of cygrpc.Operations to be
1765 started on the call.
1766 event_handler: A behavior to call to handle the events resultant from
1767 the operations on the call.
1768 context: Context object for distributed tracing.
1769 _registered_call_handle: An int representing the call handle of the
1770 method, or None if the method is not registered.
1772 Returns:
1773 A cygrpc.IntegratedCall with which to conduct an RPC.
1774 """
1775 operations_and_tags = tuple(
1776 (
1777 operation,
1778 event_handler,
1779 )
1780 for operation in operations
1781 )
1782 with state.lock:
1783 call = state.channel.integrated_call(
1784 flags,
1785 method,
1786 host,
1787 deadline,
1788 metadata,
1789 credentials,
1790 operations_and_tags,
1791 context,
1792 _registered_call_handle,
1793 )
1794 if state.managed_calls == 0:
1795 state.managed_calls = 1
1796 _run_channel_spin_thread(state)
1797 else:
1798 state.managed_calls += 1
1799 return call
1801 return create
1804class _ChannelConnectivityState:
1805 lock: threading.RLock
1806 channel: cygrpc.Channel
1807 polling: bool
1808 connectivity: grpc.ChannelConnectivity
1809 try_to_connect: bool
1810 # TODO(xuanwn): Refactor this: https://github.com/grpc/grpc/issues/31704
1811 callbacks_and_connectivities: List[
1812 Sequence[
1813 Union[
1814 Callable[[grpc.ChannelConnectivity], None],
1815 Optional[grpc.ChannelConnectivity],
1816 ]
1817 ]
1818 ]
1819 delivering: bool
1821 def __init__(self, channel: cygrpc.Channel):
1822 self.lock = threading.RLock()
1823 self.channel = channel
1824 self.polling = False
1825 self.connectivity = None
1826 self.try_to_connect = False
1827 self.callbacks_and_connectivities = []
1828 self.delivering = False
1830 def reset_postfork_child(self) -> None:
1831 self.polling = False
1832 self.connectivity = None
1833 self.delivering = False
1836def _deliveries(
1837 state: _ChannelConnectivityState,
1838) -> List[Callable[[grpc.ChannelConnectivity], None]]:
1839 callbacks_needing_update = []
1840 for callback_and_connectivity in state.callbacks_and_connectivities:
1841 callback, callback_connectivity = callback_and_connectivity
1842 if callback_connectivity is not state.connectivity:
1843 callbacks_needing_update.append(callback)
1844 callback_and_connectivity[1] = state.connectivity
1845 return callbacks_needing_update
1848def _deliver(
1849 state: _ChannelConnectivityState,
1850 initial_connectivity: grpc.ChannelConnectivity,
1851 initial_callbacks: Sequence[Callable[[grpc.ChannelConnectivity], None]],
1852) -> None:
1853 connectivity = initial_connectivity
1854 callbacks = initial_callbacks
1855 while True:
1856 for callback in callbacks:
1857 try:
1858 callback(connectivity)
1859 except Exception: # pylint: disable=broad-except
1860 _LOGGER.exception(
1861 _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE
1862 )
1863 cygrpc.block_if_fork_in_progress(state)
1864 with state.lock:
1865 callbacks = _deliveries(state)
1866 if callbacks:
1867 connectivity = state.connectivity
1868 else:
1869 state.delivering = False
1870 return
1873def _spawn_delivery(
1874 state: _ChannelConnectivityState,
1875 callbacks: Sequence[Callable[[grpc.ChannelConnectivity], None]],
1876) -> None:
1877 """Spawn a thread running the _deliver function.
1879 Should only be called while holding state.lock.
1880 """
1881 delivering_thread = cygrpc.ForkManagedThread(
1882 target=_deliver,
1883 args=(
1884 state,
1885 state.connectivity,
1886 callbacks,
1887 ),
1888 )
1889 delivering_thread.setDaemon(True)
1890 delivering_thread.start()
1891 state.delivering = True
1894# NOTE(https://github.com/grpc/grpc/issues/3064): We'd rather not poll.
1895def _poll_connectivity(
1896 state: _ChannelConnectivityState,
1897 channel: grpc.Channel,
1898 initial_try_to_connect: bool,
1899) -> None:
1900 try_to_connect = initial_try_to_connect
1901 connectivity = channel.check_connectivity_state(try_to_connect)
1902 with state.lock:
1903 state.connectivity = (
1904 _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[
1905 connectivity
1906 ]
1907 )
1908 callbacks = tuple(
1909 callback for callback, _ in state.callbacks_and_connectivities
1910 )
1911 for callback_and_connectivity in state.callbacks_and_connectivities:
1912 callback_and_connectivity[1] = state.connectivity
1913 if callbacks:
1914 _spawn_delivery(state, callbacks)
1915 while True:
1916 event = channel.watch_connectivity_state(
1917 connectivity, time.time() + 0.2
1918 )
1919 cygrpc.block_if_fork_in_progress(state)
1920 with state.lock:
1921 if (
1922 not state.callbacks_and_connectivities
1923 and not state.try_to_connect
1924 ):
1925 state.polling = False
1926 state.connectivity = None
1927 break
1928 try_to_connect = state.try_to_connect
1929 state.try_to_connect = False
1930 if event.success or try_to_connect:
1931 connectivity = channel.check_connectivity_state(try_to_connect)
1932 with state.lock:
1933 state.connectivity = (
1934 _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[
1935 connectivity
1936 ]
1937 )
1938 if not state.delivering:
1939 callbacks = _deliveries(state)
1940 if callbacks:
1941 _spawn_delivery(state, callbacks)
1944def _spawn_poll_connectivity(
1945 state: _ChannelConnectivityState, try_to_connect: bool
1946) -> None:
1947 """Spawn a thread running the _poll_connectivity function.
1949 Should only be called while holding state.lock.
1950 """
1951 polling_thread = cygrpc.ForkManagedThread(
1952 target=_poll_connectivity,
1953 args=(state, state.channel, bool(try_to_connect)),
1954 )
1955 polling_thread.setDaemon(True)
1956 polling_thread.start()
1957 state.polling = True
1960def _subscribe(
1961 state: _ChannelConnectivityState,
1962 callback: Callable[[grpc.ChannelConnectivity], None],
1963 try_to_connect: bool,
1964) -> None:
1965 with state.lock:
1966 if not state.callbacks_and_connectivities and not state.polling:
1967 _spawn_poll_connectivity(state, try_to_connect)
1968 state.callbacks_and_connectivities.append([callback, None])
1969 elif not state.delivering and state.connectivity is not None:
1970 _spawn_delivery(state, (callback,))
1971 state.try_to_connect |= bool(try_to_connect)
1972 state.callbacks_and_connectivities.append(
1973 [callback, state.connectivity]
1974 )
1975 else:
1976 state.try_to_connect |= bool(try_to_connect)
1977 state.callbacks_and_connectivities.append([callback, None])
1980def _unsubscribe(
1981 state: _ChannelConnectivityState,
1982 callback: Callable[[grpc.ChannelConnectivity], None],
1983) -> None:
1984 with state.lock:
1985 for index, (subscribed_callback, _unused_connectivity) in enumerate(
1986 state.callbacks_and_connectivities
1987 ):
1988 if callback == subscribed_callback:
1989 state.callbacks_and_connectivities.pop(index)
1990 break
1993def _augment_options(
1994 base_options: Sequence[ChannelArgumentType],
1995 compression: Optional[grpc.Compression],
1996) -> Sequence[ChannelArgumentType]:
1997 compression_option = _compression.create_channel_option(compression)
1998 return (
1999 tuple(base_options)
2000 + compression_option
2001 + (
2002 (
2003 cygrpc.ChannelArgKey.primary_user_agent_string.decode(),
2004 _USER_AGENT,
2005 ),
2006 )
2007 )
2010def _separate_channel_options(
2011 options: Sequence[ChannelArgumentType],
2012) -> Tuple[Sequence[ChannelArgumentType], Sequence[ChannelArgumentType]]:
2013 """Separates core channel options from Python channel options."""
2014 core_options = []
2015 python_options = []
2016 for pair in options:
2017 if (
2018 pair[0]
2019 == grpc.experimental.ChannelOptions.SingleThreadedUnaryStream
2020 ):
2021 python_options.append(pair)
2022 else:
2023 core_options.append(pair)
2024 return python_options, core_options
2027def _maybe_spawn_poll_connectivity_postfork(
2028 state: _ChannelConnectivityState,
2029) -> None:
2030 with state.lock:
2031 if state.callbacks_and_connectivities and not state.polling:
2032 _spawn_poll_connectivity(state, state.try_to_connect)
2035class Channel(grpc.Channel):
2036 """A cygrpc.Channel-backed implementation of grpc.Channel."""
2038 _single_threaded_unary_stream: bool
2039 _channel: cygrpc.Channel
2040 _call_state: _ChannelCallState
2041 _connectivity_state: _ChannelConnectivityState
2042 _target: str
2043 _registered_call_handles: Dict[str, int]
2045 def __init__(
2046 self,
2047 target: str,
2048 options: Sequence[ChannelArgumentType],
2049 credentials: Optional[grpc.ChannelCredentials],
2050 compression: Optional[grpc.Compression],
2051 ):
2052 """Constructor.
2054 Args:
2055 target: The target to which to connect.
2056 options: Configuration options for the channel.
2057 credentials: A cygrpc.ChannelCredentials or None.
2058 compression: An optional value indicating the compression method to be
2059 used over the lifetime of the channel.
2060 """
2061 python_options, core_options = _separate_channel_options(options)
2062 self._single_threaded_unary_stream = (
2063 _DEFAULT_SINGLE_THREADED_UNARY_STREAM
2064 )
2065 self._process_python_options(python_options)
2066 self._channel = cygrpc.Channel(
2067 _common.encode(target),
2068 _augment_options(core_options, compression),
2069 credentials,
2070 )
2071 self._target = target
2072 self._call_state = _ChannelCallState(self._channel)
2073 self._connectivity_state = _ChannelConnectivityState(self._channel)
2074 cygrpc.fork_register_channel(self)
2075 if cygrpc.g_gevent_activated:
2076 cygrpc.gevent_increment_channel_count()
2078 def _get_registered_call_handle(self, method: str) -> int:
2079 """
2080 Get the registered call handle for a method.
2082 This is a semi-private method. It is intended for use only by gRPC generated code.
2084 This method is not thread-safe.
2086 Args:
2087 method: Required, the method name for the RPC.
2089 Returns:
2090 The registered call handle pointer in the form of a Python Long.
2091 """
2092 return self._channel.get_registered_call_handle(_common.encode(method))
2094 def _process_python_options(
2095 self, python_options: Sequence[ChannelArgumentType]
2096 ) -> None:
2097 """Sets channel attributes according to python-only channel options."""
2098 for pair in python_options:
2099 if (
2100 pair[0]
2101 == grpc.experimental.ChannelOptions.SingleThreadedUnaryStream
2102 ):
2103 self._single_threaded_unary_stream = True
2105 def subscribe(
2106 self,
2107 callback: Callable[[grpc.ChannelConnectivity], None],
2108 try_to_connect: Optional[bool] = None,
2109 ) -> None:
2110 _subscribe(self._connectivity_state, callback, try_to_connect)
2112 def unsubscribe(
2113 self, callback: Callable[[grpc.ChannelConnectivity], None]
2114 ) -> None:
2115 _unsubscribe(self._connectivity_state, callback)
2117 # pylint: disable=arguments-differ
2118 def unary_unary(
2119 self,
2120 method: str,
2121 request_serializer: Optional[SerializingFunction] = None,
2122 response_deserializer: Optional[DeserializingFunction] = None,
2123 _registered_method: Optional[bool] = False,
2124 ) -> grpc.UnaryUnaryMultiCallable:
2125 _registered_call_handle = None
2126 if _registered_method:
2127 _registered_call_handle = self._get_registered_call_handle(method)
2128 return _UnaryUnaryMultiCallable(
2129 self._channel,
2130 _channel_managed_call_management(self._call_state),
2131 _common.encode(method),
2132 _common.encode(self._target),
2133 request_serializer,
2134 response_deserializer,
2135 _registered_call_handle,
2136 )
2138 # pylint: disable=arguments-differ
2139 def unary_stream(
2140 self,
2141 method: str,
2142 request_serializer: Optional[SerializingFunction] = None,
2143 response_deserializer: Optional[DeserializingFunction] = None,
2144 _registered_method: Optional[bool] = False,
2145 ) -> grpc.UnaryStreamMultiCallable:
2146 _registered_call_handle = None
2147 if _registered_method:
2148 _registered_call_handle = self._get_registered_call_handle(method)
2149 # NOTE(rbellevi): Benchmarks have shown that running a unary-stream RPC
2150 # on a single Python thread results in an appreciable speed-up. However,
2151 # due to slight differences in capability, the multi-threaded variant
2152 # remains the default.
2153 if self._single_threaded_unary_stream:
2154 return _SingleThreadedUnaryStreamMultiCallable(
2155 self._channel,
2156 _common.encode(method),
2157 _common.encode(self._target),
2158 request_serializer,
2159 response_deserializer,
2160 _registered_call_handle,
2161 )
2162 return _UnaryStreamMultiCallable(
2163 self._channel,
2164 _channel_managed_call_management(self._call_state),
2165 _common.encode(method),
2166 _common.encode(self._target),
2167 request_serializer,
2168 response_deserializer,
2169 _registered_call_handle,
2170 )
2172 # pylint: disable=arguments-differ
2173 def stream_unary(
2174 self,
2175 method: str,
2176 request_serializer: Optional[SerializingFunction] = None,
2177 response_deserializer: Optional[DeserializingFunction] = None,
2178 _registered_method: Optional[bool] = False,
2179 ) -> grpc.StreamUnaryMultiCallable:
2180 _registered_call_handle = None
2181 if _registered_method:
2182 _registered_call_handle = self._get_registered_call_handle(method)
2183 return _StreamUnaryMultiCallable(
2184 self._channel,
2185 _channel_managed_call_management(self._call_state),
2186 _common.encode(method),
2187 _common.encode(self._target),
2188 request_serializer,
2189 response_deserializer,
2190 _registered_call_handle,
2191 )
2193 # pylint: disable=arguments-differ
2194 def stream_stream(
2195 self,
2196 method: str,
2197 request_serializer: Optional[SerializingFunction] = None,
2198 response_deserializer: Optional[DeserializingFunction] = None,
2199 _registered_method: Optional[bool] = False,
2200 ) -> grpc.StreamStreamMultiCallable:
2201 _registered_call_handle = None
2202 if _registered_method:
2203 _registered_call_handle = self._get_registered_call_handle(method)
2204 return _StreamStreamMultiCallable(
2205 self._channel,
2206 _channel_managed_call_management(self._call_state),
2207 _common.encode(method),
2208 _common.encode(self._target),
2209 request_serializer,
2210 response_deserializer,
2211 _registered_call_handle,
2212 )
2214 def _unsubscribe_all(self) -> None:
2215 state = self._connectivity_state
2216 if state:
2217 with state.lock:
2218 del state.callbacks_and_connectivities[:]
2220 def _close(self) -> None:
2221 self._unsubscribe_all()
2222 self._channel.close(cygrpc.StatusCode.cancelled, "Channel closed!")
2223 cygrpc.fork_unregister_channel(self)
2224 if cygrpc.g_gevent_activated:
2225 cygrpc.gevent_decrement_channel_count()
2227 def _postfork_child(self) -> None:
2228 self._channel.cancel_calls_on_fork(
2229 cygrpc.StatusCode.cancelled, "Call cancelled in fork child"
2230 )
2231 _maybe_spawn_poll_connectivity_postfork(self._connectivity_state)
2233 def __enter__(self):
2234 return self
2236 def __exit__(self, exc_type, exc_val, exc_tb):
2237 self._close()
2238 return False
2240 def close(self) -> None:
2241 self._close()
2243 def __del__(self):
2244 # TODO(https://github.com/grpc/grpc/issues/12531): Several releases
2245 # after 1.12 (1.16 or thereabouts?) add a "self._channel.close" call
2246 # here (or more likely, call self._close() here). We don't do this today
2247 # because many valid use cases today allow the channel to be deleted
2248 # immediately after stubs are created. After a sufficient period of time
2249 # has passed for all users to be trusted to freeze out to their channels
2250 # for as long as they are in use and to close them after using them,
2251 # then deletion of this grpc._channel.Channel instance can be made to
2252 # effect closure of the underlying cygrpc.Channel instance.
2253 try:
2254 self._unsubscribe_all()
2255 except: # pylint: disable=bare-except # noqa: E722
2256 # Exceptions in __del__ are ignored by Python anyway, but they can
2257 # keep spamming logs. Just silence them.
2258 pass