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