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`.
129 self, group=None, target=None, name=None, args=(), kwargs={}, Verbose=None
131 threading.Thread.__init__(
132 self, group=None, target=target, name=None, args=args, kwargs=kwargs
134 self._target = target
138 if self._target is not None:
139 self._return = self._target(*self._args, **self._kwargs)
141 def join(self, *args):
142 threading.Thread.join(self, *args)
146 def periodically_invoke(
148 stop_after: Optional[int],
151 Periodically invoke the decorated function.
154 period_sec: the delay period in seconds between invocations
155 stop_after: total number of invocations to make or, if None,
159 a :class:`Thread` object and an :class:`Event` that, when
160 signaled, will stop the invocations.
163 It is possible to be invoked one time after the :class:`Event`
164 is set. This event can be used to stop infinite
165 invocation style or finite invocation style decorations.
169 @periodically_invoke(period_sec=0.5, stop_after=None)
170 def there(name: str, age: int) -> None:
171 print(f" ...there {name}, {age}")
173 @periodically_invoke(period_sec=1.0, stop_after=3)
174 def hello(name: str) -> None:
175 print(f"Hello, {name}")
178 def decorator_repeat(func):
179 def helper_thread(should_terminate, *args, **kwargs) -> None:
180 if stop_after is None:
182 func(*args, **kwargs)
183 res = should_terminate.wait(period_sec)
187 for _ in range(stop_after):
188 func(*args, **kwargs)
189 res = should_terminate.wait(period_sec)
194 @functools.wraps(func)
195 def wrapper_repeat(*args, **kwargs):
196 should_terminate = threading.Event()
197 should_terminate.clear()
198 newargs = (should_terminate, *args)
199 thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
201 logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
202 return (thread, should_terminate)
204 return wrapper_repeat
206 return decorator_repeat
209 if __name__ == '__main__':