More Threads!
Branch | Status |
---|
master | ![Build Status](https://github.com/klahnakoski/mo-threads/actions/workflows/build.yml/badge.svg?branch=master) |
dev | ← child thread coverage is missing |
Module threads
The main benefits over Python's threading library is:
- Multi-threaded queues do not use serialization - Serialization is
great in the general case, where you may also be communicating between
processes, but it is a needless overhead for single-process multi-threading.
It is left to the programmer to ensure the messages put on the queue are
not changed, which is not ominous demand.
- Shutdown order is deterministic and explicit - Python's threading
library is missing strict conventions for controlled and orderly shutdown.
Each thread can shutdown on its own terms, but is expected to do so expediently.
- All threads are required to accept a
please_stop
signal; are
expected to test it in a timely manner; and expected to exit when signalled. - All threads have a parent - The parent is responsible for ensuring their children get the
please_stop
signal, and are dead, before stopping themselves. This responsibility is baked into the thread spawning process,
so you need not deal with it unless you want.
- Uses Signals to simplify logical
dependencies among multiple threads, events, and timeouts.
- Logging and Profiling is Integrated - Logging and exception handling
is seamlessly integrated: This means logs are centrally handled, and thread
safe. Parent threads have access to uncaught child thread exceptions, and
the cProfiler properly aggregates results from the multiple threads.
What's it used for
A good amount of time is spent waiting for underlying C libraries and OS
services to respond to network and file access requests. Multiple
threads can make your code faster, despite the GIL, when dealing with those
requests. For example, by moving logging off the main thread, we can get
up to 15% increase in overall speed because we no longer have the main thread
waiting for disk writes or remote logging posts. Please note, this level of
speed improvement can only be realized if there is no serialization happening
at the multi-threaded queue.
Do not use Async
Actors are easier to reason about than async tasks. Mixing regular methods and co-routines (with their yield from
pollution) is dangerous because:
- calling styles between synchronous and asynchronous methods can be easily confused
- actors can use blocking methods, async can not
- there is no way to manage resource priority with co-routines.
- stack traces are lost with co-routines
- async scope easily escapes lexical scope, which promotes bugs
Python's async efforts are still immature; a re-invention of threading functionality by another name. Expect to experience a decade of problems that are already solved by threading; here is an example.
Reading
Writing threaded functions
All threaded functions must accept a please_stop
parameter, which is a Signal
to indicate the desire to stop. The function should check this signal often, and do it's best to promptly return.
Simple work loop
For threaded functions that perform small chunks of work in some loop; the chunks are small enough that they will complete soon enough. Simply check the please_stop
signal at the start of each loop:
def worker(p1, p2, please_stop):
while not please_stop:
do_some_small_chunk_of_work(p1)
One-time work
Sometimes, threads are launched to perform low priority, one-time work. You may not need to check the please_stop
signal:
def worker(p1, p2, please_stop):
do_some_work_and_exit(p1, p2)
Passing signals to others
There are many times a more complex please_stop
checks are required. For example, we want to wait on an input queue for the next task. Many of the methods in mo-threads
accept Signals
instead of timeouts so they return quickly when signalled. You may pass the please_stop
signal to these methods, or make your own. Be sure to check if the method returned because it is done, or it returned because it was signaled to stop early.
def worker(source, please_stop):
while not please_stop:
data = source.pop(till=please_stop)
if please_stop:
return
chunk_of_work(data)
Combining signals
Work might be done on some regular interval: The threaded function will sleep for a period and perform some work. In these cases you can combine Signals and wait()
on either of them:
def worker(please_stop):
while not please_stop:
next_event = Till(seconds=1000)
(next_event | please_stop).wait()
if please_stop:
return
chunk_of_work()
Spawning threads
Most threads will be declared and run in a single line. It is much like Python's threading library, except it demands a name for the thread:
thread = Thread.run("name", function, p1, p2, ...)
Sometimes you want to separate creation from starting:
thread = Thread("name", function, p1, p2, ...)
thread.start()
join()
vs release()
Once a thread is created, a few actions can be performed with the thread object:
-
join()
- Join on thread
will make the caller thread wait until thread
has stopped. Then, return the resulting value or to re-raise thread
's exception in the caller.
result = thread.join() # may raise exception
-
release()
- Will ignore any return value, and post any exception to logging. Tracking is still performed; released threads are still properly stopped. You may still join()
to guarantee the caller will wait for thread completion, but you risk being too late: The thread may have already completed and logged it's failure.
thread.release() # release thread resources asap, when done
-
stopped.wait()
- Every thread has a stopped
Signal, which can be used for coordination by other threads. This allows a thread to wait for another to be complete and then resume. No errors or return values are captured
thread.stopped.wait()
Registering Threads
Threads created without this module can call your code; You want to ensure these "alien" threads have finished their work, released the locks, and exited your code before stopping. If you register alien threads, then mo-threads
will ensure the alien work is done for a clean stop.
def my_method():
with RegisterThread():
t = Thread.current() # we can now use mo-threads on this thread
print(t.name) # a name is always given to the alien thread
Synchronization Primitives
There are three major aspects of a synchronization primitive:
- Resource - Monitors and locks can only be owned by one thread at a time
- Binary - The primitive has only two states
- Irreversible - The state of the primitive can only be set, or advanced, never reversed
The last, irreversibility is very useful, but ignored in many threading
libraries. The irreversibility allows us to model progression; and
we can allow threads to poll for progress, or be notified of progress.
These three aspects can be combined to give us 8 synchronization primitives:
- - -
- Semaphore- B -
- EventR - -
- MonitorR B -
- Lock- - I
- Iterator/generator- B I
- Signal (or Promise)R - I
- Private IteratorR B I
- Private Signal (best implemented as is_done
Boolean flag)
Lock
Class
Locks are identical to threading monitors, except for two differences:
- The
wait()
method will always acquire the lock before returning. This is an important feature, it ensures every line inside a with
block has lock acquisition, and is easier to reason about. - Exiting a lock via
__exit__()
will always signal a waiting thread to resume. This ensures no signals are missed, and every thread gets an opportunity to react to possible change. Lock
is not reentrant! This is a feature to ensure locks are not held for long periods of time.
Example
lock = Lock()
while not please_stop:
with lock:
while not todo:
lock.wait(seconds=1)
In this example, we look for stuff todo
, and if there is none, we wait for a second. During that time others can acquire the lock
and add todo
items. Upon releasing the the lock
, our example code will immediately resume to see what's available, waiting again if nothing is found.
Signal
Class
The Signal
class is a binary semaphore that can be signalled only once; subsequent signals have no effect.
- It can be signalled by any thread;
- any thread can wait on a
Signal
; and - once signalled, all waiting threads are unblocked, including all subsequent waiting threads.
- A Signal's current state can be accessed by any thread without blocking.
Signal
is used to model thread-safe state advancement. It initializes to False
, and when signalled (with go()
) becomes True
. It can not be reversed.
Signals are like a Promise, but more explicit
Signal | Promise | Python Event |
---|
s.go() | p.resolve() | e.set() |
s.then(f) | p.then(m) | |
s.wait() | await p | e.wait() |
bool(s) | | e.is_set() |
s & t | Promise.all(p, q) | |
s | t | Promise.race(p, q) | |
Here is simple worker that signals when work is done. It is assumed do_work
is executed by some other thread.
class Worker:
def __init__(self):
self.is_done = Signal()
def do_work(self):
do_some_work()
self.is_done.go()
You can attach methods to a Signal
, which will be run, just once, upon go()
. If already signalled, then the method is run immediately.
worker = Worker()
worker.is_done.then(lambda: print("done"))
You may also wait on a Signal
, which will block the current thread until the Signal
is a go
worker.is_done.wait()
print("worker thread is done")
Signals
are first class, they can be passed around and combined with other Signals. For example, using the __or__
operator (|
): either = lhs | rhs
; either
will be triggered when lhs
or rhs
is triggered.
def worker(please_stop):
while not please_stop:
user_cancel = Signal("user cancel")
...
worker(user_cancel | Till(seconds=360))
Signal
s can also be combined using logical and (&
): both = lhs & rhs
; both
is triggered only when both lhs
and rhs
are triggered:
(workerA.stopped & workerB.stopped).wait()
print("both threads are done")
Differences from Python's Event
Signal
is not reversable, while Event
has a clear()
methodSignal
allows function chaining using the then
method- Complex signals can be composed from simple signals using boolean logic
Till
Class
The Till
class (short for "until") is a special Signal
used to represent timeouts.
Till(seconds=20).wait()
Till(till=Date("21 Jan 2016").unix).wait()
Use Till
rather than sleep()
because you can combine Till
objects with other Signals
.
Beware that all Till
objects will be triggered before expiry when the main thread is asked to shutdown
Command
Class
If you find process creation is too slow, the Command
class can be used to recycle existing processes. It has the same interface as Process
, yet it manages a bash
(or cmd.exe
) session for you in the background.