Bandwidth testing is easy with Python's built-in web access, HTML parsing, and threading modules.
When Fedora Core 2 was released, I wanted to find which download mirror would be fastest when I tried downloading the CD images. It only took about an hour to whip up this mirror bandwidth tester.
This script demonstrates web download, HTML parsing, and some interesting threading issues.
Python, 151 lines
from HTMLParser import HTMLParser
MIRRORS_URL = 'http://fedora.redhat.com/download/mirrors.html'
MAX_THREADS = 50
HTTP_TIMEOUT = 60.0 # Max. seconds to wait for a response
'''Subclass of the HTMLParser object. Records the HREF attributes
of anchor tags if the scheme is 'http' and the anchor occurs in
the 'content' section of the page.'''
self.mirrorLinks = 
# True if we're currently in the 'content' section
self.isInMirrors = False
def handle_comment(self, data):
# The comments have spaces before and after, but don't count
# on that.
data = data.strip()
if 'content BEGIN' == data:
self.isInMirrors = True
elif 'content END' == data:
self.isInMirrors = False
def handle_starttag(self, tag, attrs):
attrs = dict(attrs) # Convert from tuple of tuples to dict
if 'a' == tag and 'http' == urllib.splittype(attrs['href']):
# Record the start time, so we can print a nice message at the end
processStartTime = time.time()
# Create the parser, get the 'mirrors' page from Redhat,
# and extract the URLs
print "Getting mirrors list...",
parser = UrlFinder()
print len(parser.mirrorLinks), "mirrors found."
numThreads = min(MAX_THREADS, len(parser.mirrorLinks))
print "Testing bandwidth with", numThreads, "threads..."
# Build a queue to feed the worker threads
workQueue = Queue.Queue()
for url in parser.mirrorLinks:
def TestUrl(workQueue, resultQueue):
''' Worker thread procedure. Test how long it takes to return the
mirror index page, and stuff the results into resultQueue.'''
def SubthreadProc(url, result):
''' Subthread procedure. Actually get the mirror index page
in a subthread, so that we can time out using join rather than
wait for a very slow server. Passing in a list for result
lets us simulate pass-by-reference, since callers cannot get
the return code from a Python thread.'''
startTime = time.time()
data = urllib.urlopen(url).read()
# Could be a socket error or an HTTP error--either way, we
# don't care--it's a failure to us.
elapsed = int((time.time() - startTime) * 1000)
# Contine pulling data from the work queue until it's empty
url = workQueue.get(0)
# work queue is empty--exit the thread proc.
# Create a single subthread to do the actual work
result = 
subThread = threading.Thread(target=SubthreadProc, args=(url, result))
# Daemonize the subthread so that even if a few are hanging
# around when the process is done, the process will exit.
# Run the subthread and wait for it to finish, or time out
if  == result:
# Subthread hasn't give a result yet. Consider it timed out.
elif -1 == result:
# Subthread returned an error from geturl.
# Subthread returned a time. Store it.
workers = 
resultQueue = Queue.Queue()
# Create worker threads to load-balance the retrieval
for threadNum in range(0, numThreads):
# Wait for all the workers to finish
for w in workers:
# Separate the successes from failures
timings = 
failures = 
while not resultQueue.empty():
url, result = resultQueue.get(0)
if isinstance(result, str):
# Sort by increasing time or result string
# Print the results
print "\nMirrors (ordered fastest to slowest)"
for result, url in timings:
print "%7d %s" % (result, url)
for result, url in failures:
print "%7s %s" % (result, url)
print "\nProcess completed in ", time.time() - processStartTime, " seconds."
The concept of the script is straightforward: read the mirrors page from RedHat's web site, make a list of all the mirrors, test how long it takes to download from each, and present a sorted list of the results.
The first task, reading and parsing the RedHat mirrors list, is handled with the urllib and HTMLParser modules, respectively. I chose HTMLParser over the more comprehensive parser in sgmllib because it's a bit less work to override the default parser for simple tasks. After the parser sees the content comment in the HTML source, it starts recording any tags that have a scheme of 'http'; it stops recording after it sees the end of the content comment. Currently, it happens that there aren't any absolute URLs on the mirror page outside the content block, but I didn't want to rely on that fact.
To test the bandwidth of each mirror site, I simply test how long it takes to download the index page of the mirror. This is not a perfect test, but it gives reasonably good results without depending on knowledge of the site structure.
The bandwidth test demonstrates a few important paradigms when dealing with multithreading, either in Python or other languages:
- Let the underlying libraries do as much work as possible.
- Isolate your threads from the rest of the program.
The main thread creates a work queue of URLs to be tested and a result queue for retrieving results, then starts a number of threads to do the work and waits for those threads to exit. Because the Queue class is a threadsafe container, Python guarantees that no two threads will ever get the same work unit, and the storing of results by multiple threads will never leave the queue in a bad state.
Initially, each worker thread downloaded the mirror index page directly, but this caused the process to run for long amounts of time (over three minutes) when some sites were heavily loaded. To avoid this, I defined a maximum time to attempt downloading, and made each worker thread spawn a new daemon thread to do the download. The worker thread can use Thread.join() to wait on the subthread with a timeout; timeouts are counted as failures. Note that I pass an empty list to the subthread to collect the results. Threads in Python don't have a convenient way to return a status code back to the caller; by passing a mutable object like a list, the subthread can append value to the list to indicate a result. When the join() on the subthread completes, the worker thread can tell that it timed out if the list it passed in is empty.
The worker threads put the results for each URL into a results queue. For successful tests, they put a tuple of the URL and the time it took to download; for unsuccessful results, they put a tuple of the URL and a string describing the type of failure. When the main thread has detected that all worker threads have exited, it separates successes from failures, sorts the two lists, and prints them in aligned columns.
Note that the script could be written without the second-level threads. Using them helps isolate the failure-prone download from the more reliable worker thread pool, at the cost of a few more ephemeral threads, and provides a good demonstration of how and when to use daemon threads to keep a script from hanging indefinitely at shutdown.
This script is useful to tell which mirrors are most heavily loaded, but it has shortcomings. Some HTTP-based mirrors are actually redirects to FTP mirrors, and some seem to apply different bandwidth throttles to index pages and ISO downloads. Additionally, the script can't tell which of the mirrors actually have up-to-date files; this can't easily be fixed without having knowledge of each mirror site, since mirror sites differ in their directory structure. But this at least gives the would-be upgrader an idea of where to look.