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 #include "galias.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 typedef struct {
  43   GCompareDataFunc func;
  44   gpointer         user_data;
  45 } SortData;
  46
  47 /**
  48  * g_async_queue_new:
  49  *
  50  * Creates a new asynchronous queue with the initial reference count of 1.
  51  *
  52  * Return value: the new #GAsyncQueue.
  53  **/
  54 GAsyncQueue*
  55 g_async_queue_new (void)
  56 {
  57   GAsyncQueue* retval = g_new (GAsyncQueue, 1);
  58   retval->mutex = g_mutex_new ();
  59   retval->cond = NULL;
  60   retval->queue = g_queue_new ();
  61   retval->waiting_threads = 0;
  62   retval->ref_count = 1;
  63   return retval;
  64 }
  65
  66 /**
  67  * g_async_queue_ref:
  68  * @queue: a #GAsyncQueue.
  69  *
  70  * Increases the reference count of the asynchronous @queue by 1. You
  71  * do not need to hold the lock to call this function.
  72  *
  73  * Returns: the @queue that was passed in (since 2.6)
  74  **/
  75 GAsyncQueue *
  76 g_async_queue_ref (GAsyncQueue *queue)
  77 {
  78   g_return_val_if_fail (queue, NULL);
  79   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
  80
  81   g_atomic_int_inc (&queue->ref_count);
  82
  83   return queue;
  84 }
  85
  86 /**
  87  * g_async_queue_ref_unlocked:
  88  * @queue: a #GAsyncQueue.
  89  *
  90  * Increases the reference count of the asynchronous @queue by 1.
  91  *
  92  * @Deprecated: Since 2.8, reference counting is done atomically
  93  * so g_async_queue_ref() can be used regardless of the @queue's
  94  * lock.
  95  **/
  96 void
  97 g_async_queue_ref_unlocked (GAsyncQueue *queue)
  98 {
  99   g_return_if_fail (queue);
 100   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 101
 102   g_atomic_int_inc (&queue->ref_count);
 103 }
 104
 105 /**
 106  * g_async_queue_unref_and_unlock:
 107  * @queue: a #GAsyncQueue.
 108  *
 109  * Decreases the reference count of the asynchronous @queue by 1 and
 110  * releases the lock. This function must be called while holding the
 111  * @queue's lock. If the reference count went to 0, the @queue will be
 112  * destroyed and the memory allocated will be freed.
 113  *
 114  * @Deprecated: Since 2.8, reference counting is done atomically
 115  * so g_async_queue_unref() can be used regardless of the @queue's
 116  * lock.
 117  **/
 118 void
 119 g_async_queue_unref_and_unlock (GAsyncQueue *queue)
 120 {
 121   g_return_if_fail (queue);
 122   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 123
 124   g_mutex_unlock (queue->mutex);
 125   g_async_queue_unref (queue);
 126 }
 127
 128 /**
 129  * g_async_queue_unref:
 130  * @queue: a #GAsyncQueue.
 131  *
 132  * Decreases the reference count of the asynchronous @queue by 1. If
 133  * the reference count went to 0, the @queue will be destroyed and the
 134  * memory allocated will be freed. So you are not allowed to use the
 135  * @queue afterwards, as it might have disappeared. You do not need to
 136  * hold the lock to call this function.
 137  **/
 138 void
 139 g_async_queue_unref (GAsyncQueue *queue)
 140 {
 141   g_return_if_fail (queue);
 142   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 143
 144   if (g_atomic_int_dec_and_test (&queue->ref_count))
 145     {
 146       g_return_if_fail (queue->waiting_threads == 0);
 147       g_mutex_free (queue->mutex);
 148       if (queue->cond)
 149         g_cond_free (queue->cond);
 150       g_queue_free (queue->queue);
 151       g_free (queue);
 152     }
 153 }
 154
 155 /**
 156  * g_async_queue_lock:
 157  * @queue: a #GAsyncQueue.
 158  *
 159  * Acquires the @queue's lock. After that you can only call the
 160  * <function>g_async_queue_*_unlocked()</function> function variants on that
 161  * @queue. Otherwise it will deadlock.
 162  **/
 163 void
 164 g_async_queue_lock (GAsyncQueue *queue)
 165 {
 166   g_return_if_fail (queue);
 167   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 168
 169   g_mutex_lock (queue->mutex);
 170 }
 171
 172 /**
 173  * g_async_queue_unlock:
 174  * @queue: a #GAsyncQueue.
 175  *
 176  * Releases the queue's lock.
 177  **/
 178 void
 179 g_async_queue_unlock (GAsyncQueue *queue)
 180 {
 181   g_return_if_fail (queue);
 182   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 183
 184   g_mutex_unlock (queue->mutex);
 185 }
 186
 187 /**
 188  * g_async_queue_push:
 189  * @queue: a #GAsyncQueue.
 190  * @data: @data to push into the @queue.
 191  *
 192  * Pushes the @data into the @queue. @data must not be %NULL.
 193  **/
 194 void
 195 g_async_queue_push (GAsyncQueue* queue, gpointer data)
 196 {
 197   g_return_if_fail (queue);
 198   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 199   g_return_if_fail (data);
 200
 201   g_mutex_lock (queue->mutex);
 202   g_async_queue_push_unlocked (queue, data);
 203   g_mutex_unlock (queue->mutex);
 204 }
 205
 206 /**
 207  * g_async_queue_push_unlocked:
 208  * @queue: a #GAsyncQueue.
 209  * @data: @data to push into the @queue.
 210  *
 211  * Pushes the @data into the @queue. @data must not be %NULL. This
 212  * function must be called while holding the @queue's lock.
 213  **/
 214 void
 215 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
 216 {
 217   g_return_if_fail (queue);
 218   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
 219   g_return_if_fail (data);
 220
 221   g_queue_push_head (queue->queue, data);
 222   if (queue->waiting_threads > 0)
 223     g_cond_signal (queue->cond);
 224 }
 225
 226 /**
 227  * g_async_queue_push_sorted:
 228  * @queue: a #GAsyncQueue
 229  * @data: the @data to push into the @queue
 230  * @func: the #GCompareDataFunc is used to sort @queue. This function
 231  *     is passed two elements of the @queue. The function should return
 232  *     0 if they are equal, a negative value if the first element
 233  *     should be higher in the @queue or a positive value if the first
 234  *     element should be lower in the @queue than the second element.
 235  * @user_data: user data passed to @func.
 236  *
 237  * Inserts @data into @queue using @func to determine the new
 238  * position.
 239  *
 240  * This function requires that the @queue is sorted before pushing on
 241  * new elements.
 242  *
 243  * This function will lock @queue before it sorts the queue and unlock
 244  * it when it is finished.
 245  *
 246  * For an example of @func see g_async_queue_sort().
 247  *
 248  * Since: 2.10
 249  **/
 250 void
 251 g_async_queue_push_sorted (GAsyncQueue      *queue,
 252                            gpointer          data,
 253                            GCompareDataFunc  func,
 254                            gpointer          user_data)
 255 {
 256   g_return_if_fail (queue != NULL);
 257
 258   g_mutex_lock (queue->mutex);
 259   g_async_queue_push_sorted_unlocked (queue, data, func, user_data);
 260   g_mutex_unlock (queue->mutex);
 261 }
 262
 263 static gint
 264 g_async_queue_invert_compare (gpointer  v1,
 265                               gpointer  v2,
 266                               SortData *sd)
 267 {
 268   return -sd->func (v1, v2, sd->user_data);
 269 }
 270
 271 /**
 272  * g_async_queue_push_sorted_unlocked:
 273  * @queue: a #GAsyncQueue
 274  * @data: the @data to push into the @queue
 275  * @func: the #GCompareDataFunc is used to sort @queue. This function
 276  *     is passed two elements of the @queue. The function should return
 277  *     0 if they are equal, a negative value if the first element
 278  *     should be higher in the @queue or a positive value if the first
 279  *     element should be lower in the @queue than the second element.
 280  * @user_data: user data passed to @func.
 281  *
 282  * Inserts @data into @queue using @func to determine the new
 283  * position.
 284  *
 285  * This function requires that the @queue is sorted before pushing on
 286  * new elements.
 287  *
 288  * This function is called while holding the @queue's lock.
 289  *
 290  * For an example of @func see g_async_queue_sort().
 291  *
 292  * Since: 2.10
 293  **/
 294 void
 295 g_async_queue_push_sorted_unlocked (GAsyncQueue      *queue,
 296                                     gpointer          data,
 297                                     GCompareDataFunc  func,
 298                                     gpointer          user_data)
 299 {
 300   SortData sd;
 301
 302   g_return_if_fail (queue != NULL);
 303
 304   sd.func = func;
 305   sd.user_data = user_data;
 306
 307   g_queue_insert_sorted (queue->queue,
 308                          data,
 309                          (GCompareDataFunc)g_async_queue_invert_compare,
 310                          &sd);
 311 }
 312
 313 static gpointer
 314 g_async_queue_pop_intern_unlocked (GAsyncQueue* queue, gboolean try,
 315                                    GTimeVal *end_time)
 316 {
 317   gpointer retval;
 318
 319   if (!g_queue_peek_tail (queue->queue))
 320     {
 321       if (try)
 322         return NULL;
 323
 324       if (!queue->cond)
 325         queue->cond = g_cond_new ();
 326
 327       if (!end_time)
 328         {
 329           queue->waiting_threads++;
 330           while (!g_queue_peek_tail (queue->queue))
 331             g_cond_wait(queue->cond, queue->mutex);
 332           queue->waiting_threads--;
 333         }
 334       else
 335         {
 336           queue->waiting_threads++;
 337           while (!g_queue_peek_tail (queue->queue))
 338             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
 339               break;
 340           queue->waiting_threads--;
 341           if (!g_queue_peek_tail (queue->queue))
 342             return NULL;
 343         }
 344     }
 345
 346   retval = g_queue_pop_tail (queue->queue);
 347
 348   g_assert (retval);
 349
 350   return retval;
 351 }
 352
 353 /**
 354  * g_async_queue_pop:
 355  * @queue: a #GAsyncQueue.
 356  *
 357  * Pops data from the @queue. This function blocks until data become
 358  * available.
 359  *
 360  * Return value: data from the queue.
 361  **/
 362 gpointer
 363 g_async_queue_pop (GAsyncQueue* queue)
 364 {
 365   gpointer retval;
 366
 367   g_return_val_if_fail (queue, NULL);
 368   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 369
 370   g_mutex_lock (queue->mutex);
 371   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 372   g_mutex_unlock (queue->mutex);
 373
 374   return retval;
 375 }
 376
 377 /**
 378  * g_async_queue_pop_unlocked:
 379  * @queue: a #GAsyncQueue.
 380  *
 381  * Pops data from the @queue. This function blocks until data become
 382  * available. This function must be called while holding the @queue's
 383  * lock.
 384  *
 385  * Return value: data from the queue.
 386  **/
 387 gpointer
 388 g_async_queue_pop_unlocked (GAsyncQueue* queue)
 389 {
 390   g_return_val_if_fail (queue, NULL);
 391   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 392
 393   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 394 }
 395
 396 /**
 397  * g_async_queue_try_pop:
 398  * @queue: a #GAsyncQueue.
 399  *
 400  * Tries to pop data from the @queue. If no data is available, %NULL is
 401  * returned.
 402  *
 403  * Return value: data from the queue or %NULL, when no data is
 404  * available immediately.
 405  **/
 406 gpointer
 407 g_async_queue_try_pop (GAsyncQueue* queue)
 408 {
 409   gpointer retval;
 410
 411   g_return_val_if_fail (queue, NULL);
 412   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 413
 414   g_mutex_lock (queue->mutex);
 415   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 416   g_mutex_unlock (queue->mutex);
 417
 418   return retval;
 419 }
 420
 421 /**
 422  * g_async_queue_try_pop_unlocked:
 423  * @queue: a #GAsyncQueue.
 424  *
 425  * Tries to pop data from the @queue. If no data is available, %NULL is
 426  * returned. This function must be called while holding the @queue's
 427  * lock.
 428  *
 429  * Return value: data from the queue or %NULL, when no data is
 430  * available immediately.
 431  **/
 432 gpointer
 433 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
 434 {
 435   g_return_val_if_fail (queue, NULL);
 436   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 437
 438   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 439 }
 440
 441 /**
 442  * g_async_queue_timed_pop:
 443  * @queue: a #GAsyncQueue.
 444  * @end_time: a #GTimeVal, determining the final time.
 445  *
 446  * Pops data from the @queue. If no data is received before @end_time,
 447  * %NULL is returned.
 448  *
 449  * To easily calculate @end_time a combination of g_get_current_time()
 450  * and g_time_val_add() can be used.
 451  *
 452  * Return value: data from the queue or %NULL, when no data is
 453  * received before @end_time.
 454  **/
 455 gpointer
 456 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
 457 {
 458   gpointer retval;
 459
 460   g_return_val_if_fail (queue, NULL);
 461   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 462
 463   g_mutex_lock (queue->mutex);
 464   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 465   g_mutex_unlock (queue->mutex);
 466
 467   return retval;
 468 }
 469
 470 /**
 471  * g_async_queue_timed_pop_unlocked:
 472  * @queue: a #GAsyncQueue.
 473  * @end_time: a #GTimeVal, determining the final time.
 474  *
 475  * Pops data from the @queue. If no data is received before @end_time,
 476  * %NULL is returned. This function must be called while holding the
 477  * @queue's lock.
 478  *
 479  * To easily calculate @end_time a combination of g_get_current_time()
 480  * and g_time_val_add() can be used.
 481  *
 482  * Return value: data from the queue or %NULL, when no data is
 483  * received before @end_time.
 484  **/
 485 gpointer
 486 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
 487 {
 488   g_return_val_if_fail (queue, NULL);
 489   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 490
 491   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 492 }
 493
 494 /**
 495  * g_async_queue_length:
 496  * @queue: a #GAsyncQueue.
 497  *
 498  * Returns the length of the queue, negative values mean waiting
 499  * threads, positive values mean available entries in the
 500  * @queue. Actually this function returns the number of data items in
 501  * the queue minus the number of waiting threads. Thus a return value
 502  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 503  * That can happen due to locking of the queue or due to
 504  * scheduling.
 505  *
 506  * Return value: the length of the @queue.
 507  **/
 508 gint
 509 g_async_queue_length (GAsyncQueue* queue)
 510 {
 511   gint retval;
 512
 513   g_return_val_if_fail (queue, 0);
 514   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 515
 516   g_mutex_lock (queue->mutex);
 517   retval = queue->queue->length - queue->waiting_threads;
 518   g_mutex_unlock (queue->mutex);
 519
 520   return retval;
 521 }
 522
 523 /**
 524  * g_async_queue_length_unlocked:
 525  * @queue: a #GAsyncQueue.
 526  *
 527  * Returns the length of the queue, negative values mean waiting
 528  * threads, positive values mean available entries in the
 529  * @queue. Actually this function returns the number of data items in
 530  * the queue minus the number of waiting threads. Thus a return value
 531  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 532  * That can happen due to locking of the queue or due to
 533  * scheduling. This function must be called while holding the @queue's
 534  * lock.
 535  *
 536  * Return value: the length of the @queue.
 537  **/
 538 gint
 539 g_async_queue_length_unlocked (GAsyncQueue* queue)
 540 {
 541   g_return_val_if_fail (queue, 0);
 542   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 543
 544   return queue->queue->length - queue->waiting_threads;
 545 }
 546
 547 /**
 548  * g_async_queue_sort:
 549  * @queue: a #GAsyncQueue
 550  * @func: the #GCompareDataFunc is used to sort @queue. This
 551  *     function is passed two elements of the @queue. The function
 552  *     should return 0 if they are equal, a negative value if the
 553  *     first element should be higher in the @queue or a positive
 554  *     value if the first element should be lower in the @queue than
 555  *     the second element.
 556  * @user_data: user data passed to @func
 557  *
 558  * Sorts @queue using @func.
 559  *
 560  * This function will lock @queue before it sorts the queue and unlock
 561  * it when it is finished.
 562  *
 563  * If you were sorting a list of priority numbers to make sure the
 564  * lowest priority would be at the top of the queue, you could use:
 565  * <informalexample><programlisting>
 566  *  gint32 id1;
 567  *  gint32 id2;
 568  *
 569  *  id1 = GPOINTER_TO_INT (element1);
 570  *  id2 = GPOINTER_TO_INT (element2);
 571  *
 572  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 573  * </programlisting></informalexample>
 574  *
 575  * Since: 2.10
 576  **/
 577 void
 578 g_async_queue_sort (GAsyncQueue      *queue,
 579                     GCompareDataFunc  func,
 580                     gpointer          user_data)
 581 {
 582   g_return_if_fail (queue != NULL);
 583   g_return_if_fail (func != NULL);
 584
 585   g_mutex_lock (queue->mutex);
 586   g_async_queue_sort_unlocked (queue, func, user_data);
 587   g_mutex_unlock (queue->mutex);
 588 }
 589
 590 /**
 591  * g_async_queue_sort_unlocked:
 592  * @queue: a #GAsyncQueue
 593  * @func: the #GCompareDataFunc is used to sort @queue. This
 594  *     function is passed two elements of the @queue. The function
 595  *     should return 0 if they are equal, a negative value if the
 596  *     first element should be higher in the @queue or a positive
 597  *     value if the first element should be lower in the @queue than
 598  *     the second element.
 599  * @user_data: user data passed to @func
 600  *
 601  * Sorts @queue using @func.
 602  *
 603  * This function is called while holding the @queue's lock.
 604  *
 605  * Since: 2.10
 606  **/
 607 void
 608 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 609                              GCompareDataFunc  func,
 610                              gpointer          user_data)
 611 {
 612   SortData sd;
 613
 614   g_return_if_fail (queue != NULL);
 615   g_return_if_fail (func != NULL);
 616
 617   sd.func = func;
 618   sd.user_data = user_data;
 619
 620   g_queue_sort (queue->queue,
 621                 (GCompareDataFunc)g_async_queue_invert_compare,
 622                 &sd);
 623 }
 624
 625 #define __G_ASYNCQUEUE_C__
 626 #include "galiasdef.c"