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