glib/gasyncqueue.c

   1 /* GLIB - Library of useful routines for C programming
   2  * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
   3  *
   4  * GAsyncQueue: asynchronous queue implementation, based on Gqueue.
   5  * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This 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  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 /*
  24  * MT safe
  25  */
  26
  27 #include "config.h"
  28
  29 #include "galias.h"
  30 #include "glib.h"
  31
  32
  33 struct _GAsyncQueue
  34 {
  35   GMutex *mutex;
  36   GCond *cond;
  37   GQueue *queue;
  38   guint waiting_threads;
  39   gint32 ref_count;
  40 };
  41
  42 /**
  43  * g_async_queue_new:
  44  *
  45  * Creates a new asynchronous queue with the initial reference count of 1.
  46  *
  47  * Return value: the new #GAsyncQueue.
  48  **/
  49 GAsyncQueue*
  50 g_async_queue_new ()
  51 {
  52   GAsyncQueue* retval = g_new (GAsyncQueue, 1);
  53   retval->mutex = g_mutex_new ();
  54   retval->cond = NULL;
  55   retval->queue = g_queue_new ();
  56   retval->waiting_threads = 0;
  57   retval->ref_count = 1;
  58   return retval;
  59 }
  60
  61 /**
  62  * g_async_queue_ref:
  63  * @queue: a #GAsyncQueue.
  64  *
  65  * Increases the reference count of the asynchronous @queue by 1. You
  66  * do not need to hold the lock to call this function.
  67  **/
  68 void
  69 g_async_queue_ref (GAsyncQueue *queue)
  70 {
  71   g_return_if_fail (queue);
  72   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
  73
  74   g_atomic_int_inc (&queue->ref_count);
  75 }
  76
  77 /**
  78  * g_async_queue_ref_unlocked:
  79  * @queue: a #GAsyncQueue.
  80  *
  81  * Increases the reference count of the asynchronous @queue by 1.
  82  **/
  83 void
  84 g_async_queue_ref_unlocked (GAsyncQueue *queue)
  85 {
  86   g_return_if_fail (queue);
  87   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
  88
  89   g_atomic_int_inc (&queue->ref_count);
  90 }
  91
  92 /**
  93  * g_async_queue_unref_and_unlock:
  94  * @queue: a #GAsyncQueue.
  95  *
  96  * Decreases the reference count of the asynchronous @queue by 1 and
  97  * releases the lock. This function must be called while holding the
  98  * @queue's lock. If the reference count went to 0, the @queue will be
  99  * destroyed and the memory allocated will be freed.
 100  **/
 101 void
 102 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
 103 {
 104   g_return_if_fail (queue);
 105   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 106
 107   g_mutex_unlock (queue->mutex);
 108   g_async_queue_unref (queue);
 109 }
 110
 111 /**
 112  * g_async_queue_unref:
 113  * @queue: a #GAsyncQueue.
 114  *
 115  * Decreases the reference count of the asynchronous @queue by 1. If
 116  * the reference count went to 0, the @queue will be destroyed and the
 117  * memory allocated will be freed. So you are not allowed to use the
 118  * @queue afterwards, as it might have disappeared. You do not need to
 119  * hold the lock to call this function.
 120  **/
 121 void
 122 g_async_queue_unref (GAsyncQueue *queue)
 123 {
 124   g_return_if_fail (queue);
 125   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 126
 127   if (g_atomic_int_dec_and_test (&queue->ref_count))
 128     {
 129       g_return_if_fail (queue->waiting_threads == 0);
 130       g_mutex_free (queue->mutex);
 131       if (queue->cond)
 132         g_cond_free (queue->cond);
 133       g_queue_free (queue->queue);
 134       g_free (queue);
 135     }
 136 }
 137
 138 /**
 139  * g_async_queue_lock:
 140  * @queue: a #GAsyncQueue.
 141  *
 142  * Acquires the @queue's lock. After that you can only call the
 143  * <function>g_async_queue_*_unlocked()</function> function variants on that
 144  * @queue. Otherwise it will deadlock.
 145  **/
 146 void
 147 g_async_queue_lock (GAsyncQueue *queue)
 148 {
 149   g_return_if_fail (queue);
 150   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 151
 152   g_mutex_lock (queue->mutex);
 153 }
 154
 155 /**
 156  * g_async_queue_unlock:
 157  * @queue: a #GAsyncQueue.
 158  *
 159  * Releases the queue's lock.
 160  **/
 161 void
 162 g_async_queue_unlock (GAsyncQueue *queue)
 163 {
 164   g_return_if_fail (queue);
 165   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 166
 167   g_mutex_unlock (queue->mutex);
 168 }
 169
 170 /**
 171  * g_async_queue_push:
 172  * @queue: a #GAsyncQueue.
 173  * @data: @data to push into the @queue.
 174  *
 175  * Pushes the @data into the @queue. @data must not be %NULL.
 176  **/
 177 void
 178 g_async_queue_push (GAsyncQueue* queue, gpointer data)
 179 {
 180   g_return_if_fail (queue);
 181   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 182   g_return_if_fail (data);
 183
 184   g_mutex_lock (queue->mutex);
 185   g_async_queue_push_unlocked (queue, data);
 186   g_mutex_unlock (queue->mutex);
 187 }
 188
 189 /**
 190  * g_async_queue_push_unlocked:
 191  * @queue: a #GAsyncQueue.
 192  * @data: @data to push into the @queue.
 193  *
 194  * Pushes the @data into the @queue. @data must not be %NULL. This
 195  * function must be called while holding the @queue's lock.
 196  **/
 197 void
 198 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
 199 {
 200   g_return_if_fail (queue);
 201   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 202   g_return_if_fail (data);
 203
 204   g_queue_push_head (queue->queue, data);
 205   if (queue->waiting_threads > 0)
 206     g_cond_signal (queue->cond);
 207 }
 208
 209 static gpointer
 210 g_async_queue_pop_intern_unlocked (GAsyncQueue* queue, gboolean try,
 211                                    GTimeVal *end_time)
 212 {
 213   gpointer retval;
 214
 215   if (!g_queue_peek_tail (queue->queue))
 216     {
 217       if (try)
 218         return NULL;
 219
 220       if (!queue->cond)
 221         queue->cond = g_cond_new ();
 222
 223       if (!end_time)
 224         {
 225           queue->waiting_threads++;
 226           while (!g_queue_peek_tail (queue->queue))
 227             g_cond_wait(queue->cond, queue->mutex);
 228           queue->waiting_threads--;
 229         }
 230       else
 231         {
 232           queue->waiting_threads++;
 233           while (!g_queue_peek_tail (queue->queue))
 234             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
 235               break;
 236           queue->waiting_threads--;
 237           if (!g_queue_peek_tail (queue->queue))
 238             return NULL;
 239         }
 240     }
 241
 242   retval = g_queue_pop_tail (queue->queue);
 243
 244   g_assert (retval);
 245
 246   return retval;
 247 }
 248
 249 /**
 250  * g_async_queue_pop:
 251  * @queue: a #GAsyncQueue.
 252  *
 253  * Pops data from the @queue. This function blocks until data become
 254  * available.
 255  *
 256  * Return value: data from the queue.
 257  **/
 258 gpointer
 259 g_async_queue_pop (GAsyncQueue* queue)
 260 {
 261   gpointer retval;
 262
 263   g_return_val_if_fail (queue, NULL);
 264   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 265
 266   g_mutex_lock (queue->mutex);
 267   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 268   g_mutex_unlock (queue->mutex);
 269
 270   return retval;
 271 }
 272
 273 /**
 274  * g_async_queue_pop_unlocked:
 275  * @queue: a #GAsyncQueue.
 276  *
 277  * Pops data from the @queue. This function blocks until data become
 278  * available. This function must be called while holding the @queue's
 279  * lock.
 280  *
 281  * Return value: data from the queue.
 282  **/
 283 gpointer
 284 g_async_queue_pop_unlocked (GAsyncQueue* queue)
 285 {
 286   g_return_val_if_fail (queue, NULL);
 287   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 288
 289   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 290 }
 291
 292 /**
 293  * g_async_queue_try_pop:
 294  * @queue: a #GAsyncQueue.
 295  *
 296  * Tries to pop data from the @queue. If no data is available, %NULL is
 297  * returned.
 298  *
 299  * Return value: data from the queue or %NULL, when no data is
 300  * available immediately.
 301  **/
 302 gpointer
 303 g_async_queue_try_pop (GAsyncQueue* queue)
 304 {
 305   gpointer retval;
 306
 307   g_return_val_if_fail (queue, NULL);
 308   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 309
 310   g_mutex_lock (queue->mutex);
 311   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 312   g_mutex_unlock (queue->mutex);
 313
 314   return retval;
 315 }
 316
 317 /**
 318  * g_async_queue_try_pop_unlocked:
 319  * @queue: a #GAsyncQueue.
 320  *
 321  * Tries to pop data from the @queue. If no data is available, %NULL is
 322  * returned. This function must be called while holding the @queue's
 323  * lock.
 324  *
 325  * Return value: data from the queue or %NULL, when no data is
 326  * available immediately.
 327  **/
 328 gpointer
 329 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
 330 {
 331   g_return_val_if_fail (queue, NULL);
 332   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 333
 334   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 335 }
 336
 337 /**
 338  * g_async_queue_timed_pop:
 339  * @queue: a #GAsyncQueue.
 340  * @end_time: a #GTimeVal, determining the final time.
 341  *
 342  * Pops data from the @queue. If no data is received before @end_time,
 343  * %NULL is returned.
 344  *
 345  * To easily calculate @end_time a combination of g_get_current_time()
 346  * and g_time_val_add() can be used.
 347  *
 348  * Return value: data from the queue or %NULL, when no data is
 349  * received before @end_time.
 350  **/
 351 gpointer
 352 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
 353 {
 354   gpointer retval;
 355
 356   g_return_val_if_fail (queue, NULL);
 357   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 358
 359   g_mutex_lock (queue->mutex);
 360   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 361   g_mutex_unlock (queue->mutex);
 362
 363   return retval;
 364 }
 365
 366 /**
 367  * g_async_queue_timed_pop_unlocked:
 368  * @queue: a #GAsyncQueue.
 369  * @end_time: a #GTimeVal, determining the final time.
 370  *
 371  * Pops data from the @queue. If no data is received before @end_time,
 372  * %NULL is returned. This function must be called while holding the
 373  * @queue's lock.
 374  *
 375  * To easily calculate @end_time a combination of g_get_current_time()
 376  * and g_time_val_add() can be used.
 377  *
 378  * Return value: data from the queue or %NULL, when no data is
 379  * received before @end_time.
 380  **/
 381 gpointer
 382 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
 383 {
 384   g_return_val_if_fail (queue, NULL);
 385   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 386
 387   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 388 }
 389
 390 /**
 391  * g_async_queue_length:
 392  * @queue: a #GAsyncQueue.
 393  *
 394  * Returns the length of the queue, negative values mean waiting
 395  * threads, positive values mean available entries in the
 396  * @queue. Actually this function returns the number of data items in
 397  * the queue minus the number of waiting threads. Thus a return value
 398  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 399  * That can happen due to locking of the queue or due to
 400  * scheduling.
 401  *
 402  * Return value: the length of the @queue.
 403  **/
 404 gint
 405 g_async_queue_length (GAsyncQueue* queue)
 406 {
 407   gint retval;
 408
 409   g_return_val_if_fail (queue, 0);
 410   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 411
 412   g_mutex_lock (queue->mutex);
 413   retval = queue->queue->length - queue->waiting_threads;
 414   g_mutex_unlock (queue->mutex);
 415
 416   return retval;
 417 }
 418
 419 /**
 420  * g_async_queue_length_unlocked:
 421  * @queue: a #GAsyncQueue.
 422  *
 423  * Returns the length of the queue, negative values mean waiting
 424  * threads, positive values mean available entries in the
 425  * @queue. Actually this function returns the number of data items in
 426  * the queue minus the number of waiting threads. Thus a return value
 427  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 428  * That can happen due to locking of the queue or due to
 429  * scheduling. This function must be called while holding the @queue's
 430  * lock.
 431  *
 432  * Return value: the length of the @queue.
 433  **/
 434 gint
 435 g_async_queue_length_unlocked (GAsyncQueue* queue)
 436 {
 437   g_return_val_if_fail (queue, 0);
 438   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 439
 440   return queue->queue->length - queue->waiting_threads;
 441 }
 442