Merge branch 'bug/multi-sub' into 'master'

Engine: account for multiple simultaneous subscriptions to the same path

Closes #2

See merge request GNOME/dconf!5

4 files changed, 400 insertions, 125 deletions

diff --git a/engine/dconf-engine.c b/engine/dconf-engine.c
index c0ff12d..ad891e6 100644
--- a/engine/dconf-engine.c
+++ b/engine/dconf-engine.c
@@ -128,7 +128,7 @@
  * it is willing to deal with receiving the change notifies in those
  * threads.
  *
- * Thread-safety is implemented using two locks.
+ * Thread-safety is implemented using three locks.
  *
  * The first lock (sources_lock) protects the sources.  Although the
  * sources are only ever read from, it is necessary to lock them because
@@ -143,8 +143,15 @@
  * The second lock (queue_lock) protects the various queues that are
  * used to implement the "fast" writes described above.
  *
- * If both locks are held at the same time then the sources lock must
- * have been acquired first.
+ * The third lock (subscription_count_lock) protects the two hash tables
+ * that are used to keep track of the number of subscriptions held by
+ * the client library to each path.
+ *
+ * If sources_lock and queue_lock are held at the same time then then
+ * sources_lock must have been acquired first.
+ *
+ * subscription_count_lock is never held at the same time as
+ * sources_lock or queue_lock
  */
 
 #define MAX_IN_FLIGHT 2
@@ -170,8 +177,16 @@ struct _DConfEngine
 
   gchar              *last_handled;  /* reply tag from last item in in_flight */
 
-  GHashTable         *watched_paths; /* list of paths currently being watched for changes */
-  GHashTable         *pending_paths; /* list of paths waiting to enter watched state */
+  /**
+   * establishing and active, are hash tables storing the number
+   * of subscriptions to each path in the two possible states
+   */
+  /* This lock ensures that transactions involving subscription counts are atomic */
+  GMutex              subscription_count_lock;
+  /* active on the client side, but awaiting confirmation from the writer */
+  GHashTable         *establishing;
+  /* active on the client side, and with a D-Bus match rule established */
+  GHashTable         *active;
 };
 
 /* When taking the sources lock we check if any of the databases have
@@ -225,6 +240,97 @@ dconf_engine_unlock_queues (DConfEngine *engine)
   g_mutex_unlock (&engine->queue_lock);
 }
 
+/**
+ * Adds the count of subscriptions to @path in @from_table to the
+ * corresponding count in @to_table, creating it if it did not exist.
+ * Removes the count from @from_table.
+ */
+static void
+dconf_engine_move_subscriptions (GHashTable  *from_counts,
+                                 GHashTable  *to_counts,
+                                 const gchar *path)
+{
+  guint from_count = GPOINTER_TO_UINT (g_hash_table_lookup (from_counts, path));
+  guint old_to_count = GPOINTER_TO_UINT (g_hash_table_lookup (to_counts, path));
+  // Detect overflows
+  g_assert (old_to_count <= G_MAXUINT32 - from_count);
+  guint new_to_count = old_to_count + from_count;
+  if (from_count != 0)
+    {
+      g_hash_table_remove (from_counts, path);
+      g_hash_table_replace (to_counts,
+                            g_strdup (path),
+                            GUINT_TO_POINTER (new_to_count));
+    }
+}
+
+/**
+ * Increments the reference count for the subscription to @path, or sets
+ * it to 1 if it didn’t previously exist.
+ * Returns the new reference count.
+ */
+static guint
+dconf_engine_inc_subscriptions (GHashTable  *counts,
+                                const gchar *path)
+{
+  guint old_count = GPOINTER_TO_UINT (g_hash_table_lookup (counts, path));
+  // Detect overflows
+  g_assert (old_count < G_MAXUINT32);
+  guint new_count = old_count + 1;
+  g_hash_table_replace (counts, g_strdup (path), GUINT_TO_POINTER (new_count));
+  return new_count;
+}
+
+/**
+ * Decrements the reference count for the subscription to @path, or
+ * removes it if the new value is 0. The count must exist and be greater
+ * than 0.
+ * Returns the new reference count, or 0 if it does not exist.
+ */
+static guint
+dconf_engine_dec_subscriptions (GHashTable  *counts,
+                                const gchar *path)
+{
+  guint old_count = GPOINTER_TO_UINT (g_hash_table_lookup (counts, path));
+  g_assert (old_count > 0);
+  guint new_count = old_count - 1;
+  if (new_count == 0)
+    g_hash_table_remove (counts, path);
+  else
+    g_hash_table_replace (counts, g_strdup (path), GUINT_TO_POINTER (new_count));
+  return new_count;
+}
+
+/**
+ * Returns the reference count for the subscription to @path, or 0 if it
+ * does not exist.
+ */
+static guint
+dconf_engine_count_subscriptions (GHashTable  *counts,
+                                  const gchar *path)
+{
+  return GPOINTER_TO_UINT (g_hash_table_lookup (counts, path));
+}
+
+/**
+ * Acquires the subscription counts lock, which must be held when
+ * reading or writing to the subscription counts.
+ */
+static void
+dconf_engine_lock_subscription_counts (DConfEngine *engine)
+{
+  g_mutex_lock (&engine->subscription_count_lock);
+}
+
+/**
+ * Releases the subscription counts lock
+ */
+static void
+dconf_engine_unlock_subscription_counts (DConfEngine *engine)
+{
+  g_mutex_unlock (&engine->subscription_count_lock);
+}
+
 DConfEngine *
 dconf_engine_new (const gchar    *profile,
                   gpointer        user_data,
@@ -247,8 +353,15 @@ dconf_engine_new (const gchar    *profile,
   dconf_engine_global_list = g_slist_prepend (dconf_engine_global_list, engine);
   g_mutex_unlock (&dconf_engine_global_lock);
 
-  engine->watched_paths = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, NULL);
-  engine->pending_paths = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, NULL);
+  g_mutex_init (&engine->subscription_count_lock);
+  engine->establishing = g_hash_table_new_full (g_str_hash,
+                                                g_str_equal,
+                                                g_free,
+                                                NULL);
+  engine->active = g_hash_table_new_full (g_str_hash,
+                                          g_str_equal,
+                                          g_free,
+                                          NULL);
 
   return engine;
 }
@@ -289,11 +402,22 @@ dconf_engine_unref (DConfEngine *engine)
 
       g_free (engine->last_handled);
 
+      while (!g_queue_is_empty (&engine->pending))
+        dconf_changeset_unref ((DConfChangeset *) g_queue_pop_head (&engine->pending));
+
+      while (!g_queue_is_empty (&engine->in_flight))
+        dconf_changeset_unref ((DConfChangeset *) g_queue_pop_head (&engine->in_flight));
+
       for (i = 0; i < engine->n_sources; i++)
         dconf_engine_source_free (engine->sources[i]);
 
       g_free (engine->sources);
 
+      g_hash_table_unref (engine->establishing);
+      g_hash_table_unref (engine->active);
+
+      g_mutex_clear (&engine->subscription_count_lock);
+
       if (engine->free_func)
         engine->free_func (engine->user_data);
 
@@ -835,10 +959,21 @@ dconf_engine_watch_established (DConfEngine  *engine,
        * everything under the path being watched changed.  This case is
        * very rare, anyway...
        */
+      g_debug ("SHM invalidated while establishing subscription to %s - signalling change", ow->path);
       dconf_engine_change_notify (engine, ow->path, changes, NULL, FALSE, NULL, engine->user_data);
     }
 
-  dconf_engine_set_watching (engine, ow->path, TRUE, TRUE);
+  dconf_engine_lock_subscription_counts (engine);
+  guint num_establishing = dconf_engine_count_subscriptions (engine->establishing,
+                                                             ow->path);
+  g_debug ("watch_established: \"%s\" (establishing: %d)", ow->path, num_establishing);
+  if (num_establishing > 0)
+    // Subscription(s): establishing -> active
+    dconf_engine_move_subscriptions (engine->establishing,
+                                     engine->active,
+                                     ow->path);
+
+  dconf_engine_unlock_subscription_counts (engine);
   dconf_engine_call_handle_free (handle);
 }
 
@@ -846,14 +981,20 @@ void
 dconf_engine_watch_fast (DConfEngine *engine,
                          const gchar *path)
 {
-  if (dconf_engine_is_watching (engine, path, TRUE))
-    {
-      /**
-       * Either there is already a match rule in place for this exact path,
-       * or there is already a request in progress to add a match.
-       */
-      return;
-    }
+  dconf_engine_lock_subscription_counts (engine);
+  guint num_establishing = dconf_engine_count_subscriptions (engine->establishing, path);
+  guint num_active = dconf_engine_count_subscriptions (engine->active, path);
+  g_debug ("watch_fast: \"%s\" (establishing: %d, active: %d)", path, num_establishing, num_active);
+  if (num_active > 0)
+    // Subscription: inactive -> active
+    dconf_engine_inc_subscriptions (engine->active, path);
+  else
+    // Subscription: inactive -> establishing
+    num_establishing = dconf_engine_inc_subscriptions (engine->establishing,
+                                                       path);
+  dconf_engine_unlock_subscription_counts (engine);
+  if (num_establishing > 1 || num_active > 0)
+    return;
 
   OutstandingWatch *ow;
   gint i;
@@ -888,23 +1029,36 @@ dconf_engine_watch_fast (DConfEngine *engine,
                                          "/org/freedesktop/DBus", "org.freedesktop.DBus", "AddMatch",
                                          dconf_engine_make_match_rule (engine->sources[i], path),
                                          &ow->handle, NULL);
-
-  dconf_engine_set_watching (engine, ow->path, TRUE, FALSE);
 }
 
 void
 dconf_engine_unwatch_fast (DConfEngine *engine,
                            const gchar *path)
 {
+  dconf_engine_lock_subscription_counts (engine);
+  guint num_active = dconf_engine_count_subscriptions (engine->active, path);
+  guint num_establishing = dconf_engine_count_subscriptions (engine->establishing, path);
   gint i;
+  g_debug ("unwatch_fast: \"%s\" (active: %d, establishing: %d)", path, num_active, num_establishing);
+
+  // Client code cannot unsubscribe if it is not subscribed
+  g_assert (num_active > 0 || num_establishing > 0);
+  if (num_active == 0)
+    // Subscription: establishing -> inactive
+    num_establishing = dconf_engine_dec_subscriptions (engine->establishing, path);
+  else
+    // Subscription: active -> inactive
+    num_active = dconf_engine_dec_subscriptions (engine->active, path);
+
+  dconf_engine_unlock_subscription_counts (engine);
+  if (num_active > 0 || num_establishing > 0)
+    return;
 
   for (i = 0; i < engine->n_sources; i++)
     if (engine->sources[i]->bus_type)
       dconf_engine_dbus_call_async_func (engine->sources[i]->bus_type, "org.freedesktop.DBus",
                                          "/org/freedesktop/DBus", "org.freedesktop.DBus", "RemoveMatch",
                                          dconf_engine_make_match_rule (engine->sources[i], path), NULL, NULL);
-
-  dconf_engine_set_watching (engine, path, FALSE, FALSE);
 }
 
 static void
@@ -942,16 +1096,24 @@ void
 dconf_engine_watch_sync (DConfEngine *engine,
                          const gchar *path)
 {
-  dconf_engine_handle_match_rule_sync (engine, "AddMatch", path);
-  dconf_engine_set_watching (engine, path, TRUE, TRUE);
+  dconf_engine_lock_subscription_counts (engine);
+  guint num_active = dconf_engine_inc_subscriptions (engine->active, path);
+  dconf_engine_unlock_subscription_counts (engine);
+  g_debug ("watch_sync: \"%s\" (active: %d)", path, num_active - 1);
+  if (num_active == 1)
+    dconf_engine_handle_match_rule_sync (engine, "AddMatch", path);
 }
 
 void
 dconf_engine_unwatch_sync (DConfEngine *engine,
                            const gchar *path)
 {
-  dconf_engine_handle_match_rule_sync (engine, "RemoveMatch", path);
-  dconf_engine_set_watching (engine, path, FALSE, FALSE);
+  dconf_engine_lock_subscription_counts (engine);
+  guint num_active = dconf_engine_dec_subscriptions (engine->active, path);
+  dconf_engine_unlock_subscription_counts (engine);
+  g_debug ("unwatch_sync: \"%s\" (active: %d)", path, num_active + 1);
+  if (num_active == 0)
+    dconf_engine_handle_match_rule_sync (engine, "RemoveMatch", path);
 }
 
 typedef struct
@@ -1159,7 +1321,7 @@ dconf_engine_change_fast (DConfEngine     *engine,
                           GError         **error)
 {
   GList *node;
-
+  g_debug ("change_fast");
   if (dconf_changeset_is_empty (changeset))
     return TRUE;
 
@@ -1226,6 +1388,7 @@ dconf_engine_change_sync (DConfEngine     *engine,
                           GError         **error)
 {
   GVariant *reply;
+  g_debug ("change_sync");
 
   if (dconf_changeset_is_empty (changeset))
     {
@@ -1404,47 +1567,9 @@ dconf_engine_has_outstanding (DConfEngine *engine)
 void
 dconf_engine_sync (DConfEngine *engine)
 {
+  g_debug ("sync");
   dconf_engine_lock_queues (engine);
   while (!g_queue_is_empty (&engine->in_flight))
     g_cond_wait (&engine->queue_cond, &engine->queue_lock);
   dconf_engine_unlock_queues (engine);
 }
-
-void
-dconf_engine_set_watching (DConfEngine    *engine,
-                           const gchar    *path,
-                           const gboolean  is_watching,
-                           const gboolean  is_established)
-{
-  if (is_watching)
-    {
-      if (is_established)
-        {
-          g_hash_table_add (engine->watched_paths, g_strdup (path));
-          g_hash_table_remove (engine->pending_paths, path);
-        }
-      else
-        {
-          g_hash_table_add (engine->pending_paths, g_strdup (path));
-          g_hash_table_remove (engine->watched_paths, path);
-        }
-    }
-  else
-    {
-      g_hash_table_remove (engine->watched_paths, path);
-      g_hash_table_remove (engine->pending_paths, path);
-    }
-}
-
-gboolean
-dconf_engine_is_watching (DConfEngine *engine, const gchar *path, const gboolean only_established)
-{
-  gconstpointer key = (gconstpointer) path;
-  if (g_hash_table_contains (engine->watched_paths, key))
-    return TRUE;
-
-  if (!only_established && g_hash_table_contains (engine->pending_paths, key))
-    return TRUE;
-
-  return FALSE;
-}
diff --git a/engine/dconf-engine.h b/engine/dconf-engine.h
index 06ed5a7..2485423 100644
--- a/engine/dconf-engine.h
+++ b/engine/dconf-engine.h
@@ -104,17 +104,6 @@ DConfEngine *           dconf_engine_new                                (const g
 G_GNUC_INTERNAL
 void                    dconf_engine_unref                              (DConfEngine             *engine);
 
-G_GNUC_INTERNAL
-void                    dconf_engine_set_watching                       (DConfEngine             *engine,
-                                                                         const gchar             *path,
-                                                                         const gboolean           is_watching,
-                                                                         const gboolean           is_established);
-
-G_GNUC_INTERNAL
-gboolean                dconf_engine_is_watching                        (DConfEngine             *engine,
-                                                                         const gchar             *path,
-                                                                         const gboolean           only_established);
-
 /* Read API: always handled immediately */
 G_GNUC_INTERNAL
 guint64                 dconf_engine_get_state                          (DConfEngine             *engine);
diff --git a/gsettings/dconfsettingsbackend.c b/gsettings/dconfsettingsbackend.c
index 752e013..6c8179b 100644
--- a/gsettings/dconfsettingsbackend.c
+++ b/gsettings/dconfsettingsbackend.c
@@ -232,6 +232,7 @@ dconf_engine_change_notify (DConfEngine         *engine,
 {
   GWeakRef *weak_ref = user_data;
   DConfSettingsBackend *dcsb;
+  g_debug ("change_notify: %s", prefix);
 
   dcsb = g_weak_ref_get (weak_ref);
 
diff --git a/tests/engine.c b/tests/engine.c
index aa1db1c..f2e57b2 100644
--- a/tests/engine.c
+++ b/tests/engine.c
@@ -1210,13 +1210,16 @@ test_watch_fast (void)
   c = dconf_engine_get_state (engine);
   g_assert_cmpuint (b, ==, c);
   /* The watch result was not sent, because the path was already watched */
-  dconf_mock_dbus_assert_no_async();
+  dconf_mock_dbus_assert_no_async ();
   c = dconf_engine_get_state (engine);
   g_assert_cmpuint (b, ==, c);
   /* Since the path was already being watched,
    * do not expect a second false change notification */
   g_assert_cmpstr (change_log->str, ==, "/a/b/c:1::nil;");
   dconf_engine_unwatch_fast (engine, "/a/b/c");
+  /* nothing was done, because there is still a subscription left */
+  dconf_mock_dbus_assert_no_async ();
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
   dconf_mock_dbus_async_reply (triv, NULL);
   dconf_mock_dbus_async_reply (triv, NULL);
   dconf_mock_dbus_assert_no_async ();
@@ -1229,6 +1232,185 @@ test_watch_fast (void)
   g_variant_unref (triv);
 }
 
+static void
+test_watch_fast_simultaneous_subscriptions (void)
+{
+  /**
+   * Test that creating multiple subscriptions to the same path
+   * simultaneously (before receiving replies from D-Bus) only results in
+   * a single D-Bus match rule, and that it is removed at the right time.
+   */
+  DConfEngine *engine;
+  GvdbTable *table;
+  GVariant *triv;
+
+  /* Set up */
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/HOME/.config/dconf/user", table);
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/etc/dconf/db/site", table);
+
+  triv = g_variant_ref_sink (g_variant_new ("()"));
+
+  change_log = g_string_new (NULL);
+
+  engine = dconf_engine_new (SRCDIR "/profile/dos", NULL, NULL);
+
+
+  /* Subscribe to the same path 3 times. Both AddMatch results succeed
+   * (one for each source). There is only one for each source*path.
+   */
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_engine_watch_fast (engine, "/a/b/c");
+
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_assert_no_async ();
+
+  /* Unsubscribe twice, after the AddMatch succeeds. There is still one
+   * active subscription, so no RemoveMatch request is sent. */
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+
+  dconf_mock_dbus_assert_no_async ();
+
+  /* Unsubscribe once more. The number of active subscriptions falls to 0
+   * and the D-Bus match rule is removed */
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_assert_no_async ();
+
+  /* The shm was not flagged at any point - so no change notifications
+   * should not have been sent */
+  g_assert_cmpstr (change_log->str, ==, "");
+
+  /* Clean up */
+  dconf_engine_unref (engine);
+  g_string_free (change_log, TRUE);
+  change_log = NULL;
+  g_variant_unref (triv);
+}
+
+static void
+test_watch_fast_successive_subscriptions (void)
+{
+  /**
+   * Test that subscribing to the same path multiple times successively
+   * (after waiting for any expected replies from D-Bus) results in only
+   * a single D-Bus match rule being created, and that it is created and
+   * destroyed at the right times.
+   */
+  DConfEngine *engine;
+  GvdbTable *table;
+  GVariant *triv;
+
+  /* Set up */
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/HOME/.config/dconf/user", table);
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/etc/dconf/db/site", table);
+
+  triv = g_variant_ref_sink (g_variant_new ("()"));
+
+  change_log = g_string_new (NULL);
+
+  engine = dconf_engine_new (SRCDIR "/profile/dos", NULL, NULL);
+
+  /* Subscribe to a path, and simulate a change to the database while the
+   * AddMatch request is in progress */
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_mock_shm_flag ("user");
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+
+  /* When the AddMatch requests succeeds, expect a change notification
+   * for the path */
+  dconf_mock_dbus_assert_no_async ();
+  g_assert_cmpstr (change_log->str, ==, "/a/b/c:1::nil;");
+
+  /* Subscribe to a path twice again, and simulate a change to the
+   * database */
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_mock_shm_flag ("user");
+
+  /* There was already a match rule in place, so there should be no D-Bus
+   * requests and no change notifications */
+  dconf_mock_dbus_assert_no_async ();
+  g_assert_cmpstr (change_log->str, ==, "/a/b/c:1::nil;");
+
+  /* Unsubscribe */
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_mock_dbus_assert_no_async ();
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_assert_no_async ();
+
+
+  /* Clean up */
+  dconf_engine_unref (engine);
+  g_string_free (change_log, TRUE);
+  change_log = NULL;
+  g_variant_unref (triv);
+}
+
+static void
+test_watch_fast_short_lived_subscriptions (void)
+{
+  /**
+   * Test that subscribing and then immediately unsubscribing (without
+   * waiting for replies from D-Bus) multiple times to the same path
+   * correctly triggers D-Bus requests and change notifications in cases
+   * where the D-Bus match rule was not in place when the database was
+   * changed.
+   */
+  DConfEngine *engine;
+  GvdbTable *table;
+  GVariant *triv;
+
+  /* Set up */
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/HOME/.config/dconf/user", table);
+  table = dconf_mock_gvdb_table_new ();
+  dconf_mock_gvdb_install ("/etc/dconf/db/site", table);
+
+  triv = g_variant_ref_sink (g_variant_new ("()"));
+
+  change_log = g_string_new (NULL);
+
+  engine = dconf_engine_new (SRCDIR "/profile/dos", NULL, NULL);
+
+  /* Subscribe to a path twice, and simulate a change to the database
+   * while the AddMatch request is in progress */
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_engine_watch_fast (engine, "/a/b/c");
+  dconf_mock_shm_flag ("user");
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_engine_unwatch_fast (engine, "/a/b/c");
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_async_reply (triv, NULL);
+  dconf_mock_dbus_assert_no_async ();
+
+  /* When the AddMatch requests succeed, expect a change notification
+   * to have been sent for the path, even though the client has since
+   * unsubscribed. */
+  g_assert_cmpstr (change_log->str, ==, "/a/b/c:1::nil;");
+
+
+  /* Clean up */
+  dconf_engine_unref (engine);
+  g_string_free (change_log, TRUE);
+  change_log = NULL;
+  g_variant_unref (triv);
+}
+
 static const gchar *match_request_type;
 static gboolean got_match_request[5];
 
@@ -1267,13 +1449,37 @@ test_watch_sync (void)
   engine = dconf_engine_new (SRCDIR "/profile/dos", NULL, NULL);
 
   match_request_type = "AddMatch";
+
+  /* A match rule should be added when the first subscription is established */
   dconf_engine_watch_sync (engine, "/a/b/c");
   g_assert (got_match_request[G_BUS_TYPE_SESSION]);
   g_assert (got_match_request[G_BUS_TYPE_SYSTEM]);
   got_match_request[G_BUS_TYPE_SESSION] = FALSE;
   got_match_request[G_BUS_TYPE_SYSTEM] = FALSE;
 
+  /* The match rule is now already in place, so more are not needed */
+  dconf_engine_watch_sync (engine, "/a/b/c");
+  g_assert_false (got_match_request[G_BUS_TYPE_SESSION]);
+  g_assert_false (got_match_request[G_BUS_TYPE_SYSTEM]);
+
+  dconf_engine_watch_sync (engine, "/a/b/c");
+  g_assert_false (got_match_request[G_BUS_TYPE_SESSION]);
+  g_assert_false (got_match_request[G_BUS_TYPE_SYSTEM]);
+
   match_request_type = "RemoveMatch";
+
+  /* There are 3 subscriptions, so removing 2 should not remove
+   * the match rule */
+  dconf_engine_unwatch_sync (engine, "/a/b/c");
+  g_assert_false (got_match_request[G_BUS_TYPE_SESSION]);
+  g_assert_false (got_match_request[G_BUS_TYPE_SYSTEM]);
+
+  dconf_engine_unwatch_sync (engine, "/a/b/c");
+  g_assert_false (got_match_request[G_BUS_TYPE_SESSION]);
+  g_assert_false (got_match_request[G_BUS_TYPE_SYSTEM]);
+
+  /* The match rule should be removed when the last subscription is
+   * removed */
   dconf_engine_unwatch_sync (engine, "/a/b/c");
   g_assert (got_match_request[G_BUS_TYPE_SESSION]);
   g_assert (got_match_request[G_BUS_TYPE_SYSTEM]);
@@ -1287,54 +1493,6 @@ test_watch_sync (void)
 }
 
 static void
-test_watching (void)
-{
-  DConfEngine *engine;
-  const gchar *apple = "apple";
-  const gchar *orange = "orange";
-  const gchar *banana = "banana";
-
-  engine = dconf_engine_new (SRCDIR "/profile/dos", NULL, NULL);
-
-  g_assert (!dconf_engine_is_watching(engine, apple, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, apple, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, orange, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, orange, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, banana, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, banana, FALSE));
-
-  dconf_engine_set_watching (engine, apple, FALSE, FALSE);
-  dconf_engine_set_watching (engine, orange, TRUE, FALSE);
-  dconf_engine_set_watching (engine, banana, TRUE, TRUE);
-
-  g_assert (!dconf_engine_is_watching(engine, apple, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, apple, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, orange, TRUE));
-  g_assert (dconf_engine_is_watching(engine, orange, FALSE));
-  g_assert (dconf_engine_is_watching(engine, banana, TRUE));
-  g_assert (dconf_engine_is_watching(engine, banana, FALSE));
-
-  dconf_engine_set_watching (engine, orange, TRUE, TRUE);
-  dconf_engine_set_watching (engine, banana, FALSE, FALSE);
-
-  g_assert (!dconf_engine_is_watching(engine, apple, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, apple, FALSE));
-  g_assert (dconf_engine_is_watching(engine, orange, TRUE));
-  g_assert (dconf_engine_is_watching(engine, orange, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, banana, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, banana, FALSE));
-
-  dconf_engine_set_watching (engine, orange, FALSE, FALSE);
-
-  g_assert (!dconf_engine_is_watching(engine, apple, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, apple, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, orange, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, orange, FALSE));
-  g_assert (!dconf_engine_is_watching(engine, banana, TRUE));
-  g_assert (!dconf_engine_is_watching(engine, banana, FALSE));
-}
-
-static void
 test_change_fast (void)
 {
   DConfChangeset *empty, *good_write, *bad_write, *very_good_write, *slightly_bad_write;
@@ -1818,8 +1976,10 @@ main (int argc, char **argv)
   g_test_add_func ("/engine/sources/service", test_service_source);
   g_test_add_func ("/engine/read", test_read);
   g_test_add_func ("/engine/watch/fast", test_watch_fast);
+  g_test_add_func ("/engine/watch/fast/simultaneous", test_watch_fast_simultaneous_subscriptions);
+  g_test_add_func ("/engine/watch/fast/successive", test_watch_fast_successive_subscriptions);
+  g_test_add_func ("/engine/watch/fast/short_lived", test_watch_fast_short_lived_subscriptions);
   g_test_add_func ("/engine/watch/sync", test_watch_sync);
-  g_test_add_func ("/engine/watch/watching", test_watching);
   g_test_add_func ("/engine/change/fast", test_change_fast);
   g_test_add_func ("/engine/change/sync", test_change_sync);
   g_test_add_func ("/engine/signals", test_signals);