gio/gioscheduler.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright (C) 2006-2007 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  *
  20  * Author: Alexander Larsson <alexl@redhat.com>
  21  */
  22
  23 #include "config.h"
  24
  25 #include "gioscheduler.h"
  26 #include "gcancellable.h"
  27
  28
  29 /**
  30  * SECTION:gioscheduler
  31  * @short_description: I/O Scheduler
  32  * @include: gio/gio.h
  33  *
  34  * Schedules asynchronous I/O operations. #GIOScheduler integrates
  35  * into the main event loop (#GMainLoop) and uses threads.
  36  *
  37  * <para id="io-priority"><indexterm><primary>I/O priority</primary></indexterm>
  38  * Each I/O operation has a priority, and the scheduler uses the priorities
  39  * to determine the order in which operations are executed. They are
  40  * <emphasis>not</emphasis> used to determine system-wide I/O scheduling.
  41  * Priorities are integers, with lower numbers indicating higher priority.
  42  * It is recommended to choose priorities between %G_PRIORITY_LOW and
  43  * %G_PRIORITY_HIGH, with %G_PRIORITY_DEFAULT as a default.
  44  * </para>
  45  **/
  46
  47 struct _GIOSchedulerJob {
  48   GList *active_link;
  49   GIOSchedulerJobFunc job_func;
  50   GSourceFunc cancel_func; /* Runs under job map lock */
  51   gpointer data;
  52   GDestroyNotify destroy_notify;
  53
  54   gint io_priority;
  55   GCancellable *cancellable;
  56   gulong cancellable_id;
  57   GMainContext *context;
  58 };
  59
  60 G_LOCK_DEFINE_STATIC(active_jobs);
  61 static GList *active_jobs = NULL;
  62
  63 static GThreadPool *job_thread_pool = NULL;
  64
  65 static void io_job_thread (gpointer data,
  66                            gpointer user_data);
  67
  68 static void
  69 g_io_job_free (GIOSchedulerJob *job)
  70 {
  71   if (job->cancellable)
  72     {
  73       if (job->cancellable_id)
  74         g_cancellable_disconnect (job->cancellable, job->cancellable_id);
  75       g_object_unref (job->cancellable);
  76     }
  77   g_main_context_unref (job->context);
  78   g_free (job);
  79 }
  80
  81 static gint
  82 g_io_job_compare (gconstpointer a,
  83                   gconstpointer b,
  84                   gpointer      user_data)
  85 {
  86   const GIOSchedulerJob *aa = a;
  87   const GIOSchedulerJob *bb = b;
  88
  89   /* Cancelled jobs are set prio == -1, so that
  90      they are executed as quickly as possible */
  91
  92   /* Lower value => higher priority */
  93   if (aa->io_priority < bb->io_priority)
  94     return -1;
  95   if (aa->io_priority == bb->io_priority)
  96     return 0;
  97   return 1;
  98 }
  99
 100 static gpointer
 101 init_scheduler (gpointer arg)
 102 {
 103   if (job_thread_pool == NULL)
 104     {
 105       /* TODO: thread_pool_new can fail */
 106       job_thread_pool = g_thread_pool_new (io_job_thread,
 107                                            NULL,
 108                                            10,
 109                                            FALSE,
 110                                            NULL);
 111       if (job_thread_pool != NULL)
 112         {
 113           g_thread_pool_set_sort_function (job_thread_pool,
 114                                            g_io_job_compare,
 115                                            NULL);
 116           /* It's kinda weird that this is a global setting
 117            * instead of per threadpool. However, we really
 118            * want to cache some threads, but not keep around
 119            * those threads forever. */
 120           g_thread_pool_set_max_idle_time (15 * 1000);
 121           g_thread_pool_set_max_unused_threads (2);
 122         }
 123     }
 124   return NULL;
 125 }
 126
 127 static void
 128 on_job_canceled (GCancellable    *cancellable,
 129                  gpointer         user_data)
 130 {
 131   GIOSchedulerJob *job = user_data;
 132
 133   /* This might be called more than once */
 134   job->io_priority = -1;
 135
 136   if (job_thread_pool != NULL)
 137     g_thread_pool_set_sort_function (job_thread_pool,
 138                                      g_io_job_compare,
 139                                      NULL);
 140 }
 141
 142 static void
 143 job_destroy (gpointer data)
 144 {
 145   GIOSchedulerJob *job = data;
 146
 147   if (job->destroy_notify)
 148     job->destroy_notify (job->data);
 149
 150   G_LOCK (active_jobs);
 151   active_jobs = g_list_delete_link (active_jobs, job->active_link);
 152   G_UNLOCK (active_jobs);
 153   g_io_job_free (job);
 154 }
 155
 156 static void
 157 io_job_thread (gpointer data,
 158                gpointer user_data)
 159 {
 160   GIOSchedulerJob *job = data;
 161   gboolean result;
 162
 163   if (job->cancellable)
 164     g_cancellable_push_current (job->cancellable);
 165
 166   do
 167     {
 168       result = job->job_func (job, job->cancellable, job->data);
 169     }
 170   while (result);
 171
 172   if (job->cancellable)
 173     g_cancellable_pop_current (job->cancellable);
 174
 175   job_destroy (job);
 176 }
 177
 178 /**
 179  * g_io_scheduler_push_job:
 180  * @job_func: a #GIOSchedulerJobFunc.
 181  * @user_data: data to pass to @job_func
 182  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 183  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 184  * of the request.
 185  * @cancellable: optional #GCancellable object, %NULL to ignore.
 186  *
 187  * Schedules the I/O job to run in another thread.
 188  *
 189  * @notify will be called on @user_data after @job_func has returned,
 190  * regardless whether the job was cancelled or has run to completion.
 191  *
 192  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 193  * by calling g_cancellable_cancel() or by calling
 194  * g_io_scheduler_cancel_all_jobs().
 195  **/
 196 void
 197 g_io_scheduler_push_job (GIOSchedulerJobFunc  job_func,
 198                          gpointer             user_data,
 199                          GDestroyNotify       notify,
 200                          gint                 io_priority,
 201                          GCancellable        *cancellable)
 202 {
 203   static GOnce once_init = G_ONCE_INIT;
 204   GIOSchedulerJob *job;
 205
 206   g_return_if_fail (job_func != NULL);
 207
 208   job = g_new0 (GIOSchedulerJob, 1);
 209   job->job_func = job_func;
 210   job->data = user_data;
 211   job->destroy_notify = notify;
 212   job->io_priority = io_priority;
 213
 214   if (cancellable)
 215     {
 216       job->cancellable = g_object_ref (cancellable);
 217       job->cancellable_id = g_cancellable_connect (job->cancellable, (GCallback)on_job_canceled,
 218                                                    job, NULL);
 219     }
 220
 221   job->context = g_main_context_ref_thread_default ();
 222
 223   G_LOCK (active_jobs);
 224   active_jobs = g_list_prepend (active_jobs, job);
 225   job->active_link = active_jobs;
 226   G_UNLOCK (active_jobs);
 227
 228   g_once (&once_init, init_scheduler, NULL);
 229   g_thread_pool_push (job_thread_pool, job, NULL);
 230 }
 231
 232 /**
 233  * g_io_scheduler_cancel_all_jobs:
 234  *
 235  * Cancels all cancellable I/O jobs.
 236  *
 237  * A job is cancellable if a #GCancellable was passed into
 238  * g_io_scheduler_push_job().
 239  **/
 240 void
 241 g_io_scheduler_cancel_all_jobs (void)
 242 {
 243   GList *cancellable_list, *l;
 244
 245   G_LOCK (active_jobs);
 246   cancellable_list = NULL;
 247   for (l = active_jobs; l != NULL; l = l->next)
 248     {
 249       GIOSchedulerJob *job = l->data;
 250       if (job->cancellable)
 251         cancellable_list = g_list_prepend (cancellable_list,
 252                                            g_object_ref (job->cancellable));
 253     }
 254   G_UNLOCK (active_jobs);
 255
 256   for (l = cancellable_list; l != NULL; l = l->next)
 257     {
 258       GCancellable *c = l->data;
 259       g_cancellable_cancel (c);
 260       g_object_unref (c);
 261     }
 262   g_list_free (cancellable_list);
 263 }
 264
 265 typedef struct {
 266   GSourceFunc func;
 267   gboolean ret_val;
 268   gpointer data;
 269   GDestroyNotify notify;
 270
 271   GMutex ack_lock;
 272   GCond ack_condition;
 273   gboolean ack;
 274 } MainLoopProxy;
 275
 276 static gboolean
 277 mainloop_proxy_func (gpointer data)
 278 {
 279   MainLoopProxy *proxy = data;
 280
 281   proxy->ret_val = proxy->func (proxy->data);
 282
 283   if (proxy->notify)
 284     proxy->notify (proxy->data);
 285
 286   g_mutex_lock (&proxy->ack_lock);
 287   proxy->ack = TRUE;
 288   g_cond_signal (&proxy->ack_condition);
 289   g_mutex_unlock (&proxy->ack_lock);
 290
 291   return FALSE;
 292 }
 293
 294 static void
 295 mainloop_proxy_free (MainLoopProxy *proxy)
 296 {
 297   g_mutex_clear (&proxy->ack_lock);
 298   g_cond_clear (&proxy->ack_condition);
 299   g_free (proxy);
 300 }
 301
 302 /**
 303  * g_io_scheduler_job_send_to_mainloop:
 304  * @job: a #GIOSchedulerJob
 305  * @func: a #GSourceFunc callback that will be called in the original thread
 306  * @user_data: data to pass to @func
 307  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 308  *
 309  * Used from an I/O job to send a callback to be run in the thread
 310  * that the job was started from, waiting for the result (and thus
 311  * blocking the I/O job).
 312  *
 313  * Returns: The return value of @func
 314  **/
 315 gboolean
 316 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 317                                      GSourceFunc      func,
 318                                      gpointer         user_data,
 319                                      GDestroyNotify   notify)
 320 {
 321   GSource *source;
 322   MainLoopProxy *proxy;
 323   gboolean ret_val;
 324
 325   g_return_val_if_fail (job != NULL, FALSE);
 326   g_return_val_if_fail (func != NULL, FALSE);
 327
 328   proxy = g_new0 (MainLoopProxy, 1);
 329   proxy->func = func;
 330   proxy->data = user_data;
 331   proxy->notify = notify;
 332   g_mutex_init (&proxy->ack_lock);
 333   g_cond_init (&proxy->ack_condition);
 334   g_mutex_lock (&proxy->ack_lock);
 335
 336   source = g_idle_source_new ();
 337   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 338   g_source_set_callback (source, mainloop_proxy_func, proxy,
 339                          NULL);
 340
 341   g_source_attach (source, job->context);
 342   g_source_unref (source);
 343
 344   while (!proxy->ack)
 345     g_cond_wait (&proxy->ack_condition, &proxy->ack_lock);
 346   g_mutex_unlock (&proxy->ack_lock);
 347
 348   ret_val = proxy->ret_val;
 349   mainloop_proxy_free (proxy);
 350
 351   return ret_val;
 352 }
 353
 354 /**
 355  * g_io_scheduler_job_send_to_mainloop_async:
 356  * @job: a #GIOSchedulerJob
 357  * @func: a #GSourceFunc callback that will be called in the original thread
 358  * @user_data: data to pass to @func
 359  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 360  *
 361  * Used from an I/O job to send a callback to be run asynchronously in
 362  * the thread that the job was started from. The callback will be run
 363  * when the main loop is available, but at that time the I/O job might
 364  * have finished. The return value from the callback is ignored.
 365  *
 366  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 367  * on to this function you have to ensure that it is not freed before
 368  * @func is called, either by passing %NULL as @notify to
 369  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 370  **/
 371 void
 372 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 373                                            GSourceFunc      func,
 374                                            gpointer         user_data,
 375                                            GDestroyNotify   notify)
 376 {
 377   GSource *source;
 378   MainLoopProxy *proxy;
 379
 380   g_return_if_fail (job != NULL);
 381   g_return_if_fail (func != NULL);
 382
 383   proxy = g_new0 (MainLoopProxy, 1);
 384   proxy->func = func;
 385   proxy->data = user_data;
 386   proxy->notify = notify;
 387   g_mutex_init (&proxy->ack_lock);
 388   g_cond_init (&proxy->ack_condition);
 389
 390   source = g_idle_source_new ();
 391   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 392   g_source_set_callback (source, mainloop_proxy_func, proxy,
 393                          (GDestroyNotify)mainloop_proxy_free);
 394
 395   g_source_attach (source, job->context);
 396   g_source_unref (source);
 397 }