More On Multi-Threading
This lesson includes the following topics:
Because attempts to access (and particularly to modify) the same resource from different threads can be disastrous, the threading library includes Lock objects that allow you to place a lock on resources, stopping any other thread that tries to access the resource in its tracks (in fact, stopping any thread that attempts to acquire the same lock). A threading.Lock has two states: locked and unlocked, and it is created in the unlocked state.
When a thread wants to access the resource associated with a specific Lock, it calls that Lock's acquire() method. If the Lock is currently locked, the acquiring thread is blocked until the Lock becomes unlocked and allows acquisition. If the Lock is unlocked, it is locked and acquired immediately. A Lock object becomes unlocked when its release() method is called.
In the next example, we'll modify the thread.py code from the last lesson so that the "critical resource" is the ability to sleep. Before sleeping for a tenth of a second each thread has to acquire a single lock shared between all threads. Even though each thread only has to sleep for a total of a second, because there are six threads and only one of them can be sleeping at a time, it takes the program six seconds to run.
"""
thread.py: Use threading.Lock to ensure threads run sequentially.
"""
import threading
import time
class MyThread(threading.Thread):
def __init__(self, lock, *args, **kw):
threading.Thread.__init__(self, *args, **kw)
self.lock = lock
def run(self):
for i in range(10):
self.lock.acquire()
time.sleep(0.1)
self.lock.release()
print(self.name, "finished")
lock = threading.Lock()
bgthreads = threading.active_count()
tt = [MyThread(lock) for i in range(6)]
for t in tt:
t.start()
print("Threads started")
while threading.active_count() > bgthreads:
time.sleep(2)
print("tick")
print("All threads done")
You will most probably see this:
Threads started tick tick tick Thread-1 finished Thread-2 finished Thread-3 finished Thread-4 finished Thread-5 finished Thread-6 finished tick All threads done
In different environments, however, the output from this program will typically vary each time you run it, because there are enough acquisitions and releases to allow different threads to get an advantage in the scheduling (which is not a simple deterministic round-robin). Here is the output from a run of the same program under Python 3.1.3 on MacOS 10.6:
Threads started Thread-3 finished tick Thread-6 finished tick Thread-1 finished Thread-4 finished Thread-5 finished tick Thread-2 finished tick All threads done
The simple expedient of removing the lock acquisition allows the threads to sleep in parallel, and without the limitation that only one thread can sleep at a time, all threads have terminated before the first (and last) tick from the main thread. Because the sleeps are intermingled, and again subject to random timing variations, the order of the threads finishing is unpredictable. [You should verify this assertion by making several runs of your program].
""" thread.py:UseWithout threading.Lockto ensure, threadsrusleep insequentiparallyel. """ import threading import time class MyThread(threading.Thread): def __init__(self, lock, *args, **kw): threading.Thread.__init__(self, *args, **kw)self.lock = lock def run(self): for i in range(10):self.lock.acquire()time.sleep(0.1)self.lock.release()self.lock.acquire() print(self.name, "finished") self.lock.release() lock = threading.Lock() bgthreads = threading.active_count() tt = [MyThread(lock) for i in range(6)] for t in tt: t.start() print("Threads started") while threading.active_count() > bgthreads: time.sleep(2) print("tick") print("All threads done")
Now the six threads are all sleeping pretty much in parallel, and so all terminate after one second. The main thread therefore ticks once and sees all threads already terminated, and so the program ends after two seconds. Again you should find that the order in which the "worker" threads terminate is unpredictable, because of uncontrollable timing differences. Now it is much more likely that different threads could be printing at the same time, which could lead to garbled output, so we use the locks to ensure this cannot happen. A typical output follows.
Threads started Thread-4 finished Thread-5 finished Thread-6 finished Thread-2 finished Thread-1 finished Thread-3 finished tick All threads terminated
| Note | Interactive threading experiments can be tricky in IDEs: you may find, if you experiment with threads from the interactive console, that output from a thread running in the background does not always appear immediately. This is because the IDE controls output in an attempt to ensure that your input is never interspersed with output from running code (which would make sessions extremely difficult to understand). So frequently you need to press Enter at the ">>> " prompt to allow output to become visible. A true interactive console session in a terminal window will not generally cause the same issues. |
If you are starting to enjoy the possibilities opened up by the threading library, you should definitely look at its documentation to learn about Rlock, Condition, Semaphore and Event objects.
| Modern Python | The threading module has grown considerably since Python 3.1.
threading.Lock can also be used as a context manager — with lock: — which
automatically calls acquire() on entry and release() on exit, even if
an exception is raised inside the block. This is the idiomatic modern approach and avoids the
risk of forgetting to release a lock. Additionally, threading.RLock (re-entrant lock),
threading.Condition, threading.Event, and threading.Semaphore cover a wide
range of synchronisation patterns. For CPU-bound work, note that CPython's Global Interpreter
Lock (GIL) means threads cannot execute Python bytecode in parallel on multiple cores; see the
next lesson on multi-processing for a way around this. |
This library was produced to provide programmers of threaded programs with a safe way for their threads to exchange information. The queue module defines three classes that each have the same interface but queue things in slightly different ways. queue.Queue is a FIFO (first-in first-out) queue in which the first objects added to the queue are the first to be retrieved. This is the most usual type to use for handing out work to worker threads. queue.LifoQueue objects implement a stack of sorts. The next item retrieved is the most recently-added item. Finally, queue.PriorityQueue items are always retrieved in natural sort order.
When creating a queue, you can establish a maximum length for it by providing that length as an argument. If this maximum length is not provided, the queue will be of potentially infinite length, and further items may always be added to it. With a maximum length, there are only a given number of free slots, and attempts to add to a full queue will either block the thread that is attempting the add or raise an exception to show that the queue is full (or a combination of both). The thread-safety guarantees made by the library mean that the same queue item can be accessed by multiple threads without any need to lock the queue (locking as necessary is taken care of internally by the queue methods). When a queue is empty, any attempt to extract an item will either block or raise an exception (or both).
We are making only the simplest use of queues here, by using the put() and get() methods, to present a way of writing scalable threaded programs. There are many refinements you can adopt by reading the module documentation once you understand the basics. In threaded applications, simplest is almost always best, as most of us have brains that can only conceptualize a limited amount of parallelism and have difficulty predicting situations that cause problems in practice (such as deadlocks, where Thread A is blocked waiting for Thread B, which is blocked waiting for Thread A: since neither can progress, the two threads are doomed to wait for each other forever).
queue.Queue.put(item, block=True, timeout=None) adds the given item to the queue. If block evaluates false, either the item is added immediately or an exception is raised. When block is True (the default case), either the item is added immediately or the putting thread blocks. If timeout remains None, this could leave the thread blocked indefinitely in a non-interruptible state. If a timeout (in seconds) is given, an exception will be raised if the item has not been added before the timeout expires.
queue.Queue.get(block=True, timeout=None) attempts to remove an item from the queue. If an item is immediately available, it is always returned. Otherwise, if block evaluates false, an exception is raised. When block evaluates true, the process blocks either indefinitely (when timeout is None) or until the timeout (in seconds) has expired, in which case an exception is raised if no item has arrived.
Every time an item is successfully added to a queue with put(), a task count is incremented. Removing an item with get() does not decrement the counter. To decrement the counter, the removing thread should wait until processing is complete and then call the queue's task_done() method.
If a queue is expected to end up empty, a thread can declare itself interested in the queue's exhaustion by calling its join() method. This method blocks the calling thread until all tasks have been recorded as complete. You should be confident that threads are all going to terminate correctly before using this technique, since it can lead to indefinite delays.
We'll finish the lesson by building a fairly general framework to allow you to run programs with "any number" of threads (sometimes the system places limits on the number of threads you can create).
The idea is to have a control thread that generates "work packets" for a given number of worker threads (with which it communicates by means of a queue). The worker threads compute the necessary results, and deliver them to a final output thread (by means of a second queue) which displays the results. The structure is quite general: work units can be generated by reading database tables, accepting data from web services, and the like. Computations can involve not only calculation but further database work or network communication, all of which can involve some (in computer terms) fairly extensive waiting.
The control thread is the main thread with which every program starts out (the only thread of all programs before these lessons). It creates an input and an output queue, starts the worker threads and the output thread, and thereafter distributes work packets to the worker threads until there is no more work. Since the worker threads are programmed to terminate when they receive None from the work queue, the control thread's final act is to Queue None for each worker thread and then wait for the queue to finally empty before terminating. The worker threads put a None to the output queue before terminating. The output thread counts these Nones, and terminates when enough None values have been seen to account for all workers.
The output thread simply has to extract output packets from a queue where they are placed by the worker threads. As each worker thread terminates, it posts a None to the queue. When a None has been received from each thread, the output thread terminates. The output thread is told on initialization how many worker threads there are, and each time it receives another None it decrements the worker count until eventually there are no workers left. At that point, the output thread terminates. Create output.py as shown:
"""
output.py: The output thread for the miniature framework.
"""
identity = lambda x: x
import threading
class OutThread(threading.Thread):
def __init__(self, N, q, sorting=True, *args, **kw):
"""Initialize thread and save queue reference."""
threading.Thread.__init__(self, *args, **kw)
self.queue = q
self.workers = N
self.sorting = sorting
self.output = []
def run(self):
"""Extract items from the output queue and print until all done."""
while self.workers:
p = self.queue.get()
if p is None:
self.workers -= 1
else:
# This is a real output packet
self.output.append(p)
print("".join(c for (i, c) in (sorted if self.sorting else identity)(self.output)))
print ("Output thread terminating")
In this particular case, the output thread is receiving (index, character) pairs (because the workers pass through the position argument they are given as well as the transformed character, to allow the string to be reassembled no matter in what order the threads finish). Rather than output each one as it arrives, the output thread stores them until the workers are all done, then sorts them (unless sorting is disabled with sorting=False) and the characters extracted and joined together.
The Worker threads have been cast so as to make interactions easy. The work units received from the input queue are (index, character) pairs, and the output units are also pairs. The processing is split out into a separate method to make subclassing easier—simply override the process() method. Create worker.py as shown:
"""
worker.py: a sample worker thread that receives input
through one Queue and routes output through another.
"""
from threading import Thread
class WorkerThread(Thread):
def __init__(self, iq, oq, *args, **kw):
"""Initialize thread and save Queue references."""
Thread.__init__(self, *args, **kw)
self.iq, self.oq = iq, oq
def run(self):
while True:
work = self.iq.get()
if work is None:
self.oq.put(None)
print("Worker", self.name, "done")
self.iq.task_done()
break
i, c = work
result = (i, self.process(c)) # this is the "work"
self.oq.put(result)
self.iq.task_done()
def process(self, s):
"""This defines how the string is processed to produce a result"""
return s.upper()
Although this particular worker thread is not doing particularly interesting processing (merely converting a single character to upper case), you can imagine more complex work units, perhaps with numerical inputs and the need for database lookup as well as interaction with local disk files.
Everything is started off by the control thread (which imports the output and worker threads from their respective modules). It first creates the input and output queues. These are standard FIFOs, with a limit of 50% more than the number of worker threads to avoid locking up too much memory in buffered objects. Then it creates and starts the output thread, and finally creates and starts as many worker threads as configured by the WORKERS constant. Worker threads get from the input queue and put to the output queue. The control thread then simply keeps the input queue loaded as long as it can before sending the None values required to shut the worker threads down. Once the input queue is empty, the thread terminates.
"""
control.py: Creates queues, starts output and worker threads,
and pushes inputs into the input queue.
"""
from queue import Queue
from output import OutThread
from worker import WorkerThread
WORKERS = 10
inq = Queue(maxsize=int(WORKERS*1.5))
outq = Queue(maxsize=int(WORKERS*1.5))
ot = OutThread(WORKERS, outq)
ot.start()
for i in range(WORKERS):
w = WorkerThread(inq, outq)
w.start()
instring = input("Words of wisdom: ")
for work in enumerate(instring):
inq.put(work)
for i in range(WORKERS):
inq.put(None)
inq.join()
print("Control thread terminating")
Running the program causes a prompt for input, which is then split up into individual characters and passed through the input queue to the worker threads. At present, ten threads operate in parallel, but the number can easily be varied by changing the definition of WORKERS in the source file. The output from a typical run is shown below.
Words of wisdom: Elemental forces are at work to change the way we live. Worker Thread-2 done Worker Thread-3 done Worker Thread-4 done Worker Thread-10 done Worker Thread-9 done Worker Thread-8 done Worker Thread-11 done Worker Thread-7 done Worker Thread-5 done Worker Thread-6 done Control thread terminating ELEMENTAL FORCES ARE AT WORK TO CHANGE THE WAY WE LIVE. Output thread terminating
You will appreciate the need for the sorting if you study this output, from a typical run where the output thread was created with sorting=False:
Words of wisdom: Does the string really appear correct? Worker Thread-7 done Worker Thread-6 done Worker Thread-4 done Worker Thread-2 done Worker Thread-11 done Worker Thread-3 done Worker Thread-9 done Worker Thread-5 done Worker Thread-10 done Worker Thread-8 done DOES THE STRING PEAELRYA PLAR CORERCT? Output thread terminating Control thread terminating
This ends our discussion of the queue.Queue object, and with it our somewhat lengthy study of threading.
In the last two lessons, we've made use of the threading library module to write classes whose instances run as separate threads. If enough of these are started, the waiting that each thread has to do can be filled by useful work for other threads, and so a fairly high-bandwidth network channel can be kept busy and individual hold-ups can be made to matter much less. There are a number of other schemes that have been developed to control multiple asynchronous tasks.
The oldest (and the only one currently included in the standard library) is the asyncore module. With asyncore, each client process is a "channel," and you program the channels to respond to specific network events in specific ways. Asynchat is layered on top of asyncore and allows you to specify protocol handling by looking for specific sequences in the incoming data and triggering events when those sequences are detected.
The Twisted library is a system devised by Glyph Lefkowitz that has been used to good effect by many surprisingly large enterprises (including one business that has since been purchased by Google). Operations that will potentially block (cause the process to wait) return a Deferred object, which is effectively a promise of future data. A Deferred object is asked for its result by calling specific methods; if the data is not currently available, the Twisted scheduler suspends that activity until the Deferred request can be satisfied, and returns to some other suspended task that can now be restarted.
Stackless Python was an early attempt by Christian Tismer to allow massively parallel computing in Python by the provision of so-called "micro-threads." It has been used to great effect by a gaming company to provide a space "shoot-'em-up" environment for over 50,000 simultaneous players. More recent versions allow advanced capabilities like saving a computation on one computer and restoring it on another. This was very helpful in running code on a 250-CPU cluster.
A more recent approach to asynchronous networking is the Kamaelia package, initially developed by Michael Sparks for BBCResearch in the UK. Kamaelia, as far as I am aware, pioneered the use of generator functions to interact with the task scheduling environment. This approach has also been taken in Monocle, another even more recent development by Raymond Hettinger.
All in all, if you decide to venture beyond the standard library, a wealth of choices awaits you and not all of them rely on threading.
| Modern Python | Since these lessons were written, Python's asynchronous landscape has evolved
substantially. asyncio — added in Python 3.4 and now the standard library's preferred
approach for asynchronous I/O — uses an event loop with async def coroutines and
await expressions. The asyncore module mentioned above was deprecated
in Python 3.6 and removed in Python 3.12. For CPU-bound parallelism,
concurrent.futures.ThreadPoolExecutor provides a higher-level interface to thread pools
without subclassing threading.Thread, and
concurrent.futures.ProcessPoolExecutor (covered in the next lesson) bypasses the GIL
entirely. The queue.Queue producer/consumer pattern illustrated in this lesson remains
perfectly valid and widely used today. |
Multi-threading is one way to achieve asynchronous processing. For the CPython implementation (and others relying on single-processor guarantees to speed processing) this will not help if the application is CPU-bound, as all processing must take place on a single processor, and so the application cannot benefit from multiple processors in the computer it runs on.
Next, we'll go on to consider how to share work between multiple processes, which can be done on different processors and therefore extract more work from modern multi-processor hardware.