cengal.parallel_execution.coroutines.coro_standard_services.throw_coro.versions.v_0.throw_coro
Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18""" 19Module Docstring 20Docstrings: http://www.python.org/dev/peps/pep-0257/ 21""" 22 23__author__ = "ButenkoMS <gtalk@butenkoms.space>" 24__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 25__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 26__license__ = "Apache License, Version 2.0" 27__version__ = "4.4.1" 28__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 29__email__ = "gtalk@butenkoms.space" 30# __status__ = "Prototype" 31__status__ = "Development" 32# __status__ = "Production" 33 34 35__all__ = ['ThrowCoro', 'throw_coro_on', 'try_throw_coro_on', 'athrow_coro_on', 'atry_throw_coro_on', 'throw_coro', 'try_throw_coro', 'athrow_coro', 'atry_throw_coro'] 36 37 38from cengal.parallel_execution.coroutines.coro_scheduler import * 39from cengal.parallel_execution.coroutines.coro_standard_services_internal_lib.service_with_a_direct_request import * 40from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import PutCoro 41from cengal.code_flow_control.smart_values import ValueExistence 42from cengal.introspection.inspect import get_exception, get_exception_tripple 43from typing import Any, Optional, Tuple, Dict, Set, Union, Type, List 44 45 46class ThrowCoro(TypedService[bool], ServiceWithADirectRequestMixin): 47 def __init__(self, loop: CoroSchedulerType): 48 super(ThrowCoro, self).__init__(loop) 49 self.direct_requests: List[Tuple] = list() 50 51 def single_task_registration_or_immediate_processing( 52 self, coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> Tuple[bool, bool, Optional[BaseException]]: 53 exception = None 54 result = None 55 try: 56 if tree: 57 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 58 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 59 60 result = self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 61 if tree: 62 for child_coro_id in children: 63 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 64 except: 65 exception = get_exception() 66 67 return True, result, exception 68 69 def full_processing_iteration(self): 70 direct_requests_buff = self.direct_requests 71 self.direct_requests = type(direct_requests_buff)() 72 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 73 for coro_id, ex_type, ex_value, ex_traceback, tree in direct_requests_buff: 74 try: 75 if tree: 76 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 77 78 self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 79 if tree: 80 for child_coro_id in children: 81 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 82 except: 83 ex_type, exception, tracback = get_exception_tripple() 84 if __debug__: dlog(ex_type, exception, tracback) 85 raise 86 87 self.make_dead() 88 89 def _add_direct_request(self, coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[None]: 90 self.direct_requests.append((coro_id, ex_type, ex_value, ex_traceback, tree)) 91 self.make_live() 92 return (False, None) 93 94 def in_work(self) -> bool: 95 result: bool = bool(self.direct_requests) 96 return self.thrifty_in_work(result) 97 98 99def throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 100 return make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 101 102 103def try_throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 104 return try_make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 105 106 107async def athrow_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 108 return await amake_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 109 110 111async def atry_throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 112 return await atry_make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 113 114 115def throw_coro(coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 116 return make_request_to_service(ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 117 118 119def try_throw_coro(coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 120 return try_make_request_to_service(ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 121 122 123async def athrow_coro(coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 124 return await amake_request_to_service(ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree) 125 126 127async def atry_throw_coro(coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 128 return await atry_make_request_to_service(ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree)
47class ThrowCoro(TypedService[bool], ServiceWithADirectRequestMixin): 48 def __init__(self, loop: CoroSchedulerType): 49 super(ThrowCoro, self).__init__(loop) 50 self.direct_requests: List[Tuple] = list() 51 52 def single_task_registration_or_immediate_processing( 53 self, coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> Tuple[bool, bool, Optional[BaseException]]: 54 exception = None 55 result = None 56 try: 57 if tree: 58 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 59 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 60 61 result = self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 62 if tree: 63 for child_coro_id in children: 64 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 65 except: 66 exception = get_exception() 67 68 return True, result, exception 69 70 def full_processing_iteration(self): 71 direct_requests_buff = self.direct_requests 72 self.direct_requests = type(direct_requests_buff)() 73 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 74 for coro_id, ex_type, ex_value, ex_traceback, tree in direct_requests_buff: 75 try: 76 if tree: 77 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 78 79 self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 80 if tree: 81 for child_coro_id in children: 82 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 83 except: 84 ex_type, exception, tracback = get_exception_tripple() 85 if __debug__: dlog(ex_type, exception, tracback) 86 raise 87 88 self.make_dead() 89 90 def _add_direct_request(self, coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[None]: 91 self.direct_requests.append((coro_id, ex_type, ex_value, ex_traceback, tree)) 92 self.make_live() 93 return (False, None) 94 95 def in_work(self) -> bool: 96 result: bool = bool(self.direct_requests) 97 return self.thrifty_in_work(result)
Abstract base class for generic types.
A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::
class Mapping(Generic[KT, VT]): def __getitem__(self, key: KT) -> VT: ... # Etc.
This class can then be used as follows::
def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default
52 def single_task_registration_or_immediate_processing( 53 self, coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> Tuple[bool, bool, Optional[BaseException]]: 54 exception = None 55 result = None 56 try: 57 if tree: 58 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 59 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 60 61 result = self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 62 if tree: 63 for child_coro_id in children: 64 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 65 except: 66 exception = get_exception() 67 68 return True, result, exception
70 def full_processing_iteration(self): 71 direct_requests_buff = self.direct_requests 72 self.direct_requests = type(direct_requests_buff)() 73 put_coro: PutCoro = self._loop.get_service_instance(PutCoro) 74 for coro_id, ex_type, ex_value, ex_traceback, tree in direct_requests_buff: 75 try: 76 if tree: 77 children: Set[CoroID] = put_coro.get_set_of_all_children(coro_id) 78 79 self._loop.throw_coro_by_id(coro_id, ex_type, ex_value, ex_traceback) 80 if tree: 81 for child_coro_id in children: 82 self._loop.throw_coro_by_id(child_coro_id, ex_type, ex_value, ex_traceback) 83 except: 84 ex_type, exception, tracback = get_exception_tripple() 85 if __debug__: dlog(ex_type, exception, tracback) 86 raise 87 88 self.make_dead()
95 def in_work(self) -> bool: 96 result: bool = bool(self.direct_requests) 97 return self.thrifty_in_work(result)
Will be executed twice per iteration: once before and once after the full_processing_iteration() execution
Raises: NotImplementedError: _description_
Returns: bool: _description_
Inherited Members
- cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service
- current_caller_coro_info
- iteration
- make_response
- register_response
- put_task
- resolve_request
- try_resolve_request
- in_forground_work
- thrifty_in_work
- time_left_before_next_event
- is_low_latency
- make_live
- make_dead
- service_id_impl
- service_id
- destroy
100def throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 101 return make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree)
104def try_throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 105 return try_make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree)
108async def athrow_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[bool]: 109 return await amake_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree)
112async def atry_throw_coro_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], coro_id: CoroID, ex_type: Type[Exception], ex_value: Exception = None, ex_traceback: Any = None, tree: bool = False) -> ValueExistence[Optional[bool]]: 113 return await atry_make_request_to_service_with_context(context, ThrowCoro, coro_id, ex_type, ex_value, ex_traceback, tree)