Add a PoC implementation of ActiveState subscriptions

Change-Id: I3773f135594b844e6640b8801a3b80dc994903b9
(cherry picked from commit 18bd36c8e013548f4bbd782504f253d3dbca185e)

diff --git a/src/libactd/unit_control.c b/src/libactd/unit_control.c
index d6133b2b8b14d8fa07a9be763911b25c3412f654..5de45fe31d9856b9ec91f1166a36f0443934137b 100644 (file)
--- a/src/libactd/unit_control.c
+++ b/src/libactd/unit_control.c
@@ -263,3 +263,191 @@ int actd_restart_unit_async(BusType bus_type, const char *unit, actd_unit_cb cb,
 {
        return call_uc_async((GBusType)bus_type, "Restart", unit, cb, user_data, timeout_ms);
 }
+
+// TODO: consider moving the code below to separate file
+// TODO: return values of API
+
+static void g_free_cleanup(gpointer ptr)
+{
+       if (!ptr)
+               return;
+       g_free(*(gpointer *)ptr);
+}
+
+// We hold an instance of this struct per unit that we are subscribing on.
+struct systemd_signal_subscription {
+       guint subscription_id;
+       GHashTable *callback_ptr_to_user_data_ptr;
+       enum {
+               LAST_STATUS_UNKNOWN,
+               LAST_STATUS_ACTIVE,
+               LAST_STATUS_INACTIVE,
+       } last_status;
+};
+
+static void systemd_signal_subscription_destroy(struct systemd_signal_subscription *ptr)
+{
+       if (ptr == NULL)
+               return;
+       g_hash_table_unref(ptr->callback_ptr_to_user_data_ptr);
+       g_free(ptr);
+}
+
+// This hashtable contains all the subscription data.
+// It is protected by an RWLock; callbacks take the reader lock (even though they don't touch the hashtable directly),
+// and (un)subscribe operations take the writer lock.
+static GHashTable *unit_name_to_systemd_signal_subscription = NULL;
+static GRWLock unit_name_to_systemd_signal_subscription_lock;
+// TODO: consider __attribute__(destructor) to cleanup the hashtable at program end
+
+static void reader_unlock_at_scope_end(void *x)
+{
+       g_rw_lock_reader_unlock(&unit_name_to_systemd_signal_subscription_lock);
+}
+
+static void writer_unlock_at_scope_end(void *x)
+{
+       g_rw_lock_writer_unlock(&unit_name_to_systemd_signal_subscription_lock);
+}
+
+static void call_callback(gpointer key, gpointer value, gpointer user_data)
+{
+       actd_unit_cb cb = key;
+       int status = GPOINTER_TO_INT(user_data);
+       cb(status, value);
+}
+
+static void signal_callback(GDBusConnection *connection, const gchar *sender_name, const gchar *object_path, const gchar *interface_name, const gchar *signal_name, GVariant *parameters, gpointer user_data)
+{
+       g_rw_lock_reader_lock(&unit_name_to_systemd_signal_subscription_lock);
+       __attribute__((cleanup(reader_unlock_at_scope_end))) void *unlock = unlock;
+
+       struct systemd_signal_subscription *value = user_data;
+
+       GVariant *vardict;
+       g_variant_get_child(parameters, 1, "@a{sv}", &vardict);
+
+       gchar *newstate;
+       gboolean changed = g_variant_lookup(vardict, "ActiveState", "s", &newstate);
+       if (changed) {
+               int state = -1;
+               if (value->last_status != LAST_STATUS_ACTIVE && strcmp(newstate, "active") == 0) {
+                       value->last_status = LAST_STATUS_ACTIVE;
+                       state = UNIT_CONTROL_UNIT_STATE_ACTIVE;
+               } else if (value->last_status != LAST_STATUS_INACTIVE && strcmp(newstate, "inactive") == 0) {
+                       value->last_status = LAST_STATUS_INACTIVE;
+                       state = UNIT_CONTROL_UNIT_STATE_INACTIVE;
+               }
+
+               if (state != -1)
+                       g_hash_table_foreach(value->callback_ptr_to_user_data_ptr, call_callback, GINT_TO_POINTER(state));
+       }
+}
+
+static int subscribe_to_unit(GDBusConnection *bus, const char *unit, struct systemd_signal_subscription *value)
+{
+       GVariant *msg = g_dbus_connection_call_sync(bus, SYSTEMD_SERVICE, SYSTEMD_OBJ_PATH, SYSTEMD_INTERFACE, "GetUnit", g_variant_new("(s)", unit), NULL, G_DBUS_CALL_FLAGS_NONE, timeout2glib(1000), NULL, NULL);
+       if (!msg)
+               return -EIO;
+
+       __attribute__((cleanup(g_free_cleanup))) gchar *path = NULL;
+       g_variant_get(msg, "(o)", &path);
+       g_variant_unref(msg);
+       msg = NULL;
+
+       // Note that we destroy the systemd_signal_subscription handle on unsubscribing. As per GDBus docs this is the only correct place to do so.
+       guint ret = g_dbus_connection_signal_subscribe(bus, SYSTEMD_SERVICE, "org.freedesktop.DBus.Properties", "PropertiesChanged", path, NULL, G_DBUS_SIGNAL_FLAGS_NONE, signal_callback, value, (GDestroyNotify)systemd_signal_subscription_destroy);
+       // TODO: what if ret is too big?
+       return ret;
+}
+
+int actd_register_unit_state_callback(BusType bus_type, const char *unit, actd_unit_cb cb, void *user_data)
+{
+       GDBusConnection *bus = g_bus_get_sync((GBusType)bus_type, NULL, NULL);
+       if (!bus)
+               return -EIO;
+
+       g_rw_lock_writer_lock(&unit_name_to_systemd_signal_subscription_lock);
+       __attribute__((cleanup(writer_unlock_at_scope_end))) void *unlock = unlock;
+
+       if (unit_name_to_systemd_signal_subscription != NULL) {
+               // Maybe we have already subscribed?
+               struct systemd_signal_subscription *value = g_hash_table_lookup(unit_name_to_systemd_signal_subscription, unit);
+               if (value == NULL) {
+                       // No we haven't. At least the hashtable already exists.
+               } else {
+                       // We have! Therefore, we should just add the callback.
+                       if (!g_hash_table_insert(value->callback_ptr_to_user_data_ptr, cb, user_data))
+                               return -EEXIST;
+                       return 0;
+               }
+       } else {
+               // Since there is no hashtable, we should create it.
+               // Note that the passed keys will be freed on removal.
+               // However, we do not free values on removal, as they will be freed on unsubscribe.
+               // Passing unsubscribe as destructor would be problematic, as that operation requires DBus access.
+               unit_name_to_systemd_signal_subscription = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
+       }
+
+       // If we got here, we have a hashtable, but we haven't subscribed yet. Let's do so, and fill the hashtable afterwards.
+       __attribute__((cleanup(g_free_cleanup))) char *key = g_malloc(strlen(unit) + 1);
+       memcpy(key, unit, strlen(unit));
+       key[strlen(unit)] = '\0';
+
+       struct systemd_signal_subscription *value = g_malloc(sizeof(struct systemd_signal_subscription));
+       int sub_ret = subscribe_to_unit(bus, unit, value);
+       if (sub_ret < 0) {
+               g_free(value);
+               return sub_ret;
+       }
+       value->subscription_id = sub_ret;
+       value->callback_ptr_to_user_data_ptr = g_hash_table_new(g_direct_hash, g_direct_equal);
+       value->last_status = LAST_STATUS_UNKNOWN;
+       g_hash_table_insert(value->callback_ptr_to_user_data_ptr, cb, user_data);
+       g_hash_table_insert(unit_name_to_systemd_signal_subscription, key, value);
+       value = NULL;
+       key = NULL;
+       return 0;
+}
+
+int actd_unregister_unit_state_callback(BusType bus_type, const char *unit, actd_unit_cb cb)
+{
+       GDBusConnection *bus = g_bus_get_sync((GBusType)bus_type, NULL, NULL);
+       if (!bus)
+               return -EIO;
+
+       g_rw_lock_writer_lock(&unit_name_to_systemd_signal_subscription_lock);
+       __attribute__((cleanup(writer_unlock_at_scope_end))) void *unlock = unlock;
+
+       if (unit_name_to_systemd_signal_subscription == NULL) {
+               // No hashtable, nothing to unsubscribe from.
+               return -ENOENT;
+       }
+
+       struct systemd_signal_subscription *value = g_hash_table_lookup(unit_name_to_systemd_signal_subscription, unit);
+       if (value == NULL) {
+               // We aren't actually subscribed to this unit.
+               return -ENOENT;
+       }
+
+       // Let's remove the callback from our list.
+       if (!g_hash_table_remove(value->callback_ptr_to_user_data_ptr, cb)) {
+               // ... unless it hasn't actually been added to it.
+               return -ENOENT;
+       }
+
+       // If there are no more callbacks, let's unsubscribe from this signal completely.
+       if (g_hash_table_size(value->callback_ptr_to_user_data_ptr) == 0) {
+               g_dbus_connection_signal_unsubscribe(bus, value->subscription_id);
+               value = NULL;
+               g_hash_table_remove(unit_name_to_systemd_signal_subscription, unit);
+
+               // If there are no more subscriptions, let's get rid of the main hashmap.
+               if (g_hash_table_size(unit_name_to_systemd_signal_subscription) == 0) {
+                       g_hash_table_unref(unit_name_to_systemd_signal_subscription);
+                       unit_name_to_systemd_signal_subscription = NULL;
+               }
+       }
+
+       return 0;
+}