From 497fb9e21f45ec08e1486abaee6dfa7b20b8a691 Mon Sep 17 00:00:00 2001
From: Scott Gasch <scott@gasch.org>
Date: Wed, 24 Mar 2021 18:08:54 -0700
Subject: Initial revision

---
 thread_utils.py | 137 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 137 insertions(+)
 create mode 100644 thread_utils.py

(limited to 'thread_utils.py')

diff --git a/thread_utils.py b/thread_utils.py
new file mode 100644
index 0000000..af6e0e1
--- /dev/null
+++ b/thread_utils.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+
+import functools
+import logging
+import os
+import threading
+from typing import Callable, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+def current_thread_id() -> str:
+    ppid = os.getppid()
+    pid = os.getpid()
+    tid = threading.current_thread().name
+    return f'{ppid}/{pid}/{tid}:'
+
+
+def is_current_thread_main_thread() -> bool:
+    return threading.current_thread() is threading.main_thread()
+
+
+def background_thread(
+        _funct: Optional[Callable]
+) -> Tuple[threading.Thread, threading.Event]:
+    """A function decorator to create a background thread.
+
+    *** Please note: the decorated function must take an shutdown ***
+    *** event as an input parameter and should periodically check ***
+    *** it and stop if the event is set.                          ***
+
+    Usage:
+
+        @background_thread
+        def random(a: int, b: str, stop_event: threading.Event) -> None:
+            while True:
+                print(f"Hi there {b}: {a}!")
+                time.sleep(10.0)
+                if stop_event.is_set():
+                    return
+
+
+        def main() -> None:
+            (thread, event) = random(22, "dude")
+            print("back!")
+            time.sleep(30.0)
+            event.set()
+            thread.join()
+
+    Note: in addition to any other arguments the function has, it must
+    take a stop_event as the last unnamed argument which it should
+    periodically check.  If the event is set, it means the thread has
+    been requested to terminate ASAP.
+    """
+    def wrapper(funct: Callable):
+        @functools.wraps(funct)
+        def inner_wrapper(
+                *a, **kwa
+        ) -> Tuple[threading.Thread, threading.Event]:
+            should_terminate = threading.Event()
+            should_terminate.clear()
+            newargs = (*a, should_terminate)
+            thread = threading.Thread(
+                target=funct,
+                args=newargs,
+                kwargs=kwa,
+            )
+            thread.start()
+            logger.debug(
+                f'Started thread {thread.name} tid={thread.ident}'
+            )
+            return (thread, should_terminate)
+        return inner_wrapper
+
+    if _funct is None:
+        return wrapper
+    else:
+        return wrapper(_funct)
+
+
+def periodically_invoke(
+        period_sec: float,
+        stop_after: Optional[int],
+):
+    """
+    Periodically invoke a decorated function.  Stop after N invocations
+    (or, if stop_after is None, call forever).  Delay period_sec between
+    invocations.
+
+    Returns a Thread object and an Event that, when signaled, will stop
+    the invocations.  Note that it is possible to be invoked one time
+    after the Event is set.  This event can be used to stop infinite
+    invocation style or finite invocation style decorations.
+
+        @periodically_invoke(period_sec=0.5, stop_after=None)
+        def there(name: str, age: int) -> None:
+            print(f"   ...there {name}, {age}")
+
+
+        @periodically_invoke(period_sec=1.0, stop_after=3)
+        def hello(name: str) -> None:
+            print(f"Hello, {name}")
+
+    """
+    def decorator_repeat(func):
+        def helper_thread(should_terminate, *args, **kwargs) -> None:
+            if stop_after is None:
+                while True:
+                    func(*args, **kwargs)
+                    res = should_terminate.wait(period_sec)
+                    if res:
+                        return
+            else:
+                for _ in range(stop_after):
+                    func(*args, **kwargs)
+                    res = should_terminate.wait(period_sec)
+                    if res:
+                        return
+                return
+
+        @functools.wraps(func)
+        def wrapper_repeat(*args, **kwargs):
+            should_terminate = threading.Event()
+            should_terminate.clear()
+            newargs = (should_terminate, *args)
+            thread = threading.Thread(
+                target=helper_thread,
+                args = newargs,
+                kwargs = kwargs
+            )
+            thread.start()
+            logger.debug(
+                f'Started thread {thread.name} tid={thread.ident}'
+            )
+            return (thread, should_terminate)
+        return wrapper_repeat
+    return decorator_repeat
-- 
cgit v1.3