* Author: Alexander Larsson <alexl@redhat.com>
*/
-#include <config.h>
+#include "config.h"
#include "gioscheduler.h"
+#include "gcancellable.h"
-#include "gioalias.h"
/**
* SECTION:gioscheduler
* @short_description: I/O Scheduler
+ * @include: gio/gio.h
*
- * Schedules asynchronous I/O operations for integration into the main
- * event loop (#GMainLoop).
+ * Schedules asynchronous I/O operations. #GIOScheduler integrates
+ * into the main event loop (#GMainLoop) and uses threads.
*
+ * <para id="io-priority"><indexterm><primary>I/O priority</primary></indexterm>
* Each I/O operation has a priority, and the scheduler uses the priorities
* to determine the order in which operations are executed. They are
* <emphasis>not</emphasis> used to determine system-wide I/O scheduling.
* Priorities are integers, with lower numbers indicating higher priority.
* It is recommended to choose priorities between %G_PRIORITY_LOW and
* %G_PRIORITY_HIGH, with %G_PRIORITY_DEFAULT as a default.
+ * </para>
**/
-struct _GIOJob {
- GSList *active_link;
- GIOJobFunc job_func;
- GIODataFunc cancel_func; /* Runs under job map lock */
+struct _GIOSchedulerJob {
+ GList *active_link;
+ GIOSchedulerJobFunc job_func;
+ GSourceFunc cancel_func; /* Runs under job map lock */
gpointer data;
GDestroyNotify destroy_notify;
gint io_priority;
GCancellable *cancellable;
-
- guint idle_tag;
+ gulong cancellable_id;
+ GMainContext *context;
};
G_LOCK_DEFINE_STATIC(active_jobs);
-static GSList *active_jobs = NULL;
+static GList *active_jobs = NULL;
static GThreadPool *job_thread_pool = NULL;
gpointer user_data);
static void
-g_io_job_free (GIOJob *job)
+g_io_job_free (GIOSchedulerJob *job)
{
if (job->cancellable)
- g_object_unref (job->cancellable);
+ {
+ if (job->cancellable_id)
+ g_cancellable_disconnect (job->cancellable, job->cancellable_id);
+ g_object_unref (job->cancellable);
+ }
+ g_main_context_unref (job->context);
g_free (job);
}
gconstpointer b,
gpointer user_data)
{
- const GIOJob *aa = a;
- const GIOJob *bb = b;
+ const GIOSchedulerJob *aa = a;
+ const GIOSchedulerJob *bb = b;
/* Cancelled jobs are set prio == -1, so that
they are executed as quickly as possible */
g_thread_pool_set_sort_function (job_thread_pool,
g_io_job_compare,
NULL);
- /* Its kinda weird that this is a global setting
- * instead of per threadpool. However, we really
- * want to cache some threads, but not keep around
- * those threads forever. */
- g_thread_pool_set_max_idle_time (15 * 1000);
- g_thread_pool_set_max_unused_threads (2);
}
}
return NULL;
}
static void
-remove_active_job (GIOJob *job)
+on_job_canceled (GCancellable *cancellable,
+ gpointer user_data)
{
- GIOJob *other_job;
- GSList *l;
- gboolean resort_jobs;
-
- G_LOCK (active_jobs);
- active_jobs = g_slist_delete_link (active_jobs, job->active_link);
-
- resort_jobs = FALSE;
- for (l = active_jobs; l != NULL; l = l->next)
- {
- other_job = l->data;
- if (other_job->io_priority >= 0 &&
- g_cancellable_is_cancelled (other_job->cancellable))
- {
- other_job->io_priority = -1;
- resort_jobs = TRUE;
- }
- }
- G_UNLOCK (active_jobs);
-
- if (resort_jobs &&
- job_thread_pool != NULL)
+ GIOSchedulerJob *job = user_data;
+
+ /* This might be called more than once */
+ job->io_priority = -1;
+
+ if (job_thread_pool != NULL)
g_thread_pool_set_sort_function (job_thread_pool,
g_io_job_compare,
NULL);
-
}
static void
-io_job_thread (gpointer data,
- gpointer user_data)
+job_destroy (gpointer data)
{
- GIOJob *job = data;
-
- if (job->cancellable)
- g_push_current_cancellable (job->cancellable);
- job->job_func (job, job->cancellable, job->data);
- if (job->cancellable)
- g_pop_current_cancellable (job->cancellable);
+ GIOSchedulerJob *job = data;
if (job->destroy_notify)
job->destroy_notify (job->data);
- remove_active_job (job);
+ G_LOCK (active_jobs);
+ active_jobs = g_list_delete_link (active_jobs, job->active_link);
+ G_UNLOCK (active_jobs);
g_io_job_free (job);
-
}
-static gboolean
-run_job_at_idle (gpointer data)
+static void
+io_job_thread (gpointer data,
+ gpointer user_data)
{
- GIOJob *job = data;
+ GIOSchedulerJob *job = data;
+ gboolean result;
if (job->cancellable)
- g_push_current_cancellable (job->cancellable);
-
- job->job_func (job, job->cancellable, job->data);
-
- if (job->cancellable)
- g_pop_current_cancellable (job->cancellable);
+ g_cancellable_push_current (job->cancellable);
- if (job->destroy_notify)
- job->destroy_notify (job->data);
+ do
+ {
+ result = job->job_func (job, job->cancellable, job->data);
+ }
+ while (result);
- remove_active_job (job);
- g_io_job_free (job);
+ if (job->cancellable)
+ g_cancellable_pop_current (job->cancellable);
- return FALSE;
+ job_destroy (job);
}
/**
- * g_schedule_io_job:
- * @job_func: a #GIOJobFunc.
- * @user_data: a #gpointer.
- * @notify: a #GDestroyNotify.
+ * g_io_scheduler_push_job:
+ * @job_func: a #GIOSchedulerJobFunc.
+ * @user_data: data to pass to @job_func
+ * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
* @io_priority: the <link linkend="gioscheduler">I/O priority</link>
* of the request.
* @cancellable: optional #GCancellable object, %NULL to ignore.
*
- * Schedules the I/O Job.
+ * Schedules the I/O job to run in another thread.
+ *
+ * @notify will be called on @user_data after @job_func has returned,
+ * regardless whether the job was cancelled or has run to completion.
*
+ * If @cancellable is not %NULL, it can be used to cancel the I/O job
+ * by calling g_cancellable_cancel() or by calling
+ * g_io_scheduler_cancel_all_jobs().
**/
void
-g_schedule_io_job (GIOJobFunc job_func,
- gpointer user_data,
- GDestroyNotify notify,
- gint io_priority,
- GCancellable *cancellable)
+g_io_scheduler_push_job (GIOSchedulerJobFunc job_func,
+ gpointer user_data,
+ GDestroyNotify notify,
+ gint io_priority,
+ GCancellable *cancellable)
{
static GOnce once_init = G_ONCE_INIT;
- GIOJob *job;
+ GIOSchedulerJob *job;
g_return_if_fail (job_func != NULL);
- job = g_new0 (GIOJob, 1);
+ job = g_new0 (GIOSchedulerJob, 1);
job->job_func = job_func;
job->data = user_data;
job->destroy_notify = notify;
job->io_priority = io_priority;
if (cancellable)
- job->cancellable = g_object_ref (cancellable);
+ {
+ job->cancellable = g_object_ref (cancellable);
+ job->cancellable_id = g_cancellable_connect (job->cancellable, (GCallback)on_job_canceled,
+ job, NULL);
+ }
+
+ job->context = g_main_context_ref_thread_default ();
G_LOCK (active_jobs);
- active_jobs = g_slist_prepend (active_jobs, job);
+ active_jobs = g_list_prepend (active_jobs, job);
job->active_link = active_jobs;
G_UNLOCK (active_jobs);
- if (g_thread_supported())
- {
- g_once (&once_init, init_scheduler, NULL);
- g_thread_pool_push (job_thread_pool, job, NULL);
- }
- else
- {
- /* Threads not available, instead do the i/o sync inside a
- * low prio idle handler
- */
- job->idle_tag = g_idle_add_full (G_PRIORITY_DEFAULT_IDLE + 1 + io_priority / 10,
- run_job_at_idle,
- job, NULL);
- }
+ g_once (&once_init, init_scheduler, NULL);
+ g_thread_pool_push (job_thread_pool, job, NULL);
}
/**
- * g_cancel_all_io_jobs:
+ * g_io_scheduler_cancel_all_jobs:
*
- * Cancels all cancellable I/O Jobs.
+ * Cancels all cancellable I/O jobs.
+ *
+ * A job is cancellable if a #GCancellable was passed into
+ * g_io_scheduler_push_job().
**/
void
-g_cancel_all_io_jobs (void)
+g_io_scheduler_cancel_all_jobs (void)
{
- GSList *cancellable_list, *l;
+ GList *cancellable_list, *l;
G_LOCK (active_jobs);
cancellable_list = NULL;
for (l = active_jobs; l != NULL; l = l->next)
{
- GIOJob *job = l->data;
+ GIOSchedulerJob *job = l->data;
if (job->cancellable)
- cancellable_list = g_slist_prepend (cancellable_list,
- g_object_ref (job->cancellable));
+ cancellable_list = g_list_prepend (cancellable_list,
+ g_object_ref (job->cancellable));
}
G_UNLOCK (active_jobs);
g_cancellable_cancel (c);
g_object_unref (c);
}
- g_slist_free (cancellable_list);
+ g_list_free (cancellable_list);
}
typedef struct {
- GIODataFunc func;
- gpointer data;
+ GSourceFunc func;
+ gboolean ret_val;
+ gpointer data;
GDestroyNotify notify;
- GMutex *ack_lock;
- GCond *ack_condition;
+ GMutex ack_lock;
+ GCond ack_condition;
+ gboolean ack;
} MainLoopProxy;
static gboolean
{
MainLoopProxy *proxy = data;
- proxy->func (proxy->data);
+ proxy->ret_val = proxy->func (proxy->data);
+
+ if (proxy->notify)
+ proxy->notify (proxy->data);
+
+ g_mutex_lock (&proxy->ack_lock);
+ proxy->ack = TRUE;
+ g_cond_signal (&proxy->ack_condition);
+ g_mutex_unlock (&proxy->ack_lock);
- if (proxy->ack_lock)
- {
- g_mutex_lock (proxy->ack_lock);
- g_cond_signal (proxy->ack_condition);
- g_mutex_unlock (proxy->ack_lock);
- }
-
return FALSE;
}
static void
mainloop_proxy_free (MainLoopProxy *proxy)
{
- if (proxy->ack_lock)
- {
- g_mutex_free (proxy->ack_lock);
- g_cond_free (proxy->ack_condition);
- }
-
+ g_mutex_clear (&proxy->ack_lock);
+ g_cond_clear (&proxy->ack_condition);
g_free (proxy);
}
-static void
-mainloop_proxy_notify (gpointer data)
+/**
+ * g_io_scheduler_job_send_to_mainloop:
+ * @job: a #GIOSchedulerJob
+ * @func: a #GSourceFunc callback that will be called in the original thread
+ * @user_data: data to pass to @func
+ * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
+ *
+ * Used from an I/O job to send a callback to be run in the thread
+ * that the job was started from, waiting for the result (and thus
+ * blocking the I/O job).
+ *
+ * Returns: The return value of @func
+ **/
+gboolean
+g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
+ GSourceFunc func,
+ gpointer user_data,
+ GDestroyNotify notify)
{
- MainLoopProxy *proxy = data;
+ GSource *source;
+ MainLoopProxy *proxy;
+ gboolean ret_val;
- if (proxy->notify)
- proxy->notify (proxy->data);
+ g_return_val_if_fail (job != NULL, FALSE);
+ g_return_val_if_fail (func != NULL, FALSE);
- /* If nonblocking we free here, otherwise we free in io thread */
- if (proxy->ack_lock == NULL)
- mainloop_proxy_free (proxy);
+ proxy = g_new0 (MainLoopProxy, 1);
+ proxy->func = func;
+ proxy->data = user_data;
+ proxy->notify = notify;
+ g_mutex_init (&proxy->ack_lock);
+ g_cond_init (&proxy->ack_condition);
+ g_mutex_lock (&proxy->ack_lock);
+
+ source = g_idle_source_new ();
+ g_source_set_priority (source, G_PRIORITY_DEFAULT);
+ g_source_set_callback (source, mainloop_proxy_func, proxy,
+ NULL);
+
+ g_source_attach (source, job->context);
+ g_source_unref (source);
+
+ while (!proxy->ack)
+ g_cond_wait (&proxy->ack_condition, &proxy->ack_lock);
+ g_mutex_unlock (&proxy->ack_lock);
+
+ ret_val = proxy->ret_val;
+ mainloop_proxy_free (proxy);
+
+ return ret_val;
}
/**
- * g_io_job_send_to_mainloop:
- * @job: a #GIOJob.
- * @func: a #GIODataFunc.
- * @user_data: a #gpointer.
- * @notify: a #GDestroyNotify.
- * @block: boolean flag indicating whether or not this job should block.
+ * g_io_scheduler_job_send_to_mainloop_async:
+ * @job: a #GIOSchedulerJob
+ * @func: a #GSourceFunc callback that will be called in the original thread
+ * @user_data: data to pass to @func
+ * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
*
- * Sends an I/O job to the application's main loop for processing.
+ * Used from an I/O job to send a callback to be run asynchronously in
+ * the thread that the job was started from. The callback will be run
+ * when the main loop is available, but at that time the I/O job might
+ * have finished. The return value from the callback is ignored.
+ *
+ * Note that if you are passing the @user_data from g_io_scheduler_push_job()
+ * on to this function you have to ensure that it is not freed before
+ * @func is called, either by passing %NULL as @notify to
+ * g_io_scheduler_push_job() or by using refcounting for @user_data.
**/
void
-g_io_job_send_to_mainloop (GIOJob *job,
- GIODataFunc func,
- gpointer user_data,
- GDestroyNotify notify,
- gboolean block)
+g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
+ GSourceFunc func,
+ gpointer user_data,
+ GDestroyNotify notify)
{
GSource *source;
MainLoopProxy *proxy;
- guint id;
g_return_if_fail (job != NULL);
g_return_if_fail (func != NULL);
- if (job->idle_tag)
- {
- /* We just immediately re-enter in the case of idles (non-threads)
- * Anything else would just deadlock. If you can't handle this, enable threads.
- */
- func (user_data);
- return;
- }
-
proxy = g_new0 (MainLoopProxy, 1);
proxy->func = func;
proxy->data = user_data;
proxy->notify = notify;
-
- if (block)
- {
- proxy->ack_lock = g_mutex_new ();
- proxy->ack_condition = g_cond_new ();
- }
-
+ g_mutex_init (&proxy->ack_lock);
+ g_cond_init (&proxy->ack_condition);
+
source = g_idle_source_new ();
g_source_set_priority (source, G_PRIORITY_DEFAULT);
+ g_source_set_callback (source, mainloop_proxy_func, proxy,
+ (GDestroyNotify)mainloop_proxy_free);
- g_source_set_callback (source, mainloop_proxy_func, proxy, mainloop_proxy_notify);
-
- if (block)
- g_mutex_lock (proxy->ack_lock);
-
- id = g_source_attach (source, NULL);
+ g_source_attach (source, job->context);
g_source_unref (source);
-
- if (block)
- {
- g_cond_wait (proxy->ack_condition, proxy->ack_lock);
- g_mutex_unlock (proxy->ack_lock);
-
- /* destroy notify didn't free proxy */
- mainloop_proxy_free (proxy);
- }
}
-
-#define __G_IO_SCHEDULER_C__
-#include "gioaliasdef.c"