Merge lp:~jameinel/bzr/2.5-soft-hangup-795025 into lp:bzr

[John A Meinel]

Allow 'bzr serve' to interpret SIGHUP as a graceful shutdown. (bug #795025)

Silly launchpad losing changes if you accidentally navigate away from the edit box... :( (and silly laptop for putting browser previous right next to the up arrow.)

Anyway, this should be done. It does the bits that I wanted it to, and it seems well tested.

1) sending SIGHUP will start a soft shutdown for both 'bzr serve' and 'bzr serve --inet'. In the former case, it tells all the client threads that we want a soft shutdown (so stop as soon as you are done with the current request), and then waits for them to finish, logging whether it is progressing or not.

2) This changes 'serve()' so that it defaults to closing the client connection when done serving. This makes things cleaner IMO, but it means that we will close sys.stdin during 'bzr serve --inet'. And _flush_stdout_stderr was handling some IOError, but on py2.6 on Windows, I was getting a *ValueError*, so I added that to the trap list.

3) The infrastructure is all hooked up on Windows, except that we don't have SIGHUP. For now, I've been using SIGBREAK and then self._stop_gracefully() to make sure it is working.

4) We close the primary server socket after we get the shutdown request, but before we wait for the clients to finish. That should mean that you can get a no-downtime 'bzr serve' restart by doing "kill -SIGHUP <PID> ; bzr serve". So it will shut down the current one, and bring up another one to serve new requests while the first one finishes out the existing requests. For Launchpad, it will be more complex. We'll need to teach the twisted daemon about SIGHUP, and then have it pass that same signal to all the 'bzr serve --inet' children. With the forking server, I imagine it will similarly need to pass the signal around. But something similar should be possible, making the current one quiet (down to HAProxy), close the primary socket, start a new one responding to HAProxy, etc.

5) We now need to teach the client to reconnect when it gets disconnected, and the rest should flow pretty smoothly.

[Jelmer Vernooij (jelmer)]

I've begun reviewing this, but threads make me sad and I'm not really familiar with this code (yet).
 I've noticed some really minor issues, but I'll try to follow up with some actually useful comments (once I understand the global picture a bit better).

you have an import conflict in bzrlib/smart/medium.py

206 + # have a good way (yet) to poll the spawned clients and
... and what ? :)

246 + if e.args[0] not in (errno.EBADF, errno.EINTR):
The comment above this line should probably be updated.

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 9/24/2011 11:26 PM, Jelmer Vernooij wrote:
> I've begun reviewing this, but threads make me sad and I'm not
> really familiar with this code (yet). I've noticed some really
> minor issues, but I'll try to follow up with some actually useful
> comments (once I understand the global picture a bit better).
>
> you have an import conflict in bzrlib/smart/medium.py
>
> 206 + # have a good way (yet) to poll the spawned
> clients and ... and what ? :)

I fixed that one, actually, so I just removed the comment.
SmartTCPServer now keeps track of clients and on graceful shutdown
waits for them to finish.

>
> 246 + if e.args[0] not in (errno.EBADF,
> errno.EINTR): The comment above this line should probably be
> updated.

Done. I fixed the import collision, and added gettext() calls to all
the new trace.note() functionality. As an aside, it is really easy to
forget those. It would be nice to have some sort of test/regular
procedure for making sure we get them right.

John
=:->

-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk6AMTYACgkQJdeBCYSNAAPsOACgkYEjHGH0AG7/X5Tid0537l6W
dvkAnj81J8GDMF7ijtwsqdZOSNf8+N1v
=wh/M
-----END PGP SIGNATURE-----

[Jelmer Vernooij (jelmer)]

review: Approve

[John A Meinel (jameinel)]

sent to pqm by email

=== modified file 'bzrlib/smart/__init__.py'
--- bzrlib/smart/__init__.py	2011-09-22 13:25:47 +0000
+++ bzrlib/smart/__init__.py	2011-09-26 14:29:40 +0000
@@ -1,4 +1,4 @@
-# Copyright (C) 2006 Canonical Ltd
+# Copyright (C) 2006,2011 Canonical Ltd
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by

=== modified file 'bzrlib/smart/medium.py'
--- bzrlib/smart/medium.py	2011-09-26 14:29:38 +0000
+++ bzrlib/smart/medium.py	2011-09-26 14:29:40 +0000
@@ -46,10 +46,10 @@
     urlutils,
     )
 from bzrlib.i18n import gettext
-from bzrlib.smart import client, protocol, request, vfs
+from bzrlib.smart import client, protocol, request, signals, vfs
 from bzrlib.transport import ssh
 """)
-from bzrlib import config, osutils
+from bzrlib import osutils
 
 # Throughout this module buffer size parameters are either limited to be at
 # most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
@@ -205,6 +205,8 @@
         the stream.  See also the _push_back method.
     """
 
+    _timer = time.time
+
     def __init__(self, backing_transport, root_client_path='/', timeout=None):
         """Construct new server.
 
@@ -228,6 +230,11 @@
         try:
             while not self.finished:
                 server_protocol = self._build_protocol()
+                # TODO: This seems inelegant:
+                if server_protocol is None:
+                    # We could 'continue' only to notice that self.finished is
+                    # True...
+                    break
                 self._serve_one_request(server_protocol)
         except errors.ConnectionTimeout, e:
             trace.note('%s' % (e,))
@@ -238,6 +245,12 @@
         except Exception, e:
             stderr.write("%s terminating on exception %s\n" % (self, e))
             raise
+        self._disconnect_client()
+
+    def _stop_gracefully(self):
+        """When we finish this message, stop looking for more."""
+        trace.mutter('Stopping %s' % (self,))
+        self.finished = True
 
     def _disconnect_client(self):
         """Close the current connection. We stopped due to a timeout/etc."""
@@ -267,6 +280,9 @@
         :returns: a SmartServerRequestProtocol.
         """
         self._wait_for_bytes_with_timeout(self._client_timeout)
+        if self.finished:
+            # We're stopping, so don't try to do any more work
+            return None
         bytes = self._get_line()
         protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
         protocol = protocol_factory(
@@ -281,10 +297,12 @@
         readable handle before timeout_seconds.
         :return: None
         """
-        t_end = time.time() + timeout_seconds
+        t_end = self._timer() + timeout_seconds
         poll_timeout = min(timeout_seconds, self._client_poll_timeout)
         rs = xs = None
-        while not rs and not xs and time.time() < t_end:
+        while not rs and not xs and self._timer() < t_end:
+            if self.finished:
+                return
             try:
                 rs, _, xs = select.select([fd], [], [fd], poll_timeout)
             except (select.error, socket.error) as e:
@@ -341,6 +359,18 @@
             timeout=timeout)
         sock.setblocking(True)
         self.socket = sock
+        # Get the getpeername now, as we might be closed later when we care.
+        try:
+            self._client_info = sock.getpeername()
+        except socket.error:
+            self._client_info = '<unknown>'
+
+    def __str__(self):
+        return '%s(client=%s)' % (self.__class__.__name__, self._client_info)
+
+    def __repr__(self):
+        return '%s.%s(client=%s)' % (self.__module__, self.__class__.__name__,
+            self._client_info)
 
     def _serve_one_request_unguarded(self, protocol):
         while protocol.next_read_size():
@@ -412,6 +442,17 @@
         self._in = in_file
         self._out = out_file
 
+    def serve(self):
+        """See SmartServerStreamMedium.serve"""
+        # This is the regular serve, except it adds signal trapping for soft
+        # shutdown.
+        stop_gracefully = self._stop_gracefully
+        signals.register_on_hangup(id(self), stop_gracefully)
+        try:
+            return super(SmartServerPipeStreamMedium, self).serve()
+        finally:
+            signals.unregister_on_hangup(id(self))
+
     def _serve_one_request_unguarded(self, protocol):
         while True:
             # We need to be careful not to read past the end of the current
@@ -432,6 +473,7 @@
 
     def _disconnect_client(self):
         self._in.close()
+        self._out.flush()
         self._out.close()
 
     def _wait_for_bytes_with_timeout(self, timeout_seconds):

=== modified file 'bzrlib/smart/server.py'
--- bzrlib/smart/server.py	2011-09-26 14:29:38 +0000
+++ bzrlib/smart/server.py	2011-09-26 14:29:40 +0000
@@ -20,6 +20,7 @@
 import os.path
 import socket
 import sys
+import time
 import threading
 
 from bzrlib.hooks import Hooks
@@ -31,7 +32,10 @@
 from bzrlib.i18n import gettext
 from bzrlib.lazy_import import lazy_import
 lazy_import(globals(), """
-from bzrlib.smart import medium
+from bzrlib.smart import (
+    medium,
+    signals,
+    )
 from bzrlib.transport import (
     chroot,
     pathfilter,
@@ -56,6 +60,10 @@
     # so the test suite can set it faster. (It thread.interrupt_main() will not
     # fire a KeyboardInterrupt during socket.accept)
     _ACCEPT_TIMEOUT = 1.0
+    _SHUTDOWN_POLL_TIMEOUT = 1.0
+    _LOG_WAITING_TIMEOUT = 10.0
+
+    _timer = time.time
 
     def __init__(self, backing_transport, root_client_path='/',
                  client_timeout=None):
@@ -73,6 +81,10 @@
         self.backing_transport = backing_transport
         self.root_client_path = root_client_path
         self._client_timeout = client_timeout
+        self._active_connections = []
+        # This is set to indicate we want to wait for clients to finish before
+        # we disconnect.
+        self._gracefully_stopping = False
 
     def start_server(self, host, port):
         """Create the server listening socket.
@@ -105,8 +117,14 @@
         self.port = self._sockname[1]
         self._server_socket.listen(1)
         self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
+        # Once we start accept()ing connections, we set started.
         self._started = threading.Event()
+        # Once we stop accept()ing connections (and are closing the socket) we
+        # set _stopped
         self._stopped = threading.Event()
+        # Once we have finished waiting for all clients, etc. We set
+        # _fully_stopped
+        self._fully_stopped = threading.Event()
 
     def _backing_urls(self):
         # There are three interesting urls:
@@ -145,7 +163,38 @@
         for hook in SmartTCPServer.hooks['server_stopped']:
             hook(backing_urls, self.get_url())
 
+    def _stop_gracefully(self):
+        trace.note(gettext('Requested to stop gracefully'))
+        self._should_terminate = True
+        self._gracefully_stopping = True
+        for handler, _ in self._active_connections:
+            handler._stop_gracefully()
+
+    def _wait_for_clients_to_disconnect(self):
+        self._poll_active_connections()
+        if not self._active_connections:
+            return
+        trace.note(gettext('Waiting for %d client(s) to finish')
+                   % (len(self._active_connections),))
+        t_next_log = self._timer() + self._LOG_WAITING_TIMEOUT
+        while self._active_connections:
+            now = self._timer()
+            if now >= t_next_log:
+                trace.note(gettext('Still waiting for %d client(s) to finish')
+                           % (len(self._active_connections),))
+                t_next_log = now + self._LOG_WAITING_TIMEOUT
+            self._poll_active_connections(self._SHUTDOWN_POLL_TIMEOUT)
+
     def serve(self, thread_name_suffix=''):
+        # Note: There is a temptation to do
+        #       signals.register_on_hangup(id(self), self._stop_gracefully)
+        #       However, that creates a temporary object which is a bound
+        #       method. signals._on_sighup is a WeakKeyDictionary so it
+        #       immediately gets garbage collected, because nothing else
+        #       references it. Instead, we need to keep a real reference to the
+        #       bound method for the lifetime of the serve() function.
+        stop_gracefully = self._stop_gracefully
+        signals.register_on_hangup(id(self), stop_gracefully)
         self._should_terminate = False
         # for hooks we are letting code know that a server has started (and
         # later stopped).
@@ -161,14 +210,19 @@
                         pass
                     except self._socket_error, e:
                         # if the socket is closed by stop_background_thread
-                        # we might get a EBADF here, any other socket errors
-                        # should get logged.
-                        if e.args[0] != errno.EBADF:
-                            trace.warning("listening socket error: %s", e)
+                        # we might get a EBADF here, or if we get a signal we
+                        # can get EINTR, any other socket errors should get
+                        # logged.
+                        if e.args[0] not in (errno.EBADF, errno.EINTR):
+                            trace.warning(gettext("listening socket error: %s")
+                                          % (e,))
                     else:
                         if self._should_terminate:
+                            conn.close()
                             break
                         self.serve_conn(conn, thread_name_suffix)
+                    # Cleanout any threads that have finished processing.
+                    self._poll_active_connections()
             except KeyboardInterrupt:
                 # dont log when CTRL-C'd.
                 raise
@@ -176,14 +230,18 @@
                 trace.report_exception(sys.exc_info(), sys.stderr)
                 raise
         finally:
-            self._stopped.set()
             try:
                 # ensure the server socket is closed.
                 self._server_socket.close()
             except self._socket_error:
                 # ignore errors on close
                 pass
+            self._stopped.set()
+            signals.unregister_on_hangup(id(self))
             self.run_server_stopped_hooks()
+        if self._gracefully_stopping:
+            self._wait_for_clients_to_disconnect()
+        self._fully_stopped.set()
 
     def get_url(self):
         """Return the url of the server"""
@@ -194,6 +252,23 @@
             conn, self.backing_transport, self.root_client_path,
             timeout=self._client_timeout)
 
+    def _poll_active_connections(self, timeout=0.0):
+        """Check to see if any active connections have finished.
+
+        This will iterate through self._active_connections, and update any
+        connections that are finished.
+
+        :param timeout: The timeout to pass to thread.join(). By default, we
+            set it to 0, so that we don't hang if threads are not done yet.
+        :return: None
+        """
+        still_active = []
+        for handler, thread in self._active_connections:
+            thread.join(timeout)
+            if thread.isAlive():
+                still_active.append((handler, thread))
+        self._active_connections = still_active
+
     def serve_conn(self, conn, thread_name_suffix):
         # For WIN32, where the timeout value from the listening socket
         # propagates to the newly accepted socket.
@@ -203,9 +278,7 @@
         handler = self._make_handler(conn)
         connection_thread = threading.Thread(
             None, handler.serve, name=thread_name)
-        # FIXME: This thread is never joined, it should at least be collected
-        # somewhere so that tests that want to check for leaked threads can get
-        # rid of them -- vila 20100531
+        self._active_connections.append((handler, connection_thread))
         connection_thread.setDaemon(True)
         connection_thread.start()
         return connection_thread
@@ -355,13 +428,17 @@
             transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
         self.transport = transport
 
+    def _get_stdin_stdout(self):
+        return sys.stdin, sys.stdout
+
     def _make_smart_server(self, host, port, inet, timeout):
         if timeout is None:
             c = config.GlobalStack()
             timeout = c.get('serve.client_timeout')
         if inet:
+            stdin, stdout = self._get_stdin_stdout()
             smart_server = medium.SmartServerPipeStreamMedium(
-                sys.stdin, sys.stdout, self.transport, timeout=timeout)
+                stdin, stdout, self.transport, timeout=timeout)
         else:
             if host is None:
                 host = medium.BZR_DEFAULT_INTERFACE
@@ -387,6 +464,10 @@
         self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
         ui.ui_factory = ui.SilentUIFactory()
         lockdir._DEFAULT_TIMEOUT_SECONDS = 0
+        orig = signals.install_sighup_handler()
+        def restore_signals():
+            signals.restore_sighup_handler(orig)
+        self.cleanups.append(restore_signals)
 
     def set_up(self, transport, host, port, inet, timeout):
         self._make_backing_transport(transport)
@@ -397,6 +478,7 @@
         for cleanup in reversed(self.cleanups):
             cleanup()
 
+
 def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
     """This is the default implementation of 'bzr serve'.
 

=== added file 'bzrlib/smart/signals.py'
--- bzrlib/smart/signals.py	1970-01-01 00:00:00 +0000
+++ bzrlib/smart/signals.py	2011-09-26 14:29:40 +0000
@@ -0,0 +1,114 @@
+# Copyright (C) 2011 Canonical Ltd
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+"""Signal handling for the smart server code."""
+
+import signal
+import weakref
+
+from bzrlib import trace
+
+
+# I'm pretty sure this has to be global, since signal handling is per-process.
+_on_sighup = None
+# TODO: Using a dict means that the order of calls is unordered. We could use a
+#       list and then do something like LIFO ordering. A dict was chosen so
+#       that you could have a key to easily remove your entry. However, you
+#       could just use the callable itself as the indexed part, and even in
+#       large cases, we shouldn't have more than 100 or so callbacks
+#       registered.
+def _sighup_handler(signal_number, interrupted_frame):
+    """This is the actual function that is registered for handling SIGHUP.
+
+    It will call out to all the registered functions, letting them know that a
+    graceful termination has been requested.
+    """
+    if _on_sighup is None:
+        return
+    trace.mutter('Caught SIGHUP, sending graceful shutdown requests.')
+    for ref in _on_sighup.valuerefs():
+        try:
+            cb = ref()
+            if cb is not None:
+                cb()
+        except KeyboardInterrupt:
+            raise
+        except Exception:
+            trace.mutter('Error occurred while running SIGHUP handlers:')
+            trace.log_exception_quietly()
+
+
+def install_sighup_handler():
+    """Setup a handler for the SIGHUP signal."""
+    if getattr(signal, "SIGHUP", None) is None:
+        # If we can't install SIGHUP, there is no reason (yet) to do graceful
+        # shutdown.
+        old_signal = None
+    else:
+        old_signal = signal.signal(signal.SIGHUP, _sighup_handler)
+    old_dict = _setup_on_hangup_dict()
+    return old_signal, old_dict
+
+
+def _setup_on_hangup_dict():
+    """Create something for _on_sighup.
+
+    This is done when we install the sighup handler, and for tests that want to
+    test the functionality. If this hasn'nt been called, then
+    register_on_hangup is a no-op. As is unregister_on_hangup.
+    """
+    global _on_sighup
+    old = _on_sighup
+    _on_sighup = weakref.WeakValueDictionary()
+    return old
+
+
+def restore_sighup_handler(orig):
+    """Pass in the returned value from install_sighup_handler to reset."""
+    global _on_sighup
+    old_signal, old_dict = orig
+    if old_signal is not None:
+        signal.signal(signal.SIGHUP, old_signal)
+    _on_sighup = old_dict
+
+
+# TODO: Should these be single-use callables? Meaning that once we've triggered
+#       SIGHUP and called them, they should auto-remove themselves? I don't
+#       think so. Callers need to clean up during shutdown anyway, so that we
+#       don't end up with lots of garbage in the _on_sighup dict. On the other
+#       hand, we made _on_sighup a WeakValueDictionary in case cleanups didn't
+#       get fired properly. Maybe we just assume we don't have to do it?
+def register_on_hangup(identifier, a_callable):
+    """Register for us to call a_callable as part of a graceful shutdown."""
+    if _on_sighup is None:
+        return
+    _on_sighup[identifier] = a_callable
+
+
+def unregister_on_hangup(identifier):
+    """Remove a callback from being called during sighup."""
+    if _on_sighup is None:
+        return
+    try:
+        del _on_sighup[identifier]
+    except KeyboardInterrupt:
+        raise
+    except Exception:
+        # This usually runs as a tear-down step. So we don't want to propagate
+        # most exceptions.
+        trace.mutter('Error occurred during unregister_on_hangup:')
+        trace.log_exception_quietly()
+

=== modified file 'bzrlib/tests/__init__.py'
--- bzrlib/tests/__init__.py	2011-09-19 18:25:05 +0000
+++ bzrlib/tests/__init__.py	2011-09-26 14:29:40 +0000
@@ -4077,6 +4077,7 @@
         'bzrlib.tests.test_smart',
         'bzrlib.tests.test_smart_add',
         'bzrlib.tests.test_smart_request',
+        'bzrlib.tests.test_smart_signals',
         'bzrlib.tests.test_smart_transport',
         'bzrlib.tests.test_smtp_connection',
         'bzrlib.tests.test_source',

=== modified file 'bzrlib/tests/blackbox/test_serve.py'
--- bzrlib/tests/blackbox/test_serve.py	2011-09-26 14:29:38 +0000
+++ bzrlib/tests/blackbox/test_serve.py	2011-09-26 14:29:40 +0000
@@ -314,6 +314,34 @@
             err)
         self.assertServerFinishesCleanly(process)
 
+    def test_bzr_serve_graceful_shutdown(self):
+        big_contents = 'a'*64*1024
+        self.build_tree_contents([('bigfile', big_contents)])
+        process, url = self.start_server_port(['--client-timeout=1.0'])
+        t = transport.get_transport_from_url(url)
+        m = t.get_smart_medium()
+        c = client._SmartClient(m)
+        # Start, but don't finish a response
+        resp, response_handler = c.call_expecting_body('get', 'bigfile')
+        self.assertEqual(('ok',), resp)
+        # Note: process.send_signal is a Python 2.6ism
+        process.send_signal(signal.SIGHUP)
+        # Wait for the server to notice the signal, and then read the actual
+        # body of the response. That way we know that it is waiting for the
+        # request to finish
+        self.assertEqual('Requested to stop gracefully\n',
+                         process.stderr.readline())
+        self.assertEqual('Waiting for 1 client(s) to finish\n',
+                         process.stderr.readline())
+        body = response_handler.read_body_bytes()
+        if body != big_contents:
+            self.fail('Failed to properly read the contents of "bigfile"')
+        # Now that our request is finished, the medium should notice it has
+        # been disconnected.
+        self.assertEqual('', m.read_bytes(1))
+        # And the server should be stopping
+        self.assertEqual(0, process.wait())
+
 
 class TestCmdServeChrooting(TestBzrServeBase):
 

=== added file 'bzrlib/tests/test_smart_signals.py'
--- bzrlib/tests/test_smart_signals.py	1970-01-01 00:00:00 +0000
+++ bzrlib/tests/test_smart_signals.py	2011-09-26 14:29:40 +0000
@@ -0,0 +1,189 @@
+# Copyright (C) 2011 Canonical Ltd
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+
+import os
+import signal
+import threading
+import weakref
+
+from bzrlib import tests, transport
+from bzrlib.smart import client, medium, server, signals
+
+# Windows doesn't define SIGHUP. And while we could just skip a lot of these
+# tests, we often don't actually care about interaction with 'signal', so we
+# can still run the tests for code coverage.
+SIGHUP = getattr(signal, 'SIGHUP', 1)
+
+
+class TestSignalHandlers(tests.TestCase):
+
+    def setUp(self):
+        super(TestSignalHandlers, self).setUp()
+        # This allows us to mutate the signal handler callbacks, but leave it
+        # 'pristine' after the test case.
+        # TODO: Arguably, this could be put into the base test.TestCase, along
+        #       with a tearDown that asserts that all the entries have been
+        #       removed properly. Global state is always a bit messy. A shame
+        #       that we need it for signal handling.
+        orig = signals._setup_on_hangup_dict()
+        self.assertIs(None, orig)
+        def cleanup():
+            signals._on_sighup = None
+        self.addCleanup(cleanup)
+
+    def test_registered_callback_gets_called(self):
+        calls = []
+        def call_me():
+            calls.append('called')
+        signals.register_on_hangup('myid', call_me)
+        signals._sighup_handler(SIGHUP, None)
+        self.assertEqual(['called'], calls)
+        signals.unregister_on_hangup('myid')
+
+    def test_unregister_not_present(self):
+        # We don't want unregister to fail, since it is generally run at times
+        # that shouldn't interrupt other flow.
+        signals.unregister_on_hangup('no-such-id')
+        log = self.get_log()
+        self.assertContainsRe(log, 'Error occurred during unregister_on_hangup:')
+        self.assertContainsRe(log, '(?s)Traceback.*KeyError')
+
+    def test_failing_callback(self):
+        calls = []
+        def call_me():
+            calls.append('called')
+        def fail_me():
+            raise RuntimeError('something bad happened')
+        signals.register_on_hangup('myid', call_me)
+        signals.register_on_hangup('otherid', fail_me)
+        # _sighup_handler should call both, even though it got an exception
+        signals._sighup_handler(SIGHUP, None)
+        signals.unregister_on_hangup('myid')
+        signals.unregister_on_hangup('otherid')
+        log = self.get_log()
+        self.assertContainsRe(log, '(?s)Traceback.*RuntimeError')
+        self.assertEqual(['called'], calls)
+
+    def test_unregister_during_call(self):
+        # _sighup_handler should handle if some callbacks actually remove
+        # themselves while running.
+        calls = []
+        def call_me_and_unregister():
+            signals.unregister_on_hangup('myid')
+            calls.append('called_and_unregistered')
+        def call_me():
+            calls.append('called')
+        signals.register_on_hangup('myid', call_me_and_unregister)
+        signals.register_on_hangup('other', call_me)
+        signals._sighup_handler(SIGHUP, None)
+
+    def test_keyboard_interrupt_propagated(self):
+        # In case we get 'stuck' while running a hangup function, we should
+        # not suppress KeyboardInterrupt
+        def call_me_and_raise():
+            raise KeyboardInterrupt()
+        signals.register_on_hangup('myid', call_me_and_raise)
+        self.assertRaises(KeyboardInterrupt,
+                          signals._sighup_handler, SIGHUP, None)
+        signals.unregister_on_hangup('myid')
+
+    def test_weak_references(self):
+        # TODO: This is probably a very-CPython-specific test
+        # Adding yourself to the callback should not make you immortal
+        # We overrideAttr during the test suite, so that we don't pollute the
+        # original dict. However, we can test that what we override matches
+        # what we are putting there.
+        self.assertIsInstance(signals._on_sighup,
+                              weakref.WeakValueDictionary)
+        calls = []
+        def call_me():
+            calls.append('called')
+        signals.register_on_hangup('myid', call_me)
+        del call_me
+        # Non-CPython might want to do a gc.collect() here
+        signals._sighup_handler(SIGHUP, None)
+        self.assertEqual([], calls)
+
+    def test_not_installed(self):
+        # If you haven't called bzrlib.smart.signals.install_sighup_handler,
+        # then _on_sighup should be None, and all the calls become no-ops.
+        signals._on_sighup = None
+        calls = []
+        def call_me():
+            calls.append('called')
+        signals.register_on_hangup('myid', calls)
+        signals._sighup_handler(SIGHUP, None)
+        signals.unregister_on_hangup('myid')
+        log = self.get_log()
+        self.assertEqual('', log)
+
+    def test_install_sighup_handler(self):
+        # install_sighup_handler should set up a signal handler for SIGHUP, as
+        # well as the signals._on_sighup dict.
+        signals._on_sighup = None
+        orig = signals.install_sighup_handler()
+        if getattr(signal, 'SIGHUP', None) is not None:
+            cur = signal.getsignal(SIGHUP)
+            self.assertEqual(signals._sighup_handler, cur)
+        self.assertIsNot(None, signals._on_sighup)
+        signals.restore_sighup_handler(orig)
+        self.assertIs(None, signals._on_sighup)
+
+
+class TestInetServer(tests.TestCase):
+
+    def create_file_pipes(self):
+        r, w = os.pipe()
+        rf = os.fdopen(r, 'rb')
+        wf = os.fdopen(w, 'wb')
+        return rf, wf
+
+    def test_inet_server_responds_to_sighup(self):
+        t = transport.get_transport('memory:///')
+        content = 'a'*1024*1024
+        t.put_bytes('bigfile', content)
+        factory = server.BzrServerFactory()
+        # Override stdin/stdout so that we can inject our own handles
+        client_read, server_write = self.create_file_pipes()
+        server_read, client_write = self.create_file_pipes()
+        factory._get_stdin_stdout = lambda: (server_read, server_write)
+        factory.set_up(t, None, None, inet=True, timeout=4.0)
+        self.addCleanup(factory.tear_down)
+        started = threading.Event()
+        stopped = threading.Event()
+        def serving():
+            started.set()
+            factory.smart_server.serve()
+            stopped.set()
+        server_thread = threading.Thread(target=serving)
+        server_thread.start()
+        started.wait()
+        client_medium = medium.SmartSimplePipesClientMedium(client_read,
+                            client_write, 'base')
+        client_client = client._SmartClient(client_medium)
+        resp, response_handler = client_client.call_expecting_body('get',
+            'bigfile')
+        signals._sighup_handler(SIGHUP, None)
+        self.assertTrue(factory.smart_server.finished)
+        # We can still finish reading the file content, but more than that, and
+        # the file is closed.
+        v = response_handler.read_body_bytes()
+        if v != content:
+            self.fail('Got the wrong content back, expected 1M "a"')
+        stopped.wait()
+        server_thread.join()
+

=== modified file 'bzrlib/tests/test_smart_transport.py'
--- bzrlib/tests/test_smart_transport.py	2011-09-26 14:29:38 +0000
+++ bzrlib/tests/test_smart_transport.py	2011-09-26 14:29:40 +0000
@@ -18,10 +18,14 @@
 
 # all of this deals with byte strings so this is safe
 from cStringIO import StringIO
+import doctest
 import os
 import socket
 import sys
 import threading
+import time
+
+from testtools.matchers import DocTestMatches
 
 import bzrlib
 from bzrlib import (
@@ -937,6 +941,13 @@
         server_protocol = self.build_protocol_socket('bzr request 2\n')
         self.assertProtocolTwo(server_protocol)
 
+    def test__build_protocol_returns_if_stopping(self):
+        # _build_protocol should notice that we are stopping, and return
+        # without waiting for bytes from the client.
+        server, client_sock = self.create_socket_context(None)
+        server._stop_gracefully()
+        self.assertIs(None, server._build_protocol())
+
     def test_socket_set_timeout(self):
         server, _ = self.create_socket_context(None, timeout=1.23)
         self.assertEqual(1.23, server._client_timeout)
@@ -975,6 +986,17 @@
         data = server.read_bytes(1)
         self.assertEqual('', data)
 
+    def test_socket_wait_for_bytes_with_shutdown(self):
+        server, client_sock = self.create_socket_context(None)
+        t = time.time()
+        # Override the _timer functionality, so that time never increments,
+        # this way, we can be sure we stopped because of the flag, and not
+        # because of a timeout, etc.
+        server._timer = lambda: t
+        server._client_poll_timeout = 0.1
+        server._stop_gracefully()
+        server._wait_for_bytes_with_timeout(1.0)
+
     def test_socket_serve_timeout_closes_socket(self):
         server, client_sock = self.create_socket_context(None, timeout=0.1)
         # This should timeout quickly, and then close the connection so that
@@ -1056,6 +1078,75 @@
 
 class TestSmartTCPServer(tests.TestCase):
 
+    def make_server(self):
+        """Create a SmartTCPServer that we can exercise.
+
+        Note: we don't use SmartTCPServer_for_testing because the testing
+        version overrides lots of functionality like 'serve', and we want to
+        test the raw service.
+
+        This will start the server in another thread, and wait for it to
+        indicate it has finished starting up.
+
+        :return: (server, server_thread)
+        """
+        t = _mod_transport.get_transport_from_url('memory:///')
+        server = _mod_server.SmartTCPServer(t, client_timeout=4.0)
+        server._ACCEPT_TIMEOUT = 0.1
+        # We don't use 'localhost' because that might be an IPv6 address.
+        server.start_server('127.0.0.1', 0)
+        server_thread = threading.Thread(target=server.serve,
+                                         args=(self.id(),))
+        server_thread.start()
+        # Ensure this gets called at some point
+        self.addCleanup(server._stop_gracefully)
+        server._started.wait()
+        return server, server_thread
+
+    def ensure_client_disconnected(self, client_sock):
+        """Ensure that a socket is closed, discarding all errors."""
+        try:
+            client_sock.close()
+        except Exception:
+            pass
+
+    def connect_to_server(self, server):
+        """Create a client socket that can talk to the server."""
+        client_sock = socket.socket()
+        server_info = server._server_socket.getsockname()
+        client_sock.connect(server_info)
+        self.addCleanup(self.ensure_client_disconnected, client_sock)
+        return client_sock
+
+    def connect_to_server_and_hangup(self, server):
+        """Connect to the server, and then hang up.
+        That way it doesn't sit waiting for 'accept()' to timeout.
+        """
+        # If the server has already signaled that the socket is closed, we
+        # don't need to try to connect to it. Not being set, though, the server
+        # might still close the socket while we try to connect to it. So we
+        # still have to catch the exception.
+        if server._stopped.isSet():
+            return
+        try:
+            client_sock = self.connect_to_server(server)
+            client_sock.close()
+        except socket.error, e:
+            # If the server has hung up already, that is fine.
+            pass
+
+    def say_hello(self, client_sock):
+        """Send the 'hello' smart RPC, and expect the response."""
+        client_sock.send('hello\n')
+        self.assertEqual('ok\x012\n', client_sock.recv(5))
+
+    def shutdown_server_cleanly(self, server, server_thread):
+        server._stop_gracefully()
+        self.connect_to_server_and_hangup(server)
+        server._stopped.wait()
+        server._fully_stopped.wait()
+        server_thread.join()
+
     def test_get_error_unexpected(self):
         """Error reported by server with no specific representation"""
         self.overrideEnv('BZR_NO_SMART_VFS', None)
@@ -1081,10 +1172,130 @@
 
     def test_propagates_timeout(self):
         server = _mod_server.SmartTCPServer(None, client_timeout=1.23)
-        server_socket = socket.socket()
-        handler = server._make_handler(server_socket)
+        server_sock, client_sock = portable_socket_pair()
+        handler = server._make_handler(server_sock)
         self.assertEqual(1.23, handler._client_timeout)
 
+    def test_serve_conn_tracks_connections(self):
+        server = _mod_server.SmartTCPServer(None, client_timeout=4.0)
+        server_sock, client_sock = portable_socket_pair()
+        server.serve_conn(server_sock, '-%s' % (self.id(),))
+        self.assertEqual(1, len(server._active_connections))
+        # We still want to talk on the connection. Polling should indicate it
+        # is still active.
+        server._poll_active_connections()
+        self.assertEqual(1, len(server._active_connections))
+        # Closing the socket will end the active thread, and polling will
+        # notice and remove it from the active set.
+        client_sock.close()
+        server._poll_active_connections(0.1)
+        self.assertEqual(0, len(server._active_connections))
+
+    def test_serve_closes_out_finished_connections(self):
+        server, server_thread = self.make_server()
+        # The server is started, connect to it.
+        client_sock = self.connect_to_server(server)
+        # We send and receive on the connection, so that we know the
+        # server-side has seen the connect, and started handling the
+        # results.
+        self.say_hello(client_sock)
+        self.assertEqual(1, len(server._active_connections))
+        # Grab a handle to the thread that is processing our request
+        _, server_side_thread = server._active_connections[0]
+        # Close the connection, ask the server to stop, and wait for the
+        # server to stop, as well as the thread that was servicing the
+        # client request.
+        client_sock.close()
+        # Wait for the server-side request thread to notice we are closed.
+        server_side_thread.join()
+        # Stop the server, it should notice the connection has finished.
+        self.shutdown_server_cleanly(server, server_thread)
+        # The server should have noticed that all clients are gone before
+        # exiting.
+        self.assertEqual(0, len(server._active_connections))
+
+    def test_serve_reaps_finished_connections(self):
+        server, server_thread = self.make_server()
+        client_sock1 = self.connect_to_server(server)
+        # We send and receive on the connection, so that we know the
+        # server-side has seen the connect, and started handling the
+        # results.
+        self.say_hello(client_sock1)
+        server_handler1, server_side_thread1 = server._active_connections[0]
+        client_sock1.close()
+        server_side_thread1.join()
+        # By waiting until the first connection is fully done, the server
+        # should notice after another connection that the first has finished.
+        client_sock2 = self.connect_to_server(server)
+        self.say_hello(client_sock2)
+        server_handler2, server_side_thread2 = server._active_connections[-1]
+        # There is a race condition. We know that client_sock2 has been
+        # registered, but not that _poll_active_connections has been called. We
+        # know that it will be called before the server will accept a new
+        # connection, however. So connect one more time, and assert that we
+        # either have 1 or 2 active connections (never 3), and that the 'first'
+        # connection is not connection 1
+        client_sock3 = self.connect_to_server(server)
+        self.say_hello(client_sock3)
+        # Copy the list, so we don't have it mutating behind our back
+        conns = list(server._active_connections)
+        self.assertEqual(2, len(conns))
+        self.assertNotEqual((server_handler1, server_side_thread1), conns[0])
+        self.assertEqual((server_handler2, server_side_thread2), conns[0])
+        client_sock2.close()
+        client_sock3.close()
+        self.shutdown_server_cleanly(server, server_thread)
+
+    def test_graceful_shutdown_waits_for_clients_to_stop(self):
+        server, server_thread = self.make_server()
+        # We need something big enough that it won't fit in a single recv. So
+        # the server thread gets blocked writing content to the client until we
+        # finish reading on the client.
+        server.backing_transport.put_bytes('bigfile',
+            'a'*1024*1024)
+        client_sock = self.connect_to_server(server)
+        self.say_hello(client_sock)
+        _, server_side_thread = server._active_connections[0]
+        # Start the RPC, but don't finish reading the response
+        client_medium = medium.SmartClientAlreadyConnectedSocketMedium(
+            'base', client_sock)
+        client_client = client._SmartClient(client_medium)
+        resp, response_handler = client_client.call_expecting_body('get',
+            'bigfile')
+        self.assertEqual(('ok',), resp)
+        # Ask the server to stop gracefully, and wait for it.
+        server._stop_gracefully()
+        self.connect_to_server_and_hangup(server)
+        server._stopped.wait()
+        # It should not be accepting another connection.
+        self.assertRaises(socket.error, self.connect_to_server, server)
+        # It should also not be fully stopped
+        server._fully_stopped.wait(0.01)
+        self.assertFalse(server._fully_stopped.isSet())
+        response_handler.read_body_bytes()
+        client_sock.close()
+        server_side_thread.join()
+        server_thread.join()
+        self.assertTrue(server._fully_stopped.isSet())
+        log = self.get_log()
+        self.assertThat(log, DocTestMatches("""\
+    INFO  Requested to stop gracefully
+... Stopping SmartServerSocketStreamMedium(client=('127.0.0.1', ...
+    INFO  Waiting for 1 client(s) to finish
+""", flags=doctest.ELLIPSIS|doctest.REPORT_UDIFF))
+
+    def test_stop_gracefully_tells_handlers_to_stop(self):
+        server, server_thread = self.make_server()
+        client_sock = self.connect_to_server(server)
+        self.say_hello(client_sock)
+        server_handler, server_side_thread = server._active_connections[0]
+        self.assertFalse(server_handler.finished)
+        server._stop_gracefully()
+        self.assertTrue(server_handler.finished)
+        client_sock.close()
+        self.connect_to_server_and_hangup(server)
+        server_thread.join()
+
 
 class SmartTCPTests(tests.TestCase):
     """Tests for connection/end to end behaviour using the TCP server.

=== modified file 'bzrlib/trace.py'
--- bzrlib/trace.py	2011-07-15 14:13:32 +0000
+++ bzrlib/trace.py	2011-09-26 14:29:40 +0000
@@ -559,6 +559,10 @@
     try:
         sys.stdout.flush()
         sys.stderr.flush()
+    except ValueError, e:
+        # On Windows, I get ValueError calling stdout.flush() on a closed
+        # handle
+        pass
     except IOError, e:
         import errno
         if e.errno in [errno.EINVAL, errno.EPIPE]:

=== modified file 'doc/en/release-notes/bzr-2.5.txt'
--- doc/en/release-notes/bzr-2.5.txt	2011-09-26 14:29:38 +0000
+++ doc/en/release-notes/bzr-2.5.txt	2011-09-26 14:29:40 +0000
@@ -20,6 +20,16 @@
 
 .. New commands, options, etc that users may wish to try out.
 
+* ``bzr serve`` will now disconnect clients if they have not issued an RPC
+  request after 5minutes. On POSIX platforms, this will also happen for
+  ``bzr serve --inet``. This can be overridden with the configuration
+  variable ``serve.client_timeout`` or in the command line parameter
+  ``bzr serve --client-timeout=X``. Further, it is possible to request
+  ``bzr serve [--inet]`` to shutdown gracefully by sending SIGHUP. It will
+  finish the current request, and then close the connection.
+  (John Arbash Meinel, #824797, #795025)
+
+
 Improvements
 ************
 
@@ -161,12 +171,6 @@
   The name change also affects ``bzr missing``.
   (Martin von Gagern)
 
-* ``bzr serve`` will now disconnect clients if they have not issued an RPC
-  request after 5minutes. On POSIX platforms, this will also happen for
-  ``bzr serve --inet``. This can be overridden with the configuration
-  variable ``serve.client_timeout`` or in the command line parameter
-  ``bzr serve --client-timeout=X``. (John Arbash Meinel, #824797)
-
 * ``config.Option`` can now declare ``default_from_env``, a list of
   environment variables to get a default value from. (Vincent Ladeuil)