commit c7ca5eccb5709c75797579f508c3f9ae968cd55e

Author: Andrew Beekhof <andrew@beekhof.net>

Date:   Fri Jan 31 10:53:35 2014 +1100

    Fix: Bug rhbz#1057697 - Use native DBus library for systemd async support to avoid problematic use of threads

    (cherry picked from commit 2f90aad962c63eba313ed466580703434c80bd1a)

diff --git a/lib/services/dbus.c b/lib/services/dbus.c

index 69ea6ac..a3286f2 100644

--- a/lib/services/dbus.c

+++ b/lib/services/dbus.c

@@ -4,6 +4,9 @@

 #include <dbus/dbus.h>

 #include <pcmk-dbus.h>

+#define BUS_PROPERTY_IFACE "org.freedesktop.DBus.Properties"

+

+

 static bool pcmk_dbus_error_check(DBusError *err, const char *prefix, const char *function, int line) 

 {

     if (err && dbus_error_is_set(err)) {

@@ -43,36 +46,18 @@ bool pcmk_dbus_append_arg(DBusMessage *msg, int dtype, const void *value)

     return TRUE;

 }

-DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, char **e)

+bool

+pcmk_dbus_find_error(const char *method, DBusPendingCall* pending, DBusMessage *reply, DBusError *ret)

 {

     DBusError error;

-    const char *method = NULL;

-    DBusMessage *reply = NULL;

-    DBusPendingCall* pending = NULL;

     dbus_error_init(&error);

-    CRM_ASSERT(dbus_message_get_type (msg) == DBUS_MESSAGE_TYPE_METHOD_CALL);

-    method = dbus_message_get_member (msg);

+    if(pending == NULL) {

+        error.name = "org.clusterlabs.pacemaker.NoRequest";

+        error.message = "No request sent";

-    // send message and get a handle for a reply

-    if (!dbus_connection_send_with_reply (connection, msg, &pending, -1)) { // -1 is default timeout

-        crm_err("Send with reply failed");

-        return NULL;

-    }

-    if (NULL == pending) {

-        crm_err("No pending call found");

-        return NULL;

-    }

-

-    dbus_connection_flush(connection);

-

-    /* block until we receive a reply */

-    dbus_pending_call_block(pending);

-

-    /* get the reply message */

-    reply = dbus_pending_call_steal_reply(pending);

-    if(reply == NULL) {

+    } else if(reply == NULL) {

         error.name = "org.clusterlabs.pacemaker.NoReply";

         error.message = "No reply";

@@ -80,7 +65,6 @@ DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, c

         DBusMessageIter args;

         int dtype = dbus_message_get_type(reply);

-

         switch(dtype) {

             case DBUS_MESSAGE_TYPE_METHOD_RETURN:

                 dbus_message_iter_init(reply, &args);

@@ -104,7 +88,7 @@ DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, c

             case DBUS_MESSAGE_TYPE_ERROR:

                 dbus_set_error_from_message (&error, reply);

-                crm_err("%s error '%s': %s", method, error.name, error.message);

+                crm_info("%s error '%s': %s", method, error.name, error.message);

                 break;

             default:

                 error.message = "Unknown reply type";

@@ -113,23 +97,86 @@ DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, c

         }

     }

-    if(error.name) {

-        if(e) {

-            *e = strdup(error.name);

+    if(ret && (error.name || error.message)) {

+        *ret = error;

+        return TRUE;

+    }

+

+    return FALSE;

+}

+

+DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, DBusError *error)

+{

+    const char *method = NULL;

+    DBusMessage *reply = NULL;

+    DBusPendingCall* pending = NULL;

+

+    CRM_ASSERT(dbus_message_get_type (msg) == DBUS_MESSAGE_TYPE_METHOD_CALL);

+    method = dbus_message_get_member (msg);

+

+    // send message and get a handle for a reply

+    if (!dbus_connection_send_with_reply (connection, msg, &pending, -1)) { // -1 is default timeout

+        if(error) {

+            error->message = "Call to dbus_connection_send_with_reply() failed";

+            error->name = "org.clusterlabs.pacemaker.SendFailed";

         }

+        crm_err("Error sending %s request", method);

+        return NULL;

+    }

+

+    dbus_connection_flush(connection);

+

+    if(pending) {

+        /* block until we receive a reply */

+        dbus_pending_call_block(pending);

+

+        /* get the reply message */

+        reply = dbus_pending_call_steal_reply(pending);

+    }

+

+    if(pcmk_dbus_find_error(method, pending, reply, error)) {

+        crm_trace("Was error: '%s' '%s'", error->name, error->message);

         if(reply) {

             dbus_message_unref(reply);

             reply = NULL;

         }

-    } else if(e) {

-        *e = NULL;

+    }

+    crm_trace("Was error: '%s' '%s'", error->name, error->message);

+

+    if(pending) {

+        /* free the pending message handle */

+        dbus_pending_call_unref(pending);

     }

-    /* free the pending message handle */

-    dbus_pending_call_unref(pending);

     return reply;

 }

+bool pcmk_dbus_send(DBusMessage *msg, DBusConnection *connection,

+                    void(*done)(DBusPendingCall *pending, void *user_data), void *user_data)

+{

+    DBusError error;

+    const char *method = NULL;

+    DBusPendingCall* pending = NULL;

+

+    dbus_error_init(&error);

+

+    CRM_ASSERT(dbus_message_get_type (msg) == DBUS_MESSAGE_TYPE_METHOD_CALL);

+    method = dbus_message_get_member (msg);

+

+    // send message and get a handle for a reply

+    if (!dbus_connection_send_with_reply (connection, msg, &pending, -1)) { // -1 is default timeout

+        crm_err("Send with reply failed for %s", method);

+        return FALSE;

+

+    } else if (pending == NULL) {

+        crm_err("No pending call found for %s", method);

+        return FALSE;

+

+    }

+    CRM_ASSERT(dbus_pending_call_set_notify(pending, done, user_data, NULL));

+    return TRUE;

+}

+

 bool pcmk_dbus_type_check(DBusMessage *msg, DBusMessageIter *field, int expected, const char *function, int line)

 {

     int dtype = dbus_message_iter_get_arg_type(field);

@@ -147,8 +194,6 @@ bool pcmk_dbus_type_check(DBusMessage *msg, DBusMessageIter *field, int expected

     return TRUE;

 }

-#define BUS_PROPERTY_IFACE "org.freedesktop.DBus.Properties"

-

 char *

 pcmk_dbus_get_property(

     DBusConnection *connection, const char *target, const char *obj, const gchar * iface, const char *name)

@@ -160,10 +205,11 @@ pcmk_dbus_get_property(

     /* DBusBasicValue value; */

     const char *method = "GetAll";

     char *output = NULL;

-    char *error = NULL;

+    DBusError error;

         /* desc = systemd_unit_property(path, BUS_NAME ".Unit", "Description"); */

+    dbus_error_init(&error);

     crm_info("Calling: %s on %s", method, target);

     msg = dbus_message_new_method_call(target, // target for the method call

                                        obj, // object to call on

@@ -180,7 +226,7 @@ pcmk_dbus_get_property(

     reply = pcmk_dbus_send_recv(msg, connection, &error);

     dbus_message_unref(msg);

-    if(reply == NULL) {

+    if(error.name) {

         crm_err("Call to %s for %s failed: No reply", method, iface);

         return NULL;

@@ -242,20 +288,105 @@ pcmk_dbus_get_property(

     return output;

 }

+static void pcmk_dbus_connection_dispatch(DBusConnection *connection, DBusDispatchStatus new_status, void *data){

+    crm_trace("status %d for %p", new_status, data);

+    if (new_status == DBUS_DISPATCH_DATA_REMAINS){

+        dbus_connection_dispatch(connection);

+    }

+}

+

+static int

+pcmk_dbus_watch_dispatch(gpointer userdata)

+{

+    DBusWatch *watch = userdata;

+    int flags = dbus_watch_get_flags(watch);

+    crm_trace("Dispatching %p with flags %d", watch, flags);

+    if(flags & DBUS_WATCH_READABLE) {

+        dbus_watch_handle(watch, DBUS_WATCH_READABLE);

+    } else {

+        dbus_watch_handle(watch, DBUS_WATCH_ERROR);

+    }

+    return 0;

+}

+

+static void

+pcmk_dbus_watch_destroy(gpointer userdata)

+{

+    crm_trace("Destroyed %p", userdata);

+}

+struct mainloop_fd_callbacks pcmk_dbus_cb = {

+    .dispatch = pcmk_dbus_watch_dispatch,

+    .destroy = pcmk_dbus_watch_destroy,

+};

-int dbus_watch_get_unix_fd	(	DBusWatch * 	watch	);

+static dbus_bool_t

+pcmk_dbus_watch_add(DBusWatch *watch, void *data){

+    int fd = dbus_watch_get_unix_fd(watch);

+    mainloop_io_t *client = mainloop_add_fd(

+        "dbus", G_PRIORITY_DEFAULT, fd, watch, &pcmk_dbus_cb);

-/* http://dbus.freedesktop.org/doc/api/html/group__DBusConnection.html#gaebf031eb444b4f847606aa27daa3d8e6 */

-    

-DBUS_EXPORT dbus_bool_t dbus_connection_set_watch_functions(

-    DBusConnection * 	connection,

-    DBusAddWatchFunction 	add_function,

-    DBusRemoveWatchFunction 	remove_function,

-    DBusWatchToggledFunction 	toggled_function,

-    void * 	data,

-    DBusFreeFunction 	free_data_function 

-    );

+    crm_trace("Added %p with fd=%d", watch, fd);

+    dbus_watch_set_data(watch, client, NULL);

+    return TRUE;

+}

+

+static void

+pcmk_dbus_watch_remove(DBusWatch *watch, void *data){

+    mainloop_io_t *client = dbus_watch_get_data(watch);

+

+    crm_trace("Removed %p", watch);

+    mainloop_del_fd(client);

+}

+

+static gboolean

+pcmk_dbus_timeout_dispatch(gpointer data)

+{

+    crm_trace("Timeout for %p");

+    dbus_timeout_handle(data);

+    return FALSE;

+}

+

+static dbus_bool_t

+pcmk_dbus_timeout_add(DBusTimeout *timeout, void *data){

+    guint id = g_timeout_add(dbus_timeout_get_interval(timeout), pcmk_dbus_timeout_dispatch, timeout);

+

+    if(id) {

+        dbus_timeout_set_data(timeout, GUINT_TO_POINTER(id), NULL);

+    }

+    return TRUE;

+}

+

+static void

+pcmk_dbus_timeout_remove(DBusTimeout *timeout, void *data){

+    void *vid = dbus_timeout_get_data(timeout);

+    guint id = GPOINTER_TO_UINT(vid);

+

+    if(id) {

+        g_source_remove(id);

+        dbus_timeout_set_data(timeout, 0, NULL);

+    }

+}

+

+static void

+pcmk_dbus_timeout_toggle(DBusTimeout *timeout, void *data){

+    if(dbus_timeout_get_enabled(timeout)) {

+        pcmk_dbus_timeout_add(timeout, data);

+    } else {

+        pcmk_dbus_timeout_remove(timeout, data);

+    }

+}

+

+/* Inspired by http://www.kolej.mff.cuni.cz/~vesej3am/devel/dbus-select.c */

+

+void pcmk_dbus_connection_setup_with_select(DBusConnection *c){

+	dbus_connection_set_timeout_functions(

+            c, pcmk_dbus_timeout_add, pcmk_dbus_timeout_remove, pcmk_dbus_timeout_toggle, NULL, NULL);

+	dbus_connection_set_watch_functions(c, pcmk_dbus_watch_add, pcmk_dbus_watch_remove, NULL, NULL, NULL);

+	dbus_connection_set_dispatch_status_function(c, pcmk_dbus_connection_dispatch, NULL, NULL);

+

+	pcmk_dbus_connection_dispatch(c, dbus_connection_get_dispatch_status(c), NULL);

+}

diff --git a/lib/services/pcmk-dbus.h b/lib/services/pcmk-dbus.h

index 27ac737..c8d2234 100644

--- a/lib/services/pcmk-dbus.h

+++ b/lib/services/pcmk-dbus.h

@@ -1,7 +1,13 @@

 DBusConnection *pcmk_dbus_connect(void);

+void pcmk_dbus_connection_setup_with_select(DBusConnection *c);

 void pcmk_dbus_disconnect(DBusConnection *connection);

-DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, char **error);

+bool pcmk_dbus_send(DBusMessage *msg, DBusConnection *connection,

+                    void(*done)(DBusPendingCall *pending, void *user_data), void *user_data);

+DBusMessage *pcmk_dbus_send_recv(DBusMessage *msg, DBusConnection *connection, DBusError *error);

 bool pcmk_dbus_append_arg(DBusMessage *msg, int dtype, const void *value);

 bool pcmk_dbus_type_check(DBusMessage *msg, DBusMessageIter *field, int expected, const char *function, int line);

 char *pcmk_dbus_get_property(DBusConnection *connection, const char *target, const char *obj, const gchar * iface, const char *name);

+

+bool pcmk_dbus_find_error(const char *method, DBusPendingCall* pending, DBusMessage *reply, DBusError *error);

+

diff --git a/lib/services/systemd.c b/lib/services/systemd.c

index 9aa5b03..a06d547 100644

--- a/lib/services/systemd.c

+++ b/lib/services/systemd.c

@@ -69,6 +69,7 @@ systemd_init(void)

     if (need_init) {

         need_init = 0;

         systemd_proxy = pcmk_dbus_connect();

+        pcmk_dbus_connection_setup_with_select(systemd_proxy);

     }

     if (systemd_proxy == NULL) {

         return FALSE;

@@ -122,7 +123,7 @@ systemd_unit_by_name(const gchar * arg_name, gchar ** out_unit)

     DBusMessage *reply = NULL;

     const char *method = "GetUnit";

     char *name = NULL;

-    char *error = NULL;

+    DBusError error;

 /*

   <method name="GetUnit">

@@ -144,13 +145,12 @@ systemd_unit_by_name(const gchar * arg_name, gchar ** out_unit)

         pcmk_dbus_append_arg(msg, DBUS_TYPE_STRING, &name);

+        dbus_error_init(&error);

         reply = pcmk_dbus_send_recv(msg, systemd_proxy, &error);

         dbus_message_unref(msg);

-        if(error) {

-            crm_info("Call to %s failed: %s", method, error);

-            free(error);

-            error = NULL;

+        if(error.name) {

+            crm_info("Call to %s failed: %s", method, error.name);

         } else if (dbus_message_iter_init(reply, &args)) {

@@ -192,7 +192,7 @@ systemd_unit_listall(void)

     DBusMessage *msg = NULL;

     DBusMessage *reply = NULL;

     const char *method = "ListUnits";

-    char *error = NULL;

+    DBusError error;

     if (systemd_init() == FALSE) {

         return NULL;

@@ -204,15 +204,15 @@ systemd_unit_listall(void)

         "  </method>\n"                                                 \

 */

+    dbus_error_init(&error);

     msg = systemd_new_method(BUS_NAME".Manager", method);

     CRM_ASSERT(msg != NULL);

     reply = pcmk_dbus_send_recv(msg, systemd_proxy, &error);

     dbus_message_unref(msg);

-    if(error) {

-        crm_err("Call to %s failed: %s", method, error);

-        free(error);

+    if(error.name) {

+        crm_err("Call to %s failed: %s", method, error.name);

         return NULL;

     } else if (!dbus_message_iter_init(reply, &args)) {

@@ -318,64 +318,84 @@ systemd_unit_metadata(const char *name)

     return meta;

 }

-#if 0

+static bool

+systemd_mask_error(svc_action_t *op, const char *error)

+{

+    crm_trace("Could not issue %s for %s: %s", op->action, op->rsc, error);

+    if(strstr(error, "org.freedesktop.systemd1.InvalidName")

+       || strstr(error, "org.freedesktop.systemd1.LoadFailed")

+       || strstr(error, "org.freedesktop.systemd1.NoSuchUnit")) {

+

+        if (safe_str_eq(op->action, "stop")) {

+            crm_trace("Masking %s failure for %s: unknown services are stopped", op->action, op->rsc);

+            op->rc = PCMK_OCF_OK;

+

+        } else {

+            crm_trace("Mapping %s failure for %s: unknown services are not installed", op->action, op->rsc);

+            op->rc = PCMK_OCF_NOT_INSTALLED;

+            op->status = PCMK_LRM_OP_NOT_INSTALLED;

+        }

+        return TRUE;

+    }

+

+    return FALSE;

+}

+

 static void

-systemd_unit_exec_done(GObject * source_object, GAsyncResult * res, gpointer user_data)

+systemd_async_dispatch(DBusPendingCall *pending, void *user_data)

 {

-    GError *error = NULL;

-    GVariant *_ret = NULL;

+    DBusError error;

+    DBusMessage *reply = NULL;

     svc_action_t *op = user_data;

-    GDBusProxy *proxy = G_DBUS_PROXY(source_object);

-    /* Obtain rc and stderr/out */

-    _ret = g_dbus_proxy_call_finish(proxy, res, &error);

+    dbus_error_init(&error);

+    if(pending) {

+        reply = dbus_pending_call_steal_reply(pending);

+    }

+    if(pcmk_dbus_find_error(op->action, pending, reply, &error)) {

-    if (error) {

         /* ignore "already started" or "not running" errors */

-        crm_trace("Could not issue %s for %s: %s", op->action, op->rsc, error->message);

-        if (strstr(error->message, "systemd1.LoadFailed")

-            || strstr(error->message, "systemd1.InvalidName")) {

+        if (!systemd_mask_error(op, error.name)) {

+            crm_err("%s for %s: %s", op->action, op->rsc, error.message);

+        }

-            if (safe_str_eq(op->action, "stop")) {

-                crm_trace("Masking Stop failure for %s: unknown services are stopped", op->rsc);

-                op->rc = PCMK_OCF_OK;

+    } else {

+        DBusMessageIter args;

-            } else {

-                op->rc = PCMK_OCF_NOT_INSTALLED;

-                op->status = PCMK_LRM_OP_NOT_INSTALLED;

-            }

+        if(!dbus_message_iter_init(reply, &args)) {

+            crm_err("Call to %s failed: no arguments", op->action);

-        } else {

-            crm_err("Could not issue %s for %s: %s", op->action, op->rsc, error->message);

-        }

-        g_error_free(error);

-

-    } else if(g_variant_is_of_type (_ret, G_VARIANT_TYPE("(o)"))) {

-        char *path = NULL;

+        } else if(!pcmk_dbus_type_check(reply, &args, DBUS_TYPE_OBJECT_PATH, __FUNCTION__, __LINE__)) {

+            crm_warn("Call to %s passed but return type was unexpected", op->action);

+            op->rc = PCMK_OCF_OK;

-        g_variant_get(_ret, "(o)", &path);

-        crm_info("Call to %s passed: type '%s' %s", op->action, g_variant_get_type_string(_ret),

-                 path);

-        op->rc = PCMK_OCF_OK;

+        } else {

+            const char *path = NULL;

-    } else {

-        crm_err("Call to %s passed but return type was '%s' not '(o)'", op->action, g_variant_get_type_string(_ret));

-        op->rc = PCMK_OCF_OK;

+            dbus_message_get_args (reply, NULL,

+                                   DBUS_TYPE_OBJECT_PATH, &path,

+                                   DBUS_TYPE_INVALID);

+            crm_info("Call to %s passed: %s", op->action, path);

+            op->rc = PCMK_OCF_OK;

+        }

     }

     operation_finalize(op);

-    if (_ret) {

-        g_variant_unref(_ret);

+

+    if(pending) {

+        dbus_pending_call_unref(pending);

+    }

+    if(reply) {

+        dbus_message_unref(reply);

     }

 }

-#endif

 #define SYSTEMD_OVERRIDE_ROOT "/run/systemd/system/"

 gboolean

 systemd_unit_exec(svc_action_t * op, gboolean synchronous)

 {

-    char *error = NULL;

+    DBusError error;

     char *unit = NULL;

     const char *replace_s = "replace";

     gboolean pass = FALSE;

@@ -384,7 +404,8 @@ systemd_unit_exec(svc_action_t * op, gboolean synchronous)

     DBusMessage *msg = NULL;

     DBusMessage *reply = NULL;

     DBusMessageIter args;

-    

+

+    dbus_error_init(&error);

     op->rc = PCMK_OCF_UNKNOWN_ERROR;

     CRM_ASSERT(systemd_init());

@@ -465,16 +486,6 @@ systemd_unit_exec(svc_action_t * op, gboolean synchronous)

     crm_debug("Calling %s for %s: %s", method, op->rsc, unit);

-#if 0

-    if (synchronous == FALSE) {

-        g_dbus_proxy_call(systemd_proxy, method, g_variant_new("(ss)", name, "replace"),

-                          G_DBUS_CALL_FLAGS_NONE, op->timeout, NULL, systemd_unit_exec_done, op);

-        free(unit);

-        free(name);

-        return TRUE;

-    }

-#endif

-

     msg = systemd_new_method(BUS_NAME".Manager", method);

     CRM_ASSERT(msg != NULL);

@@ -482,43 +493,47 @@ systemd_unit_exec(svc_action_t * op, gboolean synchronous)

     pcmk_dbus_append_arg(msg, DBUS_TYPE_STRING, &name);

     pcmk_dbus_append_arg(msg, DBUS_TYPE_STRING, &replace_s);

+    if (synchronous == FALSE) {

+        free(unit);

+        free(name);

+        return pcmk_dbus_send(msg, systemd_proxy, systemd_async_dispatch, op);

+    }

+

     reply = pcmk_dbus_send_recv(msg, systemd_proxy, &error);

-    dbus_message_unref(msg);

-    if(error) {

+    if(error.name) {

         /* ignore "already started" or "not running" errors */

-        if (safe_str_eq(op->action, "stop")

-            && (strstr(error, "org.freedesktop.systemd1.InvalidName")

-                || strstr(error, "org.freedesktop.systemd1.NoSuchUnit"))) {

-            crm_trace("Masking Stop failure for %s: unknown services are stopped", op->rsc);

-            op->rc = PCMK_OCF_OK;

-        } else {

-            crm_err("Could not issue %s for %s: %s (%s)", method, op->rsc, error, unit);

+        if(!systemd_mask_error(op, error.name)) {

+            crm_err("Could not issue %s for %s: %s (%s)", method, op->rsc, error.name, unit);

         }

         goto cleanup;

     } else if(!dbus_message_iter_init(reply, &args)) {

         crm_err("Call to %s failed: no arguments", method);

         goto cleanup;

-    }

-    /* (o) */

-    if(!pcmk_dbus_type_check(reply, &args, DBUS_TYPE_OBJECT_PATH, __FUNCTION__, __LINE__)) {

-        crm_err("Call to %s failed: Message has invalid arguments", method);

+    } else if(!pcmk_dbus_type_check(reply, &args, DBUS_TYPE_OBJECT_PATH, __FUNCTION__, __LINE__)) {

+        crm_warn("Call to %s passed but return type was unexpected", op->action);

+        op->rc = PCMK_OCF_OK;

     } else {

-        DBusBasicValue value;

+        const char *path = NULL;

-        dbus_message_iter_get_basic(&args, &value);

-        crm_info("Call to %s passed: %s", op->action, value.str);

+        dbus_message_get_args (reply, NULL,

+                               DBUS_TYPE_OBJECT_PATH, &path,

+                               DBUS_TYPE_INVALID);

+        crm_info("Call to %s passed: %s", op->action, path);

         op->rc = PCMK_OCF_OK;

     }

   cleanup:

-    free(error);

     free(unit);

     free(name);

+    if(msg) {

+        dbus_message_unref(msg);

+    }

+

     if(reply) {

         dbus_message_unref(reply);

     }