Merge branch '1515-gdbus-threading' into 'master'

Resolve "gio/gdbus-threading test sometimes fails in CI"

Closes #1515

See merge request GNOME/glib!1375

10 files changed, 295 insertions, 245 deletions

diff --git a/gio/gdbusconnection.c b/gio/gdbusconnection.c
index 67496be5b..1a4dae3bd 100644
--- a/gio/gdbusconnection.c
+++ b/gio/gdbusconnection.c
@@ -222,7 +222,6 @@ typedef struct
 {
   GDestroyNotify              callback;
   gpointer                    user_data;
-  GMainContext               *context;
 } CallDestroyNotifyData;
 
 static gboolean
@@ -236,8 +235,6 @@ call_destroy_notify_data_in_idle (gpointer user_data)
 static void
 call_destroy_notify_data_free (CallDestroyNotifyData *data)
 {
-  if (data->context != NULL)
-    g_main_context_unref (data->context);
   g_free (data);
 }
 
@@ -258,14 +255,11 @@ call_destroy_notify (GMainContext  *context,
   CallDestroyNotifyData *data;
 
   if (callback == NULL)
-    goto out;
+    return;
 
   data = g_new0 (CallDestroyNotifyData, 1);
   data->callback = callback;
   data->user_data = user_data;
-  data->context = context;
-  if (data->context != NULL)
-    g_main_context_ref (data->context);
 
   idle_source = g_idle_source_new ();
   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
@@ -274,11 +268,8 @@ call_destroy_notify (GMainContext  *context,
                          data,
                          (GDestroyNotify) call_destroy_notify_data_free);
   g_source_set_name (idle_source, "[gio] call_destroy_notify_data_in_idle");
-  g_source_attach (idle_source, data->context);
+  g_source_attach (idle_source, context);
   g_source_unref (idle_source);
-
- out:
-  ;
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -3703,6 +3694,14 @@ unsubscribe_id_internal (GDBusConnection *connection,
  *
  * Unsubscribes from signals.
  *
+ * Note that there may still be D-Bus traffic to process (relating to this
+ * signal subscription) in the current thread-default #GMainContext after this
+ * function has returned. You should continue to iterate the #GMainContext
+ * until the #GDestroyNotify function passed to
+ * g_dbus_connection_signal_subscribe() is called, in order to avoid memory
+ * leaks through callbacks queued on the #GMainContext after it’s stopped being
+ * iterated.
+ *
  * Since: 2.26
  */
 void
diff --git a/gio/gdbusnameowning.c b/gio/gdbusnameowning.c
index ee7d8445f..d20e6ffed 100644
--- a/gio/gdbusnameowning.c
+++ b/gio/gdbusnameowning.c
@@ -889,6 +889,13 @@ g_bus_own_name_on_connection_with_closures (GDBusConnection    *connection,
  *
  * Stops owning a name.
  *
+ * Note that there may still be D-Bus traffic to process (relating to owning
+ * and unowning the name) in the current thread-default #GMainContext after
+ * this function has returned. You should continue to iterate the #GMainContext
+ * until the #GDestroyNotify function passed to g_bus_own_name() is called, in
+ * order to avoid memory leaks through callbacks queued on the #GMainContext
+ * after it’s stopped being iterated.
+ *
  * Since: 2.26
  */
 void
diff --git a/gio/gdbusnamewatching.c b/gio/gdbusnamewatching.c
index c103bb484..bc2a9119e 100644
--- a/gio/gdbusnamewatching.c
+++ b/gio/gdbusnamewatching.c
@@ -858,6 +858,13 @@ guint g_bus_watch_name_on_connection_with_closures (
  *
  * Stops watching a name.
  *
+ * Note that there may still be D-Bus traffic to process (relating to watching
+ * and unwatching the name) in the current thread-default #GMainContext after
+ * this function has returned. You should continue to iterate the #GMainContext
+ * until the #GDestroyNotify function passed to g_bus_watch_name() is called, in
+ * order to avoid memory leaks through callbacks queued on the #GMainContext
+ * after it’s stopped being iterated.
+ *
  * Since: 2.26
  */
 void
diff --git a/gio/gdbusproxy.c b/gio/gdbusproxy.c
index 39eed1688..4c682978d 100644
--- a/gio/gdbusproxy.c
+++ b/gio/gdbusproxy.c
@@ -95,28 +95,19 @@ G_LOCK_DEFINE_STATIC (properties_lock);
 
 /* ---------------------------------------------------------------------------------------------------- */
 
-G_LOCK_DEFINE_STATIC (signal_subscription_lock);
-
-typedef struct
-{
-  volatile gint ref_count;
-  GDBusProxy *proxy;
-} SignalSubscriptionData;
-
-static SignalSubscriptionData *
-signal_subscription_ref (SignalSubscriptionData *data)
+static GWeakRef *
+weak_ref_new (GObject *object)
 {
-  g_atomic_int_inc (&data->ref_count);
-  return data;
+  GWeakRef *weak_ref = g_new0 (GWeakRef, 1);
+  g_weak_ref_init (weak_ref, object);
+  return g_steal_pointer (&weak_ref);
 }
 
 static void
-signal_subscription_unref (SignalSubscriptionData *data)
+weak_ref_free (GWeakRef *weak_ref)
 {
-  if (g_atomic_int_dec_and_test (&data->ref_count))
-    {
-      g_slice_free (SignalSubscriptionData, data);
-    }
+  g_weak_ref_clear (weak_ref);
+  g_free (weak_ref);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -152,8 +143,6 @@ struct _GDBusProxyPrivate
 
   /* mutable, protected by properties_lock */
   GDBusObject *object;
-
-  SignalSubscriptionData *signal_subscription_data;
 };
 
 enum
@@ -190,22 +179,6 @@ G_DEFINE_TYPE_WITH_CODE (GDBusProxy, g_dbus_proxy, G_TYPE_OBJECT,
                          G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE, async_initable_iface_init))
 
 static void
-g_dbus_proxy_dispose (GObject *object)
-{
-  GDBusProxy *proxy = G_DBUS_PROXY (object);
-  G_LOCK (signal_subscription_lock);
-  if (proxy->priv->signal_subscription_data != NULL)
-    {
-      proxy->priv->signal_subscription_data->proxy = NULL;
-      signal_subscription_unref (proxy->priv->signal_subscription_data);
-      proxy->priv->signal_subscription_data = NULL;
-    }
-  G_UNLOCK (signal_subscription_lock);
-
-  G_OBJECT_CLASS (g_dbus_proxy_parent_class)->dispose (object);
-}
-
-static void
 g_dbus_proxy_finalize (GObject *object)
 {
   GDBusProxy *proxy = G_DBUS_PROXY (object);
@@ -346,7 +319,6 @@ g_dbus_proxy_class_init (GDBusProxyClass *klass)
 {
   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 
-  gobject_class->dispose      = g_dbus_proxy_dispose;
   gobject_class->finalize     = g_dbus_proxy_finalize;
   gobject_class->set_property = g_dbus_proxy_set_property;
   gobject_class->get_property = g_dbus_proxy_get_property;
@@ -638,9 +610,6 @@ static void
 g_dbus_proxy_init (GDBusProxy *proxy)
 {
   proxy->priv = g_dbus_proxy_get_instance_private (proxy);
-  proxy->priv->signal_subscription_data = g_slice_new0 (SignalSubscriptionData);
-  proxy->priv->signal_subscription_data->ref_count = 1;
-  proxy->priv->signal_subscription_data->proxy = proxy;
   proxy->priv->properties = g_hash_table_new_full (g_str_hash,
                                                    g_str_equal,
                                                    g_free,
@@ -868,21 +837,12 @@ on_signal_received (GDBusConnection *connection,
                     GVariant        *parameters,
                     gpointer         user_data)
 {
-  SignalSubscriptionData *data = user_data;
+  GWeakRef *proxy_weak = user_data;
   GDBusProxy *proxy;
 
-  G_LOCK (signal_subscription_lock);
-  proxy = data->proxy;
+  proxy = G_DBUS_PROXY (g_weak_ref_get (proxy_weak));
   if (proxy == NULL)
-    {
-      G_UNLOCK (signal_subscription_lock);
-      return;
-    }
-  else
-    {
-      g_object_ref (proxy);
-      G_UNLOCK (signal_subscription_lock);
-    }
+    return;
 
   if (!proxy->priv->initialized)
     goto out;
@@ -929,8 +889,7 @@ on_signal_received (GDBusConnection *connection,
                  parameters);
 
  out:
-  if (proxy != NULL)
-    g_object_unref (proxy);
+  g_clear_object (&proxy);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -1038,7 +997,7 @@ on_properties_changed (GDBusConnection *connection,
                        GVariant        *parameters,
                        gpointer         user_data)
 {
-  SignalSubscriptionData *data = user_data;
+  GWeakRef *proxy_weak = user_data;
   gboolean emit_g_signal = FALSE;
   GDBusProxy *proxy;
   const gchar *interface_name_for_signal;
@@ -1052,18 +1011,9 @@ on_properties_changed (GDBusConnection *connection,
   changed_properties = NULL;
   invalidated_properties = NULL;
 
-  G_LOCK (signal_subscription_lock);
-  proxy = data->proxy;
+  proxy = G_DBUS_PROXY (g_weak_ref_get (proxy_weak));
   if (proxy == NULL)
-    {
-      G_UNLOCK (signal_subscription_lock);
-      goto out;
-    }
-  else
-    {
-      g_object_ref (proxy);
-      G_UNLOCK (signal_subscription_lock);
-    }
+    return;
 
   if (!proxy->priv->initialized)
     goto out;
@@ -1150,11 +1100,9 @@ on_properties_changed (GDBusConnection *connection,
     }
 
  out:
-  if (changed_properties != NULL)
-    g_variant_unref (changed_properties);
+  g_clear_pointer (&changed_properties, g_variant_unref);
   g_free (invalidated_properties);
-  if (proxy != NULL)
-    g_object_unref (proxy);
+  g_clear_object (&proxy);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -1258,8 +1206,7 @@ on_name_owner_changed_get_all_cb (GDBusConnection *connection,
     {
       G_LOCK (properties_lock);
       g_free (data->proxy->priv->name_owner);
-      data->proxy->priv->name_owner = data->name_owner;
-      data->name_owner = NULL; /* to avoid an extra copy, we steal the string */
+      data->proxy->priv->name_owner = g_steal_pointer (&data->name_owner);
       g_hash_table_remove_all (data->proxy->priv->properties);
       G_UNLOCK (properties_lock);
       if (result != NULL)
@@ -1289,23 +1236,14 @@ on_name_owner_changed (GDBusConnection *connection,
                        GVariant         *parameters,
                        gpointer          user_data)
 {
-  SignalSubscriptionData *data = user_data;
+  GWeakRef *proxy_weak = user_data;
   GDBusProxy *proxy;
   const gchar *old_owner;
   const gchar *new_owner;
 
-  G_LOCK (signal_subscription_lock);
-  proxy = data->proxy;
+  proxy = G_DBUS_PROXY (g_weak_ref_get (proxy_weak));
   if (proxy == NULL)
-    {
-      G_UNLOCK (signal_subscription_lock);
-      goto out;
-    }
-  else
-    {
-      g_object_ref (proxy);
-      G_UNLOCK (signal_subscription_lock);
-    }
+    return;
 
   /* if we are already trying to load properties, cancel that */
   if (proxy->priv->get_all_cancellable != NULL)
@@ -1415,8 +1353,7 @@ on_name_owner_changed (GDBusConnection *connection,
     }
 
  out:
-  if (proxy != NULL)
-    g_object_unref (proxy);
+  g_clear_object (&proxy);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -1762,8 +1699,8 @@ async_initable_init_first (GAsyncInitable *initable)
                                             proxy->priv->interface_name,
                                             G_DBUS_SIGNAL_FLAGS_NONE,
                                             on_properties_changed,
-                                            signal_subscription_ref (proxy->priv->signal_subscription_data),
-                                            (GDestroyNotify) signal_subscription_unref);
+                                            weak_ref_new (G_OBJECT (proxy)),
+                                            (GDestroyNotify) weak_ref_free);
     }
 
   if (!(proxy->priv->flags & G_DBUS_PROXY_FLAGS_DO_NOT_CONNECT_SIGNALS))
@@ -1778,8 +1715,8 @@ async_initable_init_first (GAsyncInitable *initable)
                                             NULL,                        /* arg0 */
                                             G_DBUS_SIGNAL_FLAGS_NONE,
                                             on_signal_received,
-                                            signal_subscription_ref (proxy->priv->signal_subscription_data),
-                                            (GDestroyNotify) signal_subscription_unref);
+                                            weak_ref_new (G_OBJECT (proxy)),
+                                            (GDestroyNotify) weak_ref_free);
     }
 
   if (proxy->priv->name != NULL &&
@@ -1794,8 +1731,8 @@ async_initable_init_first (GAsyncInitable *initable)
                                             proxy->priv->name,       /* arg0 */
                                             G_DBUS_SIGNAL_FLAGS_NONE,
                                             on_name_owner_changed,
-                                            signal_subscription_ref (proxy->priv->signal_subscription_data),
-                                            (GDestroyNotify) signal_subscription_unref);
+                                            weak_ref_new (G_OBJECT (proxy)),
+                                            (GDestroyNotify) weak_ref_free);
     }
 }
 
diff --git a/gio/tests/g-file-info-filesystem-readonly.c b/gio/tests/g-file-info-filesystem-readonly.c
index 60937b143..16fa83e33 100644
--- a/gio/tests/g-file-info-filesystem-readonly.c
+++ b/gio/tests/g-file-info-filesystem-readonly.c
@@ -25,7 +25,7 @@
 #include <gio/gio.h>
 #include <gio/gunixmounts.h>
 
-static void
+static gboolean
 run (GError **error,
      const gchar *argv0,
      ...)
@@ -34,6 +34,8 @@ run (GError **error,
   const gchar *arg;
   va_list ap;
   GSubprocess *subprocess;
+  gchar *command_line = NULL;
+  gboolean success;
 
   args = g_ptr_array_new ();
 
@@ -44,14 +46,20 @@ run (GError **error,
   g_ptr_array_add (args, NULL);
   va_end (ap);
 
+  command_line = g_strjoinv (" ", (gchar **) args->pdata);
+  g_test_message ("Running command `%s`", command_line);
+  g_free (command_line);
+
   subprocess = g_subprocess_newv ((const gchar * const *) args->pdata, G_SUBPROCESS_FLAGS_NONE, error);
   g_ptr_array_free (args, TRUE);
 
   if (subprocess == NULL)
-    return;
+    return FALSE;
 
-  g_subprocess_wait_check (subprocess, NULL, error);
+  success = g_subprocess_wait_check (subprocess, NULL, error);
   g_object_unref (subprocess);
+
+  return success;
 }
 
 static void
@@ -135,8 +143,14 @@ test_filesystem_readonly (gconstpointer with_mount_monitor)
 
   /* Use bindfs, which does not need root privileges, to mount the contents of one dir
    * into another dir (and do the mount as readonly as per passed '-o ro' option) */
-  run (&error, bindfs, "-n", "-o", "ro", dir_to_mount, dir_mountpoint, NULL);
-  g_assert_no_error (error);
+  if (!run (&error, bindfs, "-n", "-o", "ro", dir_to_mount, dir_mountpoint, NULL))
+    {
+      gchar *skip_message = g_strdup_printf ("Failed to run bindfs to set up test: %s", error->message);
+      g_test_skip (skip_message);
+      g_free (skip_message);
+      g_clear_error (&error);
+      return;
+    }
 
   /* Let's check now, that the file is in indeed in a readonly filesystem */
   file_in_mountpoint = g_strdup_printf ("%s/example.txt", dir_mountpoint);
diff --git a/gio/tests/gdbus-connection-loss.c b/gio/tests/gdbus-connection-loss.c
index 8f7023f3f..a34a9923c 100644
--- a/gio/tests/gdbus-connection-loss.c
+++ b/gio/tests/gdbus-connection-loss.c
@@ -124,14 +124,14 @@ main (int   argc,
   g_assert (g_spawn_command_line_async (path, NULL));
   g_free (path);
 
-  ensure_gdbus_testserver_up ();
-
   /* Create the connection in the main thread */
   error = NULL;
   c = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
   g_assert_no_error (error);
   g_assert (c != NULL);
 
+  ensure_gdbus_testserver_up (c, NULL);
+
   g_test_add_func ("/gdbus/connection-loss", test_connection_loss);
 
   ret = g_test_run();
diff --git a/gio/tests/gdbus-tests.c b/gio/tests/gdbus-tests.c
index 35e379bd2..bed0d24f7 100644
--- a/gio/tests/gdbus-tests.c
+++ b/gio/tests/gdbus-tests.c
@@ -86,49 +86,72 @@ _give_up (gpointer data)
   g_return_val_if_reached (TRUE);
 }
 
+typedef struct
+{
+  GMainContext *context;
+  gboolean name_appeared;
+  gboolean unwatch_complete;
+} WatchData;
+
+static void
+name_appeared_cb (GDBusConnection *connection,
+                  const gchar     *name,
+                  const gchar     *name_owner,
+                  gpointer         user_data)
+{
+  WatchData *data = user_data;
+
+  g_assert (name_owner != NULL);
+  data->name_appeared = TRUE;
+  g_main_context_wakeup (data->context);
+}
+
+static void
+watch_free_cb (gpointer user_data)
+{
+  WatchData *data = user_data;
+
+  data->unwatch_complete = TRUE;
+  g_main_context_wakeup (data->context);
+}
+
 void
-ensure_gdbus_testserver_up (void)
+ensure_gdbus_testserver_up (GDBusConnection *connection,
+                            GMainContext    *context)
 {
-  guint id;
-  gchar *name_owner;
-  GDBusConnection *connection;
-  GDBusProxy *proxy;
-  GError *error = NULL;
-
-  connection = g_bus_get_sync (G_BUS_TYPE_SESSION,
-                               NULL,
-                               &error);
-
-  g_assert_no_error (error);
-  error = NULL;
-
-  proxy = g_dbus_proxy_new_sync (connection,
-                                 G_DBUS_PROXY_FLAGS_NONE,
-                                 NULL,                      /* GDBusInterfaceInfo */
-                                 "com.example.TestService", /* name */
-                                 "/com/example/TestObject", /* object path */
-                                 "com.example.Frob",        /* interface */
-                                 NULL, /* GCancellable */
-                                 &error);
-  g_assert_no_error (error);
-
-  id = g_timeout_add_seconds (60, _give_up,
-      "waited more than ~ 60s for gdbus-testserver to take its bus name");
-
-  while (TRUE)
-    {
-      name_owner = g_dbus_proxy_get_name_owner (proxy);
-
-      if (name_owner != NULL)
-        break;
-
-      g_main_context_iteration (NULL, TRUE);
-    }
-
-  g_source_remove (id);
-  g_free (name_owner);
-  g_object_unref (proxy);
-  g_object_unref (connection);
+  GSource *timeout_source = NULL;
+  guint watch_id;
+  WatchData data = { context, FALSE, FALSE };
+
+  g_main_context_push_thread_default (context);
+
+  watch_id = g_bus_watch_name_on_connection (connection,
+                                             "com.example.TestService",
+                                             G_BUS_NAME_WATCHER_FLAGS_NONE,
+                                             name_appeared_cb,
+                                             NULL,
+                                             &data,
+                                             watch_free_cb);
+
+  timeout_source = g_timeout_source_new_seconds (60);
+  g_source_set_callback (timeout_source, _give_up,
+                         "waited more than ~ 60s for gdbus-testserver to take its bus name",
+                         NULL);
+  g_source_attach (timeout_source, context);
+
+  while (!data.name_appeared)
+    g_main_context_iteration (context, TRUE);
+
+  g_bus_unwatch_name (watch_id);
+  watch_id = 0;
+
+  while (!data.unwatch_complete)
+    g_main_context_iteration (context, TRUE);
+
+  g_source_destroy (timeout_source);
+  g_source_unref (timeout_source);
+
+  g_main_context_pop_thread_default (context);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
diff --git a/gio/tests/gdbus-tests.h b/gio/tests/gdbus-tests.h
index 00cda3713..eaef23480 100644
--- a/gio/tests/gdbus-tests.h
+++ b/gio/tests/gdbus-tests.h
@@ -114,7 +114,8 @@ GDBusConnection *_g_bus_get_priv (GBusType            bus_type,
                                   GCancellable       *cancellable,
                                   GError            **error);
 
-void ensure_gdbus_testserver_up (void);
+void ensure_gdbus_testserver_up (GDBusConnection *connection,
+                                 GMainContext    *context);
 
 G_END_DECLS
 
diff --git a/gio/tests/gdbus-threading.c b/gio/tests/gdbus-threading.c
index dd20530ff..0a2fe281f 100644
--- a/gio/tests/gdbus-threading.c
+++ b/gio/tests/gdbus-threading.c
@@ -27,59 +27,77 @@
 /* all tests rely on a global connection */
 static GDBusConnection *c = NULL;
 
-/* ---------------------------------------------------------------------------------------------------- */
-/* Ensure that signal and method replies are delivered in the right thread */
-/* ---------------------------------------------------------------------------------------------------- */
+typedef struct
+{
+  GMainContext *context;
+  gboolean timed_out;
+} TimeoutData;
 
-typedef struct {
-  GThread *thread;
-  GMainLoop *thread_loop;
-  guint signal_count;
-} DeliveryData;
+static gboolean
+timeout_cb (gpointer user_data)
+{
+  TimeoutData *data = user_data;
 
+  data->timed_out = TRUE;
+  g_main_context_wakeup (data->context);
+
+  return G_SOURCE_REMOVE;
+}
+
+/* Check that the given @connection has only one ref, waiting to let any pending
+ * unrefs complete first. This is typically used on the shared connection, to
+ * ensure it’s in a correct state before beginning the next test. */
 static void
-msg_cb_expect_success (GDBusConnection *connection,
-                       GAsyncResult    *res,
-                       gpointer         user_data)
+assert_connection_has_one_ref (GDBusConnection *connection,
+                               GMainContext    *context)
 {
-  DeliveryData *data = user_data;
-  GError *error;
-  GVariant *result;
+  GSource *timeout_source = NULL;
+  TimeoutData data = { context, FALSE };
 
-  error = NULL;
-  result = g_dbus_connection_call_finish (connection,
-                                          res,
-                                          &error);
-  g_assert_no_error (error);
-  g_assert (result != NULL);
-  g_variant_unref (result);
+  if (g_atomic_int_get (&G_OBJECT (connection)->ref_count) == 1)
+    return;
 
-  g_assert (g_thread_self () == data->thread);
+  timeout_source = g_timeout_source_new_seconds (1);
+  g_source_set_callback (timeout_source, timeout_cb, &data, NULL);
+  g_source_attach (timeout_source, context);
 
-  g_main_loop_quit (data->thread_loop);
+  while (g_atomic_int_get (&G_OBJECT (connection)->ref_count) != 1 && !data.timed_out)
+    {
+      g_debug ("refcount of %p is not right, sleeping", connection);
+      g_main_context_iteration (NULL, TRUE);
+    }
+
+  g_source_destroy (timeout_source);
+  g_source_unref (timeout_source);
+
+  if (g_atomic_int_get (&G_OBJECT (connection)->ref_count) != 1)
+    g_error ("connection %p had too many refs", connection);
 }
 
+/* ---------------------------------------------------------------------------------------------------- */
+/* Ensure that signal and method replies are delivered in the right thread */
+/* ---------------------------------------------------------------------------------------------------- */
+
+typedef struct {
+  GThread *thread;
+  GMainContext *context;
+  guint signal_count;
+  gboolean unsubscribe_complete;
+  GAsyncResult *async_result;
+} DeliveryData;
+
 static void
-msg_cb_expect_error_cancelled (GDBusConnection *connection,
-                               GAsyncResult    *res,
-                               gpointer         user_data)
+async_result_cb (GDBusConnection *connection,
+                 GAsyncResult    *res,
+                 gpointer         user_data)
 {
   DeliveryData *data = user_data;
-  GError *error;
-  GVariant *result;
 
-  error = NULL;
-  result = g_dbus_connection_call_finish (connection,
-                                          res,
-                                          &error);
-  g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
-  g_assert (!g_dbus_error_is_remote_error (error));
-  g_error_free (error);
-  g_assert (result == NULL);
+  data->async_result = g_object_ref (res);
 
-  g_assert (g_thread_self () == data->thread);
+  g_assert_true (g_thread_self () == data->thread);
 
-  g_main_loop_quit (data->thread_loop);
+  g_main_context_wakeup (data->context);
 }
 
 static void
@@ -93,33 +111,43 @@ signal_handler (GDBusConnection *connection,
 {
   DeliveryData *data = user_data;
 
-  g_assert (g_thread_self () == data->thread);
+  g_assert_true (g_thread_self () == data->thread);
 
   data->signal_count++;
 
-  g_main_loop_quit (data->thread_loop);
+  g_main_context_wakeup (data->context);
+}
+
+static void
+signal_data_free_cb (gpointer user_data)
+{
+  DeliveryData *data = user_data;
+
+  g_assert_true (g_thread_self () == data->thread);
+
+  data->unsubscribe_complete = TRUE;
+
+  g_main_context_wakeup (data->context);
 }
 
 static gpointer
 test_delivery_in_thread_func (gpointer _data)
 {
-  GMainLoop *thread_loop;
   GMainContext *thread_context;
   DeliveryData data;
   GCancellable *ca;
   guint subscription_id;
-  GDBusConnection *priv_c;
-  GError *error;
-
-  error = NULL;
+  GError *error = NULL;
+  GVariant *result_variant = NULL;
 
   thread_context = g_main_context_new ();
-  thread_loop = g_main_loop_new (thread_context, FALSE);
   g_main_context_push_thread_default (thread_context);
 
   data.thread = g_thread_self ();
-  data.thread_loop = thread_loop;
+  data.context = thread_context;
   data.signal_count = 0;
+  data.unsubscribe_complete = FALSE;
+  data.async_result = NULL;
 
   /* ---------------------------------------------------------------------------------------------------- */
 
@@ -135,9 +163,16 @@ test_delivery_in_thread_func (gpointer _data)
                           G_DBUS_CALL_FLAGS_NONE,
                           -1,
                           NULL,
-                          (GAsyncReadyCallback) msg_cb_expect_success,
+                          (GAsyncReadyCallback) async_result_cb,
                           &data);
-  g_main_loop_run (thread_loop);
+  while (data.async_result == NULL)
+    g_main_context_iteration (thread_context, TRUE);
+
+  result_variant = g_dbus_connection_call_finish (c, data.async_result, &error);
+  g_assert_no_error (error);
+  g_assert_nonnull (result_variant);
+  g_clear_pointer (&result_variant, g_variant_unref);
+  g_clear_object (&data.async_result);
 
   /*
    * Check that we never actually send a message if the GCancellable
@@ -155,9 +190,18 @@ test_delivery_in_thread_func (gpointer _data)
                           G_DBUS_CALL_FLAGS_NONE,
                           -1,
                           ca,
-                          (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
+                          (GAsyncReadyCallback) async_result_cb,
                           &data);
-  g_main_loop_run (thread_loop);
+  while (data.async_result == NULL)
+    g_main_context_iteration (thread_context, TRUE);
+
+  result_variant = g_dbus_connection_call_finish (c, data.async_result, &error);
+  g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
+  g_assert_false (g_dbus_error_is_remote_error (error));
+  g_clear_error (&error);
+  g_assert_null (result_variant);
+  g_clear_object (&data.async_result);
+
   g_object_unref (ca);
 
   /*
@@ -173,47 +217,74 @@ test_delivery_in_thread_func (gpointer _data)
                           G_DBUS_CALL_FLAGS_NONE,
                           -1,
                           ca,
-                          (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
+                          (GAsyncReadyCallback) async_result_cb,
                           &data);
   g_cancellable_cancel (ca);
-  g_main_loop_run (thread_loop);
+
+  while (data.async_result == NULL)
+    g_main_context_iteration (thread_context, TRUE);
+
+  result_variant = g_dbus_connection_call_finish (c, data.async_result, &error);
+  g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
+  g_assert_false (g_dbus_error_is_remote_error (error));
+  g_clear_error (&error);
+  g_assert_null (result_variant);
+  g_clear_object (&data.async_result);
+
   g_object_unref (ca);
 
   /*
    * Check that signals are delivered to the correct thread.
    *
-   * First we subscribe to the signal, then we create a a private
-   * connection. This should cause a NameOwnerChanged message from
-   * the message bus.
+   * First we subscribe to the signal, then we call EmitSignal(). This should
+   * cause a TestSignal emission from the testserver.
    */
   subscription_id = g_dbus_connection_signal_subscribe (c,
-                                                        "org.freedesktop.DBus",  /* sender */
-                                                        "org.freedesktop.DBus",  /* interface */
-                                                        "NameOwnerChanged",      /* member */
-                                                        "/org/freedesktop/DBus", /* path */
+                                                        "com.example.TestService", /* sender */
+                                                        "com.example.Frob",        /* interface */
+                                                        "TestSignal",              /* member */
+                                                        "/com/example/TestObject", /* path */
                                                         NULL,
                                                         G_DBUS_SIGNAL_FLAGS_NONE,
                                                         signal_handler,
                                                         &data,
-                                                        NULL);
-  g_assert (subscription_id != 0);
-  g_assert (data.signal_count == 0);
+                                                        signal_data_free_cb);
+  g_assert_cmpuint (subscription_id, !=, 0);
+  g_assert_cmpuint (data.signal_count, ==, 0);
 
-  priv_c = _g_bus_get_priv (G_BUS_TYPE_SESSION, NULL, &error);
-  g_assert_no_error (error);
-  g_assert (priv_c != NULL);
+  g_dbus_connection_call (c,
+                          "com.example.TestService", /* bus_name */
+                          "/com/example/TestObject", /* object path */
+                          "com.example.Frob",        /* interface name */
+                          "EmitSignal",              /* method name */
+                          g_variant_new_parsed ("('hello', @o '/com/example/TestObject')"),
+                          NULL,
+                          G_DBUS_CALL_FLAGS_NONE,
+                          -1,
+                          NULL,
+                          (GAsyncReadyCallback) async_result_cb,
+                          &data);
+  while (data.async_result == NULL || data.signal_count < 1)
+    g_main_context_iteration (thread_context, TRUE);
 
-  g_main_loop_run (thread_loop);
-  g_assert (data.signal_count == 1);
+  result_variant = g_dbus_connection_call_finish (c, data.async_result, &error);
+  g_assert_no_error (error);
+  g_assert_nonnull (result_variant);
+  g_clear_pointer (&result_variant, g_variant_unref);
+  g_clear_object (&data.async_result);
 
-  g_object_unref (priv_c);
+  g_assert_cmpuint (data.signal_count, ==, 1);
 
   g_dbus_connection_signal_unsubscribe (c, subscription_id);
+  subscription_id = 0;
+
+  while (!data.unsubscribe_complete)
+    g_main_context_iteration (thread_context, TRUE);
+  g_assert_true (data.unsubscribe_complete);
 
   /* ---------------------------------------------------------------------------------------------------- */
 
   g_main_context_pop_thread_default (thread_context);
-  g_main_loop_unref (thread_loop);
   g_main_context_unref (thread_context);
 
   return NULL;
@@ -229,6 +300,8 @@ test_delivery_in_thread (void)
                          NULL);
 
   g_thread_join (thread);
+
+  assert_connection_has_one_ref (c, NULL);
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
@@ -257,11 +330,11 @@ sleep_cb (GDBusProxy   *proxy,
                                      res,
                                      &error);
   g_assert_no_error (error);
-  g_assert (result != NULL);
+  g_assert_nonnull (result);
   g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
   g_variant_unref (result);
 
-  g_assert (data->thread == g_thread_self ());
+  g_assert_true (data->thread == g_thread_self ());
 
   g_main_loop_quit (data->thread_loop);
 
@@ -317,7 +390,7 @@ test_sleep_in_thread_func (gpointer _data)
             g_printerr ("S");
           //g_debug ("done invoking sync (%p)", g_thread_self ());
           g_assert_no_error (error);
-          g_assert (result != NULL);
+          g_assert_nonnull (result);
           g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
           g_variant_unref (result);
         }
@@ -441,6 +514,8 @@ test_method_calls_in_thread (void)
 
   if (g_test_verbose ())
     g_printerr ("\n");
+
+  assert_connection_has_one_ref (c, NULL);
 }
 
 #define SLEEP_MIN_USEC 1
@@ -457,8 +532,8 @@ ensure_connection_works (GDBusConnection *conn)
       "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetId", NULL, NULL, 0, -1,
       NULL, &error);
   g_assert_no_error (error);
-  g_assert (v != NULL);
-  g_assert (g_variant_is_of_type (v, G_VARIANT_TYPE ("(s)")));
+  g_assert_nonnull (v);
+  g_assert_true (g_variant_is_of_type (v, G_VARIANT_TYPE ("(s)")));
   g_variant_unref (v);
 }
 
@@ -505,24 +580,11 @@ test_threaded_singleton (void)
   for (i = 0; i < n; i++)
     {
       GThread *thread;
-      guint j;
       guint unref_delay, get_delay;
       GDBusConnection *new_conn;
 
       /* We want to be the last ref, so let it finish setting up */
-      for (j = 0; j < 100; j++)
-        {
-          guint r = (guint) g_atomic_int_get (&G_OBJECT (c)->ref_count);
-
-          if (r == 1)
-            break;
-
-          g_debug ("run %u: refcount is %u, sleeping", i, r);
-          g_usleep (1000);
-        }
-
-      if (j == 100)
-        g_error ("connection had too many refs");
+      assert_connection_has_one_ref (c, NULL);
 
       if (g_test_verbose () && (i % (n/50)) == 0)
         g_printerr ("%u%%\n", ((i * 100) / n));
@@ -586,16 +648,16 @@ main (int   argc,
 
   /* this is safe; testserver will exit once the bus goes away */
   path = g_test_build_filename (G_TEST_BUILT, "gdbus-testserver", NULL);
-  g_assert (g_spawn_command_line_async (path, NULL));
+  g_assert_true (g_spawn_command_line_async (path, NULL));
   g_free (path);
 
-  ensure_gdbus_testserver_up ();
-
   /* Create the connection in the main thread */
   error = NULL;
   c = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
   g_assert_no_error (error);
-  g_assert (c != NULL);
+  g_assert_nonnull (c);
+
+  ensure_gdbus_testserver_up (c, NULL);
 
   g_test_add_func ("/gdbus/delivery-in-thread", test_delivery_in_thread);
   g_test_add_func ("/gdbus/method-calls-in-thread", test_method_calls_in_thread);
diff --git a/gio/tests/meson.build b/gio/tests/meson.build
index 0e4ce9f93..695ad1f8f 100644
--- a/gio/tests/meson.build
+++ b/gio/tests/meson.build
@@ -302,7 +302,7 @@ if host_machine.system() != 'windows'
       },
       'gdbus-threading' : {
         'extra_sources' : extra_sources,
-        'suite' : ['slow', 'flaky'],
+        'suite' : ['slow'],
       },
       'gmenumodel' : {
         'extra_sources' : extra_sources,