From ba5db440c07c3f8c76622dd955f9d6429809cde2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Mi=C4=85sko?= Date: Thu, 8 Nov 2018 12:28:54 +0100 Subject: 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 (); } -- cgit v1.2.1