Practical threaded programming with Python

Threaded programming in Python can be done with a minimal amount of complexity by combining threads with Queues. This article explores using threads and queues together to create simple yet effective patterns for solving problems that require concurrency.       Introduction With Python, there is no shortage of options for concurrency, the standard library includes support for threading, processes, and asynchronous I/O. In many cases Python has removed much of the difficulty in using these various methods of concurrency by creating high-level modules such as asynchronous, threading, and subprocess. Outside of the standard library, there are third solutions such as twisted, stackless, and the processing module, to name a few. This article focuses exclusively on threading in Python, using practicle examples. There are many great resources online that document the threading API, but this article attempts to provide practicle examples of common threading usage patterns. It is important to first define the differences between processes and threads. Threads are different than processes in that they share state, memory, and resources. This simple difference is both a strength and a weakness for threads. On one hand, threads are lightweight and easy to communicate with, but on the other hand, they bring up a whole host of problems including deadlocks, race conditions, and sheer complexity. Fortunately, due to both the GIL and the queuing module, threading in Python is much less complex to implement than in other languages. Hello Python threads To follow along, I assume that you have Python 2.5 or greater installed, as many examples will be using newer features of the Python language that only appear in at least Python2.5. To get started with threads in Python, we will start with a simple "Hello World" example: hello_threads_example   import threading import datetime class ThreadClass(threading.Thread): def run(self): now = datetime.datetime.now() print "%s says Hello World at time: %s" % (self.getName(), now) for i in range(2): t = ThreadClass() t.start()   If you run this example, you get the following output: # python hello_threads.py Thread-1 says Hello World at time: 2008-05-13 13:22:50.252069 Thread-2 says Hello World at time: 2008-05-13 13:22:50.252576   Looking at this output, you can see that you received a Hello World statement from two threads with date stamps. If you look at the actual code, there are two import statements; one imports the datetime module and the other imports the threading module. The class ThreadClass inherits from threading.Thread and because of this, you need to define a run method that executes the code you run inside of the thread. The only thing of importance to note in the run method is that self.getName() is a method that will identify the name of the thread. The last three lines of code actually call the class and start the threads. If you notice, t.start() is what actually starts the threads. The threading module was designed with inheritance in mind, and was actually built on top of a lower-level thread module. For most situations, it would be considered a best practice to inherit from threading.Thread, as it creates a very natural API for threaded programming. Using queues with threads As I referred to earlier, threading can be complicated when threads need to share data or resources. The threading module does provide many synchronization primatives, including semaphores, condition variables, events, and locks. While these options exist, it is considered a best practice to instead concentrate on using queues. Queues are much easier to deal with, and make threaded programming considerably safer, as they effectively funnel all access to a resource to a single thread, and allow a cleaner and more readible design pattern. In the next example, you will first create a program that will serially, or one after the other, grab a URL of a website, and print out the first 1024 bytes of the page. This is a classic example of something that could be done quicker using threads. First, let's use the urllib2 module to grab these pages one at a time, and time the code: URL fetch serial   import urllib2 import time hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com", "http://ibm.com", "http://apple.com"] start = time.time() #grabs urls of hosts and prints first 1024 bytes of page for host in hosts: url = urllib2.urlopen(host) print url.read(1024) print "Elapsed Time: %s" % (time.time() - start)   When you run this, you get a lot of output to standard out, as the pages are being partially printed. But you get this at the finish: Elapsed Time: 2.40353488922   Let's look a little at this code. You import only two modules. First, the urllib2 module is what does the heavy lifting and grabs the Web pages. Second, you create a start time value by calling time.time(), and then call it again and subtract the initial value to determine how long the program takes to execute. Finally, in looking at the speed of the program, the result of "Two and a half seconds" isn't horrible, but if you had hundreds of Web pages to retrieve, it would take approximately 50 seconds, given the current average. Look at how creating a threaded version speeds things up: URL fetch threaded   #!/usr/bin/env python import Queue import threading import urllib2 import time hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com", "http://ibm.com", "http://apple.com"] queue = Queue.Queue() class ThreadUrl(threading.Thread): """Threaded Url Grab""" def __init__(self, queue): threading.Thread.__init__(self) self.queue = queue def run(self): while True: #grabs host from queue host = self.queue.get() #grabs urls of hosts and prints first 1024 bytes of page url = urllib2.urlopen(host) print url.read(1024) #signals to queue job is done self.queue.task_done() start = time.time() def main(): #spawn a pool of threads, and pass them queue instance for i in range(5): t = ThreadUrl(queue) t.setDaemon(True) t.start() #populate queue with data for host in hosts: queue.put(host) #wait on the queue until everything has been processed queue.join() main() print "Elapsed Time: %s" % (time.time() - start)   This example has a bit more code to explain, but it isn't that much more complicated than the first threading example, thanks to the use of the queuing module. This pattern is a very common and recommended way to use threads with Python. The steps are described as follows: Create an instance of Queue.Queue() and then populate it with data. Pass that instance of populated data into the threading class that you created from inheriting from threading.Thread. Spawn a pool of daemon threads. Pull one item out of the queue at a time, and use that data inside of the thread, the run method, to do the work. After the work is done, send a signal to the queue with queue.task_done() that the task has been completed. Join on the queue, which really means to wait until the queue is empty, and then exit the main program. Just a note about this pattern: By setting daemonic threads to true, it allows the main thread, or program, to exit if only daemonic threads are alive. This creates a simple way to control the flow of the program, because you can then join on the queue, or wait until the queue is empty, before exiting. The exact process is best described in the documentation for the queue module, as seen in the Resources: join() "Blocks until all items in the queue have been gotten and processed. The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer thread calls task_done() to indicate that the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks. Working with multiple queues Because the pattern demonstrated above is so effective, it is relatively simple to extend it by chaining additional thread pools with queues. In the above example, you simply printed out the first portion of a Web page. This next example instead returns the whole Web page that each thread grabs, and then places it into another queue. Then set up another pool of threads that join on the second queue, and then do work on the Web page. The work performed in this example involves parsing the Web page using a third-party Python module called Beautiful Soup. Using just a couple of lines of code, with this module, you will extract the title tag and print it out for each page you visit. Multiple queues data mining websites   import Queue import threading import urllib2 import time from BeautifulSoup import BeautifulSoup hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com", "http://ibm.com", "http://apple.com"] queue = Queue.Queue() out_queue = Queue.Queue() class ThreadUrl(threading.Thread): """Threaded Url Grab""" def __init__(self, queue, out_queue): threading.Thread.__init__(self) self.queue = queue self.out_queue = out_queue def run(self): while True: #grabs host from queue host = self.queue.get() #grabs urls of hosts and then grabs chunk of webpage url = urllib2.urlopen(host) chunk = url.read() #place chunk into out queue self.out_queue.put(chunk) #signals to queue job is done self.queue.task_done() class DatamineThread(threading.Thread): """Threaded Url Grab""" def __init__(self, out_queue): threading.Thread.__init__(self) self.out_queue = out_queue def run(self): while True: #grabs host from queue chunk = self.out_queue.get() #parse the chunk soup = BeautifulSoup(chunk) print soup.findAll(['title']) #signals to queue job is done self.out_queue.task_done() start = time.time() def main(): #spawn a pool of threads, and pass them queue instance for i in range(5): t = ThreadUrl(queue, out_queue) t.setDaemon(True) t.start() #populate queue with data for host in hosts: queue.put(host) for i in range(5): dt = DatamineThread(out_queue) dt.setDaemon(True) dt.start() #wait on the queue until everything has been processed queue.join() out_queue.join() main() print "Elapsed Time: %s" % (time.time() - start)   If you run this version of the script, you get the following output: # python url_fetch_threaded_part2.py [<title>Google</title>] [<title>Yahoo!</title>] [<title>Apple</title>] [<title>IBM United States</title>] [<title>Amazon.com: Online Shopping for Electronics, Apparel, Computers, Books, DVDs & more</title>] Elapsed Time: 3.75387597084    In looking at the code, you can see that we added another instance of a queue, and then passed that queue into the first thread pool class, ThreadURL. Next , you almost copy the exact same structure for the next thread pool class, DatamineThread. In the run method of this class, grab the Web page, chunk, from off of the queue in each thread, and then process this chunk with Beautiful Soup. In this case, you use Beautiful Soup to simply extract the title tags from each page and print them out. This example could quite easily be turned into something more useful, as you have the core for a basic search engine or data mining tool. One idea is to extract the links from each page using Beautiful Soup and then follow them.   Summary This article explored threads in Python and demonstrated the best practice of using queues to allieviate complexity and subtle errors, and to promote readable code. While this basic pattern is relatively simple, it can be used to solve a wide number of problems by chaining queues and thread pools together. In the final section, you began to explore creating a more complex processing pipeline that can serve as a model for future projects. There are quite a few excellent resources on both concurrency in general and threads in the Resources section. In closing, it is important to point out that threads are not the solution to every problem, and that processes can be quite suitable for many situations. The standard library subprocess module in particular can be much simpler to deal with if you only require forking many processes and listening for a response. Please consult the Resources section for the official documentation on this.