Thread pool mixin class for use with SocketServer.TCPServer « Python recipes « ActiveState Code

Thread pool mixin class for use with SocketServer.TCPServer
(Python recipe)
by Michael Palmer

ActiveState Code (http://code.activestate.com/recipes/574454/)

4

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

Download
Copy to clipboard

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.

Share

Tags: threads

1 comment
Rafael Zanella
3 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)