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