cengal.parallel_execution.coroutines.coro_standard_services.timer_coro_runner.versions.v_0.timer_coro_runner

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

  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
 19"""
 20Module Docstring
 21Docstrings: http://www.python.org/dev/peps/pep-0257/
 22"""
 23
 24
 25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 28__license__ = "Apache License, Version 2.0"
 29__version__ = "4.4.1"
 30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 31__email__ = "gtalk@butenkoms.space"
 32# __status__ = "Prototype"
 33__status__ = "Development"
 34# __status__ = "Production"
 35
 36
 37__all__ = ['TimerCoroRunner', 'add_timer_coro_run_from_other_service', 'discard_timer_coro_run_from_other_service', 'timer_coro_run_on', 'try_timer_coro_run_on', 'atimer_coro_run_on', 'atry_timer_coro_run_on', 'timer_coro_run', 'try_timer_coro_run', 'atimer_coro_run', 'atry_timer_coro_run']
 38
 39from cengal.parallel_execution.coroutines.coro_scheduler import *
 40from cengal.parallel_execution.coroutines.coro_standard_services_internal_lib.service_with_a_direct_request import *
 41from cengal.code_flow_control.smart_values import ValueExistence
 42from cengal.time_management.timer import Timer, TimerRequest
 43from functools import partial
 44from typing import Tuple, Optional, Union
 45from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import put_current_from_other_service, put_root_from_other_service, put_from_other_service
 46
 47
 48class TimerCoroRunnerRequest(ServiceRequest):
 49    def add(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
 50        return self._save(0, delay, coro_worker, args, kwargs)
 51    def discard(self, timer_request: TimerRequest) -> bool:
 52        return self._save(1, timer_request)
 53
 54
 55class TimerCoroRunner(DualImmediateProcessingServiceMixin, ServiceWithADirectRequestMixin, TypedService[TimerRequest]):
 56    def __init__(self, loop: CoroSchedulerType):
 57        super(TimerCoroRunner, self).__init__(loop)
 58        self.timer = Timer()
 59        self.pending_tasks_number = 0
 60        self.direct_requests = list()
 61        self._request_workers = {
 62            0: self._on_add,
 63            1: self._on_discard,
 64        }
 65
 66    def single_task_registration_or_immediate_processing_single(
 67            self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> Tuple[bool, None]:
 68        timer_request: TimerRequest = self._add_request_impl(delay, coro_worker, *args, **kwargs)
 69        self.make_live()
 70        return True, timer_request, None
 71    
 72    def _add_request_impl(self, delay: float, coro_worker: AnyWorker, *args, **kwargs):
 73        def timer_handler_func(caller_coro_id, coro_worker_: AnyWorker, *args_, **kwargs_):
 74            put_from_other_service(self, caller_coro_id, coro_worker_, *args_, **kwargs_)
 75            self.task_triggered()
 76
 77        timer_handler = partial(timer_handler_func, self.current_caller_coro_info.coro_id, coro_worker, *args, **kwargs)
 78        self.task_added()
 79        return self.timer.register(timer_handler, delay)
 80    
 81    def _add_request_external_impl(self, delay: float, coro_worker: AnyWorker, *args, **kwargs):
 82        def timer_handler_func(coro_worker_: AnyWorker, *args_, **kwargs_):
 83            try:
 84                put_root_from_other_service(self, coro_worker_, *args_, **kwargs_)
 85            finally:
 86                self.task_triggered()
 87
 88        timer_handler = partial(timer_handler_func, coro_worker, *args, **kwargs)
 89        self.task_added()
 90        return self.timer.register(timer_handler, delay)
 91    
 92    def _on_add(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 93        timer_request: TimerRequest = self._add_request_impl(delay, coro_worker, *args, **kwargs)
 94        self.make_live()
 95        return True, timer_request, None
 96    
 97    def _on_discard(self, timer_request: TimerRequest) -> Tuple[bool, TimerRequest, None]:
 98        result: bool = self.timer.discard(timer_request)
 99        if result:
100            self.task_triggered()
101        
102        return True, result, None
103    
104    def add_timer_coro_run_from_other_service(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
105        timer_request: TimerRequest = self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
106        self.make_live()
107        return timer_request
108    
109    def discard_timer_coro_run_from_other_service(self, timer_request: TimerRequest):
110        result: bool = self.timer.discard(timer_request)
111        if result:
112            self.task_triggered()
113        
114        return result
115
116    def full_processing_iteration(self):
117        if self.direct_requests:
118            direct_requests_buff = self.direct_requests
119            self.direct_requests = type(direct_requests_buff)()
120            for delay, coro_worker, args, kwargs in direct_requests_buff:
121                self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
122
123        self.timer()
124        if 0 == self.pending_tasks_number:
125            self.make_dead()
126    
127    def _add_direct_request(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[None]:
128        self.direct_requests.append((delay, coro_worker, args, kwargs))
129        self.make_live()
130        return (False, None)
131
132    def task_added(self):
133        self.pending_tasks_number += 1
134
135    def task_triggered(self):
136        self.pending_tasks_number -= 1
137
138    def in_work(self) -> bool:
139        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
140        return self.thrifty_in_work(result)
141    
142    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
143        return True, self.timer.nearest_event()
144
145
146TimerCoroRunnerRequest.default_service_type = TimerCoroRunner
147
148
149def add_timer_coro_run_from_other_service(current_service: Service, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
150    timer_coro_runner: TimerCoroRunner = current_service._loop.get_service_instance(TimerCoroRunner)
151    return timer_coro_runner.add_timer_coro_run_from_other_service(delay, coro_worker, *args, **kwargs)
152
153
154def discard_timer_coro_run_from_other_service(current_service: Service, timer_request: TimerRequest) -> bool:
155    timer_coro_runner: TimerCoroRunner = current_service._loop.get_service_instance(TimerCoroRunner)
156    return timer_coro_runner.discard_timer_coro_run_from_other_service(timer_request)
157
158
159def timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
160    """_summary_
161        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
162        
163        An example:
164
165        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
166        from cengal.parallel_execution.coroutines.coro_standard_services.timer_coro_runner import timer_coro_run_on
167        from typing import Optional, Union
168
169        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
170            try:
171                def print_hello_world(i: Interface, name: str):
172                    print(f'Hello Wrold from {name}!)
173                
174                timer_coro_run_on(10, print_hello_world, 'John Doe')
175            except CoroSchedulerContextIsNotAvailable:
176                print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)
177        
178    Args:
179        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
180        delay (float): delay in seconds
181        coro_worker (AnyWorker): coro_worker
182
183    Returns:
184        ValueExistence[Optional[CoroID]]: _description_
185    """
186    return make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)
187
188
189def try_timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
190    """_summary_
191        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
192        
193        An example:
194
195        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
196        from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
197        from typing import Optional, Union
198
199        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
200            def print_hello_world(i: Interface, name: str):
201                print(f'Hello Wrold from {name}!)
202            
203            try_timer_coro_run_on(10, print_hello_world, 'John Doe')
204        
205    Args:
206        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
207        delay (float): delay in seconds
208        coro_worker (AnyWorker): coro_worker
209
210    Returns:
211        ValueExistence[Optional[CoroID]]: _description_
212    """
213    return try_make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)
214
215
216async def atimer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
217    return await amake_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)
218
219
220async def atry_timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
221    return await atry_make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)
222
223
224def timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
225    return make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)
226
227
228def try_timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
229    return try_make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)
230
231
232async def atimer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
233    return await amake_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)
234
235
236async def atry_timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
237    return await atry_make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)

class TimerCoroRunner(cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.DualImmediateProcessingServiceMixin, cengal.parallel_execution.coroutines.coro_standard_services_internal_lib.service_with_a_direct_request.versions.v_0.service_with_a_direct_request.ServiceWithADirectRequestMixin, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.TypedService[cengal.time_management.timer.versions.v_0.timer.TimerRequest]): View Source

 56class TimerCoroRunner(DualImmediateProcessingServiceMixin, ServiceWithADirectRequestMixin, TypedService[TimerRequest]):
 57    def __init__(self, loop: CoroSchedulerType):
 58        super(TimerCoroRunner, self).__init__(loop)
 59        self.timer = Timer()
 60        self.pending_tasks_number = 0
 61        self.direct_requests = list()
 62        self._request_workers = {
 63            0: self._on_add,
 64            1: self._on_discard,
 65        }
 66
 67    def single_task_registration_or_immediate_processing_single(
 68            self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> Tuple[bool, None]:
 69        timer_request: TimerRequest = self._add_request_impl(delay, coro_worker, *args, **kwargs)
 70        self.make_live()
 71        return True, timer_request, None
 72    
 73    def _add_request_impl(self, delay: float, coro_worker: AnyWorker, *args, **kwargs):
 74        def timer_handler_func(caller_coro_id, coro_worker_: AnyWorker, *args_, **kwargs_):
 75            put_from_other_service(self, caller_coro_id, coro_worker_, *args_, **kwargs_)
 76            self.task_triggered()
 77
 78        timer_handler = partial(timer_handler_func, self.current_caller_coro_info.coro_id, coro_worker, *args, **kwargs)
 79        self.task_added()
 80        return self.timer.register(timer_handler, delay)
 81    
 82    def _add_request_external_impl(self, delay: float, coro_worker: AnyWorker, *args, **kwargs):
 83        def timer_handler_func(coro_worker_: AnyWorker, *args_, **kwargs_):
 84            try:
 85                put_root_from_other_service(self, coro_worker_, *args_, **kwargs_)
 86            finally:
 87                self.task_triggered()
 88
 89        timer_handler = partial(timer_handler_func, coro_worker, *args, **kwargs)
 90        self.task_added()
 91        return self.timer.register(timer_handler, delay)
 92    
 93    def _on_add(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 94        timer_request: TimerRequest = self._add_request_impl(delay, coro_worker, *args, **kwargs)
 95        self.make_live()
 96        return True, timer_request, None
 97    
 98    def _on_discard(self, timer_request: TimerRequest) -> Tuple[bool, TimerRequest, None]:
 99        result: bool = self.timer.discard(timer_request)
100        if result:
101            self.task_triggered()
102        
103        return True, result, None
104    
105    def add_timer_coro_run_from_other_service(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
106        timer_request: TimerRequest = self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
107        self.make_live()
108        return timer_request
109    
110    def discard_timer_coro_run_from_other_service(self, timer_request: TimerRequest):
111        result: bool = self.timer.discard(timer_request)
112        if result:
113            self.task_triggered()
114        
115        return result
116
117    def full_processing_iteration(self):
118        if self.direct_requests:
119            direct_requests_buff = self.direct_requests
120            self.direct_requests = type(direct_requests_buff)()
121            for delay, coro_worker, args, kwargs in direct_requests_buff:
122                self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
123
124        self.timer()
125        if 0 == self.pending_tasks_number:
126            self.make_dead()
127    
128    def _add_direct_request(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[None]:
129        self.direct_requests.append((delay, coro_worker, args, kwargs))
130        self.make_live()
131        return (False, None)
132
133    def task_added(self):
134        self.pending_tasks_number += 1
135
136    def task_triggered(self):
137        self.pending_tasks_number -= 1
138
139    def in_work(self) -> bool:
140        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
141        return self.thrifty_in_work(result)
142    
143    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
144        return True, self.timer.nearest_event()

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

TimerCoroRunner( loop: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable]) View Source

57    def __init__(self, loop: CoroSchedulerType):
58        super(TimerCoroRunner, self).__init__(loop)
59        self.timer = Timer()
60        self.pending_tasks_number = 0
61        self.direct_requests = list()
62        self._request_workers = {
63            0: self._on_add,
64            1: self._on_discard,
65        }

timer

pending_tasks_number

direct_requests

def add_timer_coro_run_from_other_service( self, delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.time_management.timer.versions.v_0.timer.TimerRequest: View Source

105    def add_timer_coro_run_from_other_service(self, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
106        timer_request: TimerRequest = self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
107        self.make_live()
108        return timer_request

def discard_timer_coro_run_from_other_service( self, timer_request: cengal.time_management.timer.versions.v_0.timer.TimerRequest): View Source

110    def discard_timer_coro_run_from_other_service(self, timer_request: TimerRequest):
111        result: bool = self.timer.discard(timer_request)
112        if result:
113            self.task_triggered()
114        
115        return result

def full_processing_iteration(self): View Source

117    def full_processing_iteration(self):
118        if self.direct_requests:
119            direct_requests_buff = self.direct_requests
120            self.direct_requests = type(direct_requests_buff)()
121            for delay, coro_worker, args, kwargs in direct_requests_buff:
122                self._add_request_external_impl(delay, coro_worker, *args, **kwargs)
123
124        self.timer()
125        if 0 == self.pending_tasks_number:
126            self.make_dead()

def task_added(self): View Source

133    def task_added(self):
134        self.pending_tasks_number += 1

def task_triggered(self): View Source

136    def task_triggered(self):
137        self.pending_tasks_number -= 1

def in_work(self) -> bool: View Source

139    def in_work(self) -> bool:
140        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
141        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_

def time_left_before_next_event(self) -> Tuple[bool, Union[int, float, NoneType]]: View Source

143    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
144        return True, self.timer.nearest_event()

Inherited Members

cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.DualImmediateProcessingServiceMixin: single_task_registration_or_immediate_processing; single_task_registration_or_immediate_processing_multiple; single_task_registration_or_immediate_processing_single
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; is_low_latency; make_live; make_dead; service_id_impl; service_id; destroy

def add_timer_coro_run_from_other_service( current_service: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service, delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.time_management.timer.versions.v_0.timer.TimerRequest: View Source

150def add_timer_coro_run_from_other_service(current_service: Service, delay: float, coro_worker: AnyWorker, *args, **kwargs) -> TimerRequest:
151    timer_coro_runner: TimerCoroRunner = current_service._loop.get_service_instance(TimerCoroRunner)
152    return timer_coro_runner.add_timer_coro_run_from_other_service(delay, coro_worker, *args, **kwargs)

def discard_timer_coro_run_from_other_service( current_service: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service, timer_request: cengal.time_management.timer.versions.v_0.timer.TimerRequest) -> bool: View Source

155def discard_timer_coro_run_from_other_service(current_service: Service, timer_request: TimerRequest) -> bool:
156    timer_coro_runner: TimerCoroRunner = current_service._loop.get_service_instance(TimerCoroRunner)
157    return timer_coro_runner.discard_timer_coro_run_from_other_service(timer_request)

def timer_coro_run_on( context: typing.Tuple[typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

160def timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
161    """_summary_
162        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
163        
164        An example:
165
166        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
167        from cengal.parallel_execution.coroutines.coro_standard_services.timer_coro_runner import timer_coro_run_on
168        from typing import Optional, Union
169
170        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
171            try:
172                def print_hello_world(i: Interface, name: str):
173                    print(f'Hello Wrold from {name}!)
174                
175                timer_coro_run_on(10, print_hello_world, 'John Doe')
176            except CoroSchedulerContextIsNotAvailable:
177                print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)
178        
179    Args:
180        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
181        delay (float): delay in seconds
182        coro_worker (AnyWorker): coro_worker
183
184    Returns:
185        ValueExistence[Optional[CoroID]]: _description_
186    """
187    return make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)

_summary_ context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module

An example:

from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
from cengal.parallel_execution.coroutines.coro_standard_services.timer_coro_runner import timer_coro_run_on
from typing import Optional, Union

def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
    try:
        def print_hello_world(i: Interface, name: str):
            print(f'Hello Wrold from {name}!)

        timer_coro_run_on(10, print_hello_world, 'John Doe')
    except CoroSchedulerContextIsNotAvailable:
        print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)

Args: context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_ delay (float): delay in seconds coro_worker (AnyWorker): coro_worker

Returns: ValueExistence[Optional[CoroID]]: _description_

def try_timer_coro_run_on( context: typing.Tuple[typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

190def try_timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
191    """_summary_
192        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
193        
194        An example:
195
196        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
197        from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
198        from typing import Optional, Union
199
200        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
201            def print_hello_world(i: Interface, name: str):
202                print(f'Hello Wrold from {name}!)
203            
204            try_timer_coro_run_on(10, print_hello_world, 'John Doe')
205        
206    Args:
207        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
208        delay (float): delay in seconds
209        coro_worker (AnyWorker): coro_worker
210
211    Returns:
212        ValueExistence[Optional[CoroID]]: _description_
213    """
214    return try_make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)

_summary_ context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module

An example:

from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
from typing import Optional, Union

def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
    def print_hello_world(i: Interface, name: str):
        print(f'Hello Wrold from {name}!)

    try_timer_coro_run_on(10, print_hello_world, 'John Doe')

Args: context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_ delay (float): delay in seconds coro_worker (AnyWorker): coro_worker

Returns: ValueExistence[Optional[CoroID]]: _description_

async def atimer_coro_run_on( context: typing.Tuple[typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

217async def atimer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
218    return await amake_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)

async def atry_timer_coro_run_on( context: typing.Tuple[typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

221async def atry_timer_coro_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
222    return await atry_make_request_to_service_with_context(context, TimerCoroRunner, delay, coro_worker, *args, **kwargs)

def timer_coro_run( delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

225def timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
226    return make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)

def try_timer_coro_run( delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

229def try_timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
230    return try_make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)

async def atimer_coro_run( delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

233async def atimer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[CoroID]:
234    return await amake_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)

async def atry_timer_coro_run( delay: float, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

237async def atry_timer_coro_run(delay: float, coro_worker: AnyWorker, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
238    return await atry_make_request_to_service(TimerCoroRunner, delay, coro_worker, *args, **kwargs)