Let’s Talk

Threaded data collection with Python, including examples

On today’s Internet 2.0 there are all sorts of data feeds available for consumption. From APIs to RSS feeds, it seems like nearly every site has a machine-readable output. There are many reasons why you’d want to collect this information, which I won’t go in to, so in this post I’m going to walk you through an application which consumes RSS feeds. I’ll be using the Python scripting language, and I’ll show you an evolution of the ways to go about the task:

Application introduction

Our application is going to work like this:

Database manipulation and RSS feed parsing are outside the scope of this tutorial, so we’ll start off by defining some empty functions that handle all this:

"def get_feed_list():
""" Returns a list of tuples: (id, feed_url) """
pass
def get_feed_contents(feed_url):
""" Gets feed over HTTP, returns RSS XML """
pass
def parse_feed(feed_rss):
""" Parses the feed and returns a list of items """
pass
def store_feed_items(id, items):
""" Takes a feed_id and a list of items and stored them in the DB """
pass

We’re going to have all these in a module called “functions”, which can just be a file called functions.py in the same directory ( < python3.0)

Implementation 1: Single-threaded

This is the way most people would do it at first. So simple, I’ll just post the sample code:

import functions
for id, feed_url in get_feed_list():
rss = functions.get_feed_contents(feed_url)
items = functions.parse_feed(rss)
functions.store_feed_items(id, items)

Pretty simple huh? But there are fundamental problems. Feeds are usually slow, meaning that your program will spend a lot of time waiting for feeds to come in before you can parse them. You program will also be spending time parsing feeds when it could be getting them from the internet as well. Consequently this program will be as slow as molasses. It’s like eating a bowl of peas one at a time – you’d rather just shovel them in wouldn’t you? Enter: threading.

Implementation 2

So we reckon: “If we use threads, this will make things faster?” Answer: Yes. However, there are quite a few ways of doing this. We’ll start off with this:

import threading, functions, time
def thread(id, feed_url):
rss = functions.get_feed_contents(feed_url)
items = functions.parse_feed(rss)
functions.store_feed_items(id, items)

for id, feed_url in get_feed_list():
t = threading.Thread(target=thread, kwargs={"id":id, "feed_url":feed_url})
t.start()
while threading.activeCount() > 1: time.sleep(1)

Problem: This is just going to create as many threads as there are feed items immediately and then wait for them to finish. This has the following issues:

So what do we do? Well, let’s set a limit on the number of concurrent threads:

import threading, functions, time
THREAD_LIMIT = 20
def thread(id, feed_url):
rss = functions.get_feed_contents(feed_url)
items = functions.parse_feed(rss)
functions.store_feed_items(id, items)

for id, feed_url in get_feed_list():
while threading.activeCount() > THREAD_LIMIT:
time.sleep(1)
t = threading.Thread(target=thread, kwargs={"id":id, "feed_url":feed_url})
t.start()
while threading.activeCount() > 1: time.sleep(1)

Spot the difference? We have another while loop right in the for loop. This is going to make our main thread code hang there while there are other threads still running.

There’s another problem though, and that’s with the model. In this mode, we’re continually creating new threads that live for a short time, then exit. This isn’t efficient. It would be much better to create a pool of threads which we can then re-use. Let’s kick this up a notch.

Implementaton 3: Thread pool, and a Queue

So in this version we’re going to do a few new things:

  1. Use a Queue object and populate it with the list of urls.
  2. Spawns a number of threads that will read items off this Queue.
  3. The threads will process the data and store it.

I’ll start off with the sample code then walk you through it:

import threading, functions, time, Queue
THREAD_LIMIT = 50
jobs = Queue.Queue(0) # arg1 means "No item limit"

def thread():
while True: # forever
try:
id, feed_url = jobs.get(False) # arg1 means "Don't wait for items to appear"
except Queue.Empty:
# Nothing left to do, time to die
return
rss = functions.get_feed_contents(feed_url)
items = functions.parse_feed(rss)
functions.store_feed_items(id, items)

for info in get_feed_list():
Queue.put(info)

for n in xrange(THREAD_LIMIT):
t = threading.Thread(target=thread)
t.start()
while threading.activeCount() > 1: time.sleep(1) # Wait to finish

Lines to note:

Our threads run in a loop, performing work in lines 12-14, until there is no more work, then they exit. This model will work just fine for the majority of people, however, there are (still) problems. They are:

  1. Opening 50 threads that write to a database will usually mean 50 database connections, or lots of locking. Either way, this is bad. Your data-getting threads don’t want to be sat around waiting on a DB lock when they could be getting more information. Worse, you don’t want to exceed your connection limit to the databse.
  2. In this model, you have one master thread doing nothing and 50 threads doing I/O and work. This isn’t a great idea in this application. Ideally you should only use threading (in Python) when you need to either:
    1. Wait for I/O
    2. Truly perform more than one concurrent task

Problem #2 isn’t so serious, but it would be better to have more control over the heavy lifting. However problem #1 definitely needs addressing. The solution is to shift all processing inline to the master thread, which takes care of all the processing.

Implementation 4: 1 worker, many runners

Here’s the code:

import threading, functions, time, Queue
THREAD_LIMIT = 50
jobs = Queue.Queue(0) # arg1 means "No item limit"
rss_to_process = Queue.Queue(THREAD_LIMIT) # We set a limit on this, I'll
# explain later
def thread():
while True: # forever
try:
id, feed_url = jobs.get(False) # arg1 means "Don't wait for items
# to appear"
except Queue.Empty:
# Nothing left to do, time to die
return
rss = functions.get_feed_contents(feed_url)
rss_to_process.put((id, rss), True) # This will block if our processing
# queue is too large

for info in get_feed_list(): # Load them up
jobs.put(info)

for n in xrange(THREAD_LIMIT): # Unleash the hounds
t = threading.Thread(target=thread)
t.start()
while threading.activeCount() > 1 or not rss_to_process.empty():
# That condition means we want to do this loop if there are threads
# running OR there's stuff to process
try:
id, rss = rss_to_process.get(False, 1) # Wait for up to a second for a
# result
except Queue.Empty:
continue
items = functions.parse_feed(rss)
functions.store_feed_items(id, items)

Notes:

…and there you have it, a fully fledged multithreaded data collector. Not bad for a few hours’ work. It’s not finished though, as there are plenty of things you’d want to add to it. For example:

Comments and suggestions welcome, and you can feel free to use the contact form.

Disclaimer

The code contained in this tutorial is not guaranteed to work, or even compile. It has never been executed and is not tested for syntax errors or other bugs. However it is semantically accurate and is provided “as-is”. I am not responsible for any loss of business, crashes, errors, marital crisis or nuclear wars caused as a result of using this code. Copy and paste at your own risk.