Python | Simple Multithreaded HTTP Load Generator/Timer

by Corey Goldberg - 2007

This is a module user for generating concurrent requests to an HTTP server. Each thread makes HTTP GET requests to a single URL at the specified interval. Threads are added over a given rampup time if you want to generate increasing load. Response times are printed to STDOUT and the file 'results.csv'. This might be useful for performance benchmarking or load testing a web resource.

Module Download:

load_generator.py

#!/usr/bin/env python
#  Copyright (c) 2008 Corey Goldberg (corey@goldb.org)
#
#  Multithreaded HTTP Load Generator



# URL parameters
use_ssl = False  # HTTPS/SSL support
host = 'example.com'
path = '/'

# set load parameters
threads = 1
interval = 0
rampup = 0



import time
import os
import httplib
from threading import Thread, Lock



def main():
    manager = LoadManager()
    manager.msg = (host, path)
    manager.start(threads, interval, rampup)
    
    
    
class LoadManager:
    def __init__(self):
        self.msg = ('localhost', '/')  # default
        self.start_time = time.clock()        
    def start(self, threads=1, interval=0, rampup=0):
        try:
            os.remove('results.csv')
        except:
            pass
        for i in range(threads):
            spacing = (i * (float(rampup) / float(threads)))
            time.sleep(spacing)
            agent = LoadAgent(interval, self.msg, self.start_time)
            agent.start()
            print 'started thread # ' + str(i + 1)



class LoadAgent(Thread):
    def __init__(self, interval, msg, start_time):
        Thread.__init__(self)
        self.interval = interval
        self.msg = msg
        self.start_time = start_time
        
    def run(self):
        while True:
            start = time.clock()               
            try:
                self.send(self.msg)
            except:
                print 'failed request'
            end = time.clock()
            latency = (end - start)
            expire_time = (self.interval - latency)
            self.write_result(latency, self.start_time)   
            if expire_time > 0:
                time.sleep(expire_time)
                
    def send(self, msg):
        if use_ssl:
            conn = httplib.HTTPSConnection(msg[0])
        else:
            conn = httplib.HTTPConnection(msg[0])
        try:
            #conn.set_debuglevel(1)
            conn.request('GET', msg[1])
            resp = conn.getresponse().read()
        except:
            raise
        finally:
            conn.close()
        
    def write_result(self, latency, start_time):
        lock = Lock()
        lock.acquire()
        try:
            fh = open('results.csv', 'a')
            elapsed = time.clock() - start_time
            fh.write('%.3f,%.3f\n' % (elapsed, latency))
            fh.flush()
            fh.close()
        finally:
            lock.release()
        print '%.3f' % latency
        
        

if __name__ == '__main__':
    main()