Threads in Python

Python offers multithreading on platforms that support threads, such as Win32, Linux, and other variants of Unix. An action is known as atomic when it’s guaranteed that no thread switching occurs between the start and the end of the action. In practice, in CPython, operations that look atomic (e.g., simple assignments and accesses) mostly are atomic, when executed on built-in types (augmented and multiple assignments, however, aren’t atomic). Mostly, though, it’s not a good idea to rely on atomicity. You might be dealing with an instance of a user-coded class rather than of a built-in type, so there might be implicit calls to Python code, making assumptions of atomicity unwarranted. Relying on implementation-dependent atomicity may lock your code into a specific implementation, hampering future upgrades. You’re better off using the synchronization facilities covered in the rest of this chapter, rather than relying on atomicity assumptions.

Python offers multithreading in two flavors. An older and lower-level module, _thread (named thread in v2), has low-level functionality and is not recommended for direct use in your code; we do not cover _thread in this book. The higher-level module threading, built on top of _thread, is the recommended one. The key design issue in multithreading systems is how best to coordinate multiple threads. threading supplies several synchronization objects. Alternatively, the queue module is very useful for thread synchronization, as it supplies synchronized, thread-safe queue types, handy for communication and coordination between threads. The package concurrent, covered after multiprocessing, supplies a unified interface for communication and coordination that can be implemented by pools of either threads or processes.

The threading Module

The threading module supplies multithreading functionality. The approach of threading is similar to Java’s, but locks and conditions are modeled as separate objects (in Java, such functionality is part of every object), and threads cannot be directly controlled from the outside (thus, no priorities, groups, destruction, or stopping). All methods of objects supplied by threading are atomic.

threading supplies classes dealing with threads: Thread, Condition, Event, Lock, RLock, Semaphore, BoundedSemaphore, and Timer (and, in v3 only, Barrier). threading also supplies functions, including:

lock = threading.RLock()
global_state = []
def recursive_function(some, args):
    with lock:  # acquires lock, guarantees release at end
        ...modify global_state...
        if more_changes_needed(global_state):
            recursive_function(other, args)

with c:
    while not is_ok_state(s):
        c.wait()
    do_some_work_using_state(s)

with c:
    do_something_that_modifies_state(s)
    c.notify()    # or, c.notify_all()
# no need to call c.release(), exiting `with` intrinsically does

class Periodic(threading.Timer):
  def __init__(self, interval, callable, args=(), kwargs={}):
    self.callable = callable
    threading.Timer.__init__(self, interval, self._f, args, kwargs)
 
  def _f(self, *args, **kwargs):
    Periodic(self.interval, self.callable, args, kwargs).start()
    self.callable(*args, **kwargs)

import threading
L = threading.local()
print('in main thread, setting zop to 42')
L.zop = 42
def targ():
  print('in subthread, setting zop to 23')
  L.zop = 23
  print('in subthread, zop is now', L.zop)
t = threading.Thread(target=targ)
t.start()
t.join()
print('in main thread, zop is now', L.zop)
# prints:
# in main thread, setting zop to 42
# in subthread, setting zop to 23
# in subthread, zop is now 23
# in main thread, zop is now 42

The queue Module

The queue module (named Queue in v2) supplies queue types supporting multithread access, with one main class, two subclasses, and two exception classes:

if q.empty():
    print('no work to perform')
else:
    x = q.get_nowait()
    work_on(x)

try:
    x = q.get_nowait()
except queue.Empty:
    print('no work to perform')
else:
    work_on(x)

The multiprocessing Module

The multiprocessing module supplies functions and classes you can use to code pretty much as you would for multithreading, but distributing work across processes, rather than across threads: the class Process (similar to threading.Thread) and classes for synchronization primitives (BoundedSemaphore, Condition, Event, Lock, RLock, Semaphore, and, in v3 only, Barrier—each similar to the class with the same names in module threading; also, Queue, and JoinableQueue, both similar to queue.Queue). These classes make it easy to take code written to use threading, and make a version that uses multiprocessing instead; you just need to pay attention to the differences we cover in “Differences Between Multiprocessing and Threading”.

It’s usually best to avoid sharing state among processes: use queues, instead, to explicitly pass messages among them. However, for those occasions in which you do need to share some state, multiprocessing supplies classes to access shared memory (Value and Array), and—more flexibly (including coordination among different computers on a network) though with more overhead—a Process subclass, Manager, designed to hold arbitrary data and let other processes manipulate that data via proxy objects. We cover state sharing in “Sharing State: Classes Value, Array, and Manager”.

When you’re writing new multiprocessing code, rather than porting code originally written to use threading, you can often use different approaches supplied by multiprocessing. The Pool class, in particular, can often simplify your code. We cover Pool in “Multiprocessing Pool”.

Other advanced approaches, based on Connection objects built by the Pipe factory function or wrapped in Client and Listener objects, are, on the other hand, more flexible, but potentially more complex; we do not cover them further in this book. For more thorough coverage of multiprocessing, refer to the online docs and to good third-party online tutorials.

with v.get_lock():
    v.value += 1

a = multiprocessing.Array('c', b'four score and seven')
a.value = b'five'
print(a.value)   # prints b'five'
print(a[:])      # prints b'fivex00score and seven'

p = some_manager.list()
p[:] = [1, 2, 3]
print(p == [1, 2, 3])        # prints False, as it compares with p
print(list(p) == [1, 2, 3])  # prints True, as it compares with copy

import multiprocessing as mp

def f(s):
    """Run a long time, and eventually return a result."""
    import time, random
    time.sleep(random.random()*2)  # simulate slowness
    return s+s  # some computation or other

def runner(s, d):
    print(os.getpid())
    d[s] = f(s)

def make_dict(set_of_strings):
    mgr = mp.Manager()
    d = mgr.dict()
    workers = []
    for s in set_of_strings:
        p = mp.Process(target=runner, args=(s, d))
        p.start()
        workers.append(p)
    for p in workers:
        p.join()
    return dict(d)

import multiprocessing as mp

def f(s):
    """Run a long time, and eventually return a result."""
    import time, random
    time.sleep(random.random()*2)  # simulate slowness
    return s+s  # some computation or other

def runner(s):
    print(os.getpid())
    return s, f(s)

def make_dict(set_of_strings):
    with mp.Pool() as pool:
        d = dict(pool.imap_unordered(runner, set_of_strings))
        return d

The concurrent.futures Module

The concurrent package supplies a single module, futures. concurrent.futures is in the standard library only in v3; to use it in v2, download and install the backport with pip2 install futures (or, equivalently, python2 -m pip install futures).

concurrent.futures supplies two classes, ThreadPoolExecutor (using threads as workers) and ProcessPoolExecutor (using processes as workers), which implement the same abstract interface, Executor. Instantiate either kind of pool by calling the class with one argument, max_workers, specifying how many threads or processes the pool should contain. You can omit max_workers to let the system pick the number of workers (except that you have to explicitly specify max_workers to instantiate ThreadPoolExecutor for the v2 backport, only).

An instance e of an Executor class supports the following methods:

Any instance of an Executor is also a context manager, and therefore suitable for use on a with statement (__exit__ being like shutdown(wait=True)).

For example, here’s a concurrent-based approach to perform the same task as in Example 14-1:

import concurrent.futures as cf

def f(s):
    """run a long time and eventually return a result"""
    ...

def runner(s):
    return s, f(s)

def make_dict(set_of_strings):
    with cf.ProcessPoolExecutor() as e:
        d = dict(e.map(runner, set_of_strings))
    return d

The submit method of an Executor returns a Future instance. A Future instance f supplies the methods described in Table 14-1.

The concurrent.futures module also supplies two functions:

Threaded Program Architecture

A threaded program should always try to arrange for a single thread to deal with any given object or subsystem that is external to the program (such as a file, a database, a GUI, or a network connection). Having multiple threads that deal with the same external object is possible, but can often cause gnarly problems.

When your threaded program must deal with some external object, devote a thread to such dealings, using a Queue object from which the external-interfacing thread gets work requests that other threads post. The external-interfacing thread can return results by putting them on one or more other Queue objects. The following example shows how to package this architecture into a general, reusable class, assuming that each unit of work on the external subsystem can be represented by a callable object. (In examples, remember: in v2, the module queue is spelled Queue; the class Queue in that module is spelled with an uppercase Q in both v2 and v3).

import threading, queue
class ExternalInterfacing(threading.Thread):
    def __init__(self, external_callable, **kwds):
        threading.Thread.__init__(self, **kwds)  # could use `super`
        self.daemon = True
        self.external_callable = external_callable
        self.work_request_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.start()
    def request(self, *args, **kwds):
        """called by other threads as external_callable would be"""
        self.work_request_queue.put((args,kwds))
        return self.result_queue.get()
    def run(self):
        while True:
            a, k = self.work_request_queue.get()
            self.result_queue.put(self.external_callable(*a, **k))

Once some ExternalInterfacing object ei is instantiated, any other thread may call ei.request just as it would call external_callable without such a mechanism (with or without arguments as appropriate). The advantage of the ExternalInterfacing mechanism is that all calls upon external_callable are serialized. This means they are performed by just one thread (the thread object bound to ei) in some defined sequential order, without overlap, race conditions (hard-to-debug errors that depend on which thread happens to get there first), or other anomalies that might otherwise result.

If several callables need to be serialized together, you can pass the callable as part of the work request, rather than passing it at the initialization of the class ExternalInterfacing, for greater generality. The following example shows this more general approach:

import threading, queue
class Serializer(threading.Thread):
    def __init__(self, **kwds):
        threading.Thread.__init__(self, **kwds)  # could use `super`
        self.daemon = True
        self.work_request_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.start()
    def apply(self, callable, *args, **kwds):
        """called by other threads as callable would be"""
        self.work_request_queue.put((callable, args, kwds))
        return self.result_queue.get()
    def run(self):
        while True:
            callable, args, kwds = self.work_request_queue.get()
            self.result_queue.put(callable(*args, **kwds))

Once a Serializer object ser has been instantiated, any other thread may call ser.apply(external_callable) just as it would call external_callable without such a mechanism (with or without further arguments as appropriate). The Serializer mechanism has the same advantages as ExternalInterfacing, except that all calls to the same or different callables wrapped by a single ser instance are now serialized.

The user interface of the whole program is an external subsystem, and thus should be dealt with by a single thread—specifically, the main thread of the program (this is mandatory for some user interface toolkits, and advisable even when not mandatory). A Serializer thread is therefore inappropriate. Rather, the program’s main thread should deal only with user-interface issues, and farm out actual work to worker threads that accept work requests on a Queue object and return results on another. A set of worker threads is generally known as a thread pool. As shown in the following example, all worker threads should share a single queue of requests and a single queue of results, since the main thread is the only one to post work requests and harvest results:

import threading
class Worker(threading.Thread):
    IDlock = threading.Lock()
    request_ID = 0
    def __init__(self, requests_queue, results_queue, **kwds):
        threading.Thread.__init__(self, **kwds)
        self.daemon = True
        self.work_request_queue = requests_queue
        self.result_queue = results_queue
        self.start()
    def perform_work(self, callable, *args, **kwds):
        """called by main thread as callable would be, but w/o return"""
        with self.IDlock:
            Worker.request_ID += 1
            self.work_request_queue.put(
                (Worker.request_ID, callable, args, kwds))
            return Worker.request_ID
    def run(self):
        while True:
            request_ID, callable, a, k = self.work_request_queue.get()
            self.result_queue.put((request_ID, callable(*a, **k)))

The main thread creates the two queues, and then instantiates worker threads as follows:

import queue
requests_queue = queue.Queue()
results_queue = queue.Queue()
for i in range(number_of_workers):
    worker = Worker(requests_queue, results_queue)

Whenever the main thread needs to farm out work (execute some callable object that may take substantial elapsed time to produce results), the main thread calls worker.perform_work(callable) , much as it would call callable without such a mechanism (with or without further arguments as appropriate). However, perform_work does not return the result of the call. Instead of the results, the main thread gets an id that identifies the work request. If the main thread needs the results, it can keep track of that id, since the request’s results are tagged with that id when they appear. The advantage of this mechanism is that the main thread does not block waiting for the callable’s lengthy execution to complete, but rather becomes ready again at once and can immediately return to its main business of dealing with the user interface.

The main thread must arrange to check the results_queue, since the result of each work request eventually appears there, tagged with the request’s id, when the worker thread that took that request from the queue finishes computing the result. How the main thread arranges to check for both user interface events and the results coming back from worker threads onto the results queue depends on what user interface toolkit is used or—if the user interface is text-based—on the platform on which the program runs.

A widely applicable, though not always optimal, general strategy is for the main thread to poll (check the state of the results queue periodically). On most Unix-like platforms, the function alarm of the module signal allows polling. The Tkinter GUI toolkit supplies method after, which is usable for polling. Some toolkits and platforms afford more effective strategies (letting a worker thread alert the main thread when it places some result on the results queue), but there is no generally available, cross-platform, cross-toolkit way to arrange for this. Therefore, the following artificial example ignores user interface events and just simulates work by evaluating random expressions, with random delays, on several worker threads, thus completing the previous example:

import random, time
def make_work():
    return '{} {} {}'.format(random.randrange(2,10),
        random.choice(('+', '-', '*', '/', '%', '**')),
        random.randrange(2,10))
def slow_evaluate(expression_string):
    time.sleep(random.randrange(1,5))
    return eval(expression_string)
workRequests = {}
def showResults():
    while True:
        try: id, results = results_queue.get_nowait()
        except queue.Empty: return
        print('Result {}: {} -> {}'.format(
            id, work_requests[id], results))
        del work_requests[id]
for i in range(10):
    expression_string = make_work()
    id = worker.perform_work(slow_evaluate, expression_string)
    work_requests[id] = expression_string
    print('Submitted request {}: {}'.format(id, expression_string))
    time.sleep(1)
    showResults()
while work_requests:
    time.sleep(1)
    showResults()

Process Environment

The operating system supplies each process P with an environment, a set of variables whose names are strings (most often, by convention, uppercase identifiers) and whose contents are strings. In “Environment Variables”, we cover environment variables that affect Python’s operations. Operating system shells offer ways to examine and modify the environment via shell commands and other means mentioned in “Environment Variables”.

The module os supplies the attribute environ, a mapping that represents the current process’s environment. os.environ is initialized from the process environment when Python starts. Changes to os.environ update the current process’s environment if the platform supports such updates. Keys and values in os.environ must be strings. On Windows (but not on Unix-like platforms), keys into os.environ are implicitly uppercased. For example, here’s how to try to determine which shell or command processor you’re running under:

import os
shell = os.environ.get('COMSPEC')
if shell is None: shell = os.environ.get('SHELL')
if shell is None: shell = 'an unknown command processor'
print('Running under', shell)

When a Python program changes its environment (e.g., via os.environ['X']='Y'), this does not affect the environment of the shell or command processor that started the program. As already explained—and for all programming languages including Python—changes to a process’s environment affect only the process itself, not other processes that are currently running.

Running Other Programs

You can run other programs via functions in the os module or (at a higher and usually preferable level of abstraction) with the subprocess module.

import os
os.spawnv(os.P_WAIT, editor, [editor, textfile])

The mmap Module

The mmap module supplies memory-mapped file objects. An mmap object behaves similarly to a bytestring, so you can often pass an mmap object where a bytestring is expected. However, there are differences:

• An mmap object does not supply the methods of a string object.

• An mmap object is mutable, while string objects are immutable.

• An mmap object also corresponds to an open file and behaves polymorphically to a Python file object (as covered in “File-Like Objects and Polymorphism”).

An mmap object m can be indexed or sliced, yielding bytestrings. Since m is mutable, you can also assign to an indexing or slicing of m. However, when you assign to a slice of m, the righthand side of the assignment statement must be a bytestring of exactly the same length as the slice you’re assigning to. Therefore, many of the useful tricks available with list slice assignment (covered in “Modifying a list”) do not apply to mmap slice assignment.

The mmap module supplies a factory function that is slightly different on Unix-like systems and on Windows:

fileob = open('xxx','w')
while True:
    data = input('Enter some text:')
    fileob.seek(0)
    fileob.write(data)
    fileob.truncate()
    fileob.flush()

import mmap, os, time
mx = mmap.mmap(os.open('xxx',os.O_RDWR), 1)
last = None
while True:
    mx.resize(mx.size())
    data = mx[:]
    if data != last:
        print(data)
        last = data
    time.sleep(1)