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