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. This might be useful for performance benchmarking or load testing a web resource.

Module Download:

load_generator.py

Sample Usage:

#!/usr/bin/env python

from load_generator import LoadManager

lm = LoadManager()
lm.msg = ('www.example.com', '/')
lm.start(threads=5, interval=2, rampup=2)

Sample Output:

(times in secs)

0.161
0.140
0.150
0.181
0.150
0.320
0.150
0.160
0.140
...

load_generator Module:

#!/usr/bin/env python
#
#  load_generator.py
#  Simple multithreaded HTTP load generator/timer
#  Corey Goldberg (www.goldb.org), 2007


import time
import os
import httplib
from threading import Thread


class LoadManager:
    def __init__(self):
        self.thread_refs = []
        self.msg = ('localhost', '/') # default
    
    def stop(self):
        for thread in self.thread_refs:
            thread.stop()
            
    def start(self, threads=1, interval=0, rampup=1):
        for i in range(threads):
            spacing = (i * (float(rampup) / float(threads)))
            time.sleep(spacing)
            agent = LoadAgent(interval, self.msg)
            agent.setDaemon(True)
            agent.start()
            # print 'started thread # ' + str(i + 1)
            self.thread_refs.append(agent)
        

class LoadAgent(Thread):
    def __init__(self, interval, msg):
        Thread.__init__(self)
        self.running = True
        self.interval = interval
        self.msg = msg
        
    def stop(self):
        self.running = False
        
    def run(self):
        while self.running:
            start_time = time.time()
            if self.send(self.msg):
                end_time = time.time()
                raw_latency = end_time - start_time
                expire_time = (self.interval - raw_latency)
                latency = ('%.3f' % raw_latency)
                print latency
            else:
                raw_latency = 0
                expire_time = (self.interval - raw_latency)
            if expire_time > 0:
                time.sleep(expire_time)
                
    def send(self, msg):
        conn = httplib.HTTPConnection(msg[0])
        try:
            conn.request('GET', msg[1])
            body = conn.getresponse().read()
            return True
        except:
            print 'failed request'
            return False



def main():
    # sample usage
    manager = LoadManager()
    manager.msg = ('www.example.com', '/')
    manager.start(threads=5, interval=2, rampup=2)


if __name__ == '__main__':
    main()