3 # © Copyright 2021-2022, Scott Gasch
5 """Utilities for dealing with threads + threading."""
11 from typing import Any, Callable, Optional, Tuple
13 # This module is commonly used by others in here and should avoid
14 # taking any unnecessary dependencies back on them.
16 logger = logging.getLogger(__name__)
19 def current_thread_id() -> str:
22 a string composed of the parent process' id, the current
23 process' id and the current thread identifier. The former two are
24 numbers (pids) whereas the latter is a thread id passed during thread
27 >>> ret = current_thread_id()
28 >>> (ppid, pid, tid) = ret.split('/')
37 tid = threading.current_thread().name
38 return f'{ppid}/{pid}/{tid}:'
41 def is_current_thread_main_thread() -> bool:
44 True is the current (calling) thread is the process' main
45 thread and False otherwise.
47 >>> is_current_thread_main_thread()
53 ... result = is_current_thread_main_thread()
60 >>> thread = threading.Thread(target=thunk)
67 return threading.current_thread() is threading.main_thread()
70 def background_thread(
71 _funct: Optional[Callable[..., Any]],
72 ) -> Callable[..., Tuple[threading.Thread, threading.Event]]:
73 """A function decorator to create a background thread.
78 def random(a: int, b: str, stop_event: threading.Event) -> None:
80 print(f"Hi there {b}: {a}!")
82 if stop_event.is_set():
86 (thread, event) = random(22, "dude")
94 In addition to any other arguments the function has, it must
95 take a stop_event as the last unnamed argument which it should
96 periodically check. If the event is set, it means the thread has
97 been requested to terminate ASAP.
100 def wrapper(funct: Callable):
101 @functools.wraps(funct)
102 def inner_wrapper(*a, **kwa) -> Tuple[threading.Thread, threading.Event]:
103 should_terminate = threading.Event()
104 should_terminate.clear()
105 newargs = (*a, should_terminate)
106 thread = threading.Thread(
112 logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
113 return (thread, should_terminate)
118 return wrapper # type: ignore
120 return wrapper(_funct)
123 class ThreadWithReturnValue(threading.Thread):
124 """A thread whose return value is plumbed back out as the return
125 value of :meth:`join`.
128 def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, Verbose=None):
129 threading.Thread.__init__(
130 self, group=None, target=target, name=None, args=args, kwargs=kwargs
132 self._target = target
136 if self._target is not None:
137 self._return = self._target(*self._args, **self._kwargs)
139 def join(self, *args):
140 threading.Thread.join(self, *args)
144 def periodically_invoke(
146 stop_after: Optional[int],
149 Periodically invoke the decorated function.
152 period_sec: the delay period in seconds between invocations
153 stop_after: total number of invocations to make or, if None,
157 a :class:`Thread` object and an :class:`Event` that, when
158 signaled, will stop the invocations.
161 It is possible to be invoked one time after the :class:`Event`
162 is set. This event can be used to stop infinite
163 invocation style or finite invocation style decorations.
167 @periodically_invoke(period_sec=0.5, stop_after=None)
168 def there(name: str, age: int) -> None:
169 print(f" ...there {name}, {age}")
171 @periodically_invoke(period_sec=1.0, stop_after=3)
172 def hello(name: str) -> None:
173 print(f"Hello, {name}")
176 def decorator_repeat(func):
177 def helper_thread(should_terminate, *args, **kwargs) -> None:
178 if stop_after is None:
180 func(*args, **kwargs)
181 res = should_terminate.wait(period_sec)
185 for _ in range(stop_after):
186 func(*args, **kwargs)
187 res = should_terminate.wait(period_sec)
192 @functools.wraps(func)
193 def wrapper_repeat(*args, **kwargs):
194 should_terminate = threading.Event()
195 should_terminate.clear()
196 newargs = (should_terminate, *args)
197 thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
199 logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
200 return (thread, should_terminate)
202 return wrapper_repeat
204 return decorator_repeat
207 if __name__ == '__main__':