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