wsgiref version multithread

Version imprimable

Je me suis fait une version multithread du serveur WSGI que l'on trouve avec wsgiref. Ça pourra sans doute intéresser certain d'entre vous

WSGIMultiThreadServer.py:

Code:

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
import threading 
from Queue import Queue, Empty, Full
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
 
class WSGIThreadWorker(threading.Thread):
 
    def __init__(self, pServer, pQueue, pGroup=None, pName=None, pVerbose=None):
        threading.Thread.__init__(self, group=pGroup, target=None, name=pName, args=None, kwargs=None, verbose=pVerbose)
        self.__queue = pQueue
        self.__server = pServer
 
    def run(self):
        while True:
            try:
                lRequest, lClientAddress = self.__queue.get()
                try:
                    self.__server.finish_request(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
                except:
                    self.__server.handle_error(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
                self.__queue.task_done()
            except Empty:
                # Doing something if it's the queue is empty
                pass
            except Full:
                # Doing something if it's the queue is full
                pass
 
class WSGIThreadedServer(WSGIServer):
 
    def __init__(self, pServerAddress, pRequestHandlerClass, pMaxThread = 5, pBindAndActivate=True):
        WSGIServer.__init__(self, server_address = pServerAddress, RequestHandlerClass = pRequestHandlerClass, bind_and_activate = pBindAndActivate)
        self.__maxThread = pMaxThread
        self.__queue = Queue()
        self.__threads = []
 
        for i in range(self.__maxThread):
            lNewThreadWorker = WSGIThreadWorker(self, self.__queue, pName = "WSGIThreadWorker-" + str(i+1))
            lNewThreadWorker.daemon = True
            lNewThreadWorker.start()
            self.__threads += [lNewThreadWorker]
 
    def process_request(self, pRequest, pClientAddress):
        self.__queue.put((pRequest,pClientAddress))
 
def make_multithread_server(pHost, pPort, app, pMaxThread = 5, pServerClass=WSGIThreadedServer, pHandlerClass=WSGIRequestHandler):
    server = pServerClass((pHost, pPort), pHandlerClass, pMaxThread = pMaxThread)
    server.set_app(app)
    return server

pour le tester voici test.py:

Code:

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
#!/usr/bin/env python
import os
import threading
import logging
import signal
 
logger = None
 
def ShudownGracefully(pSignal, pFrame):
    logger.info('Shutdown gracefully')
 
if logger is None:
    logger = logging.getLogger('WSGIMultiThreadServer')
    handler = logging.FileHandler('WSGIMultiThreadServer.log')
    formatter = logging.Formatter('%(asctime)-15s %(process)06d %(thread)06d %(filename)s %(lineno)d %(module)s %(name)s %(levelname)s %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)  
    logger.info('Active log')
    signal.signal(signal.SIGTERM, ShudownGracefully)
    logger.info('Register ShudownGracefully')  
 
def app(environ, start_response):
    lCurrentThread = threading.currentThread()
    logger.info('Start request')
    response_body = 'PID: %d\n' % os.getpid()
    response_body += 'TID: %d\n' % lCurrentThread.ident
    response_body += 'Thread name: %s\n' % lCurrentThread.name
    response_body += '\n'.join(['%s: %s' % (key, value) for key, value in sorted(environ.items())])
 
    # Response_body has now more than one string
    response_body = ['The Beggining\n',
                   '*' * 30 + '\n',
                   response_body,
                   '\n' + '*' * 30 ,
                   '\nThe End']
 
    # So the content-lenght is the sum of all string's lengths
    content_length = 0
    for s in response_body:
        content_length += len(s)
 
    status = '200 OK'
    response_headers = [('Content-Type', 'text/plain'), ('Content-Length', str(content_length))]
    start_response(status, response_headers)
 
    logger.info('Finish request')  
    return response_body
 
logger.info('Loaded test.py')
 
if __name__ == "__main__":
    from WSGIMultiThreadServer import make_multithread_server 
    # specify your 
    lWSGIMultiThreadServer = make_multithread_server('', 8000, app)
    # Respond to requests until process is killed
    lWSGIMultiThreadServer.serve_forever()

Ça donne un résultation du type :

Code:

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
The Beggining
******************************
PID: 81942
TID: 4342730752
Thread name: WSGIThreadWorker-5
APP_ICON_77698: ../Resources/aptana.icns
Apple_PubSub_Socket_Render: /tmp/launch-vbK50x/Render
COMMAND_MODE: unix2003
CONTENT_LENGTH: 
CONTENT_TYPE: text/plain
DBUS_LAUNCHD_SESSION_BUS_SOCKET: /tmp/launch-G71IJL/unix_domain_listener
DISPLAY: /tmp/launch-iLtHZj/org.x:0
DJANGO_SETTINGS_MODULE: TestFCGI.settings
GATEWAY_INTERFACE: CGI/1.1
HOME: /Users/me
HTTP_ACCEPT: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
HTTP_ACCEPT_CHARSET: ISO-8859-1,utf-8;q=0.7,*;q=0.7
HTTP_ACCEPT_ENCODING: gzip, deflate
HTTP_ACCEPT_LANGUAGE: en-us,en;q=0.5
HTTP_CACHE_CONTROL: max-age=0
HTTP_CONNECTION: keep-alive
HTTP_HOST: localhost:8000
HTTP_KEEP_ALIVE: 115
HTTP_USER_AGENT: Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:2.0.1) Gecko/20100101 Firefox/4.0.1
LOGNAME: me
PATH: /usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/usr/X11/bin:/opt/local/bin:/usr/local/git/bin
PATH_INFO: /
PYDEV_COMPLETER_PYTHONPATH: /Applications/Aptana Studio 3/plugins/org.python.pydev_2.0.0.2011040403/PySrc
PYDEV_CONSOLE_ENCODING: UTF-8
PYTHONIOENCODING: UTF-8
PYTHONPATH: 
QUERY_STRING: 
REMOTE_ADDR: 127.0.0.1
REMOTE_HOST: localhost
REQUEST_METHOD: GET
SCRIPT_NAME: 
SERVER_NAME: 106.1.168.192.in-addr.arpa
SERVER_PORT: 8000
SERVER_PROTOCOL: HTTP/1.1
SERVER_SOFTWARE: WSGIServer/0.1 Python/2.6.6
SHELL: /bin/bash
SSH_AUTH_SOCK: /tmp/launch-fOgohQ/Listeners
TMPDIR: /var/folders/80/80vq-Sn2FEqhl0B0FlfM4E+++TI/-Tmp-/
USER: me
__CF_USER_TEXT_ENCODING: 0x1F5:0:0
com.apple.java.jvmTask: JNI
wsgi.errors: <open file '<stderr>', mode 'w' at 0x1002601e0>
wsgi.file_wrapper: wsgiref.util.FileWrapper
wsgi.input: <socket._fileobject object at 0x1011df0d0>
wsgi.multiprocess: False
wsgi.multithread: True
wsgi.run_once: False
wsgi.url_scheme: http
wsgi.version: (1, 0)
******************************
The End

Le paramètre wsgi.multithread: True est cette fois ci réel !

J'ai bricolé ça rapidement le code peut sans aucun doute être grandement améliorer.

Si ça peut être utile à certain d'entre vous tant mieux...

07/05/2012, 03h02
larsal007

Mille fois merci

Bonsoir,
Je n'ai pas l'habitude de poster. Mais ton code a mis fin à un weekend sans fin accompagné d'une sacrée prise de tete.
Ton code m'a rendu service et je voulais te dire mille fois merci.
Je dois également mettre en place un reload automatique en cas de changement des fichiers du repertoire. Si tu as un bout de code qui traine...., on ne sait jamais.

Merci encore.

Nouvelle version Multithread et ajout version Multiprocess

Citation:

Envoyé par larsal007

Bonsoir,
Je n'ai pas l'habitude de poster. Mais ton code a mis fin à un weekend sans fin accompagné d'une sacrée prise de tete.
Ton code m'a rendu service et je voulais te dire mille fois merci.
Je dois également mettre en place un reload automatique en cas de changement des fichiers du repertoire. Si tu as un bout de code qui traine...., on ne sait jamais.

Merci encore.

Je suis content que ça t'a dépanné :)
Pour ce qui est de la détection de la modification des fichiers j'ai rien en magasin. Je sais que sous Win32 les APIs existent (je les utiliais déjà en 1997) et sous Linux tu peux regarder du coté de Pyinotify qui se base sur les fonctions de inotify.
À ma connaissance il n'y a pas de package python multiplatform pour ce genre de fonction. Mais je peux me tromper. Si c'est le cas met moi le lien sur ce poste car ça m'intéresse aussi.

Sinon depuis que j'ai posté ce bout de code j'ai retravaillé dessus. J'ai aussi fait la même classe en mode multiprocess.

WSGIMultiServer.py:

Code:

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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from threading import Thread
from Queue import Queue as ThreadQueue
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
 
# ------------------------------------------------------------------------------
# Support multithread 
# ------------------------------------------------------------------------------
 
class WSGIThreadWorker(Thread):
 
    def __init__(self, pServer, pQueue, pName):
        Thread.__init__(self, name=pName)
        self.__queue = pQueue
        self.__server = pServer
 
    def run(self):
        while True:
            try:
                lRequest, lClientAddress = self.__queue.get()
                try:
                    self.__server.finish_request(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
                except:
                    self.__server.handle_error(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
                self.__queue.task_done()
            except:
                break
 
class WSGIThreadedServer(WSGIServer):
 
    def __init__(self, pServerAddress, pRequestHandlerClass, pMaxThread = 5, pBindAndActivate=True):
        WSGIServer.__init__(self, server_address = pServerAddress, RequestHandlerClass = pRequestHandlerClass, bind_and_activate = pBindAndActivate)
        self.__maxThread = pMaxThread
        self.__queue = ThreadQueue()
        self.__threads = []
 
    def set_app(self,pApplication):
        WSGIServer.set_app(self,pApplication)
 
        for i in range(self.__maxThread):
            lNewThreadWorker = WSGIThreadWorker(pServer = self, pQueue = self.__queue, pName = "WSGIThreadWorker-" + str(i+1))
            lNewThreadWorker.daemon = True
            lNewThreadWorker.start()
            self.__threads += [lNewThreadWorker]
 
    def process_request(self, pRequest, pClientAddress):
        self.__queue.put((pRequest,pClientAddress))
 
def make_multithread_server(pHost, pPort, app, pMaxThread = 5, pServerClass=WSGIThreadedServer, pHandlerClass=WSGIRequestHandler):
    lServer = pServerClass((pHost, pPort), pHandlerClass, pMaxThread = pMaxThread)
    lServer.set_app(app)
    return lServer
 
# ------------------------------------------------------------------------------
# Support multiprocess
# ------------------------------------------------------------------------------
 
import socket
from multiprocessing import Process, Queue as ProcessQueue
from multiprocessing.reduction import reduce_handle, rebuild_handle
try:
    from setproctitle import setproctitle
except:
    import sys
    print('Error: install package "setproctitle"\nCommand: pip install setproctitle')
    sys.exit(0)
 
class WSGIProcessWorker(Process):
 
    def __init__(self, pServer, pQueue, pName):
        Process.__init__(self,name=pName)
        self.__server = pServer
        self.__queue = pQueue
 
    def run(self):
        setproctitle(self.name)
 
        while True:
            try:
                lHandle, lClientAddress = self.__queue.get()
                lFd = rebuild_handle(lHandle)
                lRequest = socket.fromfd(lFd, socket.AF_INET, socket.SOCK_STREAM)
                try:
                    self.__server.finish_request(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
                except:
                    self.__server.handle_error(lRequest, lClientAddress)
                    self.__server.close_request(lRequest)
            except:
                break
 
class WSGIProcessServer(WSGIServer):
 
    def __init__(self, pServerAddress, pRequestHandlerClass, pMaxProcess = 2, pBindAndActivate=True, pProcessName = None):
        WSGIServer.__init__(self, server_address = pServerAddress, RequestHandlerClass = pRequestHandlerClass, bind_and_activate = pBindAndActivate)
        self.__maxProcess = pMaxProcess
        self.__queue = ProcessQueue()
        self.__process = []
        self.__processName = pProcessName
 
    def serve_forever(self, pPoll_interval=0.5):
        try:
            WSGIServer.serve_forever(self, poll_interval=pPoll_interval)
        except:
            import os
            for lProcess in self.__process:
                os.kill(lProcess.pid, 9) # SIGKILL
 
    def set_app(self,pApplication):
        WSGIServer.set_app(self,pApplication)
 
        if self.__processName is not None:
            setproctitle(self.__processName)
 
        for i in range(self.__maxProcess):
            lNewProcessWorker = WSGIProcessWorker(pServer = self, pQueue = self.__queue, pName = "WSGIProcessWorker-" + str(i+1) )
            lNewProcessWorker.daemon = True
            lNewProcessWorker.start()
            self.__process += [lNewProcessWorker]
 
    def process_request(self, pRequest, pClientAddress):
        lParam = [ reduce_handle(pRequest.fileno()), pClientAddress]
        self.__queue.put(lParam)
 
def make_multiprocess_server(pHost, pPort, app, pMaxProcess = 5, pServerClass=WSGIProcessServer, pHandlerClass=WSGIRequestHandler, pProcessName = None):
    lServer = pServerClass((pHost, pPort), pHandlerClass, pMaxProcess = pMaxProcess)
    lServer.set_app(app)
    return lServer

test.py :

Code:

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
#!/usr/bin/env python
import os
import threading
import logging
import signal
 
logger = None
 
def ShudownGracefully(pSignal, pFrame):
    logger.info('Shutdown gracefully')
 
if logger is None:
    logger = logging.getLogger('TestFCGI')
    handler = logging.FileHandler('TestFCGI.log')
    formatter = logging.Formatter('%(asctime)-15s %(process)06d %(thread)06d %(filename)s %(lineno)d %(module)s %(name)s %(levelname)s %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)  
    logger.info('Active log')
    signal.signal(signal.SIGTERM, ShudownGracefully)
    logger.info('Register ShudownGracefully')  
 
def app(environ, start_response):
    lCurrentThread = threading.currentThread()
    logger.info('Start request')
    response_body = 'PID: %d\n' % os.getpid()
    response_body += 'TID: %d\n' % lCurrentThread.ident
    response_body += 'Thread name: %s\n' % lCurrentThread.name
    response_body += '\n'.join(['%s: %s' % (key, value) for key, value in sorted(environ.items())])
 
    # Response_body has now more than one string
    response_body = ['Begin\n',
                   '*' * 30 + '\n',
                   response_body,
                   '\n' + '*' * 30 ,
                   '\nEnd']
 
    # So the content-lenght is the sum of all string's lengths
    content_length = 0
    for s in response_body:
        content_length += len(s)
 
    status = '200 OK'
    response_headers = [('Content-Type', 'text/plain'), ('Content-Length', str(content_length))]
    start_response(status, response_headers)
 
    logger.info('Finish request')  
    return response_body
 
logger.info('Loaded test.py')
 
if __name__ == "__main__":
#    from WSGIMultiServer import make_multithread_server 
#    lWSGIMultiThreadServer = make_multithread_server('', 8000, app)
#    lWSGIMultiThreadServer.serve_forever()
 
    print('Main PID:%d' % os.getpid())
    from WSGIMultiServer import make_multiprocess_server 
    lWSGIMultiProcessServer = make_multiprocess_server('', 8000, app, pProcessName = 'TestFCGI')
    lWSGIMultiProcessServer.serve_forever()

TestFCGI.fcgi :

Code:

1
2
3
4
5
6
7
8
9
10
11
#!/usr/bin/env python
import sys, os
 
_PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _PROJECT_DIR)
sys.path.insert(0, os.path.dirname(_PROJECT_DIR))
 
from test import app
from flup.server.fcgi import WSGIServer
 
WSGIServer(app).run()