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 #include "gioalias.h"
  29
  30 /**
  31  * SECTION:gioscheduler
  32  * @short_description: I/O Scheduler
  33  * @include: gio/gio.h
  34  *
  35  * Schedules asynchronous I/O operations. #GIOScheduler integrates
  36  * into the main event loop (#GMainLoop) and may use threads if they
  37  * are available.
  38  *
  39  * <para id="io-priority"><indexterm><primary>I/O priority</primary></indexterm>
  40  * Each I/O operation has a priority, and the scheduler uses the priorities
  41  * to determine the order in which operations are executed. They are
  42  * <emphasis>not</emphasis> used to determine system-wide I/O scheduling.
  43  * Priorities are integers, with lower numbers indicating higher priority.
  44  * It is recommended to choose priorities between %G_PRIORITY_LOW and
  45  * %G_PRIORITY_HIGH, with %G_PRIORITY_DEFAULT as a default.
  46  * </para>
  47  **/
  48
  49 struct _GIOSchedulerJob {
  50   GSList *active_link;
  51   GIOSchedulerJobFunc job_func;
  52   GSourceFunc cancel_func; /* Runs under job map lock */
  53   gpointer data;
  54   GDestroyNotify destroy_notify;
  55
  56   gint io_priority;
  57   GCancellable *cancellable;
  58
  59   guint idle_tag;
  60 };
  61
  62 G_LOCK_DEFINE_STATIC(active_jobs);
  63 static GSList *active_jobs = NULL;
  64
  65 static GThreadPool *job_thread_pool = NULL;
  66
  67 static void io_job_thread (gpointer data,
  68                            gpointer user_data);
  69
  70 static void
  71 g_io_job_free (GIOSchedulerJob *job)
  72 {
  73   if (job->cancellable)
  74     g_object_unref (job->cancellable);
  75   g_free (job);
  76 }
  77
  78 static gint
  79 g_io_job_compare (gconstpointer a,
  80                   gconstpointer b,
  81                   gpointer      user_data)
  82 {
  83   const GIOSchedulerJob *aa = a;
  84   const GIOSchedulerJob *bb = b;
  85
  86   /* Cancelled jobs are set prio == -1, so that
  87      they are executed as quickly as possible */
  88
  89   /* Lower value => higher priority */
  90   if (aa->io_priority < bb->io_priority)
  91     return -1;
  92   if (aa->io_priority == bb->io_priority)
  93     return 0;
  94   return 1;
  95 }
  96
  97 static gpointer
  98 init_scheduler (gpointer arg)
  99 {
 100   if (job_thread_pool == NULL)
 101     {
 102       /* TODO: thread_pool_new can fail */
 103       job_thread_pool = g_thread_pool_new (io_job_thread,
 104                                            NULL,
 105                                            10,
 106                                            FALSE,
 107                                            NULL);
 108       if (job_thread_pool != NULL)
 109         {
 110           g_thread_pool_set_sort_function (job_thread_pool,
 111                                            g_io_job_compare,
 112                                            NULL);
 113           /* It's kinda weird that this is a global setting
 114            * instead of per threadpool. However, we really
 115            * want to cache some threads, but not keep around
 116            * those threads forever. */
 117           g_thread_pool_set_max_idle_time (15 * 1000);
 118           g_thread_pool_set_max_unused_threads (2);
 119         }
 120     }
 121   return NULL;
 122 }
 123
 124 static void
 125 remove_active_job (GIOSchedulerJob *job)
 126 {
 127   GIOSchedulerJob *other_job;
 128   GSList *l;
 129   gboolean resort_jobs;
 130
 131   G_LOCK (active_jobs);
 132   active_jobs = g_slist_delete_link (active_jobs, job->active_link);
 133
 134   resort_jobs = FALSE;
 135   for (l = active_jobs; l != NULL; l = l->next)
 136     {
 137       other_job = l->data;
 138       if (other_job->io_priority >= 0 &&
 139           g_cancellable_is_cancelled (other_job->cancellable))
 140         {
 141           other_job->io_priority = -1;
 142           resort_jobs = TRUE;
 143         }
 144     }
 145   G_UNLOCK (active_jobs);
 146
 147   if (resort_jobs &&
 148       job_thread_pool != NULL)
 149     g_thread_pool_set_sort_function (job_thread_pool,
 150                                      g_io_job_compare,
 151                                      NULL);
 152
 153 }
 154
 155 static void
 156 job_destroy (gpointer data)
 157 {
 158   GIOSchedulerJob *job = data;
 159
 160   if (job->destroy_notify)
 161     job->destroy_notify (job->data);
 162
 163   remove_active_job (job);
 164   g_io_job_free (job);
 165 }
 166
 167 static void
 168 io_job_thread (gpointer data,
 169                gpointer user_data)
 170 {
 171   GIOSchedulerJob *job = data;
 172   gboolean result;
 173
 174   if (job->cancellable)
 175     g_cancellable_push_current (job->cancellable);
 176
 177   do
 178     {
 179       result = job->job_func (job, job->cancellable, job->data);
 180     }
 181   while (result);
 182
 183   if (job->cancellable)
 184     g_cancellable_pop_current (job->cancellable);
 185
 186   job_destroy (job);
 187 }
 188
 189 static gboolean
 190 run_job_at_idle (gpointer data)
 191 {
 192   GIOSchedulerJob *job = data;
 193   gboolean result;
 194
 195   if (job->cancellable)
 196     g_cancellable_push_current (job->cancellable);
 197
 198   result = job->job_func (job, job->cancellable, job->data);
 199
 200   if (job->cancellable)
 201     g_cancellable_pop_current (job->cancellable);
 202
 203   return result;
 204 }
 205
 206 /**
 207  * g_io_scheduler_push_job:
 208  * @job_func: a #GIOSchedulerJobFunc.
 209  * @user_data: data to pass to @job_func
 210  * @notify: a #GDestroyNotify for @user_data, or %NULL
 211  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 212  * of the request.
 213  * @cancellable: optional #GCancellable object, %NULL to ignore.
 214  *
 215  * Schedules the I/O job to run.
 216  *
 217  * @notify will be called on @user_data after @job_func has returned,
 218  * regardless whether the job was cancelled or has run to completion.
 219  *
 220  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 221  * by calling g_cancellable_cancel() or by calling
 222  * g_io_scheduler_cancel_all_jobs().
 223  **/
 224 void
 225 g_io_scheduler_push_job (GIOSchedulerJobFunc  job_func,
 226                          gpointer             user_data,
 227                          GDestroyNotify       notify,
 228                          gint                 io_priority,
 229                          GCancellable        *cancellable)
 230 {
 231   static GOnce once_init = G_ONCE_INIT;
 232   GIOSchedulerJob *job;
 233
 234   g_return_if_fail (job_func != NULL);
 235
 236   job = g_new0 (GIOSchedulerJob, 1);
 237   job->job_func = job_func;
 238   job->data = user_data;
 239   job->destroy_notify = notify;
 240   job->io_priority = io_priority;
 241
 242   if (cancellable)
 243     job->cancellable = g_object_ref (cancellable);
 244
 245   G_LOCK (active_jobs);
 246   active_jobs = g_slist_prepend (active_jobs, job);
 247   job->active_link = active_jobs;
 248   G_UNLOCK (active_jobs);
 249
 250   if (g_thread_supported())
 251     {
 252       g_once (&once_init, init_scheduler, NULL);
 253       g_thread_pool_push (job_thread_pool, job, NULL);
 254     }
 255   else
 256     {
 257       /* Threads not available, instead do the i/o sync inside a
 258        * low prio idle handler
 259        */
 260       job->idle_tag = g_idle_add_full (G_PRIORITY_DEFAULT_IDLE + 1 + io_priority / 10,
 261                                        run_job_at_idle,
 262                                        job, job_destroy);
 263     }
 264 }
 265
 266 /**
 267  * g_io_scheduler_cancel_all_jobs:
 268  *
 269  * Cancels all cancellable I/O jobs.
 270  *
 271  * A job is cancellable if a #GCancellable was passed into
 272  * g_io_scheduler_push_job().
 273  **/
 274 void
 275 g_io_scheduler_cancel_all_jobs (void)
 276 {
 277   GSList *cancellable_list, *l;
 278
 279   G_LOCK (active_jobs);
 280   cancellable_list = NULL;
 281   for (l = active_jobs; l != NULL; l = l->next)
 282     {
 283       GIOSchedulerJob *job = l->data;
 284       if (job->cancellable)
 285         cancellable_list = g_slist_prepend (cancellable_list,
 286                                             g_object_ref (job->cancellable));
 287     }
 288   G_UNLOCK (active_jobs);
 289
 290   for (l = cancellable_list; l != NULL; l = l->next)
 291     {
 292       GCancellable *c = l->data;
 293       g_cancellable_cancel (c);
 294       g_object_unref (c);
 295     }
 296   g_slist_free (cancellable_list);
 297 }
 298
 299 typedef struct {
 300   GSourceFunc func;
 301   gboolean ret_val;
 302   gpointer data;
 303   GDestroyNotify notify;
 304
 305   GMutex *ack_lock;
 306   GCond *ack_condition;
 307 } MainLoopProxy;
 308
 309 static gboolean
 310 mainloop_proxy_func (gpointer data)
 311 {
 312   MainLoopProxy *proxy = data;
 313
 314   proxy->ret_val = proxy->func (proxy->data);
 315
 316   if (proxy->notify)
 317     proxy->notify (proxy->data);
 318
 319   if (proxy->ack_lock)
 320     {
 321       g_mutex_lock (proxy->ack_lock);
 322       g_cond_signal (proxy->ack_condition);
 323       g_mutex_unlock (proxy->ack_lock);
 324     }
 325
 326   return FALSE;
 327 }
 328
 329 static void
 330 mainloop_proxy_free (MainLoopProxy *proxy)
 331 {
 332   if (proxy->ack_lock)
 333     {
 334       g_mutex_free (proxy->ack_lock);
 335       g_cond_free (proxy->ack_condition);
 336     }
 337
 338   g_free (proxy);
 339 }
 340
 341 /**
 342  * g_io_scheduler_job_send_to_mainloop:
 343  * @job: a #GIOSchedulerJob
 344  * @func: a #GSourceFunc callback that will be called in the main thread
 345  * @user_data: data to pass to @func
 346  * @notify: a #GDestroyNotify for @user_data, or %NULL
 347  *
 348  * Used from an I/O job to send a callback to be run in the
 349  * main loop (main thread), waiting for the result (and thus
 350  * blocking the I/O job).
 351  *
 352  * Returns: The return value of @func
 353  **/
 354 gboolean
 355 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 356                                      GSourceFunc      func,
 357                                      gpointer         user_data,
 358                                      GDestroyNotify   notify)
 359 {
 360   GSource *source;
 361   MainLoopProxy *proxy;
 362   guint id;
 363   gboolean ret_val;
 364
 365   g_return_val_if_fail (job != NULL, FALSE);
 366   g_return_val_if_fail (func != NULL, FALSE);
 367
 368   if (job->idle_tag)
 369     {
 370       /* We just immediately re-enter in the case of idles (non-threads)
 371        * Anything else would just deadlock. If you can't handle this, enable threads.
 372        */
 373       ret_val = func (user_data);
 374       if (notify)
 375         notify (user_data);
 376       return ret_val;
 377     }
 378
 379   proxy = g_new0 (MainLoopProxy, 1);
 380   proxy->func = func;
 381   proxy->data = user_data;
 382   proxy->notify = notify;
 383   proxy->ack_lock = g_mutex_new ();
 384   proxy->ack_condition = g_cond_new ();
 385   g_mutex_lock (proxy->ack_lock);
 386
 387   source = g_idle_source_new ();
 388   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 389   g_source_set_callback (source, mainloop_proxy_func, proxy,
 390                          NULL);
 391
 392   id = g_source_attach (source, NULL);
 393   g_source_unref (source);
 394
 395   g_cond_wait (proxy->ack_condition, proxy->ack_lock);
 396   g_mutex_unlock (proxy->ack_lock);
 397
 398   ret_val = proxy->ret_val;
 399   mainloop_proxy_free (proxy);
 400
 401   return ret_val;
 402 }
 403
 404 /**
 405  * g_io_scheduler_job_send_to_mainloop_async:
 406  * @job: a #GIOSchedulerJob
 407  * @func: a #GSourceFunc callback that will be called in the main thread
 408  * @user_data: data to pass to @func
 409  * @notify: a #GDestroyNotify for @user_data, or %NULL
 410  *
 411  * Used from an I/O job to send a callback to be run asynchronously
 412  * in the main loop (main thread). The callback will be run when the
 413  * main loop is available, but at that time the I/O job might have
 414  * finished. The return value from the callback is ignored.
 415  *
 416  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 417  * on to this function you have to ensure that it is not freed before
 418  * @func is called, either by passing %NULL as @notify to
 419  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 420  **/
 421 void
 422 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 423                                            GSourceFunc      func,
 424                                            gpointer         user_data,
 425                                            GDestroyNotify   notify)
 426 {
 427   GSource *source;
 428   MainLoopProxy *proxy;
 429   guint id;
 430
 431   g_return_if_fail (job != NULL);
 432   g_return_if_fail (func != NULL);
 433
 434   if (job->idle_tag)
 435     {
 436       /* We just immediately re-enter in the case of idles (non-threads)
 437        * Anything else would just deadlock. If you can't handle this, enable threads.
 438        */
 439       func (user_data);
 440       if (notify)
 441         notify (user_data);
 442       return;
 443     }
 444
 445   proxy = g_new0 (MainLoopProxy, 1);
 446   proxy->func = func;
 447   proxy->data = user_data;
 448   proxy->notify = notify;
 449
 450   source = g_idle_source_new ();
 451   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 452   g_source_set_callback (source, mainloop_proxy_func, proxy,
 453                          (GDestroyNotify)mainloop_proxy_free);
 454
 455   id = g_source_attach (source, NULL);
 456   g_source_unref (source);
 457 }
 458
 459
 460 #define __G_IO_SCHEDULER_C__
 461 #include "gioaliasdef.c"