[MERGE] hpss startup/shutdown hooks

[John Arbash Meinel]

Robert Collins wrote:
> On Mon, 2007-03-26 at 01:47 -0400, Andrew Bennetts wrote:
> 
> The branch failed on PQM because of race conditions in the existing
> startup-shutdown code. I've fixed these I believe, but it is enough of a
> change that i'd like a re-review.
> 
> Cheers,
> Rob

...

>      def _restoreHooks(self):
> -        bzrlib.branch.Branch.hooks = self._preserved_hooks
> +        for klass, hooks in self._preserved_hooks.items():
> +            setattr(klass, 'hooks', hooks)

Why are you using setattr() here? Rather than just
klass.hooks = hooks

I think setattr is when the member is unknown.

...

> +    def test_server_closes_listening_sock_on_shutdown_after_request(self):
> +        """The server should close its listening socket when it's stopped."""
> +        self.setUpServer()
> +        server = self.server
> +        self.transport.has('.')
> +        self.tearDownServer()
> +        # if the listening socket has closed, we should get a BADFD error
> +        # when connecting, rather than a hang.
> +        transport = smart.SmartTCPTransport(server.get_url())
> +        self.assertRaises(errors.ConnectionError, transport.has, '.')
> +

Tests that "fail" by hanging aren't particularly nice. Especially when
they don't actually fail because of an eventual timeout.

I know we don't want to be strictly timing related (because then you
have tests which randomly fail because the machine needed to swap). Is
there a better way to test this?

>  
>  class WritableEndToEndTests(SmartTCPTests):
>      """Client to server tests that require a writable transport."""
> @@ -901,7 +926,47 @@
>          self.setUpServer(readonly=True)
>          self.assertRaises(errors.TransportNotPossible, self.transport.mkdir,
>              'foo')

v- I would generally prefer to track the 'reason' for why the callback
was called. So that you aren't accidentally getting 'server_started'
calls when you should be getting 'server_stopped'. I do realize that you
have only one hook registered, so I suppose that is good enough.
> -        
> +
> +
> +class TestServerHooks(SmartTCPTests):
> +
> +    def capture_server_call(self, backing_url, public_url):
> +        """Record a server_started|stopped hook firing."""
> +        self.hook_calls.append((backing_url, public_url))
> +
> +    def test_server_started_hook(self):
> +        """The server_started hook fires when the server is started."""
> +        self.hook_calls = []
> +        smart.SmartTCPServer.hooks.install_hook('server_started',
> +            self.capture_server_call)
> +        self.setUpServer()
> +        # at this point, the server will be starting a thread up.
> +        # there is no indicator at the moment, so bodge it by doing a request.
> +        self.transport.has('.')
> +        self.assertEqual([(self.backing_transport.base, self.transport.base)],
> +            self.hook_calls)
> +
> +    def test_server_stopped_hook_simple(self):
> +        """The server_stopped hook fires when the server is stopped."""
> +        self.hook_calls = []
> +        smart.SmartTCPServer.hooks.install_hook('server_stopped',
> +            self.capture_server_call)
> +        self.setUpServer()
> +        result = [(self.backing_transport.base, self.transport.base)]
> +        # check the stopping message isn't emitted up front.
> +        self.assertEqual([], self.hook_calls)
> +        # nor after a single message
> +        self.transport.has('.')
> +        self.assertEqual([], self.hook_calls)
> +        # clean up the server
> +        server = self.server
> +        self.tearDownServer()
> +        # now it should have fired.
> +        self.assertEqual(result, self.hook_calls)
> +
> +# TODO: test that when the server suffers an exception that it calls the
> +# server-stopped hook.
> +

...

v- Invalid markup. Did you want:

:cvar hooks: An instance ...


>  class SmartTCPServer(object):
> -    """Listens on a TCP socket and accepts connections from smart clients"""
> +    """Listens on a TCP socket and accepts connections from smart clients
> +
> +    hooks: An instance of SmartServerHooks.
> +    """
>  

v- Do I remember correctly that hooks with timeouts cause problems
because of non-blocking I/O? (You get exceptions for "operation would
block"). I *think* we can handle that as an EAGAIN, but I don't remember
exactly.

> +        from socket import error as socket_error
> +        self._socket_error = socket_error
>          self._server_socket = socket.socket()
>          self._server_socket.bind((host, port))
> -        self.port = self._server_socket.getsockname()[1]
> +        self._sockname = self._server_socket.getsockname()
> +        self.port = self._sockname[1]
>          self._server_socket.listen(1)
>          self._server_socket.settimeout(1)
>          self.backing_transport = backing_transport
> +        self._started = threading.Event()
> +        self._stopped = threading.Event()
>  
>      def serve(self):
>          # let connections timeout so that we get a chance to terminate
>          # Keep a reference to the exceptions we want to catch because the socket
>          # module's globals get set to None during interpreter shutdown.
>          from socket import timeout as socket_timeout
> -        from socket import error as socket_error
>          self._should_terminate = False
> -        while not self._should_terminate:
> -            try:
> -                self.accept_and_serve()
> -            except socket_timeout:
> -                # just check if we're asked to stop
> -                pass
> -            except socket_error, e:
> -                trace.warning("client disconnected: %s", e)
> -                pass
> +        for hook in SmartTCPServer.hooks['server_started']:
> +            hook(self.backing_transport.base, self.get_url())
> +        self._started.set()
> +        try:
> +            try:
> +                while not self._should_terminate:
> +                    try:
> +                        conn, client_addr = self._server_socket.accept()
> +                    except socket_timeout:
> +                        # just check if we're asked to stop
> +                        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)
> +                    else:
> +                        self.serve_conn(conn)

...

v- Is there a reason you are doing this as:

class C1():
  ...

class C2():
  ...

C1.hooks = C2()

Rather than doing it (IMO) the right way:

class C2():
  ...

class C1():
  hooks = C2()
  ...


Modifying classes after their definitions seems harder to follow and
understand. I understand that you are adding the doc string at least.

> +class SmartServerHooks(Hooks):
> +    """Hooks for the smart server."""
> +
> +    def __init__(self):
> +        """Create the default hooks.
> +
> +        These are all empty initially, because by default nothing should get
> +        notified.
> +        """
> +        Hooks.__init__(self)
> +        # Introduced in 0.16:
> +        # invoked whenever the server starts serving a directory.
> +        # The api signature is (backing url, public url).
> +        self['server_started'] = []
> +        # Introduced in 0.16:
> +        # invoked whenever the server stops serving a directory.
> +        # The api signature is (backing url, public url).
> +        self['server_stopped'] = []
> +
> +SmartTCPServer.hooks = SmartServerHooks()
>  

Otherwise +1 from me.

John
=:->