Merge lp:~james-w/aptdaemon/fix-455861 into lp:aptdaemon

[James Westby]

This implements locking as the io events are delivered asynchronously.

It ensures that if you call .stop() it won't close the socket just before
something else uses it.

I think that exiting quietly if the socket is already closed is the
correct behaviour.

A little testing here seems to show that this at least stops the crashes,
but it's a race condition, so it's hard to know for sure.

Thanks,

James

[Sebastian Heinlein (glatzor)]

Actually you don't need to introduce a lock here, since the event sources run in the same main loop.

I think that the root of the problem is iterating on the main loop after closing the socket, which will result in processing already queued io events which will fail.

Furthermore I increased the priority of the hangup callback, so that it is preferred to copy_stdout.

See bzr diff -r 340..343

=== modified file 'aptdaemon/debconf.py'
--- aptdaemon/debconf.py 2009-10-02 11:27:03 +0000
+++ aptdaemon/debconf.py 2010-03-17 05:59:24 +0000
@@ -89,21 +89,12 @@
         """Stop listening on the socket."""
         logging.debug("debconf.stop()")
         self.socket.close()
- # ensure outstanding gio messages are processed
- context = glib.main_context_default()
- while context.pending():
- context.iteration()
         gobject.source_remove(self._listener_id)
         self._listener_id = None

     def _accept_connection(self, source, condition):
         """Callback for new connections of the passthrough frontend."""
         log.debug("New passthrough connection")
- # ensure outstanding gio messages are processed (to ensure
- # that _hangup was run on the previous connection, see LP: #432607)
- context = glib.main_context_default()
- while context.pending():
- context.iteration()
         if self._active_conn:
             raise IOError, "Multiple debconf connections not supported"
         conn, addr = source.accept()
@@ -112,15 +103,18 @@
                                        stdin=subprocess.PIPE,
                                        stdout=subprocess.PIPE,
                                        env=self._get_debconf_env())
- w = gobject.io_add_watch(conn, gobject.IO_IN|gobject.IO_HUP|gobject.IO_ERR,
- self._copy_conn, self.helper.stdin)
- self._watch_ids.append(w)
- w= gobject.io_add_watch(self.helper.stdout, gobject.IO_IN,
- self._copy_stdout, conn)
- self._watch_ids.append(w)
- w = gobject.io_add_watch(self.helper.stdout, gobject.IO_HUP|gobject.IO_ERR,
- self._hangup)
- self._watch_ids.append(w)
+ wid = gobject.io_add_watch(conn,
+ gobject.IO_IN|gobject.IO_HUP|gobject.IO_ERR,
+ self._copy_conn, self.helper.stdin)
+ self._watch_ids.append(wid)
+ wid = gobject.io_add_watch(self.helper.stdout, gobject.IO_IN,
+ self._copy_stdout, conn)
+ self._watch_ids.append(wid)
+ wid = gobject.io_add_watch(self.helper.stdout,
+ gobject.IO_HUP|gobject.IO_ERR,
+ self._hangup,
+ priority=gobject.PRIORITY_HIGH)
+ self._watch_ids.append(wid)
         return True

[Michael Vogt (mvo)]

I think we still need some synchronization here because stop() maybe called by the aptdaemon when there are still unprocessed io events. We also need to ensure that all pending io events are processed because otherwise we may kill debconf-communicate (in _hangup) when it has not processed all messages yet, this could let to "forgetting" some of the debconf communication because it never quite made it to the socket.

[Sebastian Heinlein (glatzor)]

You could reduce the patch to the self._socket_closed setting/checking.

[Michael Vogt (mvo)]

This was fixed in trunk/ (and 0.2.x) in a slightly different way. Many thnaks for the initial patch.

[Michael Vogt (mvo)]

Set to merge (even if not the exact patch got merged)

=== modified file 'aptdaemon/debconf.py'
--- aptdaemon/debconf.py	2009-10-02 11:27:03 +0000
+++ aptdaemon/debconf.py	2010-03-15 02:18:14 +0000
@@ -33,6 +33,7 @@
 import socket
 import subprocess
 import tempfile
+import threading
 
 import gobject
 
@@ -65,6 +66,8 @@
         self._listener_id = None
         self._active_conn = None
         self._watch_ids = []
+        self._socket_lock = threading.Lock()
+        self._socket_closed = False
 
     def _get_debconf_env(self):
         """Returns a dictonary of the environment variables required by
@@ -88,40 +91,51 @@
     def stop(self):
         """Stop listening on the socket."""
         logging.debug("debconf.stop()")
-        self.socket.close()
-        # ensure outstanding gio messages are processed
-        context = glib.main_context_default()
-        while context.pending():
-            context.iteration()
-        gobject.source_remove(self._listener_id)
-        self._listener_id = None
+        self._socket_lock.acquire()
+        try:
+            self.socket.close()
+            self._socket_closed = True
+            # ensure outstanding gio messages are processed
+            context = glib.main_context_default()
+            while context.pending():
+                context.iteration()
+            gobject.source_remove(self._listener_id)
+            self._listener_id = None
+        finally:
+            self._socket_lock.release()
 
     def _accept_connection(self, source, condition):
         """Callback for new connections of the passthrough frontend."""
         log.debug("New passthrough connection")
-        # ensure outstanding gio messages are processed (to ensure
-        # that _hangup was run on the previous connection, see LP: #432607)
-        context = glib.main_context_default()
-        while context.pending():
-            context.iteration()
-        if self._active_conn:
-            raise IOError, "Multiple debconf connections not supported"
-        conn, addr = source.accept()
-        self._active_conn = conn
-        self.helper = subprocess.Popen(["debconf-communicate"],
-                                       stdin=subprocess.PIPE,
-                                       stdout=subprocess.PIPE,
-                                       env=self._get_debconf_env())
-        w = gobject.io_add_watch(conn, gobject.IO_IN|gobject.IO_HUP|gobject.IO_ERR,
-                             self._copy_conn, self.helper.stdin)
-        self._watch_ids.append(w)
-        w= gobject.io_add_watch(self.helper.stdout, gobject.IO_IN,
-                             self._copy_stdout, conn)
-        self._watch_ids.append(w)
-        w = gobject.io_add_watch(self.helper.stdout, gobject.IO_HUP|gobject.IO_ERR,
-                             self._hangup)
-        self._watch_ids.append(w)
-        return True
+        self._socket_lock.acquire()
+        try:
+            # ensure outstanding gio messages are processed (to ensure
+            # that _hangup was run on the previous connection, see LP: #432607)
+            context = glib.main_context_default()
+            while context.pending():
+                context.iteration()
+            if self._socket_closed:
+                return True
+            if self._active_conn:
+                raise IOError, "Multiple debconf connections not supported"
+            conn, addr = source.accept()
+            self._active_conn = conn
+            self.helper = subprocess.Popen(["debconf-communicate"],
+                                           stdin=subprocess.PIPE,
+                                           stdout=subprocess.PIPE,
+                                           env=self._get_debconf_env())
+            w = gobject.io_add_watch(conn, gobject.IO_IN|gobject.IO_HUP|gobject.IO_ERR,
+                                 self._copy_conn, self.helper.stdin)
+            self._watch_ids.append(w)
+            w= gobject.io_add_watch(self.helper.stdout, gobject.IO_IN,
+                                 self._copy_stdout, conn)
+            self._watch_ids.append(w)
+            w = gobject.io_add_watch(self.helper.stdout, gobject.IO_HUP|gobject.IO_ERR,
+                                 self._hangup)
+            self._watch_ids.append(w)
+            return True
+        finally:
+            self._socket_lock.release()
 
     def _hangup(self, source, condition):
         """Callback when the debconf-communicate program exists