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   GMainContext *context;
  59
  60   guint idle_tag;
  61 };
  62
  63 G_LOCK_DEFINE_STATIC(active_jobs);
  64 static GSList *active_jobs = NULL;
  65
  66 static GThreadPool *job_thread_pool = NULL;
  67
  68 static void io_job_thread (gpointer data,
  69                            gpointer user_data);
  70
  71 static void
  72 g_io_job_free (GIOSchedulerJob *job)
  73 {
  74   if (job->cancellable)
  75     g_object_unref (job->cancellable);
  76   if (job->context)
  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 remove_active_job (GIOSchedulerJob *job)
 129 {
 130   GIOSchedulerJob *other_job;
 131   GSList *l;
 132   gboolean resort_jobs;
 133
 134   G_LOCK (active_jobs);
 135   active_jobs = g_slist_delete_link (active_jobs, job->active_link);
 136
 137   resort_jobs = FALSE;
 138   for (l = active_jobs; l != NULL; l = l->next)
 139     {
 140       other_job = l->data;
 141       if (other_job->io_priority >= 0 &&
 142           g_cancellable_is_cancelled (other_job->cancellable))
 143         {
 144           other_job->io_priority = -1;
 145           resort_jobs = TRUE;
 146         }
 147     }
 148   G_UNLOCK (active_jobs);
 149
 150   if (resort_jobs &&
 151       job_thread_pool != NULL)
 152     g_thread_pool_set_sort_function (job_thread_pool,
 153                                      g_io_job_compare,
 154                                      NULL);
 155
 156 }
 157
 158 static void
 159 job_destroy (gpointer data)
 160 {
 161   GIOSchedulerJob *job = data;
 162
 163   if (job->destroy_notify)
 164     job->destroy_notify (job->data);
 165
 166   remove_active_job (job);
 167   g_io_job_free (job);
 168 }
 169
 170 static void
 171 io_job_thread (gpointer data,
 172                gpointer user_data)
 173 {
 174   GIOSchedulerJob *job = data;
 175   gboolean result;
 176
 177   if (job->cancellable)
 178     g_cancellable_push_current (job->cancellable);
 179
 180   do
 181     {
 182       result = job->job_func (job, job->cancellable, job->data);
 183     }
 184   while (result);
 185
 186   if (job->cancellable)
 187     g_cancellable_pop_current (job->cancellable);
 188
 189   job_destroy (job);
 190 }
 191
 192 static gboolean
 193 run_job_at_idle (gpointer data)
 194 {
 195   GIOSchedulerJob *job = data;
 196   gboolean result;
 197
 198   if (job->cancellable)
 199     g_cancellable_push_current (job->cancellable);
 200
 201   result = job->job_func (job, job->cancellable, job->data);
 202
 203   if (job->cancellable)
 204     g_cancellable_pop_current (job->cancellable);
 205
 206   return result;
 207 }
 208
 209 /**
 210  * g_io_scheduler_push_job:
 211  * @job_func: a #GIOSchedulerJobFunc.
 212  * @user_data: data to pass to @job_func
 213  * @notify: a #GDestroyNotify for @user_data, or %NULL
 214  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 215  * of the request.
 216  * @cancellable: optional #GCancellable object, %NULL to ignore.
 217  *
 218  * Schedules the I/O job to run.
 219  *
 220  * @notify will be called on @user_data after @job_func has returned,
 221  * regardless whether the job was cancelled or has run to completion.
 222  *
 223  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 224  * by calling g_cancellable_cancel() or by calling
 225  * g_io_scheduler_cancel_all_jobs().
 226  **/
 227 void
 228 g_io_scheduler_push_job (GIOSchedulerJobFunc  job_func,
 229                          gpointer             user_data,
 230                          GDestroyNotify       notify,
 231                          gint                 io_priority,
 232                          GCancellable        *cancellable)
 233 {
 234   static GOnce once_init = G_ONCE_INIT;
 235   GIOSchedulerJob *job;
 236
 237   g_return_if_fail (job_func != NULL);
 238
 239   job = g_new0 (GIOSchedulerJob, 1);
 240   job->job_func = job_func;
 241   job->data = user_data;
 242   job->destroy_notify = notify;
 243   job->io_priority = io_priority;
 244
 245   if (cancellable)
 246     job->cancellable = g_object_ref (cancellable);
 247
 248   job->context = g_main_context_get_thread_default ();
 249   if (job->context)
 250     g_main_context_ref (job->context);
 251
 252   G_LOCK (active_jobs);
 253   active_jobs = g_slist_prepend (active_jobs, job);
 254   job->active_link = active_jobs;
 255   G_UNLOCK (active_jobs);
 256
 257   if (g_thread_supported())
 258     {
 259       g_once (&once_init, init_scheduler, NULL);
 260       g_thread_pool_push (job_thread_pool, job, NULL);
 261     }
 262   else
 263     {
 264       /* Threads not available, instead do the i/o sync inside a
 265        * low prio idle handler
 266        */
 267       job->idle_tag = g_idle_add_full (io_priority,
 268                                        run_job_at_idle,
 269                                        job, job_destroy);
 270     }
 271 }
 272
 273 /**
 274  * g_io_scheduler_cancel_all_jobs:
 275  *
 276  * Cancels all cancellable I/O jobs.
 277  *
 278  * A job is cancellable if a #GCancellable was passed into
 279  * g_io_scheduler_push_job().
 280  **/
 281 void
 282 g_io_scheduler_cancel_all_jobs (void)
 283 {
 284   GSList *cancellable_list, *l;
 285
 286   G_LOCK (active_jobs);
 287   cancellable_list = NULL;
 288   for (l = active_jobs; l != NULL; l = l->next)
 289     {
 290       GIOSchedulerJob *job = l->data;
 291       if (job->cancellable)
 292         cancellable_list = g_slist_prepend (cancellable_list,
 293                                             g_object_ref (job->cancellable));
 294     }
 295   G_UNLOCK (active_jobs);
 296
 297   for (l = cancellable_list; l != NULL; l = l->next)
 298     {
 299       GCancellable *c = l->data;
 300       g_cancellable_cancel (c);
 301       g_object_unref (c);
 302     }
 303   g_slist_free (cancellable_list);
 304 }
 305
 306 typedef struct {
 307   GSourceFunc func;
 308   gboolean ret_val;
 309   gpointer data;
 310   GDestroyNotify notify;
 311
 312   GMutex *ack_lock;
 313   GCond *ack_condition;
 314 } MainLoopProxy;
 315
 316 static gboolean
 317 mainloop_proxy_func (gpointer data)
 318 {
 319   MainLoopProxy *proxy = data;
 320
 321   proxy->ret_val = proxy->func (proxy->data);
 322
 323   if (proxy->notify)
 324     proxy->notify (proxy->data);
 325
 326   if (proxy->ack_lock)
 327     {
 328       g_mutex_lock (proxy->ack_lock);
 329       g_cond_signal (proxy->ack_condition);
 330       g_mutex_unlock (proxy->ack_lock);
 331     }
 332
 333   return FALSE;
 334 }
 335
 336 static void
 337 mainloop_proxy_free (MainLoopProxy *proxy)
 338 {
 339   if (proxy->ack_lock)
 340     {
 341       g_mutex_free (proxy->ack_lock);
 342       g_cond_free (proxy->ack_condition);
 343     }
 344
 345   g_free (proxy);
 346 }
 347
 348 /**
 349  * g_io_scheduler_job_send_to_mainloop:
 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: a #GDestroyNotify for @user_data, or %NULL
 354  *
 355  * Used from an I/O job to send a callback to be run in the thread
 356  * that the job was started from, waiting for the result (and thus
 357  * blocking the I/O job).
 358  *
 359  * Returns: The return value of @func
 360  **/
 361 gboolean
 362 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 363                                      GSourceFunc      func,
 364                                      gpointer         user_data,
 365                                      GDestroyNotify   notify)
 366 {
 367   GSource *source;
 368   MainLoopProxy *proxy;
 369   gboolean ret_val;
 370
 371   g_return_val_if_fail (job != NULL, FALSE);
 372   g_return_val_if_fail (func != NULL, FALSE);
 373
 374   if (job->idle_tag)
 375     {
 376       /* We just immediately re-enter in the case of idles (non-threads)
 377        * Anything else would just deadlock. If you can't handle this, enable threads.
 378        */
 379       ret_val = func (user_data);
 380       if (notify)
 381         notify (user_data);
 382       return ret_val;
 383     }
 384
 385   proxy = g_new0 (MainLoopProxy, 1);
 386   proxy->func = func;
 387   proxy->data = user_data;
 388   proxy->notify = notify;
 389   proxy->ack_lock = g_mutex_new ();
 390   proxy->ack_condition = g_cond_new ();
 391   g_mutex_lock (proxy->ack_lock);
 392
 393   source = g_idle_source_new ();
 394   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 395   g_source_set_callback (source, mainloop_proxy_func, proxy,
 396                          NULL);
 397
 398   g_source_attach (source, job->context);
 399   g_source_unref (source);
 400
 401   g_cond_wait (proxy->ack_condition, proxy->ack_lock);
 402   g_mutex_unlock (proxy->ack_lock);
 403
 404   ret_val = proxy->ret_val;
 405   mainloop_proxy_free (proxy);
 406
 407   return ret_val;
 408 }
 409
 410 /**
 411  * g_io_scheduler_job_send_to_mainloop_async:
 412  * @job: a #GIOSchedulerJob
 413  * @func: a #GSourceFunc callback that will be called in the original thread
 414  * @user_data: data to pass to @func
 415  * @notify: a #GDestroyNotify for @user_data, or %NULL
 416  *
 417  * Used from an I/O job to send a callback to be run asynchronously in
 418  * the thread that the job was started from. The callback will be run
 419  * when the main loop is available, but at that time the I/O job might
 420  * have finished. The return value from the callback is ignored.
 421  *
 422  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 423  * on to this function you have to ensure that it is not freed before
 424  * @func is called, either by passing %NULL as @notify to
 425  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 426  **/
 427 void
 428 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 429                                            GSourceFunc      func,
 430                                            gpointer         user_data,
 431                                            GDestroyNotify   notify)
 432 {
 433   GSource *source;
 434   MainLoopProxy *proxy;
 435
 436   g_return_if_fail (job != NULL);
 437   g_return_if_fail (func != NULL);
 438
 439   if (job->idle_tag)
 440     {
 441       /* We just immediately re-enter in the case of idles (non-threads)
 442        * Anything else would just deadlock. If you can't handle this, enable threads.
 443        */
 444       func (user_data);
 445       if (notify)
 446         notify (user_data);
 447       return;
 448     }
 449
 450   proxy = g_new0 (MainLoopProxy, 1);
 451   proxy->func = func;
 452   proxy->data = user_data;
 453   proxy->notify = notify;
 454
 455   source = g_idle_source_new ();
 456   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 457   g_source_set_callback (source, mainloop_proxy_func, proxy,
 458                          (GDestroyNotify)mainloop_proxy_free);
 459
 460   g_source_attach (source, job->context);
 461   g_source_unref (source);
 462 }
 463
 464
 465 #define __G_IO_SCHEDULER_C__
 466 #include "gioaliasdef.c"