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         }
 117     }
 118   return NULL;
 119 }
 120
 121 static void
 122 on_job_canceled (GCancellable    *cancellable,
 123                  gpointer         user_data)
 124 {
 125   GIOSchedulerJob *job = user_data;
 126
 127   /* This might be called more than once */
 128   job->io_priority = -1;
 129
 130   if (job_thread_pool != NULL)
 131     g_thread_pool_set_sort_function (job_thread_pool,
 132                                      g_io_job_compare,
 133                                      NULL);
 134 }
 135
 136 static void
 137 job_destroy (gpointer data)
 138 {
 139   GIOSchedulerJob *job = data;
 140
 141   if (job->destroy_notify)
 142     job->destroy_notify (job->data);
 143
 144   G_LOCK (active_jobs);
 145   active_jobs = g_list_delete_link (active_jobs, job->active_link);
 146   G_UNLOCK (active_jobs);
 147   g_io_job_free (job);
 148 }
 149
 150 static void
 151 io_job_thread (gpointer data,
 152                gpointer user_data)
 153 {
 154   GIOSchedulerJob *job = data;
 155   gboolean result;
 156
 157   if (job->cancellable)
 158     g_cancellable_push_current (job->cancellable);
 159
 160   do
 161     {
 162       result = job->job_func (job, job->cancellable, job->data);
 163     }
 164   while (result);
 165
 166   if (job->cancellable)
 167     g_cancellable_pop_current (job->cancellable);
 168
 169   job_destroy (job);
 170 }
 171
 172 /**
 173  * g_io_scheduler_push_job:
 174  * @job_func: a #GIOSchedulerJobFunc.
 175  * @user_data: data to pass to @job_func
 176  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 177  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 178  * of the request.
 179  * @cancellable: optional #GCancellable object, %NULL to ignore.
 180  *
 181  * Schedules the I/O job to run in another thread.
 182  *
 183  * @notify will be called on @user_data after @job_func has returned,
 184  * regardless whether the job was cancelled or has run to completion.
 185  *
 186  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 187  * by calling g_cancellable_cancel() or by calling
 188  * g_io_scheduler_cancel_all_jobs().
 189  **/
 190 void
 191 g_io_scheduler_push_job (GIOSchedulerJobFunc  job_func,
 192                          gpointer             user_data,
 193                          GDestroyNotify       notify,
 194                          gint                 io_priority,
 195                          GCancellable        *cancellable)
 196 {
 197   static GOnce once_init = G_ONCE_INIT;
 198   GIOSchedulerJob *job;
 199
 200   g_return_if_fail (job_func != NULL);
 201
 202   job = g_new0 (GIOSchedulerJob, 1);
 203   job->job_func = job_func;
 204   job->data = user_data;
 205   job->destroy_notify = notify;
 206   job->io_priority = io_priority;
 207
 208   if (cancellable)
 209     {
 210       job->cancellable = g_object_ref (cancellable);
 211       job->cancellable_id = g_cancellable_connect (job->cancellable, (GCallback)on_job_canceled,
 212                                                    job, NULL);
 213     }
 214
 215   job->context = g_main_context_ref_thread_default ();
 216
 217   G_LOCK (active_jobs);
 218   active_jobs = g_list_prepend (active_jobs, job);
 219   job->active_link = active_jobs;
 220   G_UNLOCK (active_jobs);
 221
 222   g_once (&once_init, init_scheduler, NULL);
 223   g_thread_pool_push (job_thread_pool, job, NULL);
 224 }
 225
 226 /**
 227  * g_io_scheduler_cancel_all_jobs:
 228  *
 229  * Cancels all cancellable I/O jobs.
 230  *
 231  * A job is cancellable if a #GCancellable was passed into
 232  * g_io_scheduler_push_job().
 233  **/
 234 void
 235 g_io_scheduler_cancel_all_jobs (void)
 236 {
 237   GList *cancellable_list, *l;
 238
 239   G_LOCK (active_jobs);
 240   cancellable_list = NULL;
 241   for (l = active_jobs; l != NULL; l = l->next)
 242     {
 243       GIOSchedulerJob *job = l->data;
 244       if (job->cancellable)
 245         cancellable_list = g_list_prepend (cancellable_list,
 246                                            g_object_ref (job->cancellable));
 247     }
 248   G_UNLOCK (active_jobs);
 249
 250   for (l = cancellable_list; l != NULL; l = l->next)
 251     {
 252       GCancellable *c = l->data;
 253       g_cancellable_cancel (c);
 254       g_object_unref (c);
 255     }
 256   g_list_free (cancellable_list);
 257 }
 258
 259 typedef struct {
 260   GSourceFunc func;
 261   gboolean ret_val;
 262   gpointer data;
 263   GDestroyNotify notify;
 264
 265   GMutex ack_lock;
 266   GCond ack_condition;
 267   gboolean ack;
 268 } MainLoopProxy;
 269
 270 static gboolean
 271 mainloop_proxy_func (gpointer data)
 272 {
 273   MainLoopProxy *proxy = data;
 274
 275   proxy->ret_val = proxy->func (proxy->data);
 276
 277   if (proxy->notify)
 278     proxy->notify (proxy->data);
 279
 280   g_mutex_lock (&proxy->ack_lock);
 281   proxy->ack = TRUE;
 282   g_cond_signal (&proxy->ack_condition);
 283   g_mutex_unlock (&proxy->ack_lock);
 284
 285   return FALSE;
 286 }
 287
 288 static void
 289 mainloop_proxy_free (MainLoopProxy *proxy)
 290 {
 291   g_mutex_clear (&proxy->ack_lock);
 292   g_cond_clear (&proxy->ack_condition);
 293   g_free (proxy);
 294 }
 295
 296 /**
 297  * g_io_scheduler_job_send_to_mainloop:
 298  * @job: a #GIOSchedulerJob
 299  * @func: a #GSourceFunc callback that will be called in the original thread
 300  * @user_data: data to pass to @func
 301  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 302  *
 303  * Used from an I/O job to send a callback to be run in the thread
 304  * that the job was started from, waiting for the result (and thus
 305  * blocking the I/O job).
 306  *
 307  * Returns: The return value of @func
 308  **/
 309 gboolean
 310 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 311                                      GSourceFunc      func,
 312                                      gpointer         user_data,
 313                                      GDestroyNotify   notify)
 314 {
 315   GSource *source;
 316   MainLoopProxy *proxy;
 317   gboolean ret_val;
 318
 319   g_return_val_if_fail (job != NULL, FALSE);
 320   g_return_val_if_fail (func != NULL, FALSE);
 321
 322   proxy = g_new0 (MainLoopProxy, 1);
 323   proxy->func = func;
 324   proxy->data = user_data;
 325   proxy->notify = notify;
 326   g_mutex_init (&proxy->ack_lock);
 327   g_cond_init (&proxy->ack_condition);
 328   g_mutex_lock (&proxy->ack_lock);
 329
 330   source = g_idle_source_new ();
 331   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 332   g_source_set_callback (source, mainloop_proxy_func, proxy,
 333                          NULL);
 334
 335   g_source_attach (source, job->context);
 336   g_source_unref (source);
 337
 338   while (!proxy->ack)
 339     g_cond_wait (&proxy->ack_condition, &proxy->ack_lock);
 340   g_mutex_unlock (&proxy->ack_lock);
 341
 342   ret_val = proxy->ret_val;
 343   mainloop_proxy_free (proxy);
 344
 345   return ret_val;
 346 }
 347
 348 /**
 349  * g_io_scheduler_job_send_to_mainloop_async:
 350  * @job: a #GIOSchedulerJob
 351  * @func: a #GSourceFunc callback that will be called in the original thread
 352  * @user_data: data to pass to @func
 353  * @notify: (allow-none): a #GDestroyNotify for @user_data, or %NULL
 354  *
 355  * Used from an I/O job to send a callback to be run asynchronously in
 356  * the thread that the job was started from. The callback will be run
 357  * when the main loop is available, but at that time the I/O job might
 358  * have finished. The return value from the callback is ignored.
 359  *
 360  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 361  * on to this function you have to ensure that it is not freed before
 362  * @func is called, either by passing %NULL as @notify to
 363  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 364  **/
 365 void
 366 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 367                                            GSourceFunc      func,
 368                                            gpointer         user_data,
 369                                            GDestroyNotify   notify)
 370 {
 371   GSource *source;
 372   MainLoopProxy *proxy;
 373
 374   g_return_if_fail (job != NULL);
 375   g_return_if_fail (func != NULL);
 376
 377   proxy = g_new0 (MainLoopProxy, 1);
 378   proxy->func = func;
 379   proxy->data = user_data;
 380   proxy->notify = notify;
 381   g_mutex_init (&proxy->ack_lock);
 382   g_cond_init (&proxy->ack_condition);
 383
 384   source = g_idle_source_new ();
 385   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 386   g_source_set_callback (source, mainloop_proxy_func, proxy,
 387                          (GDestroyNotify)mainloop_proxy_free);
 388
 389   g_source_attach (source, job->context);
 390   g_source_unref (source);
 391 }