Merge lp:~aptdaemon-developers/aptdaemon/threading into lp:aptdaemon

[Sebastian Heinlein]

Moves the worker and the simulate actions to separate threads. This makes the daemon more responsive and allows to easily queue simulate and the comming PK query methods.

=== modified file 'NEWS'
--- NEWS	2011-08-27 11:37:42 +0000
+++ NEWS	2011-10-02 05:32:23 +0000
@@ -2,6 +2,8 @@
 
  * Enhancements:
 
+   - Use threading internally instead for the worker
+
    - Support multiarch package names like apt-get, e.g. xterm:amd64
 
    - Store the transaction role and sender in apt's history log
 
=== modified file 'aptdaemon/core.py'
--- aptdaemon/core.py	2011-09-15 14:23:50 +0000
+++ aptdaemon/core.py	2011-10-02 05:32:23 +0000
@@ -47,24 +47,27 @@
 import signal
 import sys
 import time
+import threading
 import uuid
 
+from defer import inline_callbacks, return_value
+from defer.utils import dbus_deferred_method
 import gobject
+gobject.threads_init()
 import dbus.exceptions
 import dbus.service
 import dbus.mainloop.glib
-import dbus.glib
+dbus.mainloop.glib.threads_init()
 from softwareproperties.AptAuth import AptAuth
 import apt_pkg
 
 from config import ConfigWriter
 import errors
 import enums
-from defer import inline_callbacks, return_value
-from defer.utils import dbus_deferred_method
 import policykit1
 from worker import AptWorker, DummyWorker
-from loop import mainloop
+from utils import locked
+
 
 # Setup i18n
 _ = lambda msg: gettext.dgettext("aptdaemon", msg)
@@ -313,6 +316,7 @@
         connect -- if the Transaction should connect to DBus (default is True)
         bus -- the DBus connection which should be used (defaults to system bus)
         """
+        self.lock = threading.Lock()
         tid = uuid.uuid4().get_hex()
         self.tid = "/org/debian/apt/transaction/%s" % tid
         if connect == True:
@@ -330,18 +334,20 @@
             packages = ([], [], [], [], [], [])
         if not kwargs:
             kwargs = {}
+        self.sender = sender
         self.queue = queue
         self.uid = uid
-        self.locale = dbus.String("")
-        self.allow_unauthenticated = dbus.Boolean(False)
-        self.remove_obsoleted_depends = dbus.Boolean(False)
+        self.kwargs = kwargs
+        # Mutable properties which need to be protected by a lock
+        self._locale = dbus.String("")
+        self._allow_unauthenticated = dbus.Boolean(False)
+        self._remove_obsoleted_depends = dbus.Boolean(False)
         self.http_proxy = dbus.String("")
-        self.terminal = dbus.String("")
-        self.debconf = dbus.String("")
-        self.kwargs = kwargs
+        self._terminal = dbus.String("")
+        self._debconf = dbus.String("")
         self._translation = None
         # The transaction which should be executed after this one
-        self.after = None
+        self._after = None
         self._role = dbus.String(role)
         self._progress = dbus.Int32(0)
         # items_done, total_items, bytes_done, total_bytes, speed, time
@@ -359,8 +365,9 @@
         self._required_medium = dbus.Struct(("", ""), signature="ss")
         self._config_file_conflict = dbus.Struct(("", ""), signature="ss")
         self._config_file_conflict_resolution = ""
-        self.cancelled = dbus.Boolean(False)
-        self.paused = dbus.Boolean(False)
+        self._cancelled = dbus.Boolean(False)
+        self._paused = dbus.Boolean(False)
+        self.feedback = threading.Event()
         self._meta_data = dbus.Dictionary(signature="sv")
         self._download = dbus.Int64(0)
         self._space = dbus.Int64(0)
@@ -376,15 +383,64 @@
         self._idle_watch = gobject.timeout_add_seconds(
             TRANSACTION_IDLE_TIMEOUT, self._remove_from_connection_no_raise)
         # Handle a disconnect of the client application
-        self.sender_alive = True
+        self._sender_alive = True
         if bus:
             self._sender_watch = bus.watch_name_owner(sender,
                                                      self._sender_owner_changed)
         else:
             self._sender_watch = None
-        self.sender = sender
-        self.output = ""
-        self.simulated = None
+        self._output = ""
+        self._simulated = None
+
+    @locked
+    def _get_cancelled(self):
+        return self._cancelled
+
+    @locked
+    def _set_cancelled(self, txt):
+        self._cancelled = txt
+
+    cancelled = property(_get_cancelled, _set_cancelled)
+
+    @locked
+    def _get_paused(self):
+        return self._paused
+
+    @locked
+    def _set_paused(self, bool):
+        self._paused = bool
+
+    paused = property(_get_paused, _set_paused)
+
+    @locked
+    def _get_simulated(self):
+        return self._simulated
+
+    @locked
+    def _set_simulated(self, txt):
+        self._simulated = txt
+
+    simulated = property(_get_simulated, _set_simulated)
+
+    @locked
+    def _get_output(self):
+        return self._output
+
+    @locked
+    def _set_output(self, txt):
+        self._output = txt
+
+    output = property(_get_output, _set_output)
+
+    @locked
+    def _get_sender_alive(self):
+        return self._sender_alive
+
+    @locked
+    def _set_sender_alive(self, alive):
+        self._sender_alive = alive
+
+    sender_alive = property(_get_sender_alive, _set_sender_alive)
 
     def _sender_owner_changed(self, connection):
         """Callback if the owner of the original sender changed, e.g.
@@ -458,9 +514,11 @@
                 raise errors.InvalidMetaDataError("The value has to be a "
                                                   "string: %s" % value)
         # Merge new data into existing one:
-        self._meta_data.update(data)
-        self.PropertyChanged("MetaData", self._meta_data)
+        with self.lock:
+            self._meta_data.update(data)
+            self.PropertyChanged("MetaData", self._meta_data)
 
+    @locked
     def _get_meta_data(self):
         return self._meta_data
 
@@ -468,22 +526,26 @@
                          doc="Allows client applications to store meta data "
                              "for the transaction in a dictionary.")
 
+    @locked
     def _set_role(self, enum):
         if self._role != enums.ROLE_UNSET:
             raise errors.TransactionRoleAlreadySet()
         self._role = dbus.String(enum)
         self.PropertyChanged("Role", self._role)
 
+    @locked
     def _get_role(self):
         return self._role
 
     role = property(_get_role, _set_role, doc="Operation type of transaction.")
 
+    @locked
     def _set_progress_details(self, details):
         # items_done, total_items, bytes_done, total_bytes, speed, time
         self._progress_details = self._convert_struct(details, "iixxdx")
         self.PropertyChanged("ProgressDetails", self._progress_details)
 
+    @locked
     def _get_progress_details(self):
         return self._progress_details
 
@@ -493,12 +555,14 @@
                                       "bytes done, total bytes, speed and "
                                       "remaining time")
 
+    @locked
     def _set_error(self, excep):
         self._error = excep
         msg = self.gettext(excep.details) % excep.details_args
         self._error_property = self._convert_struct((excep.code, msg), "ss")
         self.PropertyChanged("Error", self._error_property)
 
+    @locked
     def _get_error(self):
         return self._error
 
@@ -506,9 +570,10 @@
 
     def _set_exit(self, enum):
         self.status = enums.STATUS_FINISHED
-        self._exit = dbus.String(enum)
-        self.PropertyChanged("ExitState", self._exit)
-        self.Finished(self._exit)
+        with self.lock:
+            self._exit = dbus.String(enum)
+            self.PropertyChanged("ExitState", self._exit)
+            self.Finished(self._exit)
         if self._sender_watch:
             self._sender_watch.cancel()
         # Remove the transaction from the Bus after it is complete. A short
@@ -516,15 +581,18 @@
         gobject.timeout_add_seconds(TRANSACTION_DEL_TIMEOUT,
                                     self._remove_from_connection_no_raise)
 
+    @locked
     def _get_exit(self):
         return self._exit
 
     exit = property(_get_exit, _set_exit,
                     doc="The exit state of the transaction.")
 
+    @locked
     def _get_download(self):
         return self._download
 
+    @locked
     def _set_download(self, size):
         self._download = dbus.Int64(size)
         self.PropertyChanged("Download", self._download)
@@ -532,9 +600,11 @@
     download = property(_get_download, _set_download,
                         doc="The download size of the transaction.")
 
+    @locked
     def _get_space(self):
         return self._space
 
+    @locked
     def _set_space(self, size):
         self._space = dbus.Int64(size)
         self.PropertyChanged("Space", self._space)
@@ -542,12 +612,14 @@
     space = property(_get_space, _set_space,
                      doc="The required disk space of the transaction.")
 
+    @locked
     def _set_packages(self, packages):
         self._packages = dbus.Struct([dbus.Array(pkgs, signature="s")
                                       for pkgs in packages],
                                      signature="as")
         self.PropertyChanged("Packages", self._packages)
 
+    @locked
     def _get_packages(self):
         return self._packages
 
@@ -556,9 +628,11 @@
                             "reinstalled, removed, purged, upgraded or "
                             "downgraded.")
 
+    @locked
     def _get_unauthenticated(self):
         return self._unauthenticated
 
+    @locked
     def _set_unauthenticated(self, unauthenticated):
         self._unauthenticated = dbus.Array(unauthenticated, signature="s")
         self.PropertyChanged("Unauthenticated", self._unauthenticated)
@@ -567,9 +641,11 @@
                                doc="Unauthenticated packages in this "
                                    "transaction")
 
+    @locked
     def _get_depends(self):
         return self._depends
 
+    @locked
     def _set_depends(self, depends):
         self._depends = dbus.Struct([dbus.Array(deps, signature="s")
                                      for deps in depends],
@@ -580,9 +656,11 @@
                        doc="The additional dependencies: installs, removals, "
                            "upgrades and downgrades.")
 
+    @locked
     def _get_status(self):
         return self._status
 
+    @locked
     def _set_status(self, enum):
         self._status = dbus.String(enum)
         self.PropertyChanged("Status", self._status)
@@ -590,9 +668,11 @@
     status = property(_get_status, _set_status,
                       doc="The status of the transaction.")
 
+    @locked
     def _get_status_details(self):
         return self._status_details
 
+    @locked
     def _set_status_details(self, text):
         self._status_details = get_dbus_string(text)
         self.PropertyChanged("StatusDetails", self._status_details)
@@ -600,9 +680,11 @@
     status_details = property(_get_status_details, _set_status_details,
                               doc="The status message from apt.")
 
+    @locked
     def _get_progress(self):
         return self._progress
 
+    @locked
     def _set_progress(self, percent):
         self._progress = dbus.Int32(percent)
         self.PropertyChanged("Progress", self._progress)
@@ -610,9 +692,11 @@
     progress = property(_get_progress, _set_progress,
                       doc="The progress of the transaction in percent.")
 
+    @locked
     def _get_progress_download(self):
         return self._progress_download
 
+    @locked
     def _set_progress_download(self, progress_download):
         self._progress_download = self._convert_struct(progress_download,
                                                        "sssxxs")
@@ -626,9 +710,11 @@
                                      "partially downloaded size and a status "
                                      "message.")
 
+    @locked
     def _get_cancellable(self):
         return self._cancellable
 
+    @locked
     def _set_cancellable(self, cancellable):
         self._cancellable = dbus.Boolean(cancellable)
         self.PropertyChanged("Cancellable", self._cancellable)
@@ -637,9 +723,11 @@
                            doc="If it's currently allowed to cancel the "
                                "transaction.")
 
+    @locked
     def _get_term_attached(self):
         return self._term_attached
 
+    @locked
     def _set_term_attached(self, attached):
         self._term_attached = dbus.Boolean(attached)
         self.PropertyChanged("TerminalAttached", self._term_attached)
@@ -649,9 +737,11 @@
                                  "attached to the dpkg call of the "
                                  "transaction.")
 
+    @locked
     def _get_required_medium(self):
         return self._required_medium
 
+    @locked
     def _set_required_medium(self, medium):
         self._required_medium = self._convert_struct(medium, "ss")
         self.PropertyChanged("RequiredMedium", self._required_medium)
@@ -662,9 +752,11 @@
                                    "of a required CD/DVD to install packages "
                                    "from.")
 
+    @locked
     def _get_config_file_conflict(self):
         return self._config_file_conflict
 
+    @locked
     def _set_config_file_conflict(self, prompt):
         if prompt is None:
             self._config_file_conflict = dbus.Struct(("", ""), signature="ss")
@@ -739,6 +831,18 @@
 
     # Methods
 
+    def _get_after(self):
+        return self._after
+
+    def _set_after(self, tid):
+        self.after = tid
+
+    after = property(_get_after, _set_after)
+
+    @locked
+    def _get_locale(self):
+        return self._locale
+
     def _set_locale(self, locale_str):
         """Set the language and encoding.
 
@@ -755,11 +859,14 @@
         except ValueError:
             raise
         else:
-            self.locale = dbus.String("%s.%s" % (lang, encoding))
-            self._translation = gettext.translation("aptdaemon",
-                                                    fallback=True,
-                                                    languages=[lang])
-            self.PropertyChanged("locale", self.locale)
+            with self.lock:
+                self._locale = dbus.String("%s.%s" % (lang, encoding))
+                self._translation = gettext.translation("aptdaemon",
+                                                        fallback=True,
+                                                        languages=[lang])
+                self.PropertyChanged("locale", self._locale)
+
+    locale = property(_get_locale, _set_locale)
 
     @inline_callbacks
     def _set_http_proxy(self, url, sender):
@@ -776,6 +883,7 @@
         self.http_proxy = dbus.String(url)
         self.PropertyChanged("HttpProxy", self.http_proxy)
 
+    @locked
     def _set_remove_obsoleted_depends(self, remove_obsoleted_depends):
         """Set the handling of the removal of automatically installed
         dependencies which are now obsoleted.
@@ -788,6 +896,14 @@
         self.PropertyChanged("RemoveObsoletedDepends",
                              self.remove_obsoleted_depends)
 
+    @locked
+    def _get_remove_obsoleted_depends(self):
+        return self._remove_obsoleted_depends
+
+    remove_obsoleted_depends = property(_get_remove_obsoleted_depends,
+                                        _set_remove_obsoleted_depends)
+
+    @locked
     def _set_allow_unauthenticated(self, allow_unauthenticated):
         """Set the handling of unauthenticated packages 
 
@@ -795,8 +911,16 @@
         allow_unauthenticated -- True to allow packages that come from a 
                            repository without a valid authentication signature
         """
-        self.allow_unauthenticated = dbus.Boolean(allow_unauthenticated)
-        self.PropertyChanged("AllowUnauthenticated", self.allow_unauthenticated)
+        self._allow_unauthenticated = dbus.Boolean(allow_unauthenticated)
+        self.PropertyChanged("AllowUnauthenticated",
+                             self._allow_unauthenticated)
+
+    @locked
+    def _get_allow_unauthenticated(self):
+        return self._allow_unauthenticated
+
+    allow_unauthenticated = property(_get_allow_unauthenticated,
+                                     _set_allow_unauthenticated)
 
     # pylint: disable-msg=C0103,C0322
     @dbus.service.method(APTDAEMON_TRANSACTION_DBUS_INTERFACE,
@@ -837,12 +961,12 @@
     def _run(self, sender):
         yield self._check_foreign_user(sender)
         yield self._check_auth()
-        self.queue.put(self.tid)
+        yield self.queue.put(self.tid)
         self.status = enums.STATUS_WAITING
         next_trans = self.after
         while next_trans:
             yield next_trans._check_auth()
-            self.queue.put(next_trans.tid)
+            yield self.queue.put(next_trans.tid)
             next_trans.status = enums.STATUS_WAITING
             next_trans = next_trans.after
 
@@ -921,6 +1045,7 @@
             log_trans.debug("Setting cancel event")
             self.cancelled = True
             self.status = enums.STATUS_CANCELLING
+            self.feedback.set()
             self.paused = False
             return
         raise errors.AptDaemonError("Could not cancel transaction")
@@ -944,7 +1069,7 @@
         if self.status != enums.STATUS_SETTING_UP:
             raise errors.TransactionAlreadyRunning()
         yield self._check_foreign_user(sender)
-        self.queue.worker.simulate(self)
+        yield self.queue.worker.simulate(self)
         if self._idle_watch is not None:
             gobject.source_remove(self._idle_watch)
             self._idle_watch = None
@@ -974,13 +1099,20 @@
         try:
             slave_fd = os.open(ttyname, os.O_RDWR | os.O_NOCTTY)
             if os.isatty(slave_fd):
-                self.terminal = dbus.String(ttyname)
-                self.PropertyChanged("Terminal", self.terminal)
+                with self.lock:
+                    self._terminal = dbus.String(ttyname)
+                    self.PropertyChanged("Terminal", self._terminal)
             else:
                 raise errors.AptDaemonError("%s isn't a tty" % ttyname)
         finally:
             os.close(slave_fd)
 
+    @locked
+    def _get_terminal(self):
+        return self._terminal
+
+    terminal = property(_get_terminal, _set_terminal)
+
     def _set_debconf(self, debconf_socket):
         """Set the socket of the debconf proxy.
 
@@ -1002,8 +1134,15 @@
             raise errors.AptDaemonError("socket '%s' has to be owned by the "
                                         "owner of the "
                                         "transaction" % debconf_socket)
-        self.debconf = dbus.String(debconf_socket)
-        self.PropertyChanged("DebconfSocket", self.debconf)
+        with self.lock:
+            self._debconf = dbus.String(debconf_socket)
+            self.PropertyChanged("DebconfSocket", self._debconf)
+
+    @locked
+    def _get_debconf(self):
+        return self._debconf
+
+    debconf = property(_get_debconf, _set_debconf)
 
     # pylint: disable-msg=C0103,C0322
     @dbus_deferred_method(APTDAEMON_TRANSACTION_DBUS_INTERFACE,
@@ -1030,6 +1169,7 @@
         if not self.required_medium[0] == medium:
             raise errors.AptDaemonError("The medium '%s' isn't "
                                         "requested." % medium)
+        self.feedback.set()
         self.paused = False
 
     # pylint: disable-msg=C0103,C0322
@@ -1066,6 +1206,7 @@
         if not self.config_file_conflict[0] == config:
             raise errors.AptDaemonError("Invalid config file: %s" % config)
         self.config_file_conflict_resolution = answer
+        self.feedback.set()
         self.paused = False
 
     @inline_callbacks
@@ -1132,10 +1273,6 @@
         if self.uid != uid:
             raise errors.ForeignTransaction()
 
-    def _set_kwargs(self, kwargs):
-        """Set the kwargs which will be send to the AptWorker."""
-        self.kwargs = kwargs
-
     def gettext(self, msg):
         """Translate the given message to the language of the transaction.
         Fallback to the system default.
@@ -1184,6 +1321,7 @@
         log.debug("emitting queue changed")
         self.emit("queue-changed")
 
+    @inline_callbacks
     def put(self, tid):
         """Add an item to the queue."""
         trans = self.limbo.pop(tid)
@@ -1197,7 +1335,7 @@
             # the transaction has been started
             if not self.worker.trans:
                 trans.progress = 9
-            self.worker.simulate(trans)
+            yield self.worker.simulate(trans)
 
         if trans._idle_watch is not None:
             gobject.source_remove(trans._idle_watch)
@@ -1271,6 +1409,7 @@
         signal.signal(signal.SIGQUIT, self._sigquit)
         signal.signal(signal.SIGTERM, self._sigquit)
         self.options = options
+        self.mainloop = gobject.MainLoop()
         if connect == True:
             if bus is None:
                 bus = dbus.SystemBus()
@@ -1362,7 +1501,7 @@
                                         self._check_for_inactivity)
         log.debug("Waiting for calls")
         try:
-            mainloop.run()
+            self.mainloop.run()
         except KeyboardInterrupt:
             self.Quit(None)
 
@@ -1853,7 +1992,7 @@
         """Request a shutdown of the daemon."""
         log.info("Quitting was requested")
         log.debug("Quitting main loop...")
-        mainloop.quit()
+        self.mainloop.quit()
         log.debug("Exit")
 
     @inline_callbacks
 
=== removed file 'aptdaemon/loop.py'
--- aptdaemon/loop.py	2010-05-03 05:49:33 +0000
+++ aptdaemon/loop.py	1970-01-01 00:00:00 +0000
@@ -1,32 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-"""Main loop for aptdaemon."""
-# Copyright (C) 2008-2009 Sebastian Heinlein <[e-mail]>
-#
-# 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
-# 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.
-
-__author__  = "Sebastian Heinlein <[e-mail]>"
-
-__all__ = ("mainloop", "get_main_loop")
-
-import gobject
-
-mainloop = gobject.MainLoop()
-
-def get_main_loop():
-    """Return the gobject main loop as a singelton."""
-    return mainloop
-
-# vim:ts=4:sw=4:et
 
=== modified file 'aptdaemon/progress.py'
--- aptdaemon/progress.py	2011-09-30 15:22:13 +0000
+++ aptdaemon/progress.py	2011-10-02 05:32:23 +0000
@@ -33,7 +33,6 @@
 
 import enums
 import lock
-from loop import mainloop
 
 # Required to get translatable strings extraced by xgettext
 _ = lambda s: s
@@ -210,9 +209,6 @@
                                             len(items)) % {"files":
                                                            " ".join(items)}
             self.transaction.status_details = msg
-
-        while gobject.main_context_default().pending():
-            gobject.main_context_default().iteration()
         return True
 
     def start(self):
@@ -232,8 +228,7 @@
         self.transaction.required_medium = medium, drive
         self.transaction.paused = True
         self.transaction.status = enums.STATUS_WAITING_MEDIUM
-        while self.transaction.paused:
-            gobject.main_context_default().iteration()
+        self.transaction.feedback.wait()
         self.transaction.status = enums.STATUS_DOWNLOADING
         if self.transaction.cancelled:
             return False
@@ -309,6 +304,7 @@
             self.child_pid = pid
             os.close(self.status_child_fd)
         log.debug("Child pid: %s", pid)
+        loop = gobject.MainLoop()
         watchers = []
         flags = gobject.IO_IN | gobject.IO_ERR | gobject.IO_HUP
         if self.transaction.terminal:
@@ -318,13 +314,12 @@
         watchers.append(gobject.io_add_watch(self.master_fd, flags,
                                              self._copy_io_master, terminal_fd))
         # Monitor the child process
-        watchers.append(gobject.child_watch_add(pid, self._on_child_exit))
+        watchers.append(gobject.child_watch_add(pid, self._on_child_exit, loop))
         # Watch for status updates
         watchers.append(gobject.io_add_watch(self.status_parent_fd,
                                              gobject.IO_IN,
                                              self._on_status_update))
-        while self._child_exit == -1:
-            gobject.main_context_default().iteration()
+        loop.run()
         for id in watchers:
             gobject.source_remove(id)
         # Restore the settings of the transaction terminal
@@ -341,9 +336,10 @@
                 pass
         return os.WEXITSTATUS(self._child_exit)
 
-    def _on_child_exit(self, pid, condition):
+    def _on_child_exit(self, pid, condition, loop):
         log.debug("Child exited: %s", condition)
         self._child_exit = condition
+        loop.quit()
         return False
 
     def _on_status_update(self, source, condition):
@@ -397,8 +393,6 @@
         signal.signal(signal.SIGINT, interrupt_handler)
         # Make sure that exceptions of the child are not catched by apport
         sys.excepthook = sys.__excepthook__
-
-        mainloop.quit()
         # Switch to the language of the user
         if self.transaction.locale:
             os.putenv("LANG", self.transaction.locale)
@@ -496,8 +490,8 @@
         self.transaction.config_file_conflict = (current, new)
         self.transaction.paused = True
         self.transaction.status = enums.STATUS_WAITING_CONFIG_FILE_PROMPT
-        while self.transaction.paused:
-            gobject.main_context_default().iteration()
+        #FIXME: Should we only wait some time?
+        self.transaction.feedback.wait()
         log.debug("Sending config file answer: %s",
                   self.transaction.config_file_conflict_resolution)
         if self.transaction.config_file_conflict_resolution == "replace":
 
=== modified file 'aptdaemon/utils.py'
--- aptdaemon/utils.py	2010-05-03 05:49:33 +0000
+++ aptdaemon/utils.py	2011-10-02 05:32:23 +0000
@@ -25,8 +25,11 @@
 __all__ = ("deprecated",)
 
 import functools
+import threading
 import warnings
 
+from defer import Deferred
+
 def deprecated(func):
     """This is a decorator which can be used to mark functions
     as deprecated. It will result in a warning being emitted
@@ -48,4 +51,31 @@
         return func(*args, **kwargs)
     return new_func
 
+def locked(func):
+    """Protect the called method by a lock."""
+    def _locked(*args, **kwargs):
+        self = args[0]
+        with self.lock:
+            return func(*args, **kwargs)
+    return _locked
+
+def defer_to_thread(func):
+    """Wrap the decorated message to a Deferred which will be called
+    in a separated thread.
+    """
+    def _deferred_to_thread(*args, **kwargs):
+        def __deferred_to_thread(deferred):
+            try:
+                result = func(*args, **kwargs)
+            except Exception, error:
+                deferred.errback(error)
+            else:
+                deferred.callback(result)
+        deferred = Deferred()
+        #FIXME: Would be nice to have a thread pool and a Queue here
+        thread = threading.Thread(target=__deferred_to_thread, args=[deferred])
+        thread.run()
+        return deferred
+    return _deferred_to_thread
+
 # vim:ts=4:sw=4:et
 
=== modified file 'aptdaemon/worker.py'
--- aptdaemon/worker.py	2011-09-18 06:31:09 +0000
+++ aptdaemon/worker.py	2011-10-02 05:32:23 +0000
@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#/usr/bin/env python
 # -*- coding: utf-8 -*-
 """Provides AptWorker which processes transactions."""
 # Copyright (C) 2008-2009 Sebastian Heinlein <[e-mail]>
@@ -29,6 +29,7 @@
 import sys
 import tempfile
 import time
+import threading
 import traceback
 
 import apt
@@ -54,6 +55,7 @@
                      DaemonDpkgInstallProgress, \
                      DaemonDpkgReconfigureProgress, \
                      DaemonDpkgRecoverProgress
+from utils import locked, defer_to_thread
 
 log = logging.getLogger("AptDaemon.Worker")
 
@@ -105,6 +107,8 @@
         self._status_frozen = None
         self.plugins = {}
         self._load_plugins()
+        self.lock = threading.Lock()
+        self.worker_thread = None
 
     def _load_plugins(self):
         """Load the plugins from setuptools' entry points."""
@@ -153,10 +157,14 @@
         transaction -- core.Transcation instance to run
         """
         log.info("Processing transaction %s", transaction.tid)
-        if self.trans:
-            raise Exception("There is already a running transaction")
-        self.trans = transaction
-        gobject.idle_add(self._process_transaction, transaction)
+        with self.lock:
+            if self.trans:
+                raise Exception("There is already a running transaction")
+            self.trans = transaction
+        self.worker_thread = threading.Thread(target=self._process_transaction,
+                                              args=[transaction],
+                                              name="WorkerThread")
+        self.worker_thread.start()
 
     def _emit_transaction_done(self, trans):
         """Emit the transaction-done signal.
@@ -174,6 +182,7 @@
         trans.progress = 11
         # FIXME: Check if the transaction has been just simulated. So we could
         #        skip marking the changes a second time.
+        self.lock.acquire()
         try:
             lock.wait_for_lock(trans)
             # Prepare the package cache
@@ -236,14 +245,15 @@
         else:
             trans.exit = EXIT_SUCCESS
         finally:
+            self.lock.release()
             trans.progress = 100
             self.last_action_timestamp = time.time()
             tid = trans.tid[:]
-            self.trans = None
+            with self.lock:
+                self.trans = None
             self._emit_transaction_done(trans)
             lock.release()
             log.info("Finished transaction %s", tid)
-        return False
 
     def commit_packages(self, trans, install, reinstall, remove, purge, upgrade,
                         downgrade, simulate=False):
@@ -360,7 +370,6 @@
                                               "available."), pkg_ver, pkg_name)
             elif pkg_rel:
                 self._set_candidate_release(pkg, pkg_rel)
- 
 
     def enable_distro_comp(self, trans, component):
         """Enable given component in the sources list.
@@ -381,7 +390,8 @@
         finally:
             os.umask(old_umask)
 
-    def add_repository(self, trans, src_type, uri, dist, comps, comment, sourcesfile):
+    def add_repository(self, trans, src_type, uri, dist, comps, comment,
+                       sourcesfile):
         """Add given repository to the sources list.
 
         Keyword arguments:
@@ -444,7 +454,6 @@
         log.info("Adding vendor key from keyserver: %s %s", keyid, keyserver)
         trans.status = STATUS_DOWNLOADING
         trans.progress = 101
-        last_pulse = time.time()
         #FIXME: Use gobject.spawn_async and deferreds in the worker
         #       Alternatively we could use python-pyme directly for a better
         #       error handling. Or the --status-fd of gpg
@@ -453,12 +462,8 @@
                                  "--recv", keyid], stderr=subprocess.STDOUT,
                                  stdout=subprocess.PIPE, close_fds=True)
         while proc.poll() is None:
-            while gobject.main_context_default().pending():
-                gobject.main_context_default().iteration()
-            time.sleep(0.05)
-            if time.time() - last_pulse > 0.3:
-                trans.progress = 101
-                last_pulse = time.time()
+            time.sleep(0.5)
+            trans.progress = 101
         if proc.returncode != 0:
             stdout = unicode(proc.stdout.read(), 
                              # that can return "None", in this case, just
@@ -941,12 +946,15 @@
         frozen_dir = tempfile.mkdtemp(prefix="aptdaemon-frozen-status")
         shutil.copy(self._status_orig, frozen_dir)
         self._status_frozen = os.path.join(frozen_dir, "status")
+        self.lock.release()
         try:
             yield
         finally:
+            self.lock.acquire()
             shutil.rmtree(frozen_dir)
             self._status_frozen = None
 
+    @defer_to_thread
     def simulate(self, trans):
         """Return the dependencies which will be installed by the transaction,
         the content of the dpkg status file after the transaction would have
@@ -958,8 +966,9 @@
         log.info("Simulating trans: %s" % trans.tid)
         trans.status = STATUS_RESOLVING_DEP
         try:
-            trans.depends, trans.download, trans.space, \
-                    trans.unauthenticated = self._simulate_helper(trans)
+            with self.lock:
+                trans.depends, trans.download, trans.space, \
+                        trans.unauthenticated = self._simulate_helper(trans)
         except TransactionFailed, excep:
             trans.error = excep
         except Exception, excep:
@@ -980,7 +989,6 @@
         trans.exit = EXIT_FAILED
         trans.progress = 100
         self.last_action_timestamp = time.time()
-        raise trans.error
 
     def _simulate_helper(self, trans):
         depends = [[], [], [], [], [], [], []]
@@ -1341,11 +1349,11 @@
         trans.status = STATUS_FINISHED
         self.last_action_timestamp = time.time()
         tid = self.trans.tid[:]
-        trans = self.trans
-        self.trans = None
+        with self.lock:
+            trans = self.trans
+            self.trans = None
         self._emit_transaction_done(trans)
         log.info("Finished transaction %s", tid)
-        return False
 
     def simulate(self, trans):
         depends = [[], [], [], [], [], [], []]