Programming Python (28 page)

Besides
avoiding print collisions, thread module locks are
surprisingly useful. They can form the basis of higher-level
synchronization paradigms (e.g., semaphores) and can be used as
general thread communication devices.
^[
15
]
For instance,
Example 5-8
uses a global list of
locks to know when all child threads have finished.

"""
uses mutexes to know when threads are done in parent/main thread,
instead of time.sleep; lock stdout to avoid comingled prints;
"""
import _thread as thread
stdoutmutex = thread.allocate_lock()
exitmutexes = [thread.allocate_lock() for i in range(10)]
def counter(myId, count):
for i in range(count):
stdoutmutex.acquire()
print('[%s] => %s' % (myId, i))
stdoutmutex.release()
exitmutexes[myId].acquire()    # signal main thread
for i in range(10):
thread.start_new_thread(counter, (i, 100))
for mutex in exitmutexes:
while not mutex.locked(): pass
print('Main thread exiting.')

A lock’s
locked
method can be
used to check its state. To make this work, the main thread makes one
lock per child and tacks them onto a global
exitmutexes
list (remember, the threaded
function shares global scope with the main thread). On exit, each
thread acquires its lock on the list, and the main thread simply
watches for all locks to be acquired. This is much more accurate than
naïvely sleeping while child threads run in hopes that all will have
exited after the sleep. Run this on your own to see its output—all 10
spawned threads count up to 100 (they run in arbitrarily interleaved
order that can vary per run and platform, but their prints run
atomically and do not comingle), before the main thread exits.

Depending on how your threads run, this could be even simpler:
since threads share global memory anyhow, we can usually achieve the
same effect with a simple global list of
integers
instead of locks. In
Example 5-9
, the module’s namespace
(scope) is shared by top-level code and the threaded function, as
before.
exitmutexes
refers to the
same list object in the main thread and all threads it spawns. Because
of that, changes made in a thread are still noticed in the main thread
without resorting to extra locks.

"""
uses simple shared global data (not mutexes) to know when threads
are done in parent/main thread; threads share list but not its items,
assumes list won't move in memory once it has been created initially
"""
import _thread as thread
stdoutmutex = thread.allocate_lock()
exitmutexes = [False] * 10
def counter(myId, count):
for i in range(count):
stdoutmutex.acquire()
print('[%s] => %s' % (myId, i))
stdoutmutex.release()
exitmutexes[myId] = True  # signal main thread
for i in range(10):
thread.start_new_thread(counter, (i, 100))
while False in exitmutexes: pass
print('Main thread exiting.')

The output of this script is similar to the prior—10 threads
counting to 100 in parallel and synchronizing their prints along the
way. In fact, both of the last two counting thread scripts produce
roughly the same output as the original
thread_count.py
, albeit without
stdout
corruption and with larger counts and
different random ordering of output lines. The main difference is that
the main thread exits immediately after (and no sooner than!) the
spawned child
threads:

C:\...\PP4E\System\Threads>
python thread-count-wait2.py
...more deleted...
[4] => 98
[6] => 98
[8] => 98
[5] => 98
[0] => 99
[7] => 98
[9] => 98
[1] => 99
[3] => 99
[2] => 99
[4] => 99
[6] => 99
[8] => 99
[5] => 99
[7] => 99
[9] => 99
Main thread exiting.

Notice how the
main threads of both of the last two scripts fall into
busy-wait loops at the end, which might become significant performance
drains in tight applications. If so, simply add a
time.sleep
call in the wait loops to insert
a pause between end tests and to free up the CPU for other tasks: this
call pauses the calling thread only (in this case, the main one). You
might also try experimenting with adding sleep calls to the thread
function to simulate real work.

Passing in the lock to threaded functions as an
argument
instead of referencing it in the global
scope might be more coherent, too. When passed in, all threads
reference the same object, because they are all part of the same
process. Really, the process’s object memory is shared memory for
threads, regardless of how objects in that shared memory are
referenced (whether through global scope variables, passed argument
names, object attributes, or another way).

And while we’re at it, the
with
statement can be used to ensure thread
operations around a nested block of code, much like its use to ensure
file closure in the prior chapter. The thread lock’s
context
manager
acquires the lock on
with
statement entry and releases it on
statement exit regardless of exception outcomes. The net effect is to
save one line of code, but also to guarantee lock release when
exceptions are possible.
Example 5-10
adds all these coding
alternatives to our threaded counter script.

"""
passed in mutex object shared by all threads instead of globals;
use with context manager statement for auto acquire/release;
sleep calls added to avoid busy loops and simulate real work
"""
import _thread as thread, time
stdoutmutex = thread.allocate_lock()
numthreads  = 5
exitmutexes = [thread.allocate_lock() for i in range(numthreads)]
def counter(myId, count, mutex):                     # shared object passed in
for i in range(count):
time.sleep(1 / (myId+1))                     # diff fractions of second
with mutex:                                  # auto acquire/release: with
print('[%s] => %s' % (myId, i))
exitmutexes[myId].acquire()                      # global: signal main thread
for i in range(numthreads):
thread.start_new_thread(counter, (i, 5, stdoutmutex))
while not all(mutex.locked() for mutex in exitmutexes): time.sleep(0.25)
print('Main thread exiting.')

When run, the different sleep times per thread make them run
more independently:

C:\...\PP4E\System\Threads>
thread-count-wait3.py
[4] => 0
[3] => 0
[2] => 0
[4] => 1
[1] => 0
[3] => 1
[4] => 2
[2] => 1
[3] => 2
[4] => 3
[4] => 4
[0] => 0
[1] => 1
[2] => 2
[3] => 3
[3] => 4
[2] => 3
[1] => 2
[2] => 4
[0] => 1
[1] => 3
[1] => 4
[0] => 2
[0] => 3
[0] => 4
Main thread exiting.

Of course, threads are for much more than counting. We’ll put
shared global data to more practical use in
Adding a User Interface
, where it will serve as
completion signals from child processing threads transferring data
over a network to a main thread controlling a tkinter GUI display, and
again later in
Chapter 10
’s threadtools
and
Chapter 14
’s PyMailGUI to post
results of email operations to a GUI (watch for
Preview: GUIs and Threads
for more pointers on this
topic). Global data shared among threads also turns out to be the
basis of queues, which are discussed later in this chapter; each
thread gets or puts data using the same shared queue
object.

Example 5-11. PP4E\System\Threads\thread-classes.py

"""
thread class instances with state and run() for thread's action;
uses higher-level Java-like threading module object join method (not
mutexes or shared global vars) to know when threads are done in main
parent thread; see library manual for more details on threading;
"""
import threading
class Mythread(threading.Thread):              # subclass Thread object
def __init__(self, myId, count, mutex):
self.myId  = myId
self.count = count                     # per-thread state information
self.mutex = mutex                     # shared objects, not globals
threading.Thread.__init__(self)
def run(self):                             # run provides thread logic
for i in range(self.count):            # still sync stdout access
with self.mutex:
print('[%s] => %s' % (self.myId, i))
stdoutmutex = threading.Lock()                 # same as thread.allocate_lock()
threads = []
for i in range(10):
thread = Mythread(i, 100, stdoutmutex)     # make/start 10 threads
thread.start()                             # starts run method in a thread
threads.append(thread)
for thread in threads:
thread.join()                              # wait for thread exits
print('Main thread exiting.')

The
Thread
class
can also be used to start a simple function, or any
other type of callable object, without coding subclasses at all—if not
redefined, the
Thread
class’s
default
run
method simply calls
whatever you pass to its constructor’s
target
argument, with any provided arguments
passed to
args
(which defaults to
()
for none). This allows us to use
Thread
to run simple functions,
too, though this call form is not noticeably simpler than the basic
_thread
module. For instance, the
following code snippets sketch four different ways to spawn the same
sort of thread (see
four-threads*.py
in the examples tree; you
can run all four in the same script, but would have to also
synchronize prints to avoid overlap):

import threading, _thread
def action(i):
print(i ** 32)
# subclass with state
class Mythread(threading.Thread):
def __init__(self, i):
self.i = i
threading.Thread.__init__(self)
def run(self):                                        # redefine run for action
print(self.i ** 32)
Mythread(2).start()                                       # start invokes run()
# pass action in
thread = threading.Thread(target=(lambda: action(2)))     # run invokes target
thread.start()
# same but no lambda wrapper for state
threading.Thread(target=action, args=(2,)).start()        # callable plus its args
# basic thread module
_thread.start_new_thread(action, (2,))                    # all-function interface

As a rule of thumb, class-based threads may be better if your
threads require per-thread state, or can leverage any of OOP’s many
benefits in general. Your thread classes don’t necessarily have to
subclass
Thread
, though. In fact,
just as in the
_thread
module, the
thread’s target in
threading
may be
any type of callable object
. When combined with
techniques such as bound methods and nested scope references, the
choice between coding techniques becomes even less clear-cut:

# a non-thread class with state, OOP
class Power:
def __init__(self, i):
self.i = i
def action(self):
print(self.i ** 32)
obj = Power(2)
threading.Thread(target=obj.action).start()        # thread runs bound method
# nested scope to retain state
def action(i):
def power():
print(i ** 32)
return power
threading.Thread(target=action(2)).start()         # thread runs returned function
# both with basic thread module
_thread.start_new_thread(obj.action, ())           # thread runs a callable object
_thread.start_new_thread(action(2), ())

As usual, the threading APIs are as flexible as the Python
language itself.

Earlier, we saw
how print operations in threads need to be synchronized
with locks to avoid overlap, because the output stream is shared by
all threads. More formally, threads need to synchronize their changes
to any item that may be shared across thread in a process—both objects
and namespaces. Depending on a given program’s goals, this might
include:

For instance, even simple global variables can require
coordination if concurrent updates are possible, as in
Example 5-12
.

"prints different results on different runs on Windows 7"
import threading, time
count = 0
def adder():
global count
count = count + 1             # update a shared name in global scope
time.sleep(0.5)               # threads share object memory and global names
count = count + 1
threads = []
for i in range(100):
thread = threading.Thread(target=adder, args=())
thread.start()
threads.append(thread)
for thread in threads: thread.join()
print(count)

Here, 100 threads are spawned to update the same global scope
variable twice (with a sleep between updates to better interleave
their operations). When run on Windows 7 with Python 3.1, different
runs produce different results:

C:\...\PP4E\System\Threads>
thread-add-random.py
189
C:\...\PP4E\System\Threads>
thread-add-random.py
200
C:\...\PP4E\System\Threads>
thread-add-random.py
194
C:\...\PP4E\System\Threads>
thread-add-random.py
191

This happens because threads overlap arbitrarily in time:
statements, even the simple assignment statements like those here, are
not guaranteed to run to completion by themselves (that is, they are
not atomic). As one thread updates the global, it may be using the
partial result of another thread’s work in progress. The net effect is
this seemingly random behavior. To make this script work correctly, we
need to again use thread locks to synchronize the updates—when
Example 5-13
is run, it always prints
200 as expected.

"prints 200 each time, because shared resource access synchronized"
import threading, time
count = 0
def adder(addlock):                 # shared lock object passed in
global count
with addlock:
count = count + 1           # auto acquire/release around stmt
time.sleep(0.5)
with addlock:
count = count + 1           # only 1 thread updating at once
addlock = threading.Lock()
threads = []
for i in range(100):
thread = threading.Thread(target=adder, args=(addlock,))
thread.start()
threads.append(thread)
for thread in threads: thread.join()
print(count)

Although some basic operations in the Python language are atomic
and need not be synchronized, you’re probably better off doing so for
every potential concurrent update. Not only might the set of atomic
operations change over time, but the internal implementation of
threads in general can as well (and in fact, it may in Python 3.2, as
described ahead).

Of course, this is an artificial example (spawning 100 threads
to add twice isn’t exactly a real-world use case for threads!), but it
illustrates the issues that threads must address for any sort of
potentially concurrent updates to shared object or name. Luckily, for
many or most realistic applications, the
queue
module of the next section can make
thread synchronization an automatic artifact of program
structure.

Before we move ahead, I should point out that besides
Thread
and
Lock
, the
threading
module also includes
higher-level objects for synchronizing access to shared items (e.g.,
Semaphore
,
Condition
,
Event
)—many more, in fact, than we have
space to cover here; see the library manual for details. For more
examples of threads and forks in general, see the remainder this
chapter as well as the examples in the GUI and network scripting parts
of this book. We will thread GUIs, for instance, to avoid blocking
them, and we will thread and fork network servers to avoid denying
service to clients.

We’ll also explore the threading module’s approach to program
exits in the absence of
join
calls
in conjunction with queues—our next
topic.