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   if (job->context)
  73     g_main_context_unref (job->context);
  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           /* It's 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 /**
 189  * g_io_scheduler_push_job:
 190  * @job_func: a #GIOSchedulerJobFunc.
 191  * @user_data: data to pass to @job_func
 192  * @notify: a #GDestroyNotify for @user_data, or %NULL
 193  * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
 194  * of the request.
 195  * @cancellable: optional #GCancellable object, %NULL to ignore.
 196  *
 197  * Schedules the I/O job to run in another thread.
 198  *
 199  * @notify will be called on @user_data after @job_func has returned,
 200  * regardless whether the job was cancelled or has run to completion.
 201  *
 202  * If @cancellable is not %NULL, it can be used to cancel the I/O job
 203  * by calling g_cancellable_cancel() or by calling
 204  * g_io_scheduler_cancel_all_jobs().
 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   job->context = g_main_context_get_thread_default ();
 228   if (job->context)
 229     g_main_context_ref (job->context);
 230
 231   G_LOCK (active_jobs);
 232   active_jobs = g_slist_prepend (active_jobs, job);
 233   job->active_link = active_jobs;
 234   G_UNLOCK (active_jobs);
 235
 236   g_once (&once_init, init_scheduler, NULL);
 237   g_thread_pool_push (job_thread_pool, job, NULL);
 238 }
 239
 240 /**
 241  * g_io_scheduler_cancel_all_jobs:
 242  *
 243  * Cancels all cancellable I/O jobs.
 244  *
 245  * A job is cancellable if a #GCancellable was passed into
 246  * g_io_scheduler_push_job().
 247  **/
 248 void
 249 g_io_scheduler_cancel_all_jobs (void)
 250 {
 251   GSList *cancellable_list, *l;
 252
 253   G_LOCK (active_jobs);
 254   cancellable_list = NULL;
 255   for (l = active_jobs; l != NULL; l = l->next)
 256     {
 257       GIOSchedulerJob *job = l->data;
 258       if (job->cancellable)
 259         cancellable_list = g_slist_prepend (cancellable_list,
 260                                             g_object_ref (job->cancellable));
 261     }
 262   G_UNLOCK (active_jobs);
 263
 264   for (l = cancellable_list; l != NULL; l = l->next)
 265     {
 266       GCancellable *c = l->data;
 267       g_cancellable_cancel (c);
 268       g_object_unref (c);
 269     }
 270   g_slist_free (cancellable_list);
 271 }
 272
 273 typedef struct {
 274   GSourceFunc func;
 275   gboolean ret_val;
 276   gpointer data;
 277   GDestroyNotify notify;
 278
 279   GMutex ack_lock;
 280   GCond ack_condition;
 281   gboolean ack;
 282 } MainLoopProxy;
 283
 284 static gboolean
 285 mainloop_proxy_func (gpointer data)
 286 {
 287   MainLoopProxy *proxy = data;
 288
 289   proxy->ret_val = proxy->func (proxy->data);
 290
 291   if (proxy->notify)
 292     proxy->notify (proxy->data);
 293
 294   g_mutex_lock (&proxy->ack_lock);
 295   proxy->ack = TRUE;
 296   g_cond_signal (&proxy->ack_condition);
 297   g_mutex_unlock (&proxy->ack_lock);
 298
 299   return FALSE;
 300 }
 301
 302 static void
 303 mainloop_proxy_free (MainLoopProxy *proxy)
 304 {
 305   g_mutex_clear (&proxy->ack_lock);
 306   g_cond_clear (&proxy->ack_condition);
 307   g_free (proxy);
 308 }
 309
 310 /**
 311  * g_io_scheduler_job_send_to_mainloop:
 312  * @job: a #GIOSchedulerJob
 313  * @func: a #GSourceFunc callback that will be called in the original thread
 314  * @user_data: data to pass to @func
 315  * @notify: a #GDestroyNotify for @user_data, or %NULL
 316  *
 317  * Used from an I/O job to send a callback to be run in the thread
 318  * that the job was started from, waiting for the result (and thus
 319  * blocking the I/O job).
 320  *
 321  * Returns: The return value of @func
 322  **/
 323 gboolean
 324 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
 325                                      GSourceFunc      func,
 326                                      gpointer         user_data,
 327                                      GDestroyNotify   notify)
 328 {
 329   GSource *source;
 330   MainLoopProxy *proxy;
 331   gboolean ret_val;
 332
 333   g_return_val_if_fail (job != NULL, FALSE);
 334   g_return_val_if_fail (func != NULL, FALSE);
 335
 336   proxy = g_new0 (MainLoopProxy, 1);
 337   proxy->func = func;
 338   proxy->data = user_data;
 339   proxy->notify = notify;
 340   g_mutex_init (&proxy->ack_lock);
 341   g_cond_init (&proxy->ack_condition);
 342   g_mutex_lock (&proxy->ack_lock);
 343
 344   source = g_idle_source_new ();
 345   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 346   g_source_set_callback (source, mainloop_proxy_func, proxy,
 347                          NULL);
 348
 349   g_source_attach (source, job->context);
 350   g_source_unref (source);
 351
 352   while (!proxy->ack)
 353     g_cond_wait (&proxy->ack_condition, &proxy->ack_lock);
 354   g_mutex_unlock (&proxy->ack_lock);
 355
 356   ret_val = proxy->ret_val;
 357   mainloop_proxy_free (proxy);
 358
 359   return ret_val;
 360 }
 361
 362 /**
 363  * g_io_scheduler_job_send_to_mainloop_async:
 364  * @job: a #GIOSchedulerJob
 365  * @func: a #GSourceFunc callback that will be called in the original thread
 366  * @user_data: data to pass to @func
 367  * @notify: a #GDestroyNotify for @user_data, or %NULL
 368  *
 369  * Used from an I/O job to send a callback to be run asynchronously in
 370  * the thread that the job was started from. The callback will be run
 371  * when the main loop is available, but at that time the I/O job might
 372  * have finished. The return value from the callback is ignored.
 373  *
 374  * Note that if you are passing the @user_data from g_io_scheduler_push_job()
 375  * on to this function you have to ensure that it is not freed before
 376  * @func is called, either by passing %NULL as @notify to
 377  * g_io_scheduler_push_job() or by using refcounting for @user_data.
 378  **/
 379 void
 380 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
 381                                            GSourceFunc      func,
 382                                            gpointer         user_data,
 383                                            GDestroyNotify   notify)
 384 {
 385   GSource *source;
 386   MainLoopProxy *proxy;
 387
 388   g_return_if_fail (job != NULL);
 389   g_return_if_fail (func != NULL);
 390
 391   proxy = g_new0 (MainLoopProxy, 1);
 392   proxy->func = func;
 393   proxy->data = user_data;
 394   proxy->notify = notify;
 395   g_mutex_init (&proxy->ack_lock);
 396   g_cond_init (&proxy->ack_condition);
 397
 398   source = g_idle_source_new ();
 399   g_source_set_priority (source, G_PRIORITY_DEFAULT);
 400   g_source_set_callback (source, mainloop_proxy_func, proxy,
 401                          (GDestroyNotify)mainloop_proxy_free);
 402
 403   g_source_attach (source, job->context);
 404   g_source_unref (source);
 405 }