Progress on issue #1193577 by adding a polling .shutdown() method to

SocketServers. The core of the patch was written by Pedro Werneck, but any bugs
are mine. I've also rearranged the code for timeouts in order to avoid
interfering with the shutdown poll.

1 files changed, 66 insertions, 27 deletions

diff --git a/Lib/SocketServer.py b/Lib/SocketServer.py
index 0a194e5..2c41fbb 100644
--- a/Lib/SocketServer.py
+++ b/Lib/SocketServer.py
@@ -130,8 +130,13 @@ __version__ = "0.4"
 
 
 import socket
+import select
 import sys
 import os
+try:
+    import threading
+except ImportError:
+    import dummy_threading as threading
 
 __all__ = ["TCPServer","UDPServer","ForkingUDPServer","ForkingTCPServer",
            "ThreadingUDPServer","ThreadingTCPServer","BaseRequestHandler",
@@ -149,7 +154,8 @@ class BaseServer:
     Methods for the caller:
 
     - __init__(server_address, RequestHandlerClass)
-    - serve_forever()
+    - serve_forever(poll_interval=0.5)
+    - shutdown()
     - handle_request()  # if you do not use serve_forever()
     - fileno() -> int   # for select()
 
@@ -190,6 +196,8 @@ class BaseServer:
         """Constructor.  May be extended, do not override."""
         self.server_address = server_address
         self.RequestHandlerClass = RequestHandlerClass
+        self.__is_shut_down = threading.Event()
+        self.__serving = False
 
     def server_activate(self):
         """Called by constructor to activate the server.
@@ -199,27 +207,73 @@ class BaseServer:
         """
         pass
 
-    def serve_forever(self):
-        """Handle one request at a time until doomsday."""
-        while 1:
-            self.handle_request()
+    def serve_forever(self, poll_interval=0.5):
+        """Handle one request at a time until shutdown.
+
+        Polls for shutdown every poll_interval seconds. Ignores
+        self.timeout. If you need to do periodic tasks, do them in
+        another thread.
+        """
+        self.__serving = True
+        self.__is_shut_down.clear()
+        while self.__serving:
+            # XXX: Consider using another file descriptor or
+            # connecting to the socket to wake this up instead of
+            # polling. Polling reduces our responsiveness to a
+            # shutdown request and wastes cpu at all other times.
+            r, w, e = select.select([self], [], [], poll_interval)
+            if r:
+                self._handle_request_noblock()
+        self.__is_shut_down.set()
+
+    def shutdown(self):
+        """Stops the serve_forever loop.
+
+        Blocks until the loop has finished. This must be called while
+        serve_forever() is running in another thread, or it will
+        deadlock.
+        """
+        self.__serving = False
+        self.__is_shut_down.wait()
 
     # The distinction between handling, getting, processing and
     # finishing a request is fairly arbitrary.  Remember:
     #
     # - handle_request() is the top-level call.  It calls
-    #   await_request(), verify_request() and process_request()
-    # - get_request(), called by await_request(), is different for
-    #   stream or datagram sockets
+    #   select, get_request(), verify_request() and process_request()
+    # - get_request() is different for stream or datagram sockets
     # - process_request() is the place that may fork a new process
     #   or create a new thread to finish the request
     # - finish_request() instantiates the request handler class;
     #   this constructor will handle the request all by itself
 
     def handle_request(self):
-        """Handle one request, possibly blocking."""
+        """Handle one request, possibly blocking.
+
+        Respects self.timeout.
+        """
+        # Support people who used socket.settimeout() to escape
+        # handle_request before self.timeout was available.
+        timeout = self.socket.gettimeout()
+        if timeout is None:
+            timeout = self.timeout
+        elif self.timeout is not None:
+            timeout = min(timeout, self.timeout)
+        fd_sets = select.select([self], [], [], timeout)
+        if not fd_sets[0]:
+            self.handle_timeout()
+            return
+        self._handle_request_noblock()
+
+    def _handle_request_noblock(self):
+        """Handle one request, without blocking.
+
+        I assume that select.select has returned that the socket is
+        readable before this function was called, so there should be
+        no risk of blocking in get_request().
+        """
         try:
-            request, client_address = self.await_request()
+            request, client_address = self.get_request()
         except socket.error:
             return
         if self.verify_request(request, client_address):
@@ -229,21 +283,6 @@ class BaseServer:
                 self.handle_error(request, client_address)
                 self.close_request(request)
 
-    def await_request(self):
-        """Call get_request or handle_timeout, observing self.timeout.
-
-        Returns value from get_request() or raises socket.timeout exception if
-        timeout was exceeded.
-        """
-        if self.timeout is not None:
-            # If timeout == 0, you're responsible for your own fd magic.
-            import select
-            fd_sets = select.select([self], [], [], self.timeout)
-            if not fd_sets[0]:
-                self.handle_timeout()
-                raise socket.timeout("Listening timed out")
-        return self.get_request()
-
     def handle_timeout(self):
         """Called if no new request arrives within self.timeout.
 
@@ -307,7 +346,8 @@ class TCPServer(BaseServer):
     Methods for the caller:
 
     - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
-    - serve_forever()
+    - serve_forever(poll_interval=0.5)
+    - shutdown()
     - handle_request()  # if you don't use serve_forever()
     - fileno() -> int   # for select()
 
@@ -523,7 +563,6 @@ class ThreadingMixIn:
 
     def process_request(self, request, client_address):
         """Start a new thread to process the request."""
-        import threading
         t = threading.Thread(target = self.process_request_thread,
                              args = (request, client_address))
         if self.daemon_threads: