[dconf: 1/4] engine: Coalesce pending writes into a single changeset

[Daniel Playfair Cal <danielplayfaircal src gnome org>]

commit ba5db440c07c3f8c76622dd955f9d6429809cde2
Author: Tomasz Miąsko <tomasz miasko gmail com>
Date:   Thu Nov 8 12:28:54 2018 +0100

    engine: Coalesce pending writes into a single changeset
    
    Instead of queuing changes and sending them to a writer one by one,
    coalesce them into a single changeset.
    
    Coalescing changes requires a little bit more work on the client side,
    see implementation of `dconf_changeset_change`, but it has chance to
    substantially reduce the total amount of work necessary and avoid costly
    disk writes.

 engine/dconf-engine.c | 46 ++++++++++------------------
 tests/client.c        | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 98 insertions(+), 31 deletions(-)
---
diff --git a/engine/dconf-engine.c b/engine/dconf-engine.c
index 06d7dd6..bb0cf9f 100644
--- a/engine/dconf-engine.c
+++ b/engine/dconf-engine.c
@@ -172,7 +172,7 @@ struct _DConfEngine
 
   GMutex              queue_lock;    /* This lock is for pending, in_flight, queue_cond */
   GCond               queue_cond;    /* Signalled when the queues empty */
-  GQueue              pending;       /* DConfChangeset */
+  DConfChangeset     *pending;       /* DConfChangeset */
   GQueue              in_flight;     /* DConfChangeset */
 
   gchar              *last_handled;  /* reply tag from last item in in_flight */
@@ -402,8 +402,7 @@ 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));
+      g_clear_pointer (&engine->pending, dconf_changeset_unref);
 
       while (!g_queue_is_empty (&engine->in_flight))
         dconf_changeset_unref ((DConfChangeset *) g_queue_pop_head (&engine->in_flight));
@@ -735,8 +734,11 @@ dconf_engine_read (DConfEngine    *engine,
           /* Check the pending queue first because those were submitted
            * more recently.
            */
-          found_key = dconf_engine_find_key_in_queue (&engine->pending, key, &value) ||
-                      dconf_engine_find_key_in_queue (&engine->in_flight, key, &value);
+          if (engine->pending != NULL)
+            found_key = dconf_changeset_get (engine->pending, key, &value);
+
+          if (!found_key)
+            found_key = dconf_engine_find_key_in_queue (&engine->in_flight, key, &value);
 
           dconf_engine_unlock_queues (engine);
         }
@@ -1249,7 +1251,7 @@ dconf_engine_change_completed (DConfEngine  *engine,
 static void
 dconf_engine_manage_queue (DConfEngine *engine)
 {
-  if (!g_queue_is_empty (&engine->pending) && g_queue_get_length (&engine->in_flight) < MAX_IN_FLIGHT)
+  if (engine->pending != NULL && g_queue_get_length (&engine->in_flight) < MAX_IN_FLIGHT)
     {
       OutstandingChange *oc;
       GVariant *parameters;
@@ -1257,7 +1259,7 @@ dconf_engine_manage_queue (DConfEngine *engine)
       oc = dconf_engine_call_handle_new (engine, dconf_engine_change_completed,
                                          G_VARIANT_TYPE ("(s)"), sizeof (OutstandingChange));
 
-      oc->change = g_queue_pop_head (&engine->pending);
+      oc->change = g_steal_pointer (&engine->pending);
 
       parameters = dconf_engine_prepare_change (engine, oc->change);
 
@@ -1275,7 +1277,7 @@ dconf_engine_manage_queue (DConfEngine *engine)
       /* The in-flight queue should not be empty if we have changes
        * pending...
        */
-      g_assert (g_queue_is_empty (&engine->pending));
+      g_assert (engine->pending == NULL);
 
       g_cond_broadcast (&engine->queue_cond);
     }
@@ -1321,7 +1323,6 @@ dconf_engine_change_fast (DConfEngine     *engine,
                           gpointer         origin_tag,
                           GError         **error)
 {
-  GList *node;
   g_debug ("change_fast");
   if (dconf_changeset_is_empty (changeset))
     return TRUE;
@@ -1340,27 +1341,6 @@ dconf_engine_change_fast (DConfEngine     *engine,
    */
   dconf_engine_lock_queues (engine);
 
-  for (node = g_queue_peek_head_link (&engine->pending); node; node = node->next)
-    {
-      DConfChangeset *queued_change = node->data;
-
-      if (dconf_changeset_is_similar_to (changeset, queued_change))
-        {
-          /* We found a similar item in the queue.
-           *
-           * We want to drop the one that's in the queue already since
-           * we want our new (more recent) change to take precedence.
-           *
-           * The pending queue owned the changeset, so free it.
-           */
-          g_queue_delete_link (&engine->pending, node);
-          dconf_changeset_unref (queued_change);
-
-          /* There will only have been one, so stop looking. */
-          break;
-        }
-    }
-
   /* No matter what we're going to queue up this change, so put it in
    * the pending queue now.
    *
@@ -1371,7 +1351,11 @@ dconf_engine_change_fast (DConfEngine     *engine,
    * The change might get tossed before being sent if the loop above
    * finds it on a future call.
    */
-  g_queue_push_tail (&engine->pending, dconf_changeset_ref (changeset));
+  if (engine->pending == NULL)
+    engine->pending = dconf_changeset_new ();
+
+  dconf_changeset_change (engine->pending, changeset);
+
   dconf_engine_manage_queue (engine);
 
   dconf_engine_unlock_queues (engine);
diff --git a/tests/client.c b/tests/client.c
index 4727e0c..25f99f3 100644
--- a/tests/client.c
+++ b/tests/client.c
@@ -151,11 +151,93 @@ test_fast (void)
   check_and_free (dconf_client_read (client, "/test/value"), NULL);
   check_and_free (dconf_client_read_full (client, "/test/value", DCONF_READ_DEFAULT_VALUE, NULL), NULL);
 
+  g_assert_cmpint (g_queue_get_length (&dconf_mock_dbus_outstanding_call_handles), == , 0);
+
   /* Cleanup */
   g_signal_handlers_disconnect_by_func (client, changed, NULL);
   g_object_unref (client);
 }
 
+static gboolean changed_a, changed_b, changed_c;
+
+static void
+coalesce_changed (DConfClient         *client,
+                  const gchar         *prefix,
+                  const gchar * const *changes,
+                  const gchar         *tag,
+                  gpointer             user_data)
+{
+  changed_a = g_str_equal (prefix, "/test/a") || g_strv_contains (changes, "a");
+  changed_b = g_str_equal (prefix, "/test/b") || g_strv_contains (changes, "b");
+  changed_c = g_str_equal (prefix, "/test/c") || g_strv_contains (changes, "c");
+}
+
+static void
+test_coalesce (void)
+{
+  gint i, a, b, c;
+  gboolean should_change_a, should_change_b, should_change_c;
+  g_autoptr(DConfClient) client = NULL;
+
+  gint changes[][3] = {
+    {1, 0, 0},
+    {1, 1, 1},
+    {0, 1, 1},
+    {0, 0, 1},
+    {0, 0, 0},
+    {1, 0, 0},
+    {1, 0, 0},
+  };
+
+  client = dconf_client_new ();
+  g_signal_connect (client, "changed", G_CALLBACK (coalesce_changed), NULL);
+
+  a = b = c = 0;
+
+  for (i = 0; i != G_N_ELEMENTS (changes); ++i)
+    {
+      g_autoptr(DConfChangeset) changeset = NULL;
+
+      should_change_a = changes[i][0];
+      should_change_b = changes[i][1];
+      should_change_c = changes[i][2];
+
+      changeset = dconf_changeset_new ();
+
+      if (should_change_a)
+        dconf_changeset_set (changeset, "/test/a", g_variant_new_int32 (++a));
+      if (should_change_b)
+        dconf_changeset_set (changeset, "/test/b", g_variant_new_int32 (++b));
+      if (should_change_c)
+        dconf_changeset_set (changeset, "/test/c", g_variant_new_int32 (++c));
+
+      changed_a = changed_b = changed_c = FALSE;
+
+      g_assert_true (dconf_client_change_fast (client, changeset, NULL));
+
+      /* Notifications should be only about keys we have just written. */
+      g_assert_cmpint (should_change_a, ==, changed_a);
+      g_assert_cmpint (should_change_b, ==, changed_b);
+      g_assert_cmpint (should_change_c, ==, changed_c);
+
+      /* We should see value from the most recent write or NULL if we haven't written it yet. */
+      check_and_free (dconf_client_read (client, "/test/a"), a == 0 ? NULL : g_variant_new_int32 (a));
+      check_and_free (dconf_client_read (client, "/test/b"), b == 0 ? NULL : g_variant_new_int32 (b));
+      check_and_free (dconf_client_read (client, "/test/c"), c == 0 ? NULL : g_variant_new_int32 (c));
+    }
+
+  dconf_mock_dbus_async_reply (g_variant_new ("(s)", "1"), NULL);
+  dconf_mock_dbus_async_reply (g_variant_new ("(s)", "2"), NULL);
+  dconf_mock_dbus_async_reply (g_variant_new ("(s)", "3"), NULL);
+
+  /* There should be no more requests since all but first two have been
+   * coalesced together. */
+  dconf_mock_dbus_assert_no_async ();
+
+  /* Cleanup */
+  g_signal_handlers_disconnect_by_func (client, changed, NULL);
+}
+
 int
 main (int argc, char **argv)
 {
@@ -167,6 +249,7 @@ main (int argc, char **argv)
 
   g_test_add_func ("/client/lifecycle", test_lifecycle);
   g_test_add_func ("/client/basic-fast", test_fast);
+  g_test_add_func ("/client/coalesce", test_coalesce);
 
   return g_test_run ();
 }