[Rigo*] fifth daemon architectural design

- Introduce Activity states and busy(), unbusy() methods to allocate and deallocate Daemon activities from the Clients, concurrently. - Tokenize acquire_resources() and release_resources() to filter out older events. The same token is returned to Clients via signals whenever it makes sense (repositories_updated() is one of them). - Implement Repositories Update Activity resume functionality in Rigo. It is possible to close Rigo during a repo update and reopen it afterwards. Multiple Rigo instances are allowed as well. - Implement the ability for RigoDaemon to kindly request Rigo Clients to release their locks (either shared or exclusive) due to new activity being scheduled. All the races and possible deadlocks should be handled correctly, but due to the actual complexity, only time will tell.
2012-03-14 17:47:34 +01:00
parent b7d198c776
commit b2f3bd5b8c
3 changed files with 360 additions and 76 deletions
@@ -14,6 +14,7 @@ import os
 import sys
 import time
 import signal
+import hashlib
 from threading import Lock, Timer

 # this makes the daemon to not write the entropy pid file
@@ -213,6 +214,22 @@ class RigoDaemonService(dbus.service.Object):
    is acquired in exclusive mode.
    """

+    class BusyError(Exception):
+        """
+        Cannot acknowledge a Local Activity change.
+        """
+
+    class AlreadyAvailableError(Exception):
+        """
+        Cannot acknowledge a Local Activity change to
+        "AVAILABLE" state, because we're already ready.
+        """
+
+    class UnbusyFromDifferentActivity(Exception):
+        """
+        Unbusy request from different activity.
+        """
+
    def __init__(self):
        object_path = RigoDaemonService.OBJECT_PATH
        dbus_loop = dbus.mainloop.glib.DBusGMainLoop(set_as_default = True)
@@ -231,6 +248,7 @@ class RigoDaemonService(dbus.service.Object):

        self._current_activity_mutex = Lock()
        self._current_activity = ActivityStates.AVAILABLE
+        self._activity_token = None
        self._activity_mutex = Lock()

        self._acquired_exclusive = False
@@ -240,6 +258,41 @@ class RigoDaemonService(dbus.service.Object):
        self._entropy = Entropy()
        write_output("__init__: dbus service loaded")

+    def _busy(self, activity):
+        """
+        Switch to busy activity state, if possible.
+        Raise RigoDaemonService.BusyError if already busy.
+        Returns an Activity Token (float) that can be used
+        with acquire_exclusive() and release_exclusive().
+        """
+        with self._current_activity_mutex:
+            if self._current_activity != ActivityStates.AVAILABLE:
+                raise RigoDaemonService.BusyError()
+            self._current_activity = activity
+            token = hashlib.sha1()
+            token.update(str(time.time()))
+            self._activity_token = token.hexdigest()
+            return self._activity_token
+
+    def _unbusy(self, activity, _force=False):
+        """
+        Unbusy from previous Activity.
+        Raise RigoDaemonService.AlreadyAvailableError if already
+        AVAILABLE.
+        Raise RigoDaemonService.UnbusyFromDifferentActivity if
+        provided activity differs from the current one.
+        Return the current activity_token and atomically resets it.
+        """
+        with self._current_activity_mutex:
+            if activity != self._current_activity and not _force:
+                raise RigoDaemonService.UnbusyFromDifferentActivity()
+            if activity == ActivityStates.AVAILABLE and not _force:
+                raise RigoDaemonService.AlreadyAvalabileError()
+            self._current_activity = ActivityStates.AVAILABLE
+            token = self._activity_token
+            self._activity_token = None
+            return token
+
    def stop(self):
        """
        RigoDaemon exit method.
@@ -248,8 +301,10 @@ class RigoDaemonService(dbus.service.Object):
            write_output("stop(): called")
        with self._activity_mutex:
            self._ping_sched.kill()
-            with self._current_activity_mutex:
-                self._current_activity = ActivityStates.NOT_AVAILABLE
+            try:
+                self._busy(ActivityStates.NOT_AVAILABLE, _force=True)
+            except RigoDaemonService.AlreadyAvailableError:
+                pass
            self._close_local_resources()
            if DAEMON_DEBUG:
                write_output("stop(): activity mutex acquired, quitting")
@@ -261,8 +316,6 @@ class RigoDaemonService(dbus.service.Object):
        Repositories Update execution code.
        """
        with self._activity_mutex:
-            with self._current_activity_mutex:
-                self._current_activity = ActivityStates.UPDATING_REPOSITORIES
            self._close_local_resources()

            if not repositories:
@@ -274,25 +327,24 @@ class RigoDaemonService(dbus.service.Object):
                        repositories,))

            result = 99
+            msg = ""
            try:
                updater = self._entropy.Repositories(
                    repositories, force = force)
                result = updater.unlocked_sync()
            except AttributeError as err:
                write_output("_update_repositories error: %s" % (err,))
-                self.repositories_updated(
-                    1, _("No repositories configured"))
+                result = 1
+                msg = _("No repositories configured")
                return
            except Exception as err:
                write_output("_update_repositories error 2: %s" % (err,))
-                self.repositories_updated(
-                    2, _("Unhandled Exception"))
+                result = 2
+                msg = _("Unhandled Exception")
                return
            finally:
-                with self._current_activity_mutex:
-                    self._current_activity = \
-                        ActivityStates.AVAILABLE
-                self.repositories_updated(result, "")
+                self.repositories_updated(
+                    result, msg, self._activity_token)

    def _close_local_resources(self):
        """
@@ -302,7 +354,7 @@ class RigoDaemonService(dbus.service.Object):
        self._entropy.reopen_installed_repository()
        self._entropy.close_repositories()

-    def _acquire_exclusive(self):
+    def _acquire_exclusive(self, activity):
        """
        Acquire Exclusive access to Entropy Resources.
        Note: this is blocking and will issue the
@@ -318,9 +370,14 @@ class RigoDaemonService(dbus.service.Object):
        if acquire:
            if DAEMON_DEBUG:
                write_output("_acquire_exclusive: about to acquire lock")
-            self._entropy.lock_resources(
-                blocking=True,
+            acquired = self._entropy.lock_resources(
+                blocking=False,
                shared=False)
+            if not acquired:
+                self.resources_unlock_request(activity)
+                self._entropy.lock_resources(
+                    blocking=True,
+                    shared=False)
            if DAEMON_DEBUG:
                write_output("_acquire_exclusive: just acquired lock")

@@ -389,33 +446,68 @@ class RigoDaemonService(dbus.service.Object):
        """
        return self._current_activity

-    @dbus.service.method(BUS_NAME, in_signature='',
-        out_signature='')
-    def acquire_exclusive(self):
+    @dbus.service.method(BUS_NAME, in_signature='i',
+        out_signature='b')
+    def acquire_exclusive(self, activity):
        """
        Start the rendezvous that will give us (this process)
        exclusive access to Entropy Resources, released by
        Rigo.
        """
        if DAEMON_DEBUG:
-            write_output("acquire_exclusive: called")
-        task = ParallelTask(self._acquire_exclusive)
+            write_output("acquire_exclusive: called for activity %s" % (
+                    activity,))
+
+        try:
+            token = self._busy(activity)
+        except RigoDaemonService.BusyError:
+            # I am already busy doing other stuff, cannot
+            # satisfy request
+            return False
+
+        task = ParallelTask(self._acquire_exclusive, activity)
        task.daemon = True
        task.name = "AcquireExclusive"
        task.start()
+        return True

-    @dbus.service.method(BUS_NAME, in_signature='',
-        out_signature='')
-    def release_exclusive(self):
+    @dbus.service.method(BUS_NAME, in_signature='is',
+        out_signature='b')
+    def release_exclusive(self, activity, token):
        """
        Release exclusive access to Entropy Resources.
        """
        if DAEMON_DEBUG:
-            write_output("release_exclusive: called")
+            write_output("release_exclusive: called for activity: %s" % (
+                    activity,))
+
+        try:
+            current_token = self._unbusy(activity)
+        except RigoDaemonService.AlreadyAvailableError:
+            write_output("release_exclusive: already "
+                         "available, ignoring")
+            return True
+        except RigoDaemonService.UnbusyFromDifferentActivity:
+            write_output("release_exclusive: unbusy "
+                         "from different activity: %s"
+                         ", current: %s"% (
+                    activity, self._current_activity,))
+            return False
+
+        # determine if token is still valid or belongs to
+        # a previous run.
+        if token != current_token:
+            write_output("release_exclusive: unbusy "
+                         "with different token: %s"
+                         ", current: %s"% (
+                    token, current_token,))
+            return False
+
        task = ParallelTask(self._release_exclusive)
        task.daemon = True
        task.name = "ReleaseExclusive"
        task.start()
+        return True

    @dbus.service.method(BUS_NAME, in_signature='',
        out_signature='b')
@@ -448,16 +540,8 @@ class RigoDaemonService(dbus.service.Object):
        """
        def _deptester():
            with self._activity_mutex:
-                with self._current_activity_mutex:
-                    self._current_activity = \
-                        ActivityStates.INTERNAL_ROUTINES
-
                self._entropy.dependencies_test()

-                with self._current_activity_mutex:
-                    self._current_activity = \
-                        ActivityStates.AVAILABLE
-
        task = ParallelTask(_deptester)
        task.daemon = True
        task.name = "OutputTestThread"
@@ -487,8 +571,8 @@ class RigoDaemonService(dbus.service.Object):
        pass

    @dbus.service.signal(dbus_interface=BUS_NAME,
-        signature='is')
-    def repositories_updated(self, result, message):
+        signature='iss')
+    def repositories_updated(self, result, message, token):
        """
        Repositories have been updated.
        "result" is an integer carrying execution return status.
@@ -507,6 +591,19 @@ class RigoDaemonService(dbus.service.Object):
        if DAEMON_DEBUG:
            write_output("exclusive_acquired() issued")

+    @dbus.service.signal(dbus_interface=BUS_NAME,
+        signature='i')
+    def resources_unlock_request(self, activity):
+        """
+        Signal all the connected Clients to release their
+        Entropy Resources Lock, if possible (both shared
+        and exclusive). This is a kind request, it is
+        not expected that clients actually acknowledge us.
+        """
+        if DAEMON_DEBUG:
+            write_output("resources_unlock_request() issued for %d" % (
+                    activity,))
+
    @dbus.service.signal(dbus_interface=BUS_NAME,
        signature='')
    def ping(self):
@@ -516,10 +613,10 @@ class RigoDaemonService(dbus.service.Object):
        Entropy Resources will be released completely.
        """
        def _time_up():
-            with self._activity_mutex:
-                if DAEMON_DEBUG:
-                    write_output("time is up! issuing release_exclusive()")
-                self.release_exclusive()
+            if DAEMON_DEBUG:
+                write_output("time is up! issuing _release_exclusive()")
+            self._unbusy(None, _force=True)
+            self._release_exclusive()

        if DAEMON_DEBUG:
            write_output("ping() issued")
@@ -4,8 +4,15 @@
    <method name="activity">
       <arg name="state" type="i" direction="out"/>
    </method>
-    <method name="acquire_exclusive"/>
-    <method name="release_exclusive"/>
+    <method name="acquire_exclusive">
+      <arg name="activity" type="i" direction="in"/>
+      <arg name="accepted" type="b" direction="out"/>
+    </method>
+    <method name="release_exclusive">
+      <arg name="activity" type="i" direction="in"/>
+      <arg name="token" type="s" direction="in"/>
+      <arg name="accepted" type="b" direction="out"/>
+    </method>
    <method name="connect"/>
    <method name="output_test"/>
    <method name="update_repositories">
@@ -57,7 +57,7 @@ from rigo.ui.gtk3.utils import init_sc_css_provider, get_sc_icon_theme
 from rigo.utils import build_application_store_url, build_register_url, \
    escape_markup, prepare_markup

-from RigoDaemon.enums import ActivityStates
+from RigoDaemon.enums import ActivityStates as DaemonActivityStates

 from entropy.const import etpConst, etpUi, const_debug_write, \
    const_debug_enabled, const_convert_to_unicode, const_isunicode
@@ -70,6 +70,30 @@ from entropy.output import darkgreen, brown, darkred, red, blue

 import entropy.tools

+class LocalActivityStates:
+    (
+        READY,
+        UPDATING_REPOSITORIES_MASTER,
+        UPDATING_REPOSITORIES_SLAVE,
+        INSTALLING_APPLICATIONS,
+    ) = range(4)
+
+    class BusyError(Exception):
+        """
+        Cannot acknowledge a Local Activity change.
+        """
+
+    class AlreadyReadyError(Exception):
+        """
+        Cannot acknowledge a Local Activity change to
+        "READY" state, because we're already ready.
+        """
+
+    class UnbusyFromDifferentActivity(Exception):
+        """
+        Unbusy request from different activity.
+        """
+
 class RigoServiceController(GObject.Object):

    """
@@ -77,6 +101,12 @@ class RigoServiceController(GObject.Object):
    Handles privileged requests on our behalf.
    """

+    class InconsistentDaemonState(Exception):
+        """
+        Raised when RigoDaemon and Rigo states are not
+        coherent.
+        """
+
    __gsignals__ = {
        # we request to lock the whole UI wrt repo
        # interaction
@@ -98,6 +128,7 @@ class RigoServiceController(GObject.Object):
    _TRANSFER_OUTPUT_SIGNAL = "transfer_output"
    _EXCLUSIVE_ACQUIRED_SIGNAL = "exclusive_acquired"
    _PING_SIGNAL = "ping"
+    _RESOURCES_UNLOCK_REQUEST_SIGNAL = "resources_unlock_request"

    def __init__(self, rigo_app, activity_rwsem, entropy_client, entropy_ws):
        GObject.Object.__init__(self)
@@ -117,6 +148,8 @@ class RigoServiceController(GObject.Object):
        self._connected = False
        self._registered_signals = {}
        self._registered_signals_mutex = Lock()
+        self._local_activity = LocalActivityStates.READY
+        self._local_activity_mutex = Lock()

    def set_applications_controller(self, avc):
        """
@@ -144,6 +177,36 @@ class RigoServiceController(GObject.Object):
        """
        self._nc = nc

+    def busy(self, local_activity):
+        """
+        Become busy, switch to some local activity.
+        If an activity is already taking place,
+        LocalActivityStates.BusyError is raised.
+        """
+        with self._local_activity_mutex:
+            if self._local_activity != LocalActivityStates.READY:
+                raise LocalActivityStates.BusyError()
+            self._local_activity = local_activity
+
+    def unbusy(self, current_activity):
+        """
+        Exit from busy state, switch to local activity called "READY".
+        If we're already out of any activity, raise
+        LocalActivityStates.AlreadyReadyError()
+        """
+        with self._local_activity_mutex:
+            if self._local_activity == LocalActivityStates.READY:
+                raise LocalActivityStates.AlreadyReadyError()
+            if self._local_activity != current_activity:
+                raise LocalActivityStates.UnbusyFromDifferentActivity()
+            self._local_activity = LocalActivityStates.READY
+
+    def local_activity(self):
+        """
+        Return the current local activity (enum from LocalActivityStates)
+        """
+        return self._local_activity
+
    @property
    def _dbus_main_loop(self):
        if self.__dbus_main_loop is None:
@@ -166,22 +229,34 @@ class RigoServiceController(GObject.Object):
                    self.DBUS_INTERFACE, self.DBUS_PATH
                    )

+                # ping/pong signaling, used to let
+                # RigoDaemon release exclusive locks
+                # when no client is connected
                self.__entropy_bus.connect_to_signal(
                    self._PING_SIGNAL, self._ping_signal,
                    dbus_interface=self.DBUS_INTERFACE)

+                # Entropy stdout/stderr messages
                self.__entropy_bus.connect_to_signal(
                    self._OUTPUT_SIGNAL, self._output_signal,
                    dbus_interface=self.DBUS_INTERFACE)

+                # Entropy UrlFetchers messages
                self.__entropy_bus.connect_to_signal(
                    self._TRANSFER_OUTPUT_SIGNAL,
                    self._transfer_output_signal,
                    dbus_interface=self.DBUS_INTERFACE)

+                # RigoDaemon Entropy Resources unlock requests
+                self.__entropy_bus.connect_to_signal(
+                    self._RESOURCES_UNLOCK_REQUEST_SIGNAL,
+                    self._resources_unlock_request_signal,
+                    dbus_interface=self.DBUS_INTERFACE)
+
            return self.__entropy_bus

-    def _repositories_updated_signal(self, result, message):
+    def _repositories_updated_signal(self, result, message,
+                                     token, local_activity):
        """
        Signal coming from RigoDaemon notifying us that repositories have
        been updated.
@@ -204,16 +279,37 @@ class RigoServiceController(GObject.Object):
                        "already consumed")
                return

-        self._scale_down() # do we really need this? not really i think
-        self._release_local_resources()
-        # 1 -- ACTIVITY CRIT :: OFF
-        self._activity_rwsem.writer_release()
+        # Since we might sleep here, and we're in the
+        # MainThread, better spawning a new thread
+        # The sleep is inside _scale_down (wait on a
+        # semaphore)
+        def _scale_down():
+            # we don't expect to fail here, it would
+            # mean programming error.
+            self.unbusy(local_activity)

-        self.emit("repositories-updated")
-        if const_debug_enabled():
-            const_debug_write(
-                __name__,
-                "_repositories_updated_signal: repositories-updated")
+            activity = DaemonActivityStates.UPDATING_REPOSITORIES
+            slave_update = LocalActivityStates.UPDATING_REPOSITORIES_SLAVE
+            if local_activity == slave_update:
+                # FIXME: same??
+                self._scale_down(activity, token)
+            else:
+                self._scale_down(activity, token)
+                self._release_local_resources()
+
+            # 1 -- ACTIVITY CRIT :: OFF
+            self._activity_rwsem.writer_release()
+
+            GLib.idle_add(self.emit, "repositories-updated")
+            if const_debug_enabled():
+                const_debug_write(
+                    __name__,
+                    "_repositories_updated_signal: repositories-updated")
+
+        task = ParallelTask(_scale_down)
+        task.name = "RepositoriesUpdatedSignalScalerDown"
+        task.daemon = True
+        task.start()

    def _output_signal(self, text, header, footer, back, importance, level,
               count_c, count_t, percent):
@@ -292,6 +388,42 @@ class RigoServiceController(GObject.Object):
            self._entropy_bus,
            dbus_interface=self.DBUS_INTERFACE).pong()

+    def _resources_unlock_request_signal(self, activity):
+        """
+        RigoDaemon is asking us to release our Entropy Resources Lock.
+        An ActivityStates value is provided in order to let us decide
+        if we can acknowledge the request.
+        """
+        const_debug_write(
+            __name__,
+            "_resources_unlock_request_signal: "
+            "called, with remote activity: %s" % (activity,))
+
+        if activity == DaemonActivityStates.UPDATING_REPOSITORIES:
+            # did we ask that or is it another client?
+            if self.local_activity() == LocalActivityStates.READY:
+                # another client, bend over XD
+                # LocalActivityStates value will be atomically
+                # switched in the above thread.
+                task = ParallelTask(
+                    self._update_repositories,
+                    [], False,
+                    master=False)
+                task.daemon = True
+                task.name = "UpdateRepositoriesExternal"
+                task.start()
+                const_debug_write(
+                    __name__,
+                    "_resources_unlock_request_signal: "
+                    "somebody called repo update, starting here too")
+            else:
+                const_debug_write(
+                    __name__,
+                    "_resources_unlock_request_signal: "
+                    "it's been us calling repositories update")
+                # it's been us calling it, ignore request
+                return
+
    def activity(self):
        """
        Return RigoDaemon activity states (any of RigoDaemon.ActivityStates
@@ -352,7 +484,7 @@ class RigoServiceController(GObject.Object):
        self._avc.clear_safe()
        self._entropy.close_repositories()

-    def _scale_up(self):
+    def _scale_up(self, activity):
        """
        Acquire (in blocking mode) the Entropy Resources Lock
        in exclusive mode. Scale up privileges, and ask for
@@ -375,10 +507,13 @@ class RigoServiceController(GObject.Object):
            _acquirer,
            dbus_interface=self.DBUS_INTERFACE)

-        dbus.Interface(
+        accepted = dbus.Interface(
            self._entropy_bus,
            dbus_interface=self.DBUS_INTERFACE
-            ).acquire_exclusive()
+            ).acquire_exclusive(activity)
+        if not accepted:
+            # FIXME, tell the reason to User
+            return False

        self._entropy.unlock_resources()
        # FIXME: lock down UI here and show a please wait
@@ -393,7 +528,7 @@ class RigoServiceController(GObject.Object):
                          "_scale_up: leave")
        return True

-    def _scale_down(self):
+    def _scale_down(self, activity, release_exclusive_token):
        """
        Release RigoDaemon Entropy Resources and regain
        control here.
@@ -402,12 +537,17 @@ class RigoServiceController(GObject.Object):
        # start the rendezvous

        const_debug_write(__name__, "RigoServiceController: "
-                          "_scale_down: enter")
+                          "_scale_down: enter, for activity: %s" % (
+                activity,))

        def _acquirer(sem):
+            const_debug_write(__name__, "RigoServiceController: "
+                              "_scale_down._acquirer: enter")
            self._entropy.lock_resources(
                blocking=True,
                shared=True)
+            const_debug_write(__name__, "RigoServiceController: "
+                              "_scale_down._acquirer: leave")
            sem.release()

        task = ParallelTask(_acquirer, acquired_sem)
@@ -415,30 +555,62 @@ class RigoServiceController(GObject.Object):
        task.daemon = True
        task.start()

-        dbus.Interface(
+        # call this unconditionally and ignore any error.
+        # multiple calls are harmless since the token
+        # ensures that we actually release locks only once,
+        # without the risk of running into race conditions
+        accepted = dbus.Interface(
            self._entropy_bus,
            dbus_interface=self.DBUS_INTERFACE
-            ).release_exclusive()
+            ).release_exclusive(
+            activity,
+            release_exclusive_token)

+        # ignore accepted, unbusy() protects us from
+        # releasing during other activities.
        acquired_sem.acquire()
        # back with shared lock!

        const_debug_write(__name__, "RigoServiceController: "
                          "_scale_down: leave")

-    def _update_repositories(self, repositories, force):
+    def _update_repositories(self, repositories, force,
+                             master=True):
        """
        Ask RigoDaemon to update repositories once we're
        100% sure that the UI is locked down.
        """
-        scaled = self._scale_up()
-        if not scaled:
+        if master:
+            local_activity = LocalActivityStates.UPDATING_REPOSITORIES_MASTER
+        else:
+            local_activity = LocalActivityStates.UPDATING_REPOSITORIES_SLAVE
+        try:
+            self.busy(local_activity)
+            # will be unlocked when we get the signal back
+        except LocalActivityStates.BusyError:
+            # FIXME, notify user that we cannot do repo update
+            const_debug_write(__name__, "_update_repositories: "
+                              "LocalActivityStates.BusyError!")
            return
+
+        if master:
+            scaled = self._scale_up(
+                DaemonActivityStates.UPDATING_REPOSITORIES)
+            if not scaled:
+                self.unbusy(local_activity)
+                return
+        else:
+            # if we don't need to scale, just unlock
+            # local resources.
+            # No need to connect() because other clients have
+            # already done that
+            self._entropy.unlock_resources()
+
        self._update_repositories_unlocked(
-            repositories, force)
+            repositories, force, master)

    def _update_repositories_unlocked(self, repositories, force,
-                                      execute_method=True):
+                                      master):
        """
        Internal method handling the actual Repositories Update
        execution.
@@ -467,15 +639,21 @@ class RigoServiceController(GObject.Object):
        # 1 -- ACTIVITY CRIT :: ON
        self._activity_rwsem.writer_acquire()

+        if master:
+            loc_activity = LocalActivityStates.UPDATING_REPOSITORIES_MASTER
+        else:
+            loc_activity = LocalActivityStates.UPDATING_REPOSITORIES_SLAVE
+
        signal_sem = Semaphore(1)

-        def _repositories_updated_signal(result, message):
+        def _repositories_updated_signal(result, message, token):
            if not signal_sem.acquire(False):
                # already called, no need to call again
                return
            # this is done in order to have it called
            # only once by two different code paths
-            self._repositories_updated_signal(result, message)
+            self._repositories_updated_signal(
+                result, message, token, loc_activity)

        with self._registered_signals_mutex:
            # connect our signal
@@ -498,7 +676,7 @@ class RigoServiceController(GObject.Object):
        self._terminal.reset()
        self._release_local_resources()

-        if execute_method:
+        if master:
            dbus.Interface(
                self._entropy_bus,
                dbus_interface=self.DBUS_INTERFACE
@@ -513,9 +691,11 @@ class RigoServiceController(GObject.Object):
        Called via _update_repositories_unlocked() in order to handle
        the possible race between RigoDaemon signal and the fact that
        we just lost it.
+        This is only called in slave mode. When we didn't spawn the
+        repositories update directly.
        """
        activity = self.activity()
-        if activity == ActivityStates.UPDATING_REPOSITORIES:
+        if activity == DaemonActivityStates.UPDATING_REPOSITORIES:
            return

        # lost the signal or not, we're going to force
@@ -533,7 +713,7 @@ class RigoServiceController(GObject.Object):
        # Run in the main loop, to avoid calling a signal
        # callback in random threads.
        GLib.idle_add(self._repositories_updated_signal,
-                      0, "")
+                      0, "", activity)

    def update_repositories(self, repositories, force):
        """
@@ -2391,29 +2571,29 @@ class Rigo(Gtk.Application):
        # exclusion is handled via Entropy Resources Lock (which is a file
        # based rwsem).
        activity = self._service.activity()
-        if activity != ActivityStates.AVAILABLE:
+        if activity != DaemonActivityStates.AVAILABLE:
            msg = ""
            show_dialog = True

-            if activity == ActivityStates.NOT_AVAILABLE:
+            if activity == DaemonActivityStates.NOT_AVAILABLE:
                msg = _("Background Service is currently not available")

-            elif activity == ActivityStates.UPDATING_REPOSITORIES:
+            elif activity == DaemonActivityStates.UPDATING_REPOSITORIES:
                show_dialog = False
                task = ParallelTask(
-                    self._service._update_repositories_unlocked,
-                    [], False, execute_method=False)
+                    self._service._update_repositories,
+                    [], False, master=False)
                task.daemon = True
                task.name = "UpdateRepositoriesUnlocked"
                task.start()

-            elif activity == ActivityStates.INSTALLING_APPLICATION:
+            elif activity == DaemonActivityStates.INSTALLING_APPLICATION:
                # FIXME, jump to WORK_VIEW and show the progress.
                msg = _("Background Service is installing Applications")
-            elif activity == ActivityStates.UPGRADING_SYSTEM:
+            elif activity == DaemonActivityStates.UPGRADING_SYSTEM:
                # FIXME, jump to WORK_VIEW and show the progress.
                msg = _("Background Service is updating your system")
-            elif activity == ActivityStates.INTERNAL_ROUTINES:
+            elif activity == DaemonActivityStates.INTERNAL_ROUTINES:
                msg = _("Background Service is currently busy")
            else:
                msg = _("Background Service is incompatible with Rigo")