Welcome, guest | Sign In | My Account | Store | Cart

This is intended as a drop-in replacement for the ThreadingMixIn class in module SocketServer of the standard lib. Instead of spawning a new thread for each request, requests are processed by of pool of reusable threads.

Python, 77 lines
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
from SocketServer import ThreadingMixIn
from Queue import Queue
import threading, socket


class ThreadPoolMixIn(ThreadingMixIn):
    '''
    use a thread pool instead of a new thread on every request
    '''
    numThreads = 10
    allow_reuse_address = True  # seems to fix socket.error on server restart

    def serve_forever(self):
        '''
        Handle one request at a time until doomsday.
        '''
        # set up the threadpool
        self.requests = Queue(self.numThreads)

        for x in range(self.numThreads):
            t = threading.Thread(target = self.process_request_thread)
            t.setDaemon(1)
            t.start()

        # server main loop
        while True:
            self.handle_request()
            
        self.server_close()

    
    def process_request_thread(self):
        '''
        obtain request from queue instead of directly from server socket
        '''
        while True:
            ThreadingMixIn.process_request_thread(self, *self.requests.get())

    
    def handle_request(self):
        '''
        simply collect requests and put them on the queue for the workers.
        '''
        try:
            request, client_address = self.get_request()
        except socket.error:
            return
        if self.verify_request(request, client_address):
            self.requests.put((request, client_address))



if __name__ == '__main__':
    from SimpleHTTPServer import SimpleHTTPRequestHandler
    from SocketServer import TCPServer
    
    class ThreadedServer(ThreadPoolMixIn, TCPServer):
        pass

    def test(HandlerClass = SimpleHTTPRequestHandler,
            ServerClass = ThreadedServer, 
            protocol="HTTP/1.0"):
        '''
        Test: Run an HTTP server on port 8002
        '''

        port = 8002
        server_address = ('', port)

        HandlerClass.protocol_version = protocol
        httpd = ServerClass(server_address, HandlerClass)

        sa = httpd.socket.getsockname()
        print "Serving HTTP on", sa[0], "port", sa[1], "..."
        httpd.serve_forever()

    test()

A constant thread pool is preferable to the instantiation of one thread for every request. While there are several thread pool-based servers out there, they are usually tied to one or the other specialization. This recipe aims to provide a drop-in replacement for the ThreadingMixIn provided by module SocketServer and hopefully will facilitate adaption to various protocols and applications.

By default, this class creates 10 request worker threads. The server main loop, which does nothing more than harvest requests from the server socket, executes in the main thread. Synchronization between workers is provided by Queue.Queue. In contrast to the server classes provided in module SocketServer, this server is stoppable with Ctrl-C.

On Debian, performance is good; not tested on other platforms.

2 comments

Rafael Zanella 15 years, 9 months ago  # | flag

I've tweaked the recipe a bit to allow the main thread to quit the serve_forever():

class _ThreadPoolMixIn(SocketServer.ThreadingMixIn): ''' Uses a thread from a thread pool instead of instantiate a new one on every request. ''' numThreads = None

def __init__(self, numThreads): ''' Sets up the threadPool and "fills" it with the threads. '''

self.numThreads = numThreads

self.requests = Queue.Queue(self.numThreads)

for n in range(self.numThreads):
  t = threading.Thread(target = self.process_request_thread)
  t.setDaemon(1)
  t.start()

def process_request(self, request, client_address): ''' Simply collect requests and put them on the queue for the workers. ''' self.requests.put((request, client_address))

def process_request_thread(self): ''' Obtains request and client_address from the queue instead of directly from a call '''

# The thread starts and stays on this loop.
# The method call hangs waiting until something is inserted into self.requests
#  and .get() unblocks
while True:
  SocketServer.ThreadingMixIn.process_request_thread(self, *self.requests.get())
  # http://docs.python.org/tut/node6.html#SECTION006740000000000000000

class ThreadingPoolTCPServer(_ThreadPoolMixIn, SocketServer.TCPServer): """Calls the __init__ from both super."""

def __init__(self, server_address, RequestHandlerClass, numThreads,\ bind_and_activate=True): _ThreadPoolMixIn.__init__(self, numThreads)

SocketServer.TCPServer.__init__(self, server_address, RequestHandlerClass)

class ThreadingPoolUDPServer(_ThreadPoolMixIn, SocketServer.UDPServer): """Calls the __init__ from both super."""

def __init__(self, server_address, RequestHandlerClass, numThreads,\ bind_and_activate=True): _ThreadPoolMixIn.__init__(self, numThreads)

SocketServer.UDPServer.__init__(self, server_address, RequestHandlerClass)
GunGunGun 8 years, 10 months ago  # | flag

Hi Michael Palmer, I like your way reusing threads, can I port your script to Python 3 for personal use or can you add a Python version ?