Revert "GCancellable: Use per-instance mutex logic instead of global critical sections"

This reverts commit 3a07b2abd4006da33cdfb0916b266a21070a47d6.

Merge request !2765 has caused a thread safety regression in
`GCancellableSource` disposal. It landed too late in the cycle to fix
with any confidence before the 2.82 release, so the changes are being
reverted for the `glib-2-82` branch and 2.82.0 release.

Fixes: #3448
Re-opens: #774, #2309, #2313

gio/gcancellable.c

@@ -1,7 +1,6 @@
 /* GIO - GLib Input, Output and Streaming Library
  * 
  * Copyright (C) 2006-2007 Red Hat, Inc.
- * Copyright (C) 2022-2024 Canonical, Ltd.
  *
  * SPDX-License-Identifier: LGPL-2.1-or-later
  *
@@ -19,7 +18,6 @@
  * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
  *
  * Author: Alexander Larsson <alexl@redhat.com>
- * Author: Marco Trevisan <marco.trevisan@canonical.com>
  */
 
 #include "config.h"
@@ -47,12 +45,14 @@ enum {
 
 struct _GCancellablePrivate
 {
-  /* Atomic so that we don't require holding global mutexes for independent ops. */
+  /* Atomic so that g_cancellable_is_cancelled does not require holding the mutex. */
   gboolean cancelled;
-  int cancelled_running;
+  /* Access to fields below is protected by cancellable_mutex. */
+  guint cancelled_running : 1;
+  guint cancelled_running_waiting : 1;
+  unsigned cancelled_emissions;
+  unsigned cancelled_emissions_waiting : 1;
 
-  /* Access to fields below is protected by cancellable's mutex. */
-  GMutex mutex;
   guint fd_refcount;
   GWakeup *wakeup;
 };
@@ -62,6 +62,7 @@ static guint signals[LAST_SIGNAL] = { 0 };
 G_DEFINE_TYPE_WITH_PRIVATE (GCancellable, g_cancellable, G_TYPE_OBJECT)
 
 static GPrivate current_cancellable;
+static GMutex cancellable_mutex;
 static GCond cancellable_cond;
 
 static void
@@ -69,15 +70,9 @@ g_cancellable_finalize (GObject *object)
 {
   GCancellable *cancellable = G_CANCELLABLE (object);
 
-  /* We're at finalization phase, so only one thread can be here.
-   * Thus there's no need to lock. In case something is locking us, then we've
-   * a bug, and g_mutex_clear() will make this clear aborting.
-   */
   if (cancellable->priv->wakeup)
     GLIB_PRIVATE_CALL (g_wakeup_free) (cancellable->priv->wakeup);
 
-  g_mutex_clear (&cancellable->priv->mutex);
-
   G_OBJECT_CLASS (g_cancellable_parent_class)->finalize (object);
 }
 
@@ -159,8 +154,6 @@ static void
 g_cancellable_init (GCancellable *cancellable)
 {
   cancellable->priv = g_cancellable_get_instance_private (cancellable);
-
-  g_mutex_init (&cancellable->priv->mutex);
 }
 
 /**
@@ -272,17 +265,28 @@ g_cancellable_reset (GCancellable *cancellable)
 
   g_return_if_fail (G_IS_CANCELLABLE (cancellable));
 
+  g_mutex_lock (&cancellable_mutex);
+
   priv = cancellable->priv;
 
-  g_mutex_lock (&priv->mutex);
+  while (priv->cancelled_running || priv->cancelled_emissions > 0)
+    {
+      if (priv->cancelled_running)
+        priv->cancelled_running_waiting = TRUE;
 
-  if (g_atomic_int_compare_and_exchange (&priv->cancelled, TRUE, FALSE))
+      if (priv->cancelled_emissions > 0)
+        priv->cancelled_emissions_waiting = TRUE;
+
+      g_cond_wait (&cancellable_cond, &cancellable_mutex);
+    }
+
+  if (g_atomic_int_exchange (&priv->cancelled, FALSE))
     {
       if (priv->wakeup)
         GLIB_PRIVATE_CALL (g_wakeup_acknowledge) (priv->wakeup);
     }
 
-  g_mutex_unlock (&priv->mutex);
+  g_mutex_unlock (&cancellable_mutex);
 }
 
 /**
@@ -400,29 +404,26 @@ g_cancellable_get_fd (GCancellable *cancellable)
 gboolean
 g_cancellable_make_pollfd (GCancellable *cancellable, GPollFD *pollfd)
 {
-  GCancellablePrivate *priv;
-
   g_return_val_if_fail (pollfd != NULL, FALSE);
   if (cancellable == NULL)
     return FALSE;
   g_return_val_if_fail (G_IS_CANCELLABLE (cancellable), FALSE);
 
-  priv = cancellable->priv;
+  g_mutex_lock (&cancellable_mutex);
 
-  g_mutex_lock (&priv->mutex);
+  cancellable->priv->fd_refcount++;
 
-  if ((priv->fd_refcount++) == 0)
+  if (cancellable->priv->wakeup == NULL)
     {
-      priv->wakeup = GLIB_PRIVATE_CALL (g_wakeup_new) ();
+      cancellable->priv->wakeup = GLIB_PRIVATE_CALL (g_wakeup_new) ();
 
-      if (g_atomic_int_get (&priv->cancelled))
+      if (g_atomic_int_get (&cancellable->priv->cancelled))
-        GLIB_PRIVATE_CALL (g_wakeup_signal) (priv->wakeup);
+        GLIB_PRIVATE_CALL (g_wakeup_signal) (cancellable->priv->wakeup);
     }
 
-  g_assert (priv->wakeup);
+  GLIB_PRIVATE_CALL (g_wakeup_get_pollfd) (cancellable->priv->wakeup, pollfd);
-  GLIB_PRIVATE_CALL (g_wakeup_get_pollfd) (priv->wakeup, pollfd);
 
-  g_mutex_unlock (&priv->mutex);
+  g_mutex_unlock (&cancellable_mutex);
 
   return TRUE;
 }
@@ -446,22 +447,26 @@ g_cancellable_make_pollfd (GCancellable *cancellable, GPollFD *pollfd)
 void
 g_cancellable_release_fd (GCancellable *cancellable)
 {
+  GCancellablePrivate *priv;
+
   if (cancellable == NULL)
     return;
 
   g_return_if_fail (G_IS_CANCELLABLE (cancellable));
 
-  g_mutex_lock (&cancellable->priv->mutex);
+  priv = cancellable->priv;
 
-  g_assert (cancellable->priv->fd_refcount > 0);
+  g_mutex_lock (&cancellable_mutex);
+  g_assert (priv->fd_refcount > 0);
 
-  if ((cancellable->priv->fd_refcount--) == 1)
+  priv->fd_refcount--;
+  if (priv->fd_refcount == 0)
     {
-      GLIB_PRIVATE_CALL (g_wakeup_free) (cancellable->priv->wakeup);
+      GLIB_PRIVATE_CALL (g_wakeup_free) (priv->wakeup);
-      cancellable->priv->wakeup = NULL;
+      priv->wakeup = NULL;
     }
 
-  g_mutex_unlock (&cancellable->priv->mutex);
+  g_mutex_unlock (&cancellable_mutex);
 }
 
 /**
@@ -490,31 +495,37 @@ g_cancellable_cancel (GCancellable *cancellable)
 {
   GCancellablePrivate *priv;
 
-  if (cancellable == NULL || g_atomic_int_get (&cancellable->priv->cancelled))
+  if (cancellable == NULL || g_cancellable_is_cancelled (cancellable))
     return;
 
   priv = cancellable->priv;
 
-  g_mutex_lock (&priv->mutex);
+  g_mutex_lock (&cancellable_mutex);
 
-  if (!g_atomic_int_compare_and_exchange (&priv->cancelled, FALSE, TRUE))
+  if (g_atomic_int_exchange (&priv->cancelled, TRUE))
     {
-      g_mutex_unlock (&priv->mutex);
+      g_mutex_unlock (&cancellable_mutex);
       return;
     }
 
-  g_atomic_int_inc (&priv->cancelled_running);
+  priv->cancelled_running = TRUE;
 
   if (priv->wakeup)
     GLIB_PRIVATE_CALL (g_wakeup_signal) (priv->wakeup);
 
+  g_mutex_unlock (&cancellable_mutex);
+
   g_object_ref (cancellable);
   g_signal_emit (cancellable, signals[CANCELLED], 0);
 
-  if (g_atomic_int_dec_and_test (&priv->cancelled_running))
+  g_mutex_lock (&cancellable_mutex);
-    g_cond_broadcast (&cancellable_cond);
 
-  g_mutex_unlock (&priv->mutex);
+  priv->cancelled_running = FALSE;
+  if (priv->cancelled_running_waiting)
+    g_cond_broadcast (&cancellable_cond);
+  priv->cancelled_running_waiting = FALSE;
+
+  g_mutex_unlock (&cancellable_mutex);
 
   g_object_unref (cancellable);
 }
@@ -562,7 +573,7 @@ g_cancellable_connect (GCancellable   *cancellable,
 
   g_return_val_if_fail (G_IS_CANCELLABLE (cancellable), 0);
 
-  g_mutex_lock (&cancellable->priv->mutex);
+  g_mutex_lock (&cancellable_mutex);
 
   if (g_atomic_int_get (&cancellable->priv->cancelled))
     {
@@ -572,10 +583,23 @@ g_cancellable_connect (GCancellable   *cancellable,
       _callback = (void *)callback;
       id = 0;
 
+      cancellable->priv->cancelled_emissions++;
+
+      g_mutex_unlock (&cancellable_mutex);
+
       _callback (cancellable, data);
 
       if (data_destroy_func)
         data_destroy_func (data);
+
+      g_mutex_lock (&cancellable_mutex);
+
+      if (cancellable->priv->cancelled_emissions_waiting)
+        g_cond_broadcast (&cancellable_cond);
+
+      cancellable->priv->cancelled_emissions--;
+
+      g_mutex_unlock (&cancellable_mutex);
     }
   else
     {
@@ -583,9 +607,10 @@ g_cancellable_connect (GCancellable   *cancellable,
                                   callback, data,
                                   (GClosureNotify) data_destroy_func,
                                   G_CONNECT_DEFAULT);
+
+      g_mutex_unlock (&cancellable_mutex);
     }
 
-  g_mutex_unlock (&cancellable->priv->mutex);
 
   return id;
 }
@@ -621,26 +646,33 @@ g_cancellable_disconnect (GCancellable  *cancellable,
   if (handler_id == 0 ||  cancellable == NULL)
     return;
 
+  g_mutex_lock (&cancellable_mutex);
+
   priv = cancellable->priv;
 
-  g_mutex_lock (&priv->mutex);
+  while (priv->cancelled_running || priv->cancelled_emissions)
+    {
+      if (priv->cancelled_running)
+        priv->cancelled_running_waiting = TRUE;
 
-  while (g_atomic_int_get (&priv->cancelled_running) != 0)
+      if (priv->cancelled_emissions)
-    g_cond_wait (&cancellable_cond, &priv->mutex);
+        priv->cancelled_emissions_waiting = TRUE;
 
-  g_mutex_unlock (&priv->mutex);
+      g_cond_wait (&cancellable_cond, &cancellable_mutex);
+    }
 
   g_signal_handler_disconnect (cancellable, handler_id);
+
+  g_mutex_unlock (&cancellable_mutex);
 }
 
 typedef struct {
   GSource       source;
 
-  /* Atomic: */
   GCancellable *cancellable;
   gulong        cancelled_handler;
-  /* Atomic: */
+  /* Protected by cancellable_mutex: */
-  gboolean      cancelled_callback_called;
+  gboolean      resurrected_during_cancellation;
 } GCancellableSource;
 
 /*
@@ -660,35 +692,27 @@ cancellable_source_cancelled (GCancellable *cancellable,
 {
   GSource *source = user_data;
   GCancellableSource *cancellable_source = (GCancellableSource *) source;
-  gboolean callback_was_not_called;
+
+  g_mutex_lock (&cancellable_mutex);
+
+  /* Drop the reference added in cancellable_source_dispose(); see the comment there.
+   * The reference must be dropped after unlocking @cancellable_mutex since
+   * it could be the final reference, and the dispose function takes
+   * @cancellable_mutex. */
+  if (cancellable_source->resurrected_during_cancellation)
+    {
+      cancellable_source->resurrected_during_cancellation = FALSE;
+      g_mutex_unlock (&cancellable_mutex);
+      g_source_unref (source);
+      return;
+    }
 
   g_source_ref (source);
+  g_mutex_unlock (&cancellable_mutex);
   g_source_set_ready_time (source, 0);
-
-  callback_was_not_called = g_atomic_int_compare_and_exchange (
-    &cancellable_source->cancelled_callback_called, FALSE, TRUE);
-  g_assert (callback_was_not_called);
-
   g_source_unref (source);
 }
 
-static gboolean
-cancellable_source_prepare (GSource *source,
-                            gint    *timeout)
-{
-  GCancellableSource *cancellable_source = (GCancellableSource *) source;
-  GCancellable *cancellable;
-
-  if (timeout)
-    *timeout = -1;
-
-  cancellable = g_atomic_pointer_get (&cancellable_source->cancellable);
-  if (cancellable && !g_atomic_int_get (&cancellable->priv->cancelled_running))
-    g_atomic_int_set (&cancellable_source->cancelled_callback_called, FALSE);
-
-  return FALSE;
-}
-
 static gboolean
 cancellable_source_dispatch (GSource     *source,
 			     GSourceFunc  callback,
@@ -705,13 +729,12 @@ static void
 cancellable_source_dispose (GSource *source)
 {
   GCancellableSource *cancellable_source = (GCancellableSource *)source;
-  GCancellable *cancellable;
 
-  cancellable = g_atomic_pointer_exchange (&cancellable_source->cancellable, NULL);
+  g_mutex_lock (&cancellable_mutex);
 
-  if (cancellable)
+  if (cancellable_source->cancellable)
     {
-      if (g_atomic_int_get (&cancellable->priv->cancelled_running))
+      if (cancellable_source->cancellable->priv->cancelled_running)
         {
           /* There can be a race here: if thread A has called
            * g_cancellable_cancel() and has got as far as committing to call
@@ -723,21 +746,21 @@ cancellable_source_dispose (GSource *source)
            * will then be left in a state where it’s committed to using a
            * dangling GCancellableSource pointer.
            *
-           * Eliminate that race by waiting to ensure that our cancelled
+           * Eliminate that race by resurrecting the #GSource temporarily, and
-           * callback has been called, keeping a temporary ref, so that
+           * then dropping that reference in cancellable_source_cancelled(),
-           * there's no risk that we're unreffing something that is still
+           * which should be guaranteed to fire because we’re inside a
-           * going to be used.
+           * @cancelled_running block.
            */
-
           g_source_ref (source);
-          while (!g_atomic_int_get (&cancellable_source->cancelled_callback_called))
+          cancellable_source->resurrected_during_cancellation = TRUE;
-            ;
-          g_source_unref (source);
         }
 
-      g_clear_signal_handler (&cancellable_source->cancelled_handler, cancellable);
+      g_clear_signal_handler (&cancellable_source->cancelled_handler,
-      g_object_unref (cancellable);
+                              cancellable_source->cancellable);
+      g_clear_object (&cancellable_source->cancellable);
     }
+
+  g_mutex_unlock (&cancellable_mutex);
 }
 
 static gboolean
@@ -766,7 +789,7 @@ cancellable_source_closure_callback (GCancellable *cancellable,
 
 static GSourceFuncs cancellable_source_funcs =
 {
-  cancellable_source_prepare,
+  NULL,
   NULL,
   cancellable_source_dispatch,
   NULL,

gio/tests/cancellable.c

@@ -261,6 +261,11 @@ threaded_dispose_thread_cb (gpointer user_data)
 static void
 test_cancellable_source_threaded_dispose (void)
 {
+#ifdef _GLIB_ADDRESS_SANITIZER
+  g_test_incomplete ("FIXME: Leaks lots of GCancellableSource objects, see glib#2309");
+  (void) cancelled_cb;
+  (void) threaded_dispose_thread_cb;
+#else
   ThreadedDisposeData data;
   GThread *thread = NULL;
   guint i;
@@ -270,10 +275,6 @@ test_cancellable_source_threaded_dispose (void)
                   "(in one thread) and cancelling the GCancellable it refers "
                   "to (in another thread)");
   g_test_bug ("https://gitlab.gnome.org/GNOME/glib/issues/1841");
-#ifdef _GLIB_ADDRESS_SANITIZER
-  g_test_message ("We also ensure that no GCancellableSource are leaked");
-  g_test_bug ("https://gitlab.gnome.org/GNOME/glib/issues/2309");
-#endif
 
   /* Create a new thread and wait until it’s ready to execute. Each iteration of
    * the test will pass it a new #GCancellableSource. */
@@ -334,6 +335,7 @@ test_cancellable_source_threaded_dispose (void)
   g_cond_clear (&data.cond);
 
   g_ptr_array_unref (cancellables_pending_unref);
+#endif
 }
 
 static void

gio/tests/gmenumodel.c

@@ -1007,16 +1007,17 @@ test_dbus_roundtrip (void)
 static void
 test_dbus_peer_roundtrip (void)
 {
-  PeerConnection peer;
-
 #ifdef _GLIB_ADDRESS_SANITIZER
-  g_test_message ("Ensure that no GCancellableSource are leaked");
+  g_test_incomplete ("FIXME: Leaks a GCancellableSource, see glib#2313");
-  g_test_bug ("https://gitlab.gnome.org/GNOME/glib/issues/2313");
+  (void) peer_connection_up;
-#endif
+  (void) peer_connection_down;
+#else
+  PeerConnection peer;
 
   peer_connection_up (&peer);
   do_roundtrip (peer.server_connection, peer.client_connection);
   peer_connection_down (&peer);
+#endif
 }
 
 static gint items_changed_count;
@@ -1145,16 +1146,17 @@ test_dbus_subscriptions (void)
 static void
 test_dbus_peer_subscriptions (void)
 {
-  PeerConnection peer;
-
 #ifdef _GLIB_ADDRESS_SANITIZER
-  g_test_message ("Ensure that no GCancellableSource are leaked");
+  g_test_incomplete ("FIXME: Leaks a GCancellableSource, see glib#2313");
-  g_test_bug ("https://gitlab.gnome.org/GNOME/glib/issues/2313");
+  (void) peer_connection_up;
-#endif
+  (void) peer_connection_down;
+#else
+  PeerConnection peer;
 
   peer_connection_up (&peer);
   do_subscriptions (peer.server_connection, peer.client_connection);
   peer_connection_down (&peer);
+#endif
 }
 
 static void