[gnome-software: 6/29] lib: Add a worker thread implementation

[Philip Withnall <pwithnall src gnome org>]

commit ecf25739b2bab13efbd858174b13a4f25e2f48ac
Author: Philip Withnall <pwithnall endlessos org>
Date:   Thu Oct 14 20:26:39 2021 +0100

    lib: Add a worker thread implementation
    
    This creates a single worker thread and allows queueing of `GTask`s for
    it to execute. It’s kind of like being able to call
    `g_task_run_in_thread()` with a fixed thread rather than a thread pool.
    
    This will be used in several of the plugins which require a threaded
    design, to implement message passing from the plugin vfunc interface to
    a single worker thread which will do the work of the plugin.
    
    This will replace a global thread pool where multiple plugin vfuncs from
    multiple plugins could be executed in parallel in transient worker
    threads. This model made it hard to synchronise state, or get locking
    correct, in plugins.
    
    Signed-off-by: Philip Withnall <pwithnall endlessos org>
    
    Helps: #1472

 lib/gnome-software.h   |   1 +
 lib/gs-worker-thread.c | 410 +++++++++++++++++++++++++++++++++++++++++++++++++
 lib/gs-worker-thread.h |  41 +++++
 lib/meson.build        |   4 +-
 4 files changed, 455 insertions(+), 1 deletion(-)
---
diff --git a/lib/gnome-software.h b/lib/gnome-software.h
index 824b9524c..32eb6a594 100644
--- a/lib/gnome-software.h
+++ b/lib/gnome-software.h
@@ -27,3 +27,4 @@
 #include <gs-plugin-vfuncs.h>
 #include <gs-remote-icon.h>
 #include <gs-utils.h>
+#include <gs-worker-thread.h>
diff --git a/lib/gs-worker-thread.c b/lib/gs-worker-thread.c
new file mode 100644
index 000000000..00ea0e9be
--- /dev/null
+++ b/lib/gs-worker-thread.c
@@ -0,0 +1,410 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*-
+ * vi:set noexpandtab tabstop=8 shiftwidth=8:
+ *
+ * Copyright (C) 2021 Endless OS Foundation LLC
+ *
+ * Author: Philip Withnall <pwithnall endlessos org>
+ *
+ * SPDX-License-Identifier: GPL-2.0+
+ */
+
+/**
+ * SECTION:gs-worker-thread
+ * @short_description: A worker thread which executes queued #GTasks until stopped
+ *
+ * #GsWorkerThread is a thread-safe wrapper around a #GTask queue and a single
+ * worker thread which executes tasks on that queue.
+ *
+ * Tasks can be added to the queue using gs_worker_thread_queue(). The worker
+ * thread (which is created when #GsWorkerThread is constructed) will execute
+ * them in (priority, queue order) order. Each #GTaskThreadFunc is responsible
+ * for calling `g_task_return_*()` on its #GTask to complete that task.
+ *
+ * It is intended that gs_worker_thread_queue() is an alternative to using
+ * g_task_run_in_thread(). g_task_run_in_thread() queues tasks into a single
+ * process-wide thread pool, so they are mixed in with other tasks, and it can
+ * become hard to ensure the thread pool isn’t overwhelmed and that tasks are
+ * executed in the right order.
+ *
+ * The worker thread will continue executing tasks until
+ * gs_worker_thread_shutdown_async() is called. This must be called before the
+ * final reference to the #GsWorkerThread is dropped.
+ *
+ * Since: 42
+ */
+
+#include "config.h"
+
+#include <glib.h>
+#include <glib-object.h>
+
+#include "gs-worker-thread.h"
+
+typedef enum {
+       GS_WORKER_THREAD_STATE_RUNNING = 0,
+       GS_WORKER_THREAD_STATE_SHUTTING_DOWN = 1,
+       GS_WORKER_THREAD_STATE_SHUT_DOWN = 2,
+} GsWorkerThreadState;
+
+struct _GsWorkerThread
+{
+       GObject                  parent;
+
+       gchar                   *name;  /* (nullable) (owned) */
+
+       GsWorkerThreadState      worker_state;  /* (atomic) */
+       GMainContext            *worker_context;  /* (owned); may be NULL before setup or after shutdown */
+       GThread                 *worker_thread;  /* (atomic); may be NULL before setup or after shutdown */
+};
+
+typedef enum {
+       PROP_NAME = 1,
+} GsWorkerThreadProperty;
+
+static GParamSpec *props[PROP_NAME + 1] = { NULL, };
+
+G_DEFINE_TYPE (GsWorkerThread, gs_worker_thread, G_TYPE_OBJECT)
+
+static void
+gs_worker_thread_get_property (GObject    *object,
+                               guint       prop_id,
+                               GValue     *value,
+                               GParamSpec *pspec)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (object);
+
+       switch ((GsWorkerThreadProperty) prop_id) {
+       case PROP_NAME:
+               g_value_set_string (value, self->name);
+               break;
+       default:
+               G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+               break;
+       }
+}
+
+static void
+gs_worker_thread_set_property (GObject      *object,
+                               guint         prop_id,
+                               const GValue *value,
+                               GParamSpec   *pspec)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (object);
+
+       switch ((GsWorkerThreadProperty) prop_id) {
+       case PROP_NAME:
+               /* Construct only */
+               g_assert (self->name == NULL);
+               self->name = g_value_dup_string (value);
+               break;
+       default:
+               G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+               break;
+       }
+}
+
+static void
+gs_worker_thread_dispose (GObject *object)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (object);
+
+       /* Should have stopped by now. */
+       g_assert (self->worker_thread == NULL);
+
+       g_clear_pointer (&self->name, g_free);
+       g_clear_pointer (&self->worker_context, g_main_context_unref);
+
+       G_OBJECT_CLASS (gs_worker_thread_parent_class)->dispose (object);
+}
+
+static gpointer thread_cb (gpointer data);
+
+static void
+gs_worker_thread_constructed (GObject *object)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (object);
+
+       G_OBJECT_CLASS (gs_worker_thread_parent_class)->constructed (object);
+
+       /* Start up a worker thread and its #GMainContext. The worker will run
+        * and process events on @worker_context until @worker_state changes
+        * from %GS_WORKER_THREAD_STATE_RUNNING. */
+       self->worker_state = GS_WORKER_THREAD_STATE_RUNNING;
+       self->worker_context = g_main_context_new ();
+       self->worker_thread = g_thread_new (self->name, thread_cb, self);
+}
+
+static void
+gs_worker_thread_class_init (GsWorkerThreadClass *klass)
+{
+       GObjectClass *object_class = G_OBJECT_CLASS (klass);
+
+       object_class->constructed = gs_worker_thread_constructed;
+       object_class->get_property = gs_worker_thread_get_property;
+       object_class->set_property = gs_worker_thread_set_property;
+       object_class->dispose = gs_worker_thread_dispose;
+
+       /**
+        * GsWorkerThread:name: (not nullable):
+        *
+        * Name for the worker thread to use in debug output. This must be set.
+        *
+        * Since: 42
+        */
+       props[PROP_NAME] =
+               g_param_spec_string ("name",
+                                    "Name",
+                                    "Name for the worker thread to use in debug output.",
+                                    NULL,
+                                    G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS | 
G_PARAM_EXPLICIT_NOTIFY);
+
+       g_object_class_install_properties (object_class, G_N_ELEMENTS (props), props);
+}
+
+static gpointer
+thread_cb (gpointer data)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (data);
+       g_autoptr(GMainContextPusher) pusher = g_main_context_pusher_new (self->worker_context);
+
+       while (g_atomic_int_get (&self->worker_state) != GS_WORKER_THREAD_STATE_SHUT_DOWN)
+               g_main_context_iteration (self->worker_context, TRUE);
+
+       return NULL;
+}
+
+static void
+gs_worker_thread_init (GsWorkerThread *self)
+{
+}
+
+/**
+ * gs_worker_thread_new:
+ * @name: (not nullable): name for the worker thread
+ *
+ * Create and start a new #GsWorkerThread.
+ *
+ * @name will be used to set the thread name and in debug output.
+ *
+ * Returns: (transfer full): a new #GsWorkerThread
+ * Since: 42
+ */
+GsWorkerThread *
+gs_worker_thread_new (const gchar *name)
+{
+       g_return_val_if_fail (name != NULL, NULL);
+
+       return g_object_new (GS_TYPE_WORKER_THREAD,
+                            "name", name,
+                            NULL);
+}
+
+/* Essentially a wrapper around these elements to avoid the caller having to
+ * return `G_SOURCE_REMOVE` from their `work_func` every time. */
+typedef struct {
+       GTaskThreadFunc work_func;
+       GTask *task;  /* (owned) */
+} WorkData;
+
+static void
+work_data_free (WorkData *data)
+{
+       g_clear_object (&data->task);
+       g_free (data);
+}
+
+G_DEFINE_AUTOPTR_CLEANUP_FUNC (WorkData, work_data_free)
+
+static gboolean
+work_run_cb (gpointer _data)
+{
+       WorkData *data = _data;
+       GTask *task = data->task;
+       gpointer source_object = g_task_get_source_object (task);
+       gpointer task_data = g_task_get_task_data (task);
+       GCancellable *cancellable = g_task_get_cancellable (task);
+
+       data->work_func (task, source_object, task_data, cancellable);
+
+       return G_SOURCE_REMOVE;
+}
+
+/**
+ * gs_worker_thread_queue:
+ * @self: a #GsWorkerThread
+ * @priority: (default G_PRIORITY_DEFAULT): priority to queue the task at,
+ *   typically #G_PRIORITY_DEFAULT
+ * @work_func: (not nullable): function to run the task
+ * @task: (transfer full) (not nullable): the #GTask containing context data to
+ *   pass to @work_func
+ *
+ * Queue @task to be run in the worker thread at the given @priority.
+ *
+ * This function takes ownership of @task.
+ *
+ * When the task is run, @work_func will be executed and passed @task and the
+ * source object, task data and cancellable set on @task.
+ *
+ * @work_func is responsible for calling `g_task_return_*()` on @task once the
+ * task is complete.
+ *
+ * If a task is cancelled using its #GCancellable after it’s queued to the
+ * #GsWorkerThread, @work_func will still be executed. @work_func is responsible
+ * for checking whether the #GCancellable has been cancelled.
+ *
+ * It is an error to call this function after gs_worker_thread_shutdown_async()
+ * has called.
+ *
+ * Since: 42
+ */
+void
+gs_worker_thread_queue (GsWorkerThread  *self,
+                        gint             priority,
+                        GTaskThreadFunc  work_func,
+                        GTask           *task)
+{
+       g_autoptr(WorkData) data = NULL;
+
+       g_return_if_fail (GS_IS_WORKER_THREAD (self));
+       g_return_if_fail (work_func != NULL);
+       g_return_if_fail (G_IS_TASK (task));
+
+       g_assert (g_atomic_int_get (&self->worker_state) == GS_WORKER_THREAD_STATE_RUNNING ||
+                 g_task_get_source_tag (task) == gs_worker_thread_shutdown_async);
+
+       data = g_new0 (WorkData, 1);
+       data->work_func = work_func;
+       data->task = g_steal_pointer (&task);
+
+       g_main_context_invoke_full (self->worker_context, priority,
+                                   work_run_cb, g_steal_pointer (&data), (GDestroyNotify) work_data_free);
+}
+
+/**
+ * gs_worker_thread_is_in_worker_context:
+ * @self: a #GsWorkerThread
+ *
+ * Returns whether the calling thread is the worker thread.
+ *
+ * This is intended to be used as a precondition check to ensure that worker
+ * code is not accidentally run from the wrong thread.
+ *
+ * |[
+ * static void
+ * do_work (MyPlugin *self)
+ * {
+ *   g_assert (gs_worker_thread_is_in_worker_context (self->worker_thread));
+ *
+ *   // do some work
+ * }
+ * ]|
+ *
+ * Returns: %TRUE if running in the worker context, %FALSE otherwise
+ * Since: 42
+ */
+gboolean
+gs_worker_thread_is_in_worker_context (GsWorkerThread *self)
+{
+       return g_main_context_is_owner (self->worker_context);
+}
+
+static void shutdown_cb (GTask        *task,
+                         gpointer      source_object,
+                         gpointer      task_data,
+                         GCancellable *cancellable);
+
+/**
+ * gs_worker_thread_shutdown_async:
+ * @self: a #GsWorkerThread
+ * @cancellable: (nullable): a #GCancellable, or %NULL
+ * @callback: callback for once the asynchronous operation is complete
+ * @user_data: data to pass to @callback
+ *
+ * Shut down the worker thread.
+ *
+ * The thread will finish processing whatever task it’s currently processing
+ * (if any), will return %G_IO_ERROR_CANCELLED for all remaining queued
+ * tasks, and will then join the main process.
+ *
+ * This is a no-op if called subsequently.
+ *
+ * Since: 42
+ */
+void
+gs_worker_thread_shutdown_async (GsWorkerThread      *self,
+                                 GCancellable        *cancellable,
+                                 GAsyncReadyCallback  callback,
+                                 gpointer             user_data)
+{
+       g_autoptr(GTask) task = NULL;
+
+       g_return_if_fail (GS_IS_WORKER_THREAD (self));
+       g_return_if_fail (cancellable == NULL || G_IS_CANCELLABLE (cancellable));
+
+       task = g_task_new (self, cancellable, callback, user_data);
+       g_task_set_source_tag (task, gs_worker_thread_shutdown_async);
+
+       /* Already called? */
+       if (g_atomic_int_get (&self->worker_state) != GS_WORKER_THREAD_STATE_RUNNING) {
+               g_task_return_boolean (task, TRUE);
+               return;
+       }
+
+       /* Signal the worker thread to stop processing tasks. */
+       g_atomic_int_set (&self->worker_state, GS_WORKER_THREAD_STATE_SHUTTING_DOWN);
+       gs_worker_thread_queue (self, G_MAXINT  /* lowest priority */,
+                               shutdown_cb, g_steal_pointer (&task));
+}
+
+static void
+shutdown_cb (GTask        *task,
+             gpointer      source_object,
+             gpointer      task_data,
+             GCancellable *cancellable)
+{
+       GsWorkerThread *self = GS_WORKER_THREAD (source_object);
+       gboolean updated_state;
+
+       updated_state = g_atomic_int_compare_and_exchange (&self->worker_state,
+                                                          GS_WORKER_THREAD_STATE_SHUTTING_DOWN,
+                                                          GS_WORKER_THREAD_STATE_SHUT_DOWN);
+       g_assert (updated_state);
+
+       /* Tidy up. We can’t join the thread here as this function is executing
+        * within the thread and that would deadlock. */
+       g_clear_pointer (&self->worker_context, g_main_context_unref);
+
+       g_task_return_boolean (task, TRUE);
+}
+
+/**
+ * gs_worker_thread_shutdown_finish:
+ * @self: a #GsWorkerThread
+ * @result: a #GAsyncResult
+ * @error: return location for a #GError, or %NULL
+ *
+ * Finish an asynchronous shutdown operation started with
+ * gs_worker_thread_shutdown_async();
+ *
+ * Returns: %TRUE on success, %FALSE otherwise
+ * Since: 42
+ */
+gboolean
+gs_worker_thread_shutdown_finish (GsWorkerThread  *self,
+                                  GAsyncResult    *result,
+                                  GError         **error)
+{
+       gboolean success;
+
+       g_return_val_if_fail (GS_IS_WORKER_THREAD (self), FALSE);
+       g_return_val_if_fail (g_async_result_is_tagged (result, gs_worker_thread_shutdown_async), FALSE);
+       g_return_val_if_fail (g_task_is_valid (result, self), FALSE);
+       g_return_val_if_fail (error == NULL || *error == NULL, FALSE);
+
+       success = g_task_propagate_boolean (G_TASK (result), error);
+
+       if (success)
+               g_thread_join (g_steal_pointer (&self->worker_thread));
+
+       return success;
+}
diff --git a/lib/gs-worker-thread.h b/lib/gs-worker-thread.h
new file mode 100644
index 000000000..36b28df89
--- /dev/null
+++ b/lib/gs-worker-thread.h
@@ -0,0 +1,41 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*-
+ * vi:set noexpandtab tabstop=8 shiftwidth=8:
+ *
+ * Copyright (C) 2021 Endless OS Foundation LLC
+ *
+ * Author: Philip Withnall <pwithnall endlessos org>
+ *
+ * SPDX-License-Identifier: GPL-2.0+
+ */
+
+#pragma once
+
+#include <glib.h>
+#include <glib-object.h>
+#include <gio/gio.h>
+
+G_BEGIN_DECLS
+
+#define GS_TYPE_WORKER_THREAD (gs_worker_thread_get_type ())
+
+G_DECLARE_FINAL_TYPE (GsWorkerThread, gs_worker_thread, GS, WORKER_THREAD, GObject)
+
+GsWorkerThread *gs_worker_thread_new                   (const gchar *name);
+
+void            gs_worker_thread_queue                 (GsWorkerThread  *self,
+                                                        gint             priority,
+                                                        GTaskThreadFunc  work_func,
+                                                        GTask           *task);
+
+gboolean        gs_worker_thread_is_in_worker_context  (GsWorkerThread *self);
+
+void            gs_worker_thread_shutdown_async        (GsWorkerThread      *self,
+                                                        GCancellable        *cancellable,
+                                                        GAsyncReadyCallback  callback,
+                                                        gpointer             user_data);
+
+gboolean        gs_worker_thread_shutdown_finish       (GsWorkerThread  *self,
+                                                        GAsyncResult    *result,
+                                                        GError         **error);
+
+G_END_DECLS
diff --git a/lib/meson.build b/lib/meson.build
index 64ae55438..a80da81e0 100644
--- a/lib/meson.build
+++ b/lib/meson.build
@@ -25,7 +25,8 @@ libgnomesoftware_public_headers = [
   'gs-plugin-types.h',
   'gs-plugin-vfuncs.h',
   'gs-remote-icon.h',
-  'gs-utils.h'
+  'gs-utils.h',
+  'gs-worker-thread.h',
 ]
 
 install_headers(libgnomesoftware_public_headers,
@@ -97,6 +98,7 @@ libgnomesoftware = library(
     'gs-remote-icon.c',
     'gs-test.c',
     'gs-utils.c',
+    'gs-worker-thread.c',
   ] + libgnomesoftware_enums + [gs_build_ident_h],
   soversion: gs_plugin_api_version,
   include_directories : libgnomesoftware_include_directories,