Add configurable timeout (default: no timeout) to be used when the

WSGI application is called. Only applies to forked servers!

10 files changed, 91 insertions, 18 deletions

diff --git a/ChangeLog b/ChangeLog
index f94b94a..abb31b0 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,8 @@
+2009-10-21  Allan Saddi  <allan@saddi.com>
+
+	* Add configurable timeout (default: no timeout) to be used when the
+	  WSGI application is called. Only applies to forked servers!
+
 2009-06-05  Allan Saddi  <allan@saddi.com>
 
 	* Fix bug in scgi servers that occurs when SCRIPT_NAME is missing.
diff --git a/flup/server/ajp.py b/flup/server/ajp.py
index 687cfb0..64e377c 100644
--- a/flup/server/ajp.py
+++ b/flup/server/ajp.py
@@ -142,8 +142,8 @@ class WSGIServer(BaseAJPServer, ThreadedServer):
         for key in ('jobClass', 'jobArgs'):
             if kw.has_key(key):
                 del kw[key]
-        ThreadedServer.__init__(self, jobClass=Connection, jobArgs=(self,),
-                                **kw)
+        ThreadedServer.__init__(self, jobClass=Connection,
+                                jobArgs=(self, None), **kw)
 
     def run(self):
         """
diff --git a/flup/server/ajp_base.py b/flup/server/ajp_base.py
index a2e9963..f28e57d 100644
--- a/flup/server/ajp_base.py
+++ b/flup/server/ajp_base.py
@@ -36,6 +36,7 @@ import logging
 import errno
 import datetime
 import time
+import traceback
 
 # Unfortunately, for now, threads are required.
 import thread
@@ -591,21 +592,31 @@ class Request(object):
             data = data[toWrite:]
             bytesLeft -= toWrite
 
+class TimeoutException(Exception):
+    pass
+
 class Connection(object):
     """
     A single Connection with the server. Requests are not multiplexed over the
     same connection, so at any given time, the Connection is either
     waiting for a request, or processing a single request.
     """
-    def __init__(self, sock, addr, server):
+    def __init__(self, sock, addr, server, timeout):
         self.server = server
         self._sock = sock
         self._addr = addr
+        self._timeout = timeout
 
         self._request = None
 
         self.logger = logging.getLogger(LoggerName)
 
+    def timeout_handler(self, signum, frame):
+        self.logger.error('Timeout Exceeded')
+        self.logger.error("\n".join(traceback.format_stack(frame)))
+
+        raise TimeoutException
+
     def run(self):
         self.logger.debug('Connection starting up (%s:%d)',
                           self._addr[0], self._addr[1])
@@ -702,11 +713,20 @@ class Connection(object):
         if req.input.bytesAvailForAdd():
             self.processInput()
 
+        # If there is a timeout
+        if self._timeout:
+            old_alarm = signal.signal(signal.SIGALRM, self.timeout_handler)
+            signal.alarm(self._timeout)
+            
         # Run Request.
         req.run()
 
         self._request = None
 
+        # Restore old handler if timeout was given
+        if self._timeout:
+            signal.signal(signal.SIGALRM, old_alarm)
+
     def _shutdown(self, pkt):
         """Not sure what to do with this yet."""
         self.logger.info('Received shutdown request from server')
diff --git a/flup/server/ajp_fork.py b/flup/server/ajp_fork.py
index 86b0451..96af6fd 100644
--- a/flup/server/ajp_fork.py
+++ b/flup/server/ajp_fork.py
@@ -108,7 +108,7 @@ class WSGIServer(BaseAJPServer, PreforkServer):
     """
     def __init__(self, application, scriptName='', environ=None,
                  bindAddress=('localhost', 8009), allowedServers=None,
-                 loggingLevel=logging.INFO, debug=False, **kw):
+                 loggingLevel=logging.INFO, debug=False, timeout=None, **kw):
         """
         scriptName is the initial portion of the URL path that "belongs"
         to your application. It is used to determine PATH_INFO (which doesn't
@@ -141,7 +141,8 @@ class WSGIServer(BaseAJPServer, PreforkServer):
         for key in ('multithreaded', 'multiprocess', 'jobClass', 'jobArgs'):
             if kw.has_key(key):
                 del kw[key]
-        PreforkServer.__init__(self, jobClass=Connection, jobArgs=(self,), **kw)
+        PreforkServer.__init__(self, jobClass=Connection,
+                               jobArgs=(self, timeout), **kw)
 
     def run(self):
         """
diff --git a/flup/server/fcgi.py b/flup/server/fcgi.py
index bc621ee..660bafa 100644
--- a/flup/server/fcgi.py
+++ b/flup/server/fcgi.py
@@ -93,7 +93,7 @@ class WSGIServer(BaseFCGIServer, ThreadedServer):
             if kw.has_key(key):
                 del kw[key]
         ThreadedServer.__init__(self, jobClass=self._connectionClass,
-                                jobArgs=(self,), **kw)
+                                jobArgs=(self, None), **kw)
 
     def _isClientAllowed(self, addr):
         return self._web_server_addrs is None or \
diff --git a/flup/server/fcgi_base.py b/flup/server/fcgi_base.py
index 72396a9..231a56f 100644
--- a/flup/server/fcgi_base.py
+++ b/flup/server/fcgi_base.py
@@ -533,6 +533,9 @@ class Record(object):
         if self.paddingLength:
             self._sendall(sock, '\x00'*self.paddingLength)
             
+class TimeoutException(Exception):
+    pass
+
 class Request(object):
     """
     Represents a single FastCGI request.
@@ -542,8 +545,9 @@ class Request(object):
     be called by your handler. However, server, params, stdin, stdout,
     stderr, and data are free for your handler's use.
     """
-    def __init__(self, conn, inputStreamClass):
+    def __init__(self, conn, inputStreamClass, timeout):
         self._conn = conn
+        self._timeout = timeout
 
         self.server = conn.server
         self.params = {}
@@ -552,8 +556,20 @@ class Request(object):
         self.stderr = OutputStream(conn, self, FCGI_STDERR, buffered=True)
         self.data = inputStreamClass(conn)
 
+    def timeout_handler(self, signum, frame):
+        self.stderr.write('Timeout Exceeded\n')
+        self.stderr.write("\n".join(traceback.format_stack(frame)))
+        self.stderr.flush()
+
+        raise TimeoutException
+
     def run(self):
         """Runs the handler, flushes the streams, and ends the request."""
+        # If there is a timeout
+        if self._timeout:
+            old_alarm = signal.signal(signal.SIGALRM, self.timeout_handler)
+            signal.alarm(self._timeout)
+            
         try:
             protocolStatus, appStatus = self.server.handler(self)
         except:
@@ -567,6 +583,10 @@ class Request(object):
         if __debug__: _debug(1, 'protocolStatus = %d, appStatus = %d' %
                              (protocolStatus, appStatus))
 
+        # Restore old handler if timeout was given
+        if self._timeout:
+            signal.signal(signal.SIGALRM, old_alarm)
+
         try:
             self._flush()
             self._end(appStatus, protocolStatus)
@@ -615,10 +635,11 @@ class Connection(object):
     _multiplexed = False
     _inputStreamClass = InputStream
 
-    def __init__(self, sock, addr, server):
+    def __init__(self, sock, addr, server, timeout):
         self._sock = sock
         self._addr = addr
         self.server = server
+        self._timeout = timeout
 
         # Active Requests for this Connection, mapped by request ID.
         self._requests = {}
@@ -739,7 +760,8 @@ class Connection(object):
         """Handle an FCGI_BEGIN_REQUEST from the web server."""
         role, flags = struct.unpack(FCGI_BeginRequestBody, inrec.contentData)
 
-        req = self.server.request_class(self, self._inputStreamClass)
+        req = self.server.request_class(self, self._inputStreamClass,
+                                        self._timeout)
         req.requestId, req.role, req.flags = inrec.requestId, role, flags
         req.aborted = False
 
@@ -807,8 +829,9 @@ class MultiplexedConnection(Connection):
     _multiplexed = True
     _inputStreamClass = MultiplexedInputStream
 
-    def __init__(self, sock, addr, server):
-        super(MultiplexedConnection, self).__init__(sock, addr, server)
+    def __init__(self, sock, addr, server, timeout):
+        super(MultiplexedConnection, self).__init__(sock, addr, server,
+                                                    timeout)
 
         # Used to arbitrate access to self._requests.
         lock = threading.RLock()
diff --git a/flup/server/fcgi_fork.py b/flup/server/fcgi_fork.py
index 1cb1588..41ca6ed 100644
--- a/flup/server/fcgi_fork.py
+++ b/flup/server/fcgi_fork.py
@@ -70,7 +70,8 @@ class WSGIServer(BaseFCGIServer, PreforkServer):
     """
     def __init__(self, application, environ=None,
                  bindAddress=None, umask=None, multiplexed=False,
-                 debug=False, roles=(FCGI_RESPONDER,), forceCGI=False, **kw):
+                 debug=False, roles=(FCGI_RESPONDER,), forceCGI=False,
+                 timeout=None, **kw):
         """
         environ, if present, must be a dictionary-like object. Its
         contents will be copied into application's environ. Useful
@@ -99,7 +100,7 @@ class WSGIServer(BaseFCGIServer, PreforkServer):
             if kw.has_key(key):
                 del kw[key]
         PreforkServer.__init__(self, jobClass=self._connectionClass,
-                               jobArgs=(self,), **kw)
+                               jobArgs=(self, timeout), **kw)
 
         try:
             import resource
diff --git a/flup/server/scgi.py b/flup/server/scgi.py
index 450a9d3..6c70e7c 100644
--- a/flup/server/scgi.py
+++ b/flup/server/scgi.py
@@ -138,8 +138,8 @@ class WSGIServer(BaseSCGIServer, ThreadedServer):
         for key in ('jobClass', 'jobArgs'):
             if kw.has_key(key):
                 del kw[key]
-        ThreadedServer.__init__(self, jobClass=Connection, jobArgs=(self,),
-                                **kw)
+        ThreadedServer.__init__(self, jobClass=Connection,
+                                jobArgs=(self, None), **kw)
 
     def run(self):
         """
diff --git a/flup/server/scgi_base.py b/flup/server/scgi_base.py
index e4f8f4c..091ddc1 100644
--- a/flup/server/scgi_base.py
+++ b/flup/server/scgi_base.py
@@ -37,6 +37,7 @@ import signal
 import datetime
 import os
 import warnings
+import traceback
 
 # Threads are required. If you want a non-threaded (forking) version, look at
 # SWAP <http://www.idyll.org/~t/www-tools/wsgi/>.
@@ -197,18 +198,28 @@ class Request(object):
                           handlerTime.seconds +
                           handlerTime.microseconds / 1000000.0)
 
+class TimeoutException(Exception):
+    pass
+
 class Connection(object):
     """
     Represents a single client (web server) connection. A single request
     is handled, after which the socket is closed.
     """
-    def __init__(self, sock, addr, server):
+    def __init__(self, sock, addr, server, timeout):
         self._sock = sock
         self._addr = addr
         self.server = server
+        self._timeout = timeout
 
         self.logger = logging.getLogger(LoggerName)
 
+    def timeout_handler(self, signum, frame):
+        self.logger.error('Timeout Exceeded')
+        self.logger.error("\n".join(traceback.format_stack(frame)))
+
+        raise TimeoutException
+
     def run(self):
         if len(self._addr) == 2:
             self.logger.debug('Connection starting up (%s:%d)',
@@ -263,12 +274,22 @@ class Connection(object):
         # Allocate Request
         req = Request(self, environ, input, output)
 
+        # If there is a timeout
+        if self._timeout:
+            old_alarm = signal.signal(signal.SIGALRM, self.timeout_handler)
+            signal.alarm(self._timeout)
+            
         # Run it.
         req.run()
 
         output.close()
         input.close()
 
+        # Restore old handler if timeout was given
+        if self._timeout:
+            signal.signal(signal.SIGALRM, old_alarm)
+
+
 class BaseSCGIServer(object):
     # What Request class to use.
     requestClass = Request
diff --git a/flup/server/scgi_fork.py b/flup/server/scgi_fork.py
index 7de39a2..755dd3f 100644
--- a/flup/server/scgi_fork.py
+++ b/flup/server/scgi_fork.py
@@ -97,7 +97,7 @@ class WSGIServer(BaseSCGIServer, PreforkServer):
     def __init__(self, application, scriptName=NoDefault, environ=None,
                  bindAddress=('localhost', 4000), umask=None,
                  allowedServers=None,
-                 loggingLevel=logging.INFO, debug=False, **kw):
+                 loggingLevel=logging.INFO, debug=False, timeout=None, **kw):
         """
         scriptName is the initial portion of the URL path that "belongs"
         to your application. It is used to determine PATH_INFO (which doesn't
@@ -137,7 +137,9 @@ class WSGIServer(BaseSCGIServer, PreforkServer):
         for key in ('multithreaded', 'multiprocess', 'jobClass', 'jobArgs'):
             if kw.has_key(key):
                 del kw[key]
-        PreforkServer.__init__(self, jobClass=Connection, jobArgs=(self,), **kw)
+        
+        PreforkServer.__init__(self, jobClass=Connection,
+                               jobArgs=(self, timeout), **kw)
 
     def run(self):
         """