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   GSList *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   GMainContext *context;
  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_main_context_unref (job->context);
  73   g_free (job);
  74 }
  75
  76 static gint
  77 g_io_job_compare (gconstpointer a,
  78                   gconstpointer b,
  79                   gpointer      user_data)
  80 {
  81   const GIOSchedulerJob *aa = a;
  82   const GIOSchedulerJob *bb = b;
  83
  84   /* Cancelled jobs are set prio == -1, so that
  85      they are executed as quickly as possible */
  86
  87   /* Lower value => higher priority */
  88   if (aa->io_priority < bb->io_priority)
  89     return -1;
  90   if (aa->io_priority == bb->io_priority)
  91     return 0;
  92   return 1;
  93 }
  94
  95 static gpointer
  96 init_scheduler (gpointer arg)
  97 {
  98   if (job_thread_pool == NULL)
  99     {
 100       /* TODO: thread_pool_new can fail */
 101       job_thread_pool = g_thread_pool_new (io_job_thread,
 102                                            NULL,
 103                                            10,
 104                                            FALSE,
 105                                            NULL);
 106       if (job_thread_pool != NULL)
 107         {
 108           g_thread_pool_set_sort_function (job_thread_pool,
 109                                            g_io_job_compare,
 110                                            NULL);
 111           /* It's kinda weird that this is a global setting
 112            * instead of per threadpool. However, we really
 113            * want to cache some threads, but not keep around
 114            * those threads forever. */
 115           g_thread_pool_set_max_idle_time (15 * 1000);
 116           g_thread_pool_set_max_unused_threads (2);
 117         }
 118     }
 119   return NULL;
 120 }
 121
 122 static void
 123 remove_active_job (GIOSchedulerJob *job)
 124 {
 125   GIOSchedulerJob *other_job;
 126   GSList *l;
 127   gboolean resort_jobs;
 128
 129   G_LOCK (active_jobs);
 130   active_jobs = g_slist_delete_link (active_jobs, job->active_link);
 131
 132   resort_jobs = FALSE;
 133   for (l = active_jobs; l != NULL; l = l->next)
 134     {
 135       other_job = l->data;
 136       if (other_job->io_priority >= 0 &&
 137           g_cancellable_is_cancelled (other_job->cancellable))
 138         {
 139           other_job->io_priority = -1;
 140           resort_jobs = TRUE;
 141         }
 142     }
 143   G_UNLOCK (active_jobs);
 144
 145   if (resort_jobs &&
 146       job_thread_pool != NULL)
 147     g_thread_pool_set_sort_function (job_thread_pool,
 148                                      g_io_job_compare,
 149                                      NULL);
 150
 151 }
 152
 153 static void
 154 job_destroy (gpointer data)
 155 {
 156   GIOSchedulerJob *job = data;
 157
 158   if (job->destroy_notify)
 159     job->destroy_notify (job->data);
 160
 161   remove_active_job (job);
 162   g_io_job_free (job);
 163 }
 164
 165 static void
 166 io_job_thread (gpointer data,
 167                gpointer user_data)
 168 {
 169   GIOSchedulerJob *job = data;
 170   gboolean result;
 171
 172   if (job->cancellable)
 173     g_cancellable_push_current (job->cancellable);
 174
 175   do
 176     {
 177       result = job->job_func (job, job->cancellable, job->data);
 178     }
 179   while (result);
 180
 181   if (job->cancellable)
 182     g_cancellable_pop_current (job->cancellable);
 183
 184   job_destroy (job);
 185 }
 186
 187 /**
 188  * g_io_scheduler_push_job:
 189  * @job_func: a #GIOSchedulerJobFunc.
 190  * @user_data: data to pass to @job_func
 191  * @notify: a #GDestroyNotify for @user_data, or %NULL
 192  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 193  * of the request.
 194  * @cancellable: optional #GCancellable object, %NULL to ignore.
 195  *
 196  * Schedules the I/O job to run in another thread.
 197  *
 198  * @notify will be called on @user_data after @job_func has returned,
 199  * regardless whether the job was cancelled or has run to completion.
 200  *
 201  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 202  * by calling g_cancellable_cancel() or by calling
 203  * g_io_scheduler_cancel_all_jobs().
 204  **/
 205 void
 206 g_io_scheduler_push_job (GIOSchedulerJobFunc  job_func,
 207                          gpointer             user_data,
 208                          GDestroyNotify       notify,
 209                          gint                 io_priority,
 210                          GCancellable        *cancellable)
 211 {
 212   static GOnce once_init = G_ONCE_INIT;
 213   GIOSchedulerJob *job;
 214
 215   g_return_if_fail (job_func != NULL);
 216
 217   job = g_new0 (GIOSchedulerJob, 1);
 218   job->job_func = job_func;
 219   job->data = user_data;
 220   job->destroy_notify = notify;
 221   job->io_priority = io_priority;
 222
 223   if (cancellable)
 224     job->cancellable = g_object_ref (cancellable);
 225
 226   job->context = g_main_context_ref_thread_default ();
 227
 228   G_LOCK (active_jobs);
 229   active_jobs = g_slist_prepend (active_jobs, job);
 230   job->active_link = active_jobs;
 231   G_UNLOCK (active_jobs);
 232
 233   g_once (&once_init, init_scheduler, NULL);
 234   g_thread_pool_push (job_thread_pool, job, NULL);
 235 }
 236
 237 /**
 238  * g_io_scheduler_cancel_all_jobs:
 239  *
 240  * Cancels all cancellable I/O jobs.
 241  *
 242  * A job is cancellable if a #GCancellable was passed into
 243  * g_io_scheduler_push_job().
 244  **/
 245 void
 246 g_io_scheduler_cancel_all_jobs (void)
 247 {
 248   GSList *cancellable_list, *l;
 249
 250   G_LOCK (active_jobs);
 251   cancellable_list = NULL;
 252   for (l = active_jobs; l != NULL; l = l->next)
 253     {
 254       GIOSchedulerJob *job = l->data;
 255       if (job->cancellable)
 256         cancellable_list = g_slist_prepend (cancellable_list,
 257                                             g_object_ref (job->cancellable));
 258     }
 259   G_UNLOCK (active_jobs);
 260
 261   for (l = cancellable_list; l != NULL; l = l->next)
 262     {
 263       GCancellable *c = l->data;
 264       g_cancellable_cancel (c);
 265       g_object_unref (c);
 266     }
 267   g_slist_free (cancellable_list);
 268 }
 269
 270 typedef struct {
 271   GSourceFunc func;
 272   gboolean ret_val;
 273   gpointer data;
 274   GDestroyNotify notify;
 275
 276   GMutex ack_lock;
 277   GCond ack_condition;
 278   gboolean ack;
 279 } MainLoopProxy;
 280
 281 static gboolean
 282 mainloop_proxy_func (gpointer data)
 283 {
 284   MainLoopProxy *proxy = data;
 285
 286   proxy->ret_val = proxy->func (proxy->data);
 287
 288   if (proxy->notify)
 289     proxy->notify (proxy->data);
 290
 291   g_mutex_lock (&proxy->ack_lock);
 292   proxy->ack = TRUE;
 293   g_cond_signal (&proxy->ack_condition);
 294   g_mutex_unlock (&proxy->ack_lock);
 295
 296   return FALSE;
 297 }
 298
 299 static void
 300 mainloop_proxy_free (MainLoopProxy *proxy)
 301 {
 302   g_mutex_clear (&proxy->ack_lock);
 303   g_cond_clear (&proxy->ack_condition);
 304   g_free (proxy);
 305 }
 306
 307 /**
 308  * g_io_scheduler_job_send_to_mainloop:
 309  * @job: a #GIOSchedulerJob
 310  * @func: a #GSourceFunc callback that will be called in the original thread
 311  * @user_data: data to pass to @func
 312  * @notify: a #GDestroyNotify for @user_data, or %NULL
 313  *
 314  * Used from an I/O job to send a callback to be run in the thread
 315  * that the job was started from, waiting for the result (and thus
 316  * blocking the I/O job).
 317  *
 318  * Returns: The return value of @func
 319  **/
 320 gboolean
 321 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 322                                      GSourceFunc      func,
 323                                      gpointer         user_data,
 324                                      GDestroyNotify   notify)
 325 {
 326   GSource *source;
 327   MainLoopProxy *proxy;
 328   gboolean ret_val;
 329
 330   g_return_val_if_fail (job != NULL, FALSE);
 331   g_return_val_if_fail (func != NULL, FALSE);
 332
 333   proxy = g_new0 (MainLoopProxy, 1);
 334   proxy->func = func;
 335   proxy->data = user_data;
 336   proxy->notify = notify;
 337   g_mutex_init (&proxy->ack_lock);
 338   g_cond_init (&proxy->ack_condition);
 339   g_mutex_lock (&proxy->ack_lock);
 340
 341   source = g_idle_source_new ();
 342   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 343   g_source_set_callback (source, mainloop_proxy_func, proxy,
 344                          NULL);
 345
 346   g_source_attach (source, job->context);
 347   g_source_unref (source);
 348
 349   while (!proxy->ack)
 350     g_cond_wait (&proxy->ack_condition, &proxy->ack_lock);
 351   g_mutex_unlock (&proxy->ack_lock);
 352
 353   ret_val = proxy->ret_val;
 354   mainloop_proxy_free (proxy);
 355
 356   return ret_val;
 357 }
 358
 359 /**
 360  * g_io_scheduler_job_send_to_mainloop_async:
 361  * @job: a #GIOSchedulerJob
 362  * @func: a #GSourceFunc callback that will be called in the original thread
 363  * @user_data: data to pass to @func
 364  * @notify: a #GDestroyNotify for @user_data, or %NULL
 365  *
 366  * Used from an I/O job to send a callback to be run asynchronously in
 367  * the thread that the job was started from. The callback will be run
 368  * when the main loop is available, but at that time the I/O job might
 369  * have finished. The return value from the callback is ignored.
 370  *
 371  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 372  * on to this function you have to ensure that it is not freed before
 373  * @func is called, either by passing %NULL as @notify to
 374  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 375  **/
 376 void
 377 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 378                                            GSourceFunc      func,
 379                                            gpointer         user_data,
 380                                            GDestroyNotify   notify)
 381 {
 382   GSource *source;
 383   MainLoopProxy *proxy;
 384
 385   g_return_if_fail (job != NULL);
 386   g_return_if_fail (func != NULL);
 387
 388   proxy = g_new0 (MainLoopProxy, 1);
 389   proxy->func = func;
 390   proxy->data = user_data;
 391   proxy->notify = notify;
 392   g_mutex_init (&proxy->ack_lock);
 393   g_cond_init (&proxy->ack_condition);
 394
 395   source = g_idle_source_new ();
 396   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 397   g_source_set_callback (source, mainloop_proxy_func, proxy,
 398                          (GDestroyNotify)mainloop_proxy_free);
 399
 400   g_source_attach (source, job->context);
 401   g_source_unref (source);
 402 }