diff --git a/rigo/RigoDaemon/app.py b/rigo/RigoDaemon/app.py index d4b41f920..8cc4b1099 100755 --- a/rigo/RigoDaemon/app.py +++ b/rigo/RigoDaemon/app.py @@ -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") diff --git a/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml b/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml index b621e4247..7bd0f9627 100644 --- a/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml +++ b/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml @@ -4,8 +4,15 @@ - - + + + + + + + + + diff --git a/rigo/rigo_app.py b/rigo/rigo_app.py index 7be55e636..f062c0a9e 100644 --- a/rigo/rigo_app.py +++ b/rigo/rigo_app.py @@ -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")