Programming Python (29 page)

Example 5-14. PP4E\System\Threads\queuetest.py

"producer and consumer threads communicating with a shared queue"
numconsumers = 2                  # how many consumers to start
numproducers = 4                  # how many producers to start
nummessages  = 4                  # messages per producer to put
import _thread as thread, queue, time
safeprint = thread.allocate_lock()    # else prints may overlap
dataQueue = queue.Queue()             # shared global, infinite size
def producer(idnum):
for msgnum in range(nummessages):
time.sleep(idnum)
dataQueue.put('[producer id=%d, count=%d]' % (idnum, msgnum))
def consumer(idnum):
while True:
time.sleep(0.1)
try:
data = dataQueue.get(block=False)
except queue.Empty:
pass
else:
with safeprint:
print('consumer', idnum, 'got =>', data)
if __name__ == '__main__':
for i in range(numconsumers):
thread.start_new_thread(consumer, (i,))
for i in range(numproducers):
thread.start_new_thread(producer, (i,))
time.sleep(((numproducers-1) * nummessages) + 1)
print('Main thread exit.')

Notice how the queue is
assigned to a global variable; because of that, it is
shared by all of the spawned threads (all of them run in the same
process and in the same global scope). Since these threads change an
object instead of a variable name, it would work just as well to pass
the queue object in to the threaded functions as an argument—the queue
is a shared object in memory, regardless of how it is referenced (see
queuetest2.py
in the examples tree for
a full version that does this):

dataQueue = queue.Queue()             # shared object, infinite size
def producer(idnum, dataqueue):
for msgnum in range(nummessages):
time.sleep(idnum)
dataqueue.put('[producer id=%d, count=%d]' % (idnum, msgnum))
def consumer(idnum, dataqueue): ...
if __name__ == '__main__':
for i in range(numproducers):
thread.start_new_thread(producer, (i, dataQueue))
for i in range(numproducers):
thread.start_new_thread(producer, (i, dataQueue))

Also notice how this
script exits when the main thread does, even though
consumer threads are still running in their infinite loops. This works
fine on Windows (and most other platforms)—with the basic
_thread
module, the program ends silently
when the main thread does. This is why we’ve had to sleep in some
examples to give threads time to do their work, but is also why we do
not need to be concerned about exiting while consumer threads are
still running here.

In the alternative
threading
module, though, the program will not exit if any spawned threads are
running, unless they are set to be
daemon
threads. Specifically, the entire program exits when only daemon
threads are left. Threads inherit a default initial daemonic value
from the thread that creates them. The initial thread of a Python
program is considered not daemonic, though alien threads created
outside this module’s control are considered daemonic (including some
threads created in C code). To override inherited defaults, a thread
object’s
daemon
flag can be set
manually. In other words, nondaemon threads prevent program exit, and
programs by default do not exit until all
threading
-managed threads finish.

This is either a feature or nonfeature, depending on your
program—it allows spawned worker threads to finish their tasks in the
absence of
join
calls or sleeps,
but it can prevent programs like the one in
Example 5-14
from shutting down when
they wish. To make this example work with
threading
, use the following alternative
code (see
queuetest3.py
in the examples tree for
a complete version of this, as well as
thread-count-threading.py
, also in the
tree, for a case where this refusal to exit can come in handy):

import threading, queue, time
def producer(idnum, dataqueue): ...
def consumer(idnum, dataqueue): ...
if __name__ == '__main__':
for i in range(numconsumers):
thread = threading.Thread(target=consumer, args=(i, dataQueue))
thread.daemon = True  # else cannot exit!
thread.start()
waitfor = []
for i in range(numproducers):
thread = threading.Thread(target=producer, args=(i, dataQueue))
waitfor.append(thread)
thread.start()
for thread in waitfor: thread.join()    # or time.sleep() long enough here
print('Main thread exit.')

We’ll revisit the daemons and exits issue in
Chapter 10
while studying GUIs; as we’ll see,
it’s no different in that context, except that the main thread is
usually the GUI itself.

Now, as coded in
Example 5-14
, the
following is the output of this example when run on my
Windows machine. Notice that even though the queue automatically
coordinates the communication of data between the threads, this script
still must use a lock to manually synchronize access to the standard
output stream; queues synchronize data passing, but some programs may
still need to use locks for other purposes. As in prior examples, if
the
safeprint
lock is not used, the
printed lines from one consumer may be intermixed with those of
another. It is not impossible that a consumer may be paused in the
middle of a print operation:

C:\...\PP4E\System\Threads>
queuetest.py
consumer 1 got => [producer id=0, count=0]
consumer 0 got => [producer id=0, count=1]
consumer 1 got => [producer id=0, count=2]
consumer 0 got => [producer id=0, count=3]
consumer 1 got => [producer id=1, count=0]
consumer 1 got => [producer id=2, count=0]
consumer 0 got => [producer id=1, count=1]
consumer 1 got => [producer id=3, count=0]
consumer 0 got => [producer id=1, count=2]
consumer 1 got => [producer id=2, count=1]
consumer 1 got => [producer id=1, count=3]
consumer 1 got => [producer id=3, count=1]
consumer 0 got => [producer id=2, count=2]
consumer 1 got => [producer id=2, count=3]
consumer 1 got => [producer id=3, count=2]
consumer 1 got => [producer id=3, count=3]
Main thread exit.

Try adjusting the parameters at the top of this script to
experiment with different scenarios. A single consumer, for instance,
would simulate a GUI’s main thread. Here is the output of a
single-consumer run—producers still add to the queue in fairly random
fashion, because threads run in parallel with each other and with the
consumer:

C:\...\PP4E\System\Threads>
queuetest.py
consumer 0 got => [producer id=0, count=0]
consumer 0 got => [producer id=0, count=1]
consumer 0 got => [producer id=0, count=2]
consumer 0 got => [producer id=0, count=3]
consumer 0 got => [producer id=1, count=0]
consumer 0 got => [producer id=2, count=0]
consumer 0 got => [producer id=1, count=1]
consumer 0 got => [producer id=3, count=0]
consumer 0 got => [producer id=1, count=2]
consumer 0 got => [producer id=2, count=1]
consumer 0 got => [producer id=1, count=3]
consumer 0 got => [producer id=3, count=1]
consumer 0 got => [producer id=2, count=2]
consumer 0 got => [producer id=2, count=3]
consumer 0 got => [producer id=3, count=2]
consumer 0 got => [producer id=3, count=3]
Main thread exit.

In addition to the basics used in our script, queues may be
fixed or infinite in size, and get and put calls may or may not block;
see the Python library manual for more details on queue interface
options. Since we just simulated a typical GUI structure, though,
let’s explore the notion a bit
further.

• The
  main thread
  handles all GUI updates
  and runs a timer-based loop that wakes up periodically to check for
  new data on the queue to be displayed on-screen. In Python’s tkinter
  toolkit, for instance, the widget
  after(
msecs
,
func
,
*args
)
method can be used to schedule
  queue-check events. Because such events are dispatched by the GUI’s
  event processor, all GUI updates occur only in this main thread (and
  often must, due to the lack of thread safety in GUI
  toolkits).
  

• The
  child threads
  don’t do anything
  GUI-related. They just produce data and put it on the queue to be
  picked up by the main thread. Alternatively, child threads can place
  a callback function on the queue, to be picked up and run by the
  main thread. It’s not generally sufficient to simply pass in a GUI
  update callback function from the main thread to the child thread
  and run it from there; the function in shared memory will still be
  executed in the child thread, and potentially in parallel with other
  threads.
  

Thread Timers versus GUI Timers

Interestingly, the
threading
module
exports a general timer function, which, like the
tkinter widget
after
method, can be
used to run another function after a timer has expired:

Timer(N.M, somefunc).start()   # after N.M seconds run somefunc

Timer
objects have a
start()
method to set the timer as well as a
cancel()
method to cancel the
scheduled event, and they implement the wait state in a spawned
thread. For example, the following prints a message after 5.5
seconds:

>>>
import sys
>>>
from threading import Timer
>>>
t = Timer(5.5, lambda: print('Spam!'))
# spawned thread
>>>
t.start()
>>>
Spam!

This may be useful in a variety of contexts, but it doesn’t
quite apply to GUIs: because the time-delayed function call is run in
a spawned thread, not in the main GUI thread, it should not generally
perform GUI updates. Because the tkinter
after
method is run from the main thread’s
event processing loop instead, it runs in the main GUI thread and can
freely update the GUI.

As a preview, for instance, the following displays a pop-up
message window in 5.5 seconds in the main thread of a tkinter GUI (you
might also have to run
win.mainloop()
in some interfaces):

>>>
from tkinter import Tk
>>>
from tkinter.messagebox import showinfo
>>>
win = Tk()
>>>
win.after(5500, lambda: showinfo('Popup', 'Spam!'))

The last call here schedules the function to be run once in the
main GUI thread, but it does not pause the caller during the wait, and
so does not block the GUI. It’s equivalent to this simpler
form:

>>>
win.after(5500, showinfo, 'Popup', 'Spam')

Stay tuned for much more on tkinter in the next part of this
book, and watch for the full story on its
after
timer events in
Chapter 9
and the roles of threads in
GUIs in
Chapter 10
.