Merge lp:~jameinel/launchpad/lp-serve-child-hangup into lp:launchpad

[John A Meinel]

Children spawned from launchpad-forking-service should hangup if nobody has talked to them in a couple minutes.

This branch builds on my earlier branch, which has the children spawned by LPForkingService clean themselves up more readily.

This adds code so that the blocking calls will be unblocked after 2 minutes. At the moment, if the Conch server requests a fork, but is unable to actually connect to the child, that child ends up hanging forever on a file handle that it will never open.

Originally, I got this working using a loop an O_NONBLOCK. But after doing some testing, fcntl.fcntl(...) was unable to actually change the file descriptor to be blocking again. It looks good in man, and it does unset the flag, but I still get EAGAIN failures in the smart server code.

I don't really like spawning a thread to send SIGUSR1 back to myself, but it seemed the best tradeoff. If we want, we could even make it SIGTERM or something, since we know we are going to kill the process if the connection fails.

I figured 2 minutes was reasonable. This is the time from a successful ssh handshake until we actually connect to a newly spawned child. Even the worst-case time that I've seen was about 30s for a child to be spawned. So this gives us a 4x margin of error.

This is cleanup related to bug #717345, but it doesn't directly fix the problem. That will be in my next branch.

[Martin Pool (mbp)]

On 17 February 2011 08:44, John A Meinel <email address hidden> wrote:
> John A Meinel has proposed merging lp:~jameinel/launchpad/lp-serve-child-hangup into lp:launchpad with lp:~jameinel/launchpad/lp-forking-serve-cleaner-childre as a prerequisite.
>
> Requested reviews:
>  Launchpad code reviewers (launchpad-reviewers)
>
> For more details, see:
> https://code.launchpad.net/~jameinel/launchpad/lp-serve-child-hangup/+merge/50055
>
> This branch builds on my earlier branch, which has the children spawned by LPForkingService clean themselves up more readily.
>
> This adds code so that the blocking calls will be unblocked after 2 minutes. At the moment, if the Conch server requests a fork, but is unable to actually connect to the child, that child ends up hanging forever on a file handle that it will never open.
>
> Originally, I got this working using a loop an O_NONBLOCK. But after doing some testing, fcntl.fcntl(...) was unable to actually change the file descriptor to be blocking again. It looks good in man, and it does unset the flag, but I still get EAGAIN failures in the smart server code.
>
> I don't really like spawning a thread to send SIGUSR1 back to myself, but it seemed the best tradeoff. If we want, we could even make it SIGTERM or something, since we know we are going to kill the process if the connection fails.

I don't really like that either, especially this is already getting to
be a fairly complex set of processes.

Why can't you make the file handles nonblocking and just leave them
so? Is it because they're later inherited by some other process?

If you do need a signal to interrupt things, how about just using alarm(2)?

[John A Meinel (jameinel)]

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

...

>> I don't really like spawning a thread to send SIGUSR1 back to myself, but it seemed the best tradeoff. If we want, we could even make it SIGTERM or something, since we know we are going to kill the process if the connection fails.
>
> I don't really like that either, especially this is already getting to
> be a fairly complex set of processes.
>
> Why can't you make the file handles nonblocking and just leave them
> so? Is it because they're later inherited by some other process?

Because the Smart server code expects blocking file handles.
SmartServerPipeStreamMedium IIRC.

>
> If you do need a signal to interrupt things, how about just using alarm(2)?

Because alarm doesn't seem to be exposed to python, and I want the
ability to cancel the callback. threading.Timer() gives me that exact
functionality.

I would be fine using SIGALRM if you prefer that to SIGUSR1.

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

iEYEARECAAYFAk1cS8UACgkQJdeBCYSNAAMupgCgjFH3eTQXY6vxSoZkp3inrYs7
CjQAnREQQ/NXmuuyz9Rkl8tUMNIFEED6
=iiu5
-----END PGP SIGNATURE-----

[Robert Collins (lifeless)]

What about using dup2 to duplicate the fd, make one copy nonblocking
and use that until we're up and running?

[John A Meinel (jameinel)]

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

On 2/16/2011 4:28 PM, Robert Collins wrote:
> What about using dup2 to duplicate the fd, make one copy nonblocking
> and use that until we're up and running?
>

The issue is we want non-blocking in the 'os.open()' call, until
everything is open, then we want to reset it to blocking.

You can use tricks with fcntl.fcntl(... F_SETFD) according to the man
pages. But testing showed it doesn't actually set the file back to
blocking mode.

I would expect something similar from dup2.

John
=:->

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

iEYEARECAAYFAk1cUS8ACgkQJdeBCYSNAAND1ACfZtlcchz8RU9L+0XIw/MZdsep
XscAmwRiaaebSzfpYa6dvAxNGU4jQNfP
=Fj9R
-----END PGP SIGNATURE-----

[Robert Collins (lifeless)]

On Thu, Feb 17, 2011 at 11:38 AM, John A Meinel <email address hidden> wrote:
> I would expect something similar from dup2.

I would expect dup2 to workaround those bugs :) - its why I suggested it.

-Rob

[Martin Pool (mbp)]

On 17 February 2011 10:01, Robert Collins <email address hidden> wrote:
> On Thu, Feb 17, 2011 at 11:38 AM, John A Meinel <email address hidden> wrote:
>> I would expect something similar from dup2.
>
> I would expect dup2 to workaround those bugs :) - its why I suggested it.

We want to do a nonblocking open, so we don't have an fd to dup until
we've done that.

After opening it, maybe we could dup it, then make the new copy
blocking, then close the original. It's possible that would work
around it. On the other hand if in general we can't make open fds
blocking, it may not fix it.

[John A Meinel (jameinel)]

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

On 2/16/2011 5:43 PM, Martin Pool wrote:
> On 17 February 2011 10:01, Robert Collins <email address hidden> wrote:
>> On Thu, Feb 17, 2011 at 11:38 AM, John A Meinel <email address hidden> wrote:
>>> I would expect something similar from dup2.
>>
>> I would expect dup2 to workaround those bugs :) - its why I suggested it.
>
> We want to do a nonblocking open, so we don't have an fd to dup until
> we've done that.
>
> After opening it, maybe we could dup it, then make the new copy
> blocking, then close the original. It's possible that would work
> around it. On the other hand if in general we can't make open fds
> blocking, it may not fix it.
>

Right, it *might* work to, open non-blocking, set it to blocking, then
dup2 it and use the new duplicated one. I can try it, but honestly, the
code is much clearer with the Timer rather than the
blocking/non-blocking dance. I have to write a loop that times out, etc.
Rather than just getting a SIG* after 120s of inactivity.

If you prefer SIGALRM, or alarm(2), then we can use ctypes/pyrex/whatever.

John
=:->

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

iEYEARECAAYFAk1dN90ACgkQJdeBCYSNAAP5nwCgkPAFSuCwQP6+l96tNXjB1PAP
XBQAnj5JBVESjYtPmxJiz13EZ7iJeAt3
=KHsN
-----END PGP SIGNATURE-----

[John A Meinel (jameinel)]

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

...

>
> If you do need a signal to interrupt things, how about just using alarm(2)?
>

It turns out it is exposed, but only conditionally, and not where I
expected it.

import signal
signal.alarm()

If you prefer, I'd be happy to use that instead. Should I be catching
SIGALRM, or should I just let it propagate to kill the child?

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

iEYEARECAAYFAk1j/c0ACgkQJdeBCYSNAANc0QCeLE9a+01g55AbrIa5EVWGmNdh
tukAn3Ybnl0SCYTqoxNKIu0VHvz95+r7
=LF2C
-----END PGP SIGNATURE-----

[Martin Pool (mbp)]

> It turns out it is exposed, but only conditionally, and not where I
> expected it.
>
> import signal
> signal.alarm()
>
> If you prefer, I'd be happy to use that instead. Should I be catching
> SIGALRM, or should I just let it propagate to kill the child?

Since it's supposed to be an unusual case that the connection fails
half way through, I think just letting it kill the client should be
fine. The parent should wait on them and log their exit status.

=== modified file 'bzrplugins/lpserve/__init__.py'
--- bzrplugins/lpserve/__init__.py	2011-02-17 21:47:07 +0000
+++ bzrplugins/lpserve/__init__.py	2011-03-02 13:07:04 +0000
@@ -15,6 +15,7 @@
 
 
 import errno
+import fcntl
 import logging
 import os
 import resource
@@ -31,6 +32,7 @@
 from bzrlib.option import Option
 from bzrlib import (
     commands,
+    errors,
     lockdir,
     osutils,
     trace,
@@ -309,6 +311,11 @@
     SLEEP_FOR_CHILDREN_TIMEOUT = 1.0
     WAIT_FOR_REQUEST_TIMEOUT = 1.0 # No request should take longer than this to
                                    # be read
+    CHILD_CONNECT_TIMEOUT = 120   # If we get a fork() request, but nobody
+                                  # connects just exit
+                                  # On a heavily loaded server, it could take a
+                                  # couple secs, but it should never take
+                                  # minutes
 
     _fork_function = os.fork
 
@@ -324,6 +331,7 @@
         # Map from pid => (temp_path_for_handles, request_socket)
         self._child_processes = {}
         self._children_spawned = 0
+        self._child_connect_timeout = self.CHILD_CONNECT_TIMEOUT
 
     def _create_master_socket(self):
         self._server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -372,28 +380,94 @@
         signal.signal(signal.SIGCHLD, signal.SIG_DFL)
         signal.signal(signal.SIGTERM, signal.SIG_DFL)
 
+    def _compute_paths(self, base_path):
+        stdin_path = os.path.join(base_path, 'stdin')
+        stdout_path = os.path.join(base_path, 'stdout')
+        stderr_path = os.path.join(base_path, 'stderr')
+        return (stdin_path, stdout_path, stderr_path)
+
     def _create_child_file_descriptors(self, base_path):
-        stdin_path = os.path.join(base_path, 'stdin')
-        stdout_path = os.path.join(base_path, 'stdout')
-        stderr_path = os.path.join(base_path, 'stderr')
+        stdin_path, stdout_path, stderr_path = self._compute_paths(base_path)
         os.mkfifo(stdin_path)
         os.mkfifo(stdout_path)
         os.mkfifo(stderr_path)
 
-    def _bind_child_file_descriptors(self, base_path):
-        stdin_path = os.path.join(base_path, 'stdin')
-        stdout_path = os.path.join(base_path, 'stdout')
-        stderr_path = os.path.join(base_path, 'stderr')
+    def _set_blocking(self, fd):
+        """Change the file descriptor to unset the O_NONBLOCK flag."""
+        flags = fcntl.fcntl(fd, fcntl.F_GETFD)
+        flags = flags & (~os.O_NONBLOCK)
+        fcntl.fcntl(fd, fcntl.F_SETFD, flags)
+
+    def _get_child_connect_timeout(self):
+        """signal.alarm only supports 1s granularity.
+
+        We have to make sure we don't ever send 0, which would not generate an
+        alarm.
+        """
+        timeout = int(self._child_connect_timeout)
+        if timeout <= 0:
+            timeout = 1
+        return timeout
+
+    def _open_handles(self, base_path):
+        """Open the given file handles.
+
+        This will attempt to open all of these file handles, but will not block
+        while opening them, timing out after self._child_connect_timeout
+        seconds.
+
+        :param base_path: The directory where all FIFOs are located
+        :return: (stdin_fid, stdout_fid, stderr_fid)
+        """
+        stdin_path, stdout_path, stderr_path = self._compute_paths(base_path)
         # These open calls will block until another process connects (which
         # must connect in the same order)
-        stdin_fid = os.open(stdin_path, os.O_RDONLY)
-        stdout_fid = os.open(stdout_path, os.O_WRONLY)
-        stderr_fid = os.open(stderr_path, os.O_WRONLY)
+        fids = []
+        to_open = [(stdin_path, os.O_RDONLY), (stdout_path, os.O_WRONLY),
+                   (stderr_path, os.O_WRONLY)]
+        signal.alarm(self._get_child_connect_timeout())
+        tstart = time.time()
+        for path, flags in to_open:
+            try:
+                fids.append(os.open(path, flags))
+            except OSError, e:
+                if e.errno == errno.EINTR:
+                    error = ('After %.3fs we failed to open %s, exiting'
+                             % (time.time() - tstart, path,))
+                    trace.warning(error)
+                    for fid in fids:
+                        try:
+                            os.close(fid)
+                        except OSError:
+                            pass
+                    self._cleanup_fifos(base_path)
+                    raise errors.BzrError(error)
+                raise
+        # If we get to here, that means all the handles were opened
+        # successfully, so cancel the wakeup call.
+        signal.alarm(0)
+        return fids
+
+    def _cleanup_fifos(self, base_path):
+        """Remove the FIFO objects and directory from disk."""
+        stdin_path, stdout_path, stderr_path = self._compute_paths(base_path)
+        # Now that we've opened the handles, delete everything so that we don't
+        # leave garbage around. Because the open() is done in blocking mode, we
+        # know that someone has already connected to them, and we don't want
+        # anyone else getting confused and connecting.
+        # See [Decision #5]
+        os.remove(stdin_path)
+        os.remove(stdout_path)
+        os.remove(stderr_path)
+        os.rmdir(base_path)
+
+    def _bind_child_file_descriptors(self, base_path):
         # Note: by this point bzrlib has opened stderr for logging
         #       (as part of starting the service process in the first place).
         #       As such, it has a stream handler that writes to stderr. logging
         #       tries to flush and close that, but the file is already closed.
         #       This just supresses that exception
+        stdin_fid, stdout_fid, stderr_fid = self._open_handles(base_path)
         logging.raiseExceptions = False
         sys.stdin.close()
         sys.stdout.close()
@@ -407,15 +481,7 @@
         ui.ui_factory.stdin = sys.stdin
         ui.ui_factory.stdout = sys.stdout
         ui.ui_factory.stderr = sys.stderr
-        # Now that we've opened the handles, delete everything so that we don't
-        # leave garbage around. Because the open() is done in blocking mode, we
-        # know that someone has already connected to them, and we don't want
-        # anyone else getting confused and connecting.
-        # See [Decision #5]
-        os.remove(stderr_path)
-        os.remove(stdout_path)
-        os.remove(stdin_path)
-        os.rmdir(base_path)
+        self._cleanup_fifos(base_path)
 
     def _close_child_file_descriptors(self):
         sys.stdin.close()
@@ -724,6 +790,15 @@
             self._should_terminate.set()
             conn.sendall('ok\nquit command requested... exiting\n')
             conn.close()
+        elif request.startswith('child_connect_timeout '):
+            try:
+                value = int(request.split(' ', 1)[1])
+            except ValueError, e:
+                conn.sendall('FAILURE: %r\n' % (e,))
+            else:
+                self._child_connect_timeout = value
+                conn.sendall('ok\n')
+            conn.close()
         elif request.startswith('fork ') or request.startswith('fork-env '):
             command_argv, env = self._parse_fork_request(conn, client_addr,
                                                          request)
 
=== modified file 'bzrplugins/lpserve/test_lpserve.py'
--- bzrplugins/lpserve/test_lpserve.py	2010-11-08 22:43:58 +0000
+++ bzrplugins/lpserve/test_lpserve.py	2011-03-02 13:07:04 +0000
@@ -3,6 +3,7 @@
 
 import errno
 import os
+import shutil
 import signal
 import socket
 import subprocess
@@ -13,6 +14,7 @@
 from testtools import content
 
 from bzrlib import (
+    errors,
     osutils,
     tests,
     trace,
@@ -263,6 +265,46 @@
                                                 one_byte_at_a_time=True)
         self.assertStartsWith(response, 'FAILURE\n')
 
+    def test_child_connection_timeout(self):
+        self.assertEqual(self.service.CHILD_CONNECT_TIMEOUT,
+                         self.service._child_connect_timeout)
+        response = self.send_message_to_service('child_connect_timeout 1\n')
+        self.assertEqual('ok\n', response)
+        self.assertEqual(1, self.service._child_connect_timeout)
+
+    def test_child_connection_timeout_bad_float(self):
+        self.assertEqual(self.service.CHILD_CONNECT_TIMEOUT,
+                         self.service._child_connect_timeout)
+        response = self.send_message_to_service('child_connect_timeout 1.2\n')
+        self.assertStartsWith(response, 'FAILURE:')
+
+    def test_child_connection_timeout_no_val(self):
+        response = self.send_message_to_service('child_connect_timeout \n')
+        self.assertStartsWith(response, 'FAILURE:')
+
+    def test_child_connection_timeout_bad_val(self):
+        response = self.send_message_to_service('child_connect_timeout b\n')
+        self.assertStartsWith(response, 'FAILURE:')
+
+    def test__open_handles_will_timeout(self):
+        # signal.alarm() has only 1-second granularity. :(
+        self.service._child_connect_timeout = 1
+        tempdir = tempfile.mkdtemp(prefix='testlpserve-')
+        self.addCleanup(shutil.rmtree, tempdir, ignore_errors=True)
+        os.mkfifo(os.path.join(tempdir, 'stdin'))
+        os.mkfifo(os.path.join(tempdir, 'stdout'))
+        os.mkfifo(os.path.join(tempdir, 'stderr'))
+        def noop_on_alarm(signal, frame):
+            return
+        signal.signal(signal.SIGALRM, noop_on_alarm)
+        self.addCleanup(signal.signal, signal.SIGALRM, signal.SIG_DFL)
+        e = self.assertRaises(errors.BzrError,
+            self.service._open_handles, tempdir)
+        self.assertContainsRe(str(e), r'After \d+.\d+s we failed to open.*')
+        # Even though it timed out, we still cleanup the temp dir
+        self.assertFalse(os.path.exists(tempdir))
+
+
 
 class TestCaseWithSubprocess(tests.TestCaseWithTransport):
     """Override the bzr start_bzr_subprocess command.
@@ -452,9 +494,9 @@
         stderr_path = os.path.join(path, 'stderr')
         # The ordering must match the ordering of the service or we get a
         # deadlock.
-        child_stdin = open(stdin_path, 'wb')
-        child_stdout = open(stdout_path, 'rb')
-        child_stderr = open(stderr_path, 'rb')
+        child_stdin = open(stdin_path, 'wb', 0)
+        child_stdout = open(stdout_path, 'rb', 0)
+        child_stderr = open(stderr_path, 'rb', 0)
         return child_stdin, child_stdout, child_stderr
 
     def communicate_with_fork(self, path, stdin=None):
@@ -484,6 +526,24 @@
         self.assertEqual('', stderr_content)
         self.assertReturnCode(0, sock)
 
+    def DONT_test_fork_lp_serve_multiple_hello(self):
+        # This ensures that the fifos are all set to blocking mode
+        # We can't actually run this test, because by default 'bzr serve
+        # --inet' does not flush after each message. So we end up blocking
+        # forever waiting for the server to finish responding to the first
+        # request.
+        path, _, sock = self.send_fork_request('lp-serve --inet 2')
+        child_stdin, child_stdout, child_stderr = self._get_fork_handles(path)
+        child_stdin.write('hello\n')
+        child_stdin.flush()
+        self.assertEqual('ok\x012\n', child_stdout.read())
+        child_stdin.write('hello\n')
+        self.assertEqual('ok\x012\n', child_stdout.read())
+        child_stdin.close()
+        self.assertEqual('', child_stderr.read())
+        child_stdout.close()
+        child_stderr.close()
+
     def test_fork_replay(self):
         path, _, sock = self.send_fork_request('launchpad-replay')
         stdout_content, stderr_content = self.communicate_with_fork(path,
@@ -540,6 +600,29 @@
     def test_sigint_exits_nicely(self):
         self._check_exits_nicely(signal.SIGINT)
 
+    def test_child_exits_eventually(self):
+        # We won't ever bind to the socket the child wants, and after some
+        # time, the child should exit cleanly.
+        # First, tell the subprocess that we want children to exit quickly.
+        # *sigh* signal.alarm only has 1s resolution, so this test is slow.
+        response = self.send_message_to_service('child_connect_timeout 1\n')
+        self.assertEqual('ok\n', response)
+        # Now request a fork
+        path, pid, sock = self.send_fork_request('rocks')
+        # # Open one handle, but not all of them
+        stdin_path = os.path.join(path, 'stdin')
+        stdout_path = os.path.join(path, 'stdout')
+        stderr_path = os.path.join(path, 'stderr')
+        child_stdin = open(stdin_path, 'wb')
+        # We started opening the child, but stop before we get all handles
+        # open. After 1 second, the child should get signaled and die.
+        # The master process should notice, and tell us the status of the
+        # exited child.
+        val = sock.recv(4096)
+        self.assertEqual('exited\n%s\n' % (signal.SIGALRM,), val)
+        # The master process should clean up after the now deceased child.
+        self.failIfExists(path)
+
 
 class TestCaseWithLPForkingServiceDaemon(
     TestCaseWithLPForkingServiceSubprocess):