From efde0c2ec3d9c86fa548e86d4742d0845b318432 Mon Sep 17 00:00:00 2001 From: Fabio Erculiani Date: Sun, 18 Mar 2012 07:57:57 +0100 Subject: [PATCH] [Rigo/RigoDaemon] greatly simplify resource passing rendezvous Completely move the arbitration to RigoDaemon, making Rigo passively accepting the former requests. Moreover, complete support for bottom notification area and start implementing app management events. --- rigo/RigoDaemon/app.py | 244 +++---- rigo/RigoDaemon/config.py | 2 +- rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml | 11 +- rigo/RigoDaemon/enums.py | 2 +- rigo/rigo/enums.py | 13 +- rigo/rigo/ui/gtk3/controllers/application.py | 4 +- rigo/rigo/ui/gtk3/widgets/apptreeview.py | 2 - rigo/rigo_app.py | 713 +++++++++++++------ rigo/test_polkit.py | 71 ++ 9 files changed, 672 insertions(+), 390 deletions(-) create mode 100644 rigo/test_polkit.py diff --git a/rigo/RigoDaemon/app.py b/rigo/RigoDaemon/app.py index 89fdb46b3..2492d3722 100755 --- a/rigo/RigoDaemon/app.py +++ b/rigo/RigoDaemon/app.py @@ -14,7 +14,6 @@ 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 @@ -249,7 +248,6 @@ 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 @@ -266,17 +264,11 @@ class RigoDaemonService(dbus.service.Object): """ 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): """ @@ -285,7 +277,6 @@ class RigoDaemonService(dbus.service.Object): 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: @@ -293,9 +284,6 @@ class RigoDaemonService(dbus.service.Object): 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): """ @@ -315,27 +303,32 @@ class RigoDaemonService(dbus.service.Object): entropy.tools.kill_threads() os.kill(os.getpid(), signal.SIGTERM) - def _update_repositories(self, repositories, force): + def _update_repositories(self, repositories, force, activity): """ Repositories Update execution code. """ with self._activity_mutex: - self._close_local_resources() - - if not repositories: - repositories = list( - SystemSettings()['repositories']['available']) - - if DAEMON_DEBUG: - write_output("_update_repositories(): %s" % ( - repositories,)) + # ask clients to release their locks + self._acquire_exclusive(activity) result = 99 msg = "" try: + self._close_local_resources() + self._entropy_setup() + self.activity_started(activity) + + if not repositories: + repositories = list( + SystemSettings()['repositories']['available']) + if DAEMON_DEBUG: + write_output("_update_repositories(): %s" % ( + repositories,)) + updater = self._entropy.Repositories( repositories, force = force) result = updater.unlocked_sync() + except AttributeError as err: write_output("_update_repositories error: %s" % (err,)) result = 1 @@ -346,9 +339,17 @@ class RigoDaemonService(dbus.service.Object): result = 2 msg = _("Unhandled Exception") return + finally: - self.repositories_updated( - result, msg, self._activity_token) + try: + self._unbusy(activity) + except RigoDaemonService.AlreadyAvailableError: + write_output("_update_repositories._unbusy: already " + "available, wtf !?!?") + # wtf?? + self._release_exclusive(activity) + self.activity_completed(activity, result == 0) + self.repositories_updated(result, msg) def _close_local_resources(self): """ @@ -378,6 +379,8 @@ class RigoDaemonService(dbus.service.Object): blocking=False, shared=False) if not acquired: + if DAEMON_DEBUG: + write_output("_acquire_exclusive: asking to unlock") self.resources_unlock_request(activity) self._entropy.lock_resources( blocking=True, @@ -385,61 +388,66 @@ class RigoDaemonService(dbus.service.Object): if DAEMON_DEBUG: write_output("_acquire_exclusive: just acquired lock") - self.exclusive_acquired() - - def _release_exclusive(self): + def _release_exclusive(self, activity): """ Release Exclusive access to Entropy Resources. """ - with self._activity_mutex: - # make sure to not release locks as long - # as there is activity - with self._acquired_exclusive_mutex: - if self._acquired_exclusive: - self._entropy.unlock_resources() - # now we got the exclusive lock - self._acquired_exclusive = False + # make sure to not release locks as long + # as there is activity + with self._acquired_exclusive_mutex: + if self._acquired_exclusive: + self.resources_lock_request(activity) + self._entropy.unlock_resources() + # now we got the exclusive lock + self._acquired_exclusive = False + + def _entropy_setup(self): + """ + Notify us that a new client is now connected. + Here we reload Entropy configuration and other resources. + """ + if DAEMON_DEBUG: + write_output("_entropy_setup(): called") + acquired = self._ping_sched_startup.acquire(False) + if acquired: + if DAEMON_DEBUG: + write_output("_entropy_setup(): starting ping() signaling") + self._ping_sched.start() + + initconfig_entropy_constants(etpConst['systemroot']) + self._entropy.Settings().clear() + self._entropy._validate_repositories() + self._close_local_resources() + if DAEMON_DEBUG: + write_output("_entropy_setup(): complete") ### DBUS METHODS @dbus.service.method(BUS_NAME, in_signature='asb', - out_signature='') + out_signature='b') def update_repositories(self, repositories, force): """ Request RigoDaemon to update the given repositories. At the end of the execution, the "repositories_updated" signal will be raised. """ + activity = ActivityStates.UPDATING_REPOSITORIES + try: + self._busy(activity) + except RigoDaemonService.BusyError: + # I am already busy doing other stuff, cannot + # satisfy request + return False + if DAEMON_DEBUG: write_output("update_repositories called: %s" % ( repositories,)) task = ParallelTask(self._update_repositories, repositories, - force) + force, activity) task.daemon = True task.name = "UpdateRepositoriesThread" task.start() - - @dbus.service.method(BUS_NAME, in_signature='', - out_signature='') - def connect(self): - """ - Notify us that a new client is now connected. - Here we reload Entropy configuration and other resources. - """ - if DAEMON_DEBUG: - write_output("connect(): called") - acquired = self._ping_sched_startup.acquire(False) - if acquired: - if DAEMON_DEBUG: - write_output("connect(): starting ping() signaling") - self._ping_sched.start() - with self._activity_mutex: - initconfig_entropy_constants(etpConst['systemroot']) - self._entropy.Settings().clear() - self._entropy._validate_repositories() - self._close_local_resources() - if DAEMON_DEBUG: - write_output("A new client is now connected !") + return True @dbus.service.method(BUS_NAME, in_signature='', out_signature='i') @@ -450,69 +458,6 @@ class RigoDaemonService(dbus.service.Object): """ return self._current_activity - @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 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='is', - out_signature='b') - def release_exclusive(self, activity, token): - """ - Release exclusive access to Entropy Resources. - """ - if DAEMON_DEBUG: - 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') def is_exclusive(self): @@ -575,8 +520,8 @@ class RigoDaemonService(dbus.service.Object): pass @dbus.service.signal(dbus_interface=BUS_NAME, - signature='iss') - def repositories_updated(self, result, message, token): + signature='is') + def repositories_updated(self, result, message): """ Repositories have been updated. "result" is an integer carrying execution return status. @@ -585,16 +530,6 @@ class RigoDaemonService(dbus.service.Object): write_output("repositories_updated() issued, args:" " %s" % (locals(),)) - @dbus.service.signal(dbus_interface=BUS_NAME, - signature='') - def exclusive_acquired(self): - """ - Entropy Resources have been eventually acquired in - blocking mode. - """ - if DAEMON_DEBUG: - write_output("exclusive_acquired() issued") - @dbus.service.signal(dbus_interface=BUS_NAME, signature='i') def resources_unlock_request(self, activity): @@ -608,6 +543,42 @@ class RigoDaemonService(dbus.service.Object): write_output("resources_unlock_request() issued for %d" % ( activity,)) + @dbus.service.signal(dbus_interface=BUS_NAME, + signature='i') + def resources_lock_request(self, activity): + """ + Signal all the connected Clients to re-acquire shared + Entropy Resources Lock. This is actually a mandatory + request in order to run into really bad, inconsistent states. + """ + if DAEMON_DEBUG: + write_output("resources_lock_request() issued for %d" % ( + activity,)) + + @dbus.service.signal(dbus_interface=BUS_NAME, + signature='i') + def activity_started(self, activity): + """ + Signal all the connected Clients that a scheduled activity + has begun (and we're running with exclusive Entropy Resources + access). + """ + if DAEMON_DEBUG: + write_output("activity_started() issued for %d" % ( + activity,)) + + @dbus.service.signal(dbus_interface=BUS_NAME, + signature='ib') + def activity_completed(self, activity, success): + """ + Signal all the connected Clients that a scheduled activity + has been carried out. + """ + if DAEMON_DEBUG: + write_output("activity_completed() issued for %d," + " success: %s" % ( + activity, success,)) + @dbus.service.signal(dbus_interface=BUS_NAME, signature='') def ping(self): @@ -620,7 +591,10 @@ class RigoDaemonService(dbus.service.Object): if DAEMON_DEBUG: write_output("time is up! issuing _release_exclusive()") self._unbusy(None, _force=True) - self._release_exclusive() + with self._activity_mutex: + self._release_exclusive() + if DAEMON_DEBUG: + write_output("issued _release_exclusive()") if DAEMON_DEBUG: write_output("ping() issued") diff --git a/rigo/RigoDaemon/config.py b/rigo/RigoDaemon/config.py index 39b0585ff..069568cdd 100644 --- a/rigo/RigoDaemon/config.py +++ b/rigo/RigoDaemon/config.py @@ -20,4 +20,4 @@ class PolicyActions: # PolicyKit update action UPDATE_REPOSITORIES = "org.sabayon.RigoDaemon.update" UPGRADE_SYSTEM = "org.sabayon.RigoDaemon.upgrade" - MANAGE_APP = "org.sabayon.RigoDaemon.manage" + MANAGE_APPLICATIONS = "org.sabayon.RigoDaemon.manage" diff --git a/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml b/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml index 7bd0f9627..b20607ecb 100644 --- a/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml +++ b/rigo/RigoDaemon/dbus/org.sabayon.Rigo.xml @@ -4,20 +4,11 @@ - - - - - - - - - - + diff --git a/rigo/RigoDaemon/enums.py b/rigo/RigoDaemon/enums.py index c32c4656d..dcc3f36e6 100644 --- a/rigo/RigoDaemon/enums.py +++ b/rigo/RigoDaemon/enums.py @@ -15,7 +15,7 @@ class ActivityStates: AVAILABLE, NOT_AVAILABLE, UPDATING_REPOSITORIES, - INSTALLING_APPLICATION, + MANAGING_APPLICATIONS, UPGRADING_SYSTEM, INTERNAL_ROUTINES ) = range(6) diff --git a/rigo/rigo/enums.py b/rigo/rigo/enums.py index 21e334605..918783390 100644 --- a/rigo/rigo/enums.py +++ b/rigo/rigo/enums.py @@ -75,16 +75,21 @@ class RigoViewStates: class LocalActivityStates: ( READY, - UPDATING_REPOSITORIES_MASTER, - UPDATING_REPOSITORIES_SLAVE, - INSTALLING_APPLICATIONS, - ) = range(4) + UPDATING_REPOSITORIES, + MANAGING_APPLICATIONS, + ) = range(3) class BusyError(Exception): """ Cannot acknowledge a Local Activity change. """ + class SameError(Exception): + """ + Cannot set the same Local Activity. + The proposed activity equals the current one. + """ + class AlreadyReadyError(Exception): """ Cannot acknowledge a Local Activity change to diff --git a/rigo/rigo/ui/gtk3/controllers/application.py b/rigo/rigo/ui/gtk3/controllers/application.py index 3f40b8387..8903682a2 100644 --- a/rigo/rigo/ui/gtk3/controllers/application.py +++ b/rigo/rigo/ui/gtk3/controllers/application.py @@ -83,7 +83,7 @@ class ApplicationViewController(GObject.Object): "application-request-action" : (GObject.SignalFlags.RUN_LAST, None, (GObject.TYPE_PYOBJECT, - str), + GObject.TYPE_PYOBJECT), ), } @@ -724,7 +724,7 @@ class ApplicationViewController(GObject.Object): Remove the given Application. """ self.emit("application-request-action", - app, AppActions.REMOVE) + app.get_installed(), AppActions.REMOVE) def _on_app_install(self, widget, app): """ diff --git a/rigo/rigo/ui/gtk3/widgets/apptreeview.py b/rigo/rigo/ui/gtk3/widgets/apptreeview.py index 9623e2c39..726a864c6 100644 --- a/rigo/rigo/ui/gtk3/widgets/apptreeview.py +++ b/rigo/rigo/ui/gtk3/widgets/apptreeview.py @@ -502,8 +502,6 @@ class AppTreeView(Gtk.TreeView): else: perform_action = AppActions.INSTALL - store.notify_action_request(app, path) - self._apc.emit("application-request-action", self.appmodel.get_application(app), perform_action) diff --git a/rigo/rigo_app.py b/rigo/rigo_app.py index 90d057a08..96623c9be 100644 --- a/rigo/rigo_app.py +++ b/rigo/rigo_app.py @@ -27,6 +27,7 @@ import time from threading import Lock, Semaphore import dbus +import dbus.exceptions sys.path.insert(0, "../lib") sys.path.insert(1, "../client") @@ -165,9 +166,10 @@ class RigoServiceController(GObject.Object): _OUTPUT_SIGNAL = "output" _REPOSITORIES_UPDATED_SIGNAL = "repositories_updated" _TRANSFER_OUTPUT_SIGNAL = "transfer_output" - _EXCLUSIVE_ACQUIRED_SIGNAL = "exclusive_acquired" _PING_SIGNAL = "ping" _RESOURCES_UNLOCK_REQUEST_SIGNAL = "resources_unlock_request" + _RESOURCES_LOCK_REQUEST_SIGNAL = "resources_lock_request" + _ACTIVITY_STARTED_SIGNAL = "activity_started" def __init__(self, rigo_app, activity_rwsem, auth, entropy_client, entropy_ws): @@ -176,8 +178,10 @@ class RigoServiceController(GObject.Object): self._activity_rwsem = activity_rwsem self._auth = auth self._nc = None + self._bottom_nc = None self._wc = None self._avc = None + self._apc = None self._terminal = None self._entropy = entropy_client self._entropy_ws = entropy_ws @@ -185,8 +189,6 @@ class RigoServiceController(GObject.Object): self.__system_bus = None self.__entropy_bus = None self.__entropy_bus_mutex = Lock() - self._connected_mutex = Lock() - self._connected = False self._registered_signals = {} self._registered_signals_mutex = Lock() @@ -196,12 +198,20 @@ class RigoServiceController(GObject.Object): self._please_wait_box = None self._please_wait_mutex = Lock() + self._application_request_mutex = Lock() + def set_applications_controller(self, avc): """ Bind ApplicationsViewController object to this class. """ self._avc = avc + def set_application_controller(self, apc): + """ + Bind ApplicationViewController object to this class. + """ + self._apc = apc + def set_terminal(self, terminal): """ Bind a TerminalWidget to this object, in order to be used with @@ -222,15 +232,56 @@ class RigoServiceController(GObject.Object): """ self._nc = nc + def set_bottom_notification_controller(self, bottom_nc): + """ + Bind a BottomNotificationViewController to this object. + """ + self._bottom_nc = bottom_nc + + def setup(self, shared_locked): + """ + Execute object setup once initialization phase is complete. + This phase is comprehensive of all the set_* method calls. + """ + if self._apc is not None: + # connect application request events + self._apc.connect( + "application-request-action", + self._on_application_request_action) + + # since we handle the lock/unlock of entropy + # resources here, we need to know what's the + # initial state + self._resources_locked = Lock() + if not shared_locked: + self._resources_locked.acquire() + + def service_available(self): + """ + Return whether the RigoDaemon dbus service is + available. + """ + try: + self._entropy_bus + return True + except dbus.exceptions.DBusException: + return False + def busy(self, local_activity): """ Become busy, switch to some local activity. If an activity is already taking place, LocalActivityStates.BusyError is raised. + If the active activity equals the requested one, + LocalActivityStates.SameError is raised. """ with self._local_activity_mutex: + if self._local_activity == local_activity: + raise LocalActivityStates.SameError() if self._local_activity != LocalActivityStates.READY: raise LocalActivityStates.BusyError() + GLib.idle_add(self._bottom_nc.set_activity, + local_activity) self._local_activity = local_activity def unbusy(self, current_activity): @@ -244,6 +295,8 @@ class RigoServiceController(GObject.Object): raise LocalActivityStates.AlreadyReadyError() if self._local_activity != current_activity: raise LocalActivityStates.UnbusyFromDifferentActivity() + GLib.idle_add(self._bottom_nc.set_activity, + LocalActivityStates.READY) self._local_activity = LocalActivityStates.READY def local_activity(self): @@ -298,10 +351,40 @@ class RigoServiceController(GObject.Object): self._resources_unlock_request_signal, dbus_interface=self.DBUS_INTERFACE) + # RigoDaemon Entropy Resources lock requests + self.__entropy_bus.connect_to_signal( + self._RESOURCES_LOCK_REQUEST_SIGNAL, + self._resources_lock_request_signal, + dbus_interface=self.DBUS_INTERFACE) + + # RigoDaemon tells us that a new activity + # has just begun + self.__entropy_bus.connect_to_signal( + self._ACTIVITY_STARTED_SIGNAL, + self._activity_started_signal, + dbus_interface=self.DBUS_INTERFACE) + return self.__entropy_bus - def _repositories_updated_signal(self, result, message, - token, local_activity): + ### GOBJECT EVENTS + + def _on_application_request_action(self, apc, app, app_action): + """ + This event comes from ApplicationViewController notifying + that user would like to schedule the given action for App. + "app" is an Application object, "app_action" is an AppActions + enum value. + """ + if const_debug_enabled(): + const_debug_write( + __name__, + "_on_application_request_action: " + "%s -> %s" % (app, app_action)) + self.application_request(app, app_action) + + ### DBUS SIGNALS + + def _repositories_updated_signal(self, result, message): """ Signal coming from RigoDaemon notifying us that repositories have been updated. @@ -324,33 +407,20 @@ class RigoServiceController(GObject.Object): "already consumed") return - # 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) + local_activity = LocalActivityStates.UPDATING_REPOSITORIES + # we don't expect to fail here, it would + # mean programming error. + self.unbusy(local_activity) - activity = DaemonActivityStates.UPDATING_REPOSITORIES - self._scale_down(activity, token) - self._release_local_resources() + # 1 -- ACTIVITY CRIT :: OFF + self._activity_rwsem.writer_release() - # 1 -- ACTIVITY CRIT :: OFF - self._activity_rwsem.writer_release() + self.emit("repositories-updated", + result, message) - GLib.idle_add(self.emit, "repositories-updated", - result, message) - 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() + const_debug_write( + __name__, + "_repositories_updated_signal: repositories-updated") def _output_signal(self, text, header, footer, back, importance, level, count_c, count_t, percent): @@ -433,6 +503,44 @@ class RigoServiceController(GObject.Object): self._entropy_bus, dbus_interface=self.DBUS_INTERFACE).pong() + def _resources_lock_request_signal(self, activity): + """ + RigoDaemon is asking us to acquire a shared Entropy Resources + lock. First we check if we have released it. + """ + const_debug_write( + __name__, + "_resources_lock_request_signal: " + "called, with remote activity: %s" % (activity,)) + + def _resources_lock(): + const_debug_write( + __name__, + "_resources_lock_request_signal._resources_lock: " + "enter (sleep)") + + self._entropy.lock_resources( + blocking=True, + shared=True) + self._release_local_resources() + + const_debug_write( + __name__, + "_resources_lock_request_signal._resources_lock: " + "regained shared lock") + + # it's a muted, ask the MainLoop to release it + # or it will explode + GLib.idle_add(self._resources_locked.release) + + # check if we actually released them + locked = self._resources_locked.acquire(False) + if not locked: + task = ParallelTask(_resources_lock) + task.name = "ResourceLockAfterRelease" + task.daemon = True + task.start() + def _resources_unlock_request_signal(self, activity): """ RigoDaemon is asking us to release our Entropy Resources Lock. @@ -446,7 +554,12 @@ class RigoServiceController(GObject.Object): if activity == DaemonActivityStates.UPDATING_REPOSITORIES: # did we ask that or is it another client? - if self.local_activity() == LocalActivityStates.READY: + local_activity = self.local_activity() + if local_activity == LocalActivityStates.READY: + locked = self._resources_locked.acquire(False) + if locked: + self._entropy.unlock_resources() + # another client, bend over XD # LocalActivityStates value will be atomically # switched in the above thread. @@ -461,13 +574,63 @@ class RigoServiceController(GObject.Object): __name__, "_resources_unlock_request_signal: " "somebody called repo update, starting here too") - else: + elif local_activity == \ + LocalActivityStates.UPDATING_REPOSITORIES: + locked = self._resources_locked.acquire(False) + if locked: + self._entropy.unlock_resources() + + self._entropy.unlock_resources() + const_debug_write( __name__, "_resources_unlock_request_signal: " "it's been us calling repositories update") # it's been us calling it, ignore request return + else: + const_debug_write( + __name__, + "_resources_unlock_request_signal: " + "not accepting RigoDaemon resources unlock request, " + "local activity: %s" % (local_activity,)) + + def _activity_started_signal(self, activity): + """ + RigoDaemon is telling us that the scheduled activity, + either by us or by another Rigo, has just begun and + that it, RigoDaemon, has now exclusive access to + Entropy Resources. + """ + const_debug_write( + __name__, + "_activity_started_signal: " + "called, with remote activity: %s" % (activity,)) + # reset please wait notification then + self._please_wait(None) + + ### GP PUBLIC METHODS + + def application_request(self, app, app_action): + """ + Start Application Action (install/remove). + """ + task = ParallelTask(self._application_request, + app, app_action) + task.name = "ApplicationRequest{%s, %s}" % ( + app, app_action,) + task.daemon = True + task.start() + + def update_repositories(self, repositories, force): + """ + Start Entropy Repositories Update + """ + task = ParallelTask(self._update_repositories, + repositories, force) + task.name = "UpdateRepositoriesThread" + task.daemon = True + task.start() def activity(self): """ @@ -514,21 +677,6 @@ class RigoServiceController(GObject.Object): """ GLib.idle_add(self.output_test) - def _connect(self): - """ - Inform RigoDaemon that a new Client is now connected. - This MUST be called after Entropy Resources Lock is - acquired in exclusive mode. - """ - with self._connected_mutex: - if self._connected: - return - # inform daemon that a new instance is now connected - dbus.Interface( - self._entropy_bus, - dbus_interface=self.DBUS_INTERFACE).connect() - self._connected = True - def _release_local_resources(self): """ Release all the local resources (like repositories) @@ -597,6 +745,10 @@ class RigoServiceController(GObject.Object): action_id = None if daemon_activity == DaemonActivityStates.UPDATING_REPOSITORIES: action_id = PolicyActions.UPDATE_REPOSITORIES + elif daemon_activity == DaemonActivityStates.MANAGING_APPLICATIONS: + action_id = PolicyActions.MANAGE_APPLICATIONS + elif daemon_activity == DaemonActivityStates.UPGRADING_SYSTEM: + action_id = PolicyActions.UPGRADE_SYSTEM if action_id is None: raise AttributeError("unsupported daemon activity") @@ -614,111 +766,23 @@ class RigoServiceController(GObject.Object): def _scale_up(self, activity): """ - Acquire (in blocking mode) the Entropy Resources Lock - in exclusive mode. Scale up privileges, and ask for - root password if not done yet. + Make sure User is authorized to perform a privileged + operation. """ granted = self._authorize(activity) if not granted: - return False - - acquired_sem = Semaphore(0) - - const_debug_write(__name__, "RigoServiceController: " - "_scale_up: enter") - - def _acquirer(): const_debug_write(__name__, "RigoServiceController: " - "_scale_up: acquired!") - acquired_sem.release() - - # start the rendezvous - sig_match = self._entropy_bus.connect_to_signal( - self._EXCLUSIVE_ACQUIRED_SIGNAL, - _acquirer, - dbus_interface=self.DBUS_INTERFACE) - - accepted = dbus.Interface( - self._entropy_bus, - dbus_interface=self.DBUS_INTERFACE - ).acquire_exclusive(activity) - if not accepted: - def _notify(): - if self._nc is not None: - box = self.ServiceNotificationBox( - _("Another activity is currently in progress"), - Gtk.MessageType.ERROR) - box.add_destroy_button(_("K thanks")) - self._nc.append(box) - GLib.idle_add(_notify) + "_scale_up: abort") return False - self._entropy.unlock_resources() - # lock down UI here and show a please wait - # state, or the user won't understand what's happening - # we assume we already have the activity_rwsem writer - # lock here - ms = _("Waiting for Entropy Resources availability, please wait") + ms = _("Waiting for RigoDaemon, please wait...") + # this will be reset when activity_started() arrives self._please_wait(ms) - try: - acquired_sem.acquire() # CANBLOCK - finally: - self._please_wait(None) - sig_match.remove() - - # we successfully passed the resource to RigoDaemon - self._connect() const_debug_write(__name__, "RigoServiceController: " "_scale_up: leave") return True - def _scale_down(self, activity, release_exclusive_token): - """ - Release RigoDaemon Entropy Resources and regain - control here. - """ - acquired_sem = Semaphore(0) - # start the rendezvous - - const_debug_write(__name__, "RigoServiceController: " - "_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) - task.name = "RigoDaemonResourcesReleaser" - task.daemon = True - task.start() - - # 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( - 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, master=True): """ @@ -728,10 +792,7 @@ class RigoServiceController(GObject.Object): # 1 -- ACTIVITY CRIT :: ON self._activity_rwsem.writer_acquire() # CANBLOCK - if master: - local_activity = LocalActivityStates.UPDATING_REPOSITORIES_MASTER - else: - local_activity = LocalActivityStates.UPDATING_REPOSITORIES_SLAVE + local_activity = LocalActivityStates.UPDATING_REPOSITORIES try: self.busy(local_activity) # will be unlocked when we get the signal back @@ -741,6 +802,12 @@ class RigoServiceController(GObject.Object): # 1 -- ACTIVITY CRIT :: OFF self._activity_rwsem.writer_release() return + except LocalActivityStates.SameError: + const_debug_write(__name__, "_update_repositories: " + "LocalActivityStates.SameError!") + # 1 -- ACTIVITY CRIT :: OFF + self._activity_rwsem.writer_release() + return if master: scaled = self._scale_up( @@ -750,16 +817,23 @@ class RigoServiceController(GObject.Object): # 1 -- ACTIVITY CRIT :: OFF self._activity_rwsem.writer_release() 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( + accepted = self._update_repositories_unlocked( repositories, force, master) + if not accepted: + self.unbusy(local_activity) + # 1 -- ACTIVITY CRIT :: OFF + self._activity_rwsem.writer_release() + + def _notify(): + box = self.ServiceNotificationBox( + _("Another activity is currently in progress"), + Gtk.MessageType.ERROR) + box.add_destroy_button(_("K thanks")) + self._nc.append(box) + GLib.idle_add(_notify) + def _update_repositories_unlocked(self, repositories, force, master): """ @@ -787,21 +861,16 @@ class RigoServiceController(GObject.Object): const_debug_write(__name__, "RigoServiceController: " "rigo UI now locked!") - 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, token): + def _repositories_updated_signal(result, message): 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, token, loc_activity) + result, message) with self._registered_signals_mutex: # connect our signal @@ -824,8 +893,9 @@ class RigoServiceController(GObject.Object): self._terminal.reset() self._release_local_resources() + accepted = True if master: - dbus.Interface( + accepted = dbus.Interface( self._entropy_bus, dbus_interface=self.DBUS_INTERFACE ).update_repositories(repositories, force) @@ -834,6 +904,8 @@ class RigoServiceController(GObject.Object): self._update_repositories_signal_check( sig_match, signal_sem) + return accepted + def _update_repositories_signal_check(self, sig_match, signal_sem): """ Called via _update_repositories_unlocked() in order to handle @@ -863,16 +935,63 @@ class RigoServiceController(GObject.Object): GLib.idle_add(self._repositories_updated_signal, 0, "", activity) - def update_repositories(self, repositories, force): + def _application_request(self, app, app_action, master=True): """ - Local method used to start Entropy repositories - update. + Forward Application Request (install or remove) to RigoDaemon. + Make sure there isn't any other ongoing activity. + """ + # Need to serialize access to this method because + # we're going to acquire several resources in a non-atomic + # way wrt access to this method. + with self._application_request_mutex: + + acquire_writer = True + # since we need to writer_acquire(), which is blocking + # better try to allocate the local activity first + local_activity = LocalActivityStates.MANAGING_APPLICATIONS + try: + self.busy(local_activity) + except LocalActivityStates.BusyError: + const_debug_write(__name__, "_application_request: " + "LocalActivityStates.BusyError!") + # doing other stuff, cannot go ahead + return + except LocalActivityStates.SameError: + const_debug_write(__name__, "_application_request: " + "LocalActivityStates.SameError, " + "no need to acquire writer") + # we're already doing this activity, do not acquire + # activity_rwsem + acquire_writer = False + + if acquire_writer: + # 2 -- ACTIVITY CRIT :: ON + const_debug_write(__name__, "_application_request: " + "about to acquire writer end of " + "activity rwsem") + self._activity_rwsem.writer_acquire() # CANBLOCK + + # acquire_exclusive should be removed and the same + # done by RigoDaemon, see the other window + """ + if master: + scaled = self._scale_up( + DaemonActivityStates.MANAGING_APPLICATIONS) + if not scaled: + self.unbusy(local_activity) + # 1 -- ACTIVITY CRIT :: OFF + self._activity_rwsem.writer_release() + 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, master) """ - task = ParallelTask(self._update_repositories, - repositories, force) - task.name = "UpdateRepositoriesThread" - task.daemon = True - task.start() class WorkViewController(GObject.Object): @@ -1029,6 +1148,85 @@ class WorkViewController(GObject.Object): class NotificationViewController(GObject.Object): + """ + Base Class for NotificationBox Controller. + """ + + def __init__(self, notification_box): + GObject.Object.__init__(self) + self._box = notification_box + self._context_id_map = {} + + def setup(self): + """ + Controller Setup code + """ + + def append(self, box, timeout=None, context_id=None): + """ + Append a notification to the Notification area. + context_id is used to automatically drop any other + notification exposing the same context identifier. + """ + context_id = box.get_context_id() + if context_id is not None: + old_box = self._context_id_map.get(context_id) + if old_box is not None: + old_box.destroy() + self._context_id_map[context_id] = box + box.render() + self._box.pack_start(box, False, False, 0) + box.show() + self._box.show() + if timeout is not None: + GLib.timeout_add_seconds(timeout, self.remove, box) + + def append_safe(self, box, timeout=None): + """ + Thread-safe version of append(). + """ + def _append(): + self.append(box, timeout=timeout) + GLib.idle_add(_append) + + def remove(self, box): + """ + Remove a NotificationBox from this notification + area, if there. + """ + if box in self._box.get_children(): + context_id = box.get_context_id() + if context_id is not None: + self._context_id_map.pop(context_id, None) + box.destroy() + + def remove_safe(self, box): + """ + Thread-safe version of remove(). + """ + GLib.idle_add(self.remove, box) + + def clear(self, managed=True): + """ + Clear all the notifications. + """ + for child in self._box.get_children(): + if child.is_managed() and not managed: + continue + context_id = child.get_context_id() + if context_id is not None: + self._context_id_map.pop(context_id, None) + child.destroy() + + def clear_safe(self, managed=True): + """ + Thread-safe version of clear(). + """ + GLib.idle_add(self.clear, managed) + + +class UpperNotificationViewController(NotificationViewController): + """ Notification area widget controller. This class features the handling of some built-in @@ -1036,20 +1234,24 @@ class NotificationViewController(GObject.Object): but also accepts external NotificationBox instances as well. """ - def __init__(self, activity_rwsem, entropy_client, entropy_ws, rigo_service, - avc, notification_box): - GObject.Object.__init__(self) + def __init__(self, activity_rwsem, entropy_client, entropy_ws, + rigo_service, avc, notification_box): + + NotificationViewController.__init__( + self, notification_box) + + self._avc = avc + self._service = rigo_service self._activity_rwsem = activity_rwsem self._entropy = entropy_client self._entropy_ws = entropy_ws - self._service = rigo_service - self._avc = avc - self._box = notification_box self._updates = None self._security_updates = None - self._context_id_map = {} def setup(self): + """ + Reimplemented from NotificationViewController. + """ GLib.timeout_add_seconds(1, self._calculate_updates) GLib.idle_add(self._check_connectivity) @@ -1164,8 +1366,6 @@ class NotificationViewController(GObject.Object): """ # FIXME, lxnay complete print("On Upgrade Request Received", args) - # FIXME, this is for testing, REMOVE !!!! - self._service.update_repositories([], True) def _on_update(self, box): """ @@ -1181,67 +1381,78 @@ class NotificationViewController(GObject.Object): """ self._avc.set_many_safe(self._updates) - def append(self, box, timeout=None, context_id=None): - """ - Append a notification to the Notification area. - context_id is used to automatically drop any other - notification exposing the same context identifier. - """ - context_id = box.get_context_id() - if context_id is not None: - old_box = self._context_id_map.get(context_id) - if old_box is not None: - old_box.destroy() - self._context_id_map[context_id] = box - box.render() - self._box.pack_start(box, False, False, 0) - box.show() - self._box.show() - if timeout is not None: - GLib.timeout_add_seconds(timeout, self.remove, box) +class BottomNotificationViewController(NotificationViewController): - def append_safe(self, box, timeout=None): - """ - Thread-safe version of append(). - """ - def _append(): - self.append(box, timeout=timeout) - GLib.idle_add(_append) + """ + Bottom Notification Area. + This area is only used to show Activity controls to User. + For example, during repositories update, this area just + shows one notification box stating that the above activity is in + progress, making possible to switch to the Work View anytime. + """ - def remove(self, box): - """ - Remove a NotificationBox from this notification - area, if there. - """ - if box in self._box.get_children(): - context_id = box.get_context_id() - if context_id is not None: - self._context_id_map.pop(context_id, None) - box.destroy() + UNIQUE_CONTEXT_ID = "BottomNotificationBoxContextId" - def remove_safe(self, box): - """ - Thread-safe version of remove(). - """ - GLib.idle_add(self.remove, box) + __gsignals__ = { + "show-work-view" : (GObject.SignalFlags.RUN_LAST, + None, + tuple(), + ), + } - def clear(self, managed=True): - """ - Clear all the notifications. - """ - for child in self._box.get_children(): - if child.is_managed() and not managed: - continue - context_id = child.get_context_id() - if context_id is not None: - self._context_id_map.pop(context_id, None) - child.destroy() + def __init__(self, notification_box): - def clear_safe(self, managed=True): + NotificationViewController.__init__( + self, notification_box) + + def _on_work_view_show(self, widget): """ - Thread-safe version of clear(). + User is asking to show the Work View. """ - GLib.idle_add(self.clear, managed) + self.emit("show-work-view") + + def _append_repositories_update(self): + """ + Add a NotificationBox related to Repositories Update + Activity in progress. + """ + msg = _("Repositories Update in progress...") + box = NotificationBox( + msg, message_type=Gtk.MessageType.INFO, + context_id=self.UNIQUE_CONTEXT_ID) + box.add_button(_("Show me"), self._on_work_view_show) + self.append(box) + + def _append_installing_apps(self): + """ + Add a NotificationBox related to Applications Install + Activity in progress. + """ + msg = _("Application Install in progress...") + box = NotificationBox( + msg, message_type=Gtk.MessageType.INFO, + context_id=self.UNIQUE_CONTEXT_ID) + box.add_button(_("Show me"), self._on_work_view_show) + self.append(box) + + def set_activity(self, local_activity): + """ + Set a current local Activity, showing the + most appropriate NotificationBox. + This method must be called from the MainLoop. + """ + if local_activity == LocalActivityStates.READY: + self.clear() + return + + if local_activity == LocalActivityStates.UPDATING_REPOSITORIES: + self._append_repositories_update() + return + elif local_activity == LocalActivityStates.MANAGING_APPLICATIONS: + self._append_installing_apps() + return + + raise NotImplementedError() class Rigo(Gtk.Application): @@ -1314,14 +1525,17 @@ class Rigo(Gtk.Application): self._search_entry = self._builder.get_object("searchEntry") self._static_view = self._builder.get_object("staticViewVbox") self._notification = self._builder.get_object("notificationBox") + self._bottom_notification = \ + self._builder.get_object("bottomNotificationBox") self._work_view = self._builder.get_object("workViewVbox") self._work_view.set_name("rigo-view") self._app_view_c = ApplicationViewController( self._entropy, self._entropy_ws, self._builder) - self._view = AppTreeView(self._entropy, self._app_view_c, icons, True, - AppListStore.ICON_SIZE, store=None) + self._view = AppTreeView( + self._entropy, self._app_view_c, icons, + True, AppListStore.ICON_SIZE, store=None) self._scrolled_view.add(self._view) self._app_store = AppListStore( @@ -1357,15 +1571,25 @@ class Rigo(Gtk.Application): self._avc.connect("view-filled", self._on_view_filled) self._avc.connect("view-want-change", self._on_view_change) - self._nc = NotificationViewController( + self._nc = UpperNotificationViewController( self._activity_rwsem, self._entropy, self._entropy_ws, self._service, self._avc, self._notification) + # Bottom NotificationBox controller. + # Bottom notifications are only used for + # providing Activity control to User during + # the Activity itself. + self._bottom_nc = BottomNotificationViewController( + self._bottom_notification) + self._service.set_bottom_notification_controller( + self._bottom_nc) + self._app_view_c.set_notification_controller(self._nc) self._app_view_c.set_applications_controller(self._avc) self._service.set_applications_controller(self._avc) + self._service.set_application_controller(self._app_view_c) self._service.set_notification_controller(self._nc) self._service.connect("repositories-updating", self._on_repo_updating) @@ -1375,6 +1599,8 @@ class Rigo(Gtk.Application): self._service, self._work_view) self._service.set_work_controller(self._work_view_c) + self._bottom_nc.connect("show-work-view", self._on_show_work_view) + def is_ui_locked(self): """ Return whether the UI is currently locked. @@ -1389,6 +1615,13 @@ class Rigo(Gtk.Application): self._search_entry.set_sensitive(False) self._change_view_state(state, lock=True) + def _on_show_work_view(self, widget): + """ + We've been explicitly asked to switch to WORK_VIEW_STATE + """ + self._change_view_state(RigoViewStates.WORK_VIEW_STATE, + _ignore_lock=True) + def _on_repo_updated(self, widget, result, message): """ Emitted by RigoServiceController when we're allowed to @@ -1496,14 +1729,14 @@ class Rigo(Gtk.Application): """ self._work_view.hide() - def _change_view_state(self, state, lock=False): + def _change_view_state(self, state, lock=False, _ignore_lock=False): """ Change Rigo Application UI state. You can pass a custom widget that will be shown in case of static view state. """ with self._state_mutex: - if self._current_state_lock: + if self._current_state_lock and not _ignore_lock: const_debug_write( __name__, "cannot change view state, UI locked") @@ -1579,6 +1812,15 @@ class Rigo(Gtk.Application): Gtk.main_quit() return + if not self._service.service_available(): + self._show_ok_dialog( + None, + escape_markup(_("Rigo")), + escape_markup(_("RigoDaemon service is not available"))) + entropy.tools.kill_threads() + Gtk.main_quit() + return + acquired = not self._entropy.wait_resources( max_lock_count=1, shared=True) @@ -1619,7 +1861,7 @@ class Rigo(Gtk.Application): task.name = "UpdateRepositoriesUnlocked" task.start() - elif activity == DaemonActivityStates.INSTALLING_APPLICATION: + elif activity == DaemonActivityStates.MANAGING_APPLICATIONS: # FIXME, jump to WORK_VIEW and show the progress. msg = _("Background Service is installing Applications") elif activity == DaemonActivityStates.UPGRADING_SYSTEM: @@ -1654,6 +1896,7 @@ class Rigo(Gtk.Application): self._avc.setup() self._nc.setup() self._work_view_c.setup() + self._service.setup(acquired) self._window.show() def run(self): diff --git a/rigo/test_polkit.py b/rigo/test_polkit.py new file mode 100644 index 000000000..d346fb4d2 --- /dev/null +++ b/rigo/test_polkit.py @@ -0,0 +1,71 @@ +import os +from threading import Semaphore +from gi.repository import GLib, Polkit, GObject + +class AuthenticationController(object): + + """ + This class handles User authentication required + for privileged activies, like Repository updates + and Application management. + """ + + def __init__(self, mainloop): + self._authenticated = False + self._authenticated_sem = Semaphore(1) + self._mainloop = mainloop + + def authenticate(self, action_id, authentication_callback): + """ + Authenticate current User asking Administrator + passwords. + authentication_callback is the function that + is called after the authentication procedure, + providing one boolean argument describing the + process result: True for authenticated, False + for not authenticated. + This method must be called from the MainLoop. + If authentication has been already + """ + self._authenticated_sem.acquire() + if self._authenticated: + try: + authentication_callback(True) + finally: + self._authenticated_sem.release() + return + + def _polkit_auth_callback(authority, res, loop): + authenticated = False + try: + result = authority.check_authorization_finish(res) + if result.get_is_authorized(): + authenticated = True + elif result.get_is_challenge(): + authenticated = True + except GObject.GError as err: + raise err + finally: + self._authenticated = authenticated + self._authenticated_sem.release() + authentication_callback(authenticated) + + # authenticated_sem will be released in the callback + authority = Polkit.Authority.get() + subject = Polkit.UnixProcess.new(os.getppid()) + authority.check_authorization( + subject, + action_id, + None, + Polkit.CheckAuthorizationFlags.ALLOW_USER_INTERACTION, + None, # Gio.Cancellable() + _polkit_auth_callback, + self._mainloop) + +def callback(result): + print "Auth status", result + +mainloop = GLib.MainLoop() +ctrl = AuthenticationController(mainloop) +GLib.idle_add(ctrl.authenticate, "org.sabayon.RigoDaemon.update", callback) +mainloop.run()