libgnomevfs/gnome-vfs-job-queue.c

   1 /* -*- Mode: C; indent-tabs-mode: t; c-basic-offset: 8; tab-width: 8 -*- */
   2 /* gnome-vfs-job-queue.c - Job queue for asynchronous GnomeVFSJobs
   3    (version for POSIX threads).
   4
   5    Copyright (C) 2001 Free Software Foundation
   6
   7    The Gnome Library is free software; you can redistribute it and/or
   8    modify it under the terms of the GNU Library General Public License as
   9    published by the Free Software Foundation; either version 2 of the
  10    License, or (at your option) any later version.
  11
  12    The Gnome Library is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15    Library General Public License for more details.
  16
  17    You should have received a copy of the GNU Library General Public
  18    License along with the Gnome Library; see the file COPYING.LIB.  If not,
  19    write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20    Boston, MA 02111-1307, USA.
  21
  22    Author: László Péter <laca@ireland.sun.com> */
  23
  24 #include <config.h>
  25 #include "gnome-vfs-job-queue.h"
  26 #include "gnome-vfs-job-slave.h"
  27 #include <libgnomevfs/gnome-vfs-job-limit.h>
  28
  29 #include <glib/gtree.h>
  30 #include <unistd.h>
  31
  32 #undef QUEUE_DEBUG
  33
  34 #ifdef QUEUE_DEBUG
  35 #define Q_DEBUG(x) g_print x
  36 #else
  37 #define Q_DEBUG(x)
  38 #endif
  39
  40 /* See the comment at job_can_start () for
  41    an explanation of the following macros */
  42 #ifndef DEFAULT_THREAD_COUNT_LIMIT
  43 #define DEFAULT_THREAD_COUNT_LIMIT 10
  44 #endif
  45
  46 #define LIMIT_FUNCTION_LOWER_BOUND 2 /* must NOT be more than DEFAULT_THREAD_COUNT_LIMIT */
  47 #define LIMIT_FUNCTION_SPEED 7       /* must be more than 0 */
  48
  49 #if LIMIT_FUNCTION_LOWER_BOUND > DEFAULT_THREAD_COUNT_LIMIT
  50 #error LIMIT_FUNCTION_LOWER_BOUND must not be more than DEFAULT_THREAD_COUNT_LIMIT
  51 #endif
  52
  53 #if LIMIT_FUNCTION_SPEED <= 0
  54 #error LIMIT_FUNCTION_SPEED must be more than 0
  55 #endif
  56
  57 /* The maximum number of threads to use for async ops */
  58 static int thread_count_limit;
  59
  60 /* This is the maximum number of threads reserved for higher priority jobs */
  61 static float max_decrease;
  62
  63 typedef GTree JobQueueType;
  64
  65 /* This mutex protects these */
  66 static GStaticMutex job_queue_lock = G_STATIC_MUTEX_INIT;
  67 static JobQueueType *job_queue;
  68 static int running_job_count;
  69 static int job_id;
  70 #ifdef QUEUE_DEBUG
  71   static int job_queue_length;
  72 #endif
  73 /* end mutex guard */
  74
  75 typedef struct JobQueueKey {
  76         int job_id;
  77         int priority;
  78 } JobQueueKey;
  79
  80 static int
  81 key_compare (gconstpointer cast_to_key1, gconstpointer cast_to_key2, gpointer user_data)
  82 {
  83         JobQueueKey *key1 = (JobQueueKey *)cast_to_key1;
  84         JobQueueKey *key2 = (JobQueueKey *)cast_to_key2;
  85
  86         /* Lower priority job comes first */
  87         if (key1->priority > key2->priority) {
  88                 return 1;
  89         }
  90
  91         if (key1->priority < key2->priority) {
  92                 return -1;
  93         }
  94
  95         /* If the 2 priorities are the same then the
  96            job with the lower job_id comes first.
  97
  98            job_ids are positive so this won't overflow.
  99         */
 100         return key1->job_id - key2->job_id;
 101 }
 102
 103 static void
 104 value_destroy (gpointer cast_to_job)
 105 {
 106         _gnome_vfs_job_destroy ((GnomeVFSJob *)cast_to_job);
 107 }
 108
 109 static JobQueueType *
 110 job_queue_new (void)
 111 {
 112         return g_tree_new_full (key_compare, NULL, g_free, value_destroy);
 113 }
 114
 115 static void
 116 job_queue_destroy (void)
 117 {
 118         g_tree_destroy (job_queue);
 119         job_queue = NULL;
 120 }
 121
 122 static void
 123 job_queue_add (GnomeVFSJob *job)
 124 {
 125         JobQueueKey *key = g_new (JobQueueKey, 1);
 126         key->job_id = ++job_id;
 127         key->priority = job->priority;
 128
 129         g_tree_insert (job_queue, key, job);
 130 #ifdef QUEUE_DEBUG
 131         job_queue_length++;
 132 #endif
 133 }
 134
 135 static int
 136 find_first_value (gpointer key, gpointer value, gpointer data)
 137 {
 138         *((GnomeVFSJob **)data) = value;
 139         return TRUE;
 140 }
 141
 142 static GnomeVFSJob *
 143 job_queue_get_first (void)
 144 {
 145         GnomeVFSJob *job = NULL;
 146
 147         if (job_queue) {
 148                 g_tree_foreach (job_queue, find_first_value, &job);
 149         }
 150
 151         return job;
 152 }
 153
 154 static int
 155 find_first_key (gpointer key, gpointer value, gpointer data)
 156 {
 157         *((JobQueueKey **)data) = key;
 158         return TRUE;
 159 }
 160
 161 static void
 162 job_queue_delete_first (void)
 163 {
 164         JobQueueKey *key = NULL;
 165
 166         g_tree_foreach (job_queue, find_first_key, &key);
 167         g_tree_steal (job_queue, key);
 168
 169         g_free (key);
 170 #ifdef QUEUE_DEBUG
 171         job_queue_length--;
 172 #endif
 173 }
 174
 175 void
 176 _gnome_vfs_job_queue_init (void)
 177 {
 178         static gboolean queue_initialized = FALSE;
 179
 180         if (queue_initialized != TRUE) {
 181                 Q_DEBUG (("initializing the job queue (thread limit: %d)\n", DEFAULT_THREAD_COUNT_LIMIT));
 182                 thread_count_limit = DEFAULT_THREAD_COUNT_LIMIT;
 183                 max_decrease = (float)thread_count_limit - LIMIT_FUNCTION_LOWER_BOUND;
 184                 job_queue = job_queue_new ();
 185                 queue_initialized = TRUE;
 186         }
 187 }
 188
 189 /* This function implements a scheduling policy where a certain number
 190    of threads is reserved for high priority jobs so they can start
 191    immediately if needed.  The lower the priority of the running jobs
 192    the more threads are reserved.  So the actual limit on running jobs
 193    is a function of the priority of the job to be started.
 194    This function converges to LIMIT_FUNCTION_LOWER_BOUND (i.e. this
 195    will be the limit belonging to the lowest priority jobs.)
 196    The speed of convergence is determined by LIMIT_FUNCTION_SPEED.
 197    For negative priority jobs the limit equals to thread_count_limit.
 198
 199    Note that thread_count_limit can be queried/set runtime using the
 200    gnome_vfs_async_job_{get,set}_limit functions.
 201
 202    The formula is as follows:
 203
 204    max_decrease = thread_count_limit - LIMIT_FUNCTION_LOWER_BOUND
 205
 206    This is the maximum difference between the limit function and the
 207    thread_count_limit.
 208
 209                                                 max_decrease * p
 210    max jobs = thread_count_limit  - floor (--------------------------)
 211                                             LIMIT_FUNCTION_SPEED + p
 212
 213    This table shows some limits belonging to the default parameters:
 214
 215    priority of the  | max number
 216    job to start     | of jobs
 217    -----------------+-----------
 218    <1               | 10
 219    1                | 9
 220    2                | 9
 221    3                | 8
 222    5                | 7
 223    10               | 6
 224    20               | 5
 225    50               | 3
 226    1000             | 3
 227
 228    For example a job with a priority of 3 will NOT be started if
 229    there are at least 8 jobs already running.
 230 */
 231 static gboolean
 232 job_can_start (int priority)
 233 {
 234         int transformed_priority;
 235         int actual_limit;
 236
 237         /* Move the highest priority to the zero point */
 238         transformed_priority = priority + GNOME_VFS_PRIORITY_MIN;
 239
 240         if (running_job_count >= thread_count_limit) {
 241                 /* Max number of jobs are already running */
 242                 return FALSE;
 243         } else if (transformed_priority >= 0) {
 244                 /* Let's not allow low (i.e. positive) priority jobs to use up all the threads.
 245                    We reserve some threads for higher priority jobs.
 246                    The lower the priority to more threads are reserved.
 247
 248                    The actual limit should the the thread count limit less a proportion
 249                    of the maximum decrease.
 250                 */
 251
 252                 actual_limit = thread_count_limit - (int)(max_decrease * transformed_priority /
 253                                                           (LIMIT_FUNCTION_SPEED + transformed_priority));
 254
 255                 if (actual_limit <= running_job_count) {
 256                         return FALSE;
 257                 }
 258         }
 259         return TRUE;
 260 }
 261
 262 void
 263 _gnome_vfs_job_queue_run (void)
 264 {
 265         GnomeVFSJob *job_to_run;
 266
 267         g_static_mutex_lock (&job_queue_lock);
 268
 269         running_job_count--;
 270         Q_DEBUG (("job finished;\t\t\t\t       %d jobs running, %d waiting\n",
 271                  running_job_count,
 272                  job_queue_length));
 273
 274         job_to_run = job_queue_get_first ();
 275         if (job_to_run != NULL) {
 276                 /* The queue is not empty */
 277                 if (job_can_start (job_to_run->priority)) {
 278                         running_job_count++;
 279                         job_queue_delete_first ();
 280                         Q_DEBUG (("taking a %2d priority job from the queue;"
 281                                  "       %d jobs running, %d waiting\n",
 282                                  job_to_run->priority,
 283                                  running_job_count,
 284                                  job_queue_length));
 285                         g_static_mutex_unlock (&job_queue_lock);
 286                         _gnome_vfs_job_create_slave (job_to_run);
 287                 } else {
 288                         g_static_mutex_unlock (&job_queue_lock);
 289                         Q_DEBUG (("waiting job is too low priority (%2d) to start;"
 290                                  " %d jobs running, %d waiting\n",
 291                                  job_to_run->priority,
 292                                  running_job_count,
 293                                  job_queue_length));
 294                 }
 295         } else {
 296                 g_static_mutex_unlock (&job_queue_lock);
 297                 Q_DEBUG (("the queue is empty;\t\t\t       %d jobs running\n", running_job_count));
 298         }
 299 }
 300
 301 gboolean
 302 _gnome_vfs_job_schedule (GnomeVFSJob *job)
 303 {
 304         g_static_mutex_lock (&job_queue_lock);
 305         if (!job_can_start (job->priority)) {
 306                 job_queue_add (job);
 307                 Q_DEBUG (("adding a %2d priority job to the queue;"
 308                          "\t       %d jobs running, %d waiting\n",
 309                          job->priority,
 310                          running_job_count,
 311                          job_queue_length));
 312                 g_static_mutex_unlock (&job_queue_lock);
 313         } else {
 314                 running_job_count++;
 315                 Q_DEBUG (("starting a %2d priority job;\t\t       %d jobs running, %d waiting\n",
 316                         job->priority,
 317                         running_job_count,
 318                         job_queue_length));
 319                 g_static_mutex_unlock (&job_queue_lock);
 320                 _gnome_vfs_job_create_slave (job);
 321         }
 322         return TRUE;
 323 }
 324
 325 /**
 326  * gnome_vfs_async_set_job_limit:
 327  * @limit: maximuum number of allowable threads
 328  *
 329  * Restrict the number of worker threads used by Async operations
 330  * to @limit.
 331  **/
 332 void
 333 gnome_vfs_async_set_job_limit (int limit)
 334 {
 335         if (limit < LIMIT_FUNCTION_LOWER_BOUND) {
 336                 g_warning ("Attempt to set the thread_count_limit below %d",
 337                            LIMIT_FUNCTION_LOWER_BOUND);
 338                 return;
 339         }
 340         g_static_mutex_lock (&job_queue_lock);
 341         thread_count_limit = limit;
 342         max_decrease = (float)thread_count_limit - LIMIT_FUNCTION_LOWER_BOUND;
 343         Q_DEBUG (("changing the thread count limit to %d\n", limit));
 344         g_static_mutex_unlock (&job_queue_lock);
 345 }
 346
 347 /**
 348  * gnome_vfs_async_get_job_limit:
 349  *
 350  * Get the current maximuum allowable number of
 351  * worker threads for Asynch operations.
 352  *
 353  * Return value: current maximuum number of threads
 354  **/
 355 int
 356 gnome_vfs_async_get_job_limit (void)
 357 {
 358         return thread_count_limit;
 359 }
 360
 361 void
 362 _gnome_vfs_job_queue_shutdown (void)
 363 {
 364         g_static_mutex_lock (&job_queue_lock);
 365
 366         job_queue_destroy ();
 367
 368         g_static_mutex_unlock (&job_queue_lock);
 369 }