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   if (queue->waiting_threads > 0)
 312     g_cond_signal (queue->cond);
 313 }
 314
 315 static gpointer
 316 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue,
 317                                    gboolean     try,
 318                                    GTimeVal    *end_time)
 319 {
 320   gpointer retval;
 321
 322   if (!g_queue_peek_tail_link (queue->queue))
 323     {
 324       if (try)
 325         return NULL;
 326
 327       if (!queue->cond)
 328         queue->cond = g_cond_new ();
 329
 330       if (!end_time)
 331         {
 332           queue->waiting_threads++;
 333           while (!g_queue_peek_tail_link (queue->queue))
 334             g_cond_wait (queue->cond, queue->mutex);
 335           queue->waiting_threads--;
 336         }
 337       else
 338         {
 339           queue->waiting_threads++;
 340           while (!g_queue_peek_tail_link (queue->queue))
 341             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
 342               break;
 343           queue->waiting_threads--;
 344           if (!g_queue_peek_tail_link (queue->queue))
 345             return NULL;
 346         }
 347     }
 348
 349   retval = g_queue_pop_tail (queue->queue);
 350
 351   g_assert (retval);
 352
 353   return retval;
 354 }
 355
 356 /**
 357  * g_async_queue_pop:
 358  * @queue: a #GAsyncQueue.
 359  *
 360  * Pops data from the @queue. This function blocks until data become
 361  * available.
 362  *
 363  * Return value: data from the queue.
 364  **/
 365 gpointer
 366 g_async_queue_pop (GAsyncQueue* queue)
 367 {
 368   gpointer retval;
 369
 370   g_return_val_if_fail (queue, NULL);
 371   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 372
 373   g_mutex_lock (queue->mutex);
 374   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 375   g_mutex_unlock (queue->mutex);
 376
 377   return retval;
 378 }
 379
 380 /**
 381  * g_async_queue_pop_unlocked:
 382  * @queue: a #GAsyncQueue.
 383  *
 384  * Pops data from the @queue. This function blocks until data become
 385  * available. This function must be called while holding the @queue's
 386  * lock.
 387  *
 388  * Return value: data from the queue.
 389  **/
 390 gpointer
 391 g_async_queue_pop_unlocked (GAsyncQueue* queue)
 392 {
 393   g_return_val_if_fail (queue, NULL);
 394   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 395
 396   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
 397 }
 398
 399 /**
 400  * g_async_queue_try_pop:
 401  * @queue: a #GAsyncQueue.
 402  *
 403  * Tries to pop data from the @queue. If no data is available, %NULL is
 404  * returned.
 405  *
 406  * Return value: data from the queue or %NULL, when no data is
 407  * available immediately.
 408  **/
 409 gpointer
 410 g_async_queue_try_pop (GAsyncQueue* queue)
 411 {
 412   gpointer retval;
 413
 414   g_return_val_if_fail (queue, NULL);
 415   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 416
 417   g_mutex_lock (queue->mutex);
 418   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 419   g_mutex_unlock (queue->mutex);
 420
 421   return retval;
 422 }
 423
 424 /**
 425  * g_async_queue_try_pop_unlocked:
 426  * @queue: a #GAsyncQueue.
 427  *
 428  * Tries to pop data from the @queue. If no data is available, %NULL is
 429  * returned. This function must be called while holding the @queue's
 430  * lock.
 431  *
 432  * Return value: data from the queue or %NULL, when no data is
 433  * available immediately.
 434  **/
 435 gpointer
 436 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
 437 {
 438   g_return_val_if_fail (queue, NULL);
 439   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 440
 441   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
 442 }
 443
 444 /**
 445  * g_async_queue_timed_pop:
 446  * @queue: a #GAsyncQueue.
 447  * @end_time: a #GTimeVal, determining the final time.
 448  *
 449  * Pops data from the @queue. If no data is received before @end_time,
 450  * %NULL is returned.
 451  *
 452  * To easily calculate @end_time a combination of g_get_current_time()
 453  * and g_time_val_add() can be used.
 454  *
 455  * Return value: data from the queue or %NULL, when no data is
 456  * received before @end_time.
 457  **/
 458 gpointer
 459 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
 460 {
 461   gpointer retval;
 462
 463   g_return_val_if_fail (queue, NULL);
 464   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 465
 466   g_mutex_lock (queue->mutex);
 467   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 468   g_mutex_unlock (queue->mutex);
 469
 470   return retval;
 471 }
 472
 473 /**
 474  * g_async_queue_timed_pop_unlocked:
 475  * @queue: a #GAsyncQueue.
 476  * @end_time: a #GTimeVal, determining the final time.
 477  *
 478  * Pops data from the @queue. If no data is received before @end_time,
 479  * %NULL is returned. This function must be called while holding the
 480  * @queue's lock.
 481  *
 482  * To easily calculate @end_time a combination of g_get_current_time()
 483  * and g_time_val_add() can be used.
 484  *
 485  * Return value: data from the queue or %NULL, when no data is
 486  * received before @end_time.
 487  **/
 488 gpointer
 489 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
 490 {
 491   g_return_val_if_fail (queue, NULL);
 492   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 493
 494   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
 495 }
 496
 497 /**
 498  * g_async_queue_length:
 499  * @queue: a #GAsyncQueue.
 500  *
 501  * Returns the length of the queue, negative values mean waiting
 502  * threads, positive values mean available entries in the
 503  * @queue. Actually this function returns the number of data items in
 504  * the queue minus the number of waiting threads. Thus a return value
 505  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 506  * That can happen due to locking of the queue or due to
 507  * scheduling.
 508  *
 509  * Return value: the length of the @queue.
 510  **/
 511 gint
 512 g_async_queue_length (GAsyncQueue* queue)
 513 {
 514   gint retval;
 515
 516   g_return_val_if_fail (queue, 0);
 517   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 518
 519   g_mutex_lock (queue->mutex);
 520   retval = queue->queue->length - queue->waiting_threads;
 521   g_mutex_unlock (queue->mutex);
 522
 523   return retval;
 524 }
 525
 526 /**
 527  * g_async_queue_length_unlocked:
 528  * @queue: a #GAsyncQueue.
 529  *
 530  * Returns the length of the queue, negative values mean waiting
 531  * threads, positive values mean available entries in the
 532  * @queue. Actually this function returns the number of data items in
 533  * the queue minus the number of waiting threads. Thus a return value
 534  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.
 535  * That can happen due to locking of the queue or due to
 536  * scheduling. This function must be called while holding the @queue's
 537  * lock.
 538  *
 539  * Return value: the length of the @queue.
 540  **/
 541 gint
 542 g_async_queue_length_unlocked (GAsyncQueue* queue)
 543 {
 544   g_return_val_if_fail (queue, 0);
 545   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
 546
 547   return queue->queue->length - queue->waiting_threads;
 548 }
 549
 550 /**
 551  * g_async_queue_sort:
 552  * @queue: a #GAsyncQueue
 553  * @func: the #GCompareDataFunc is used to sort @queue. This
 554  *     function is passed two elements of the @queue. The function
 555  *     should return 0 if they are equal, a negative value if the
 556  *     first element should be higher in the @queue or a positive
 557  *     value if the first element should be lower in the @queue than
 558  *     the second element.
 559  * @user_data: user data passed to @func
 560  *
 561  * Sorts @queue using @func.
 562  *
 563  * This function will lock @queue before it sorts the queue and unlock
 564  * it when it is finished.
 565  *
 566  * If you were sorting a list of priority numbers to make sure the
 567  * lowest priority would be at the top of the queue, you could use:
 568  * <informalexample><programlisting>
 569  *  gint32 id1;
 570  *  gint32 id2;
 571  *
 572  *  id1 = GPOINTER_TO_INT (element1);
 573  *  id2 = GPOINTER_TO_INT (element2);
 574  *
 575  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 576  * </programlisting></informalexample>
 577  *
 578  * Since: 2.10
 579  **/
 580 void
 581 g_async_queue_sort (GAsyncQueue      *queue,
 582                     GCompareDataFunc  func,
 583                     gpointer          user_data)
 584 {
 585   g_return_if_fail (queue != NULL);
 586   g_return_if_fail (func != NULL);
 587
 588   g_mutex_lock (queue->mutex);
 589   g_async_queue_sort_unlocked (queue, func, user_data);
 590   g_mutex_unlock (queue->mutex);
 591 }
 592
 593 /**
 594  * g_async_queue_sort_unlocked:
 595  * @queue: a #GAsyncQueue
 596  * @func: the #GCompareDataFunc is used to sort @queue. This
 597  *     function is passed two elements of the @queue. The function
 598  *     should return 0 if they are equal, a negative value if the
 599  *     first element should be higher in the @queue or a positive
 600  *     value if the first element should be lower in the @queue than
 601  *     the second element.
 602  * @user_data: user data passed to @func
 603  *
 604  * Sorts @queue using @func.
 605  *
 606  * This function is called while holding the @queue's lock.
 607  *
 608  * Since: 2.10
 609  **/
 610 void
 611 g_async_queue_sort_unlocked (GAsyncQueue      *queue,
 612                              GCompareDataFunc  func,
 613                              gpointer          user_data)
 614 {
 615   SortData sd;
 616
 617   g_return_if_fail (queue != NULL);
 618   g_return_if_fail (func != NULL);
 619
 620   sd.func = func;
 621   sd.user_data = user_data;
 622
 623   g_queue_sort (queue->queue,
 624                 (GCompareDataFunc)g_async_queue_invert_compare,
 625                 &sd);
 626 }
 627
 628 /*
 629  * Private API
 630  */
 631
 632 GMutex*
 633 _g_async_queue_get_mutex (GAsyncQueue* queue)
 634 {
 635   g_return_val_if_fail (queue, NULL);
 636   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
 637
 638   return queue->mutex;
 639 }
 640
 641 #define __G_ASYNCQUEUE_C__
 642 #include "galiasdef.c"