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  * Portions copyright (c) 2006 Nokia Corporation.  All rights reserved.

     7  *

     8  * This library is free software; you can redistribute it and/or

     9  * modify it under the terms of the GNU Lesser General Public

    10  * License as published by the Free Software Foundation; either

    11  * version 2 of the License, or (at your option) any later version.

    12  *

    13  * This library is distributed in the hope that it will be useful,

    14  * but WITHOUT ANY WARRANTY; without even the implied warranty of

    15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

    16  * Lesser General Public License for more details.

    17  *

    18  * You should have received a copy of the GNU Lesser General Public

    19  * License along with this library; if not, write to the

    20  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,

    21  * Boston, MA 02111-1307, USA.

    22  */

    23 

    24 /*

    25  * MT safe

    26  */

    27 

    28 #include "config.h"

    29 

    30 #include "glib.h"

    31 #include "galias.h"

    32 

    33 #ifdef __SYMBIAN32__

    34 #include <glib_wsd.h>

    35 #endif /* __SYMBIAN32__ */

    36 

    37 #if EMULATOR

    38 #define g_thread_functions_for_glib_use (*_g_thread_functions_for_glib_use())

    39 #define g_thread_use_default_impl (*_g_thread_use_default_impl())

    40 #endif /* EMULATOR */

    41 

    42 struct _GAsyncQueue

    43 {

    44   GMutex *mutex;

    45   GCond *cond;

    46   GQueue *queue;

    47   guint waiting_threads;

    48   gint32 ref_count;

    49 };

    50 

    51 typedef struct {

    52   GCompareDataFunc func;

    53   gpointer         user_data;

    54 } SortData;

    55 

    56 /**

    57  * g_async_queue_new:

    58  * 

    59  * Creates a new asynchronous queue with the initial reference count of 1.

    60  * 

    61  * Return value: the new #GAsyncQueue.

    62  **/

    63 EXPORT_C GAsyncQueue*

    64 g_async_queue_new (void)

    65 {

    66   GAsyncQueue* retval = g_new (GAsyncQueue, 1);

    67   retval->mutex = g_mutex_new ();

    68   retval->cond = NULL;

    69   retval->queue = g_queue_new ();

    70   retval->waiting_threads = 0;

    71   retval->ref_count = 1;

    72   return retval;

    73 }

    74 

    75 /**

    76  * g_async_queue_ref:

    77  * @queue: a #GAsyncQueue.

    78  *

    79  * Increases the reference count of the asynchronous @queue by 1. You

    80  * do not need to hold the lock to call this function.

    81  *

    82  * Returns: the @queue that was passed in (since 2.6)

    83  **/

    84 EXPORT_C GAsyncQueue *

    85 g_async_queue_ref (GAsyncQueue *queue)

    86 {

    87   g_return_val_if_fail (queue, NULL);

    88   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

    89   

    90   g_atomic_int_inc (&queue->ref_count);

    91 

    92   return queue;

    93 }

    94 

    95 /**

    96  * g_async_queue_ref_unlocked:

    97  * @queue: a #GAsyncQueue.

    98  * 

    99  * Increases the reference count of the asynchronous @queue by 1.

   100  *

   101  * @Deprecated: Since 2.8, reference counting is done atomically

   102  * so g_async_queue_ref() can be used regardless of the @queue's

   103  * lock.

   104  **/

   105 EXPORT_C void 

   106 g_async_queue_ref_unlocked (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_atomic_int_inc (&queue->ref_count);

   112 }

   113 

   114 /**

   115  * g_async_queue_unref_and_unlock:

   116  * @queue: a #GAsyncQueue.

   117  * 

   118  * Decreases the reference count of the asynchronous @queue by 1 and

   119  * releases the lock. This function must be called while holding the

   120  * @queue's lock. If the reference count went to 0, the @queue will be

   121  * destroyed and the memory allocated will be freed.

   122  *

   123  * @Deprecated: Since 2.8, reference counting is done atomically

   124  * so g_async_queue_unref() can be used regardless of the @queue's

   125  * lock.

   126  **/

   127 EXPORT_C void 

   128 g_async_queue_unref_and_unlock (GAsyncQueue *queue)

   129 {

   130   g_return_if_fail (queue);

   131   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   132 

   133   g_mutex_unlock (queue->mutex);

   134   g_async_queue_unref (queue);

   135 }

   136 

   137 /**

   138  * g_async_queue_unref:

   139  * @queue: a #GAsyncQueue.

   140  * 

   141  * Decreases the reference count of the asynchronous @queue by 1. If

   142  * the reference count went to 0, the @queue will be destroyed and the

   143  * memory allocated will be freed. So you are not allowed to use the

   144  * @queue afterwards, as it might have disappeared. You do not need to

   145  * hold the lock to call this function.

   146  **/

   147 EXPORT_C void 

   148 g_async_queue_unref (GAsyncQueue *queue)

   149 {

   150   g_return_if_fail (queue);

   151   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   152   

   153   if (g_atomic_int_dec_and_test (&queue->ref_count))

   154     {

   155       g_return_if_fail (queue->waiting_threads == 0);

   156       g_mutex_free (queue->mutex);

   157       if (queue->cond)

   158 	g_cond_free (queue->cond);

   159       g_queue_free (queue->queue);

   160       g_free (queue);

   161     }

   162 }

   163 

   164 /**

   165  * g_async_queue_lock:

   166  * @queue: a #GAsyncQueue.

   167  * 

   168  * Acquires the @queue's lock. After that you can only call the

   169  * <function>g_async_queue_*_unlocked()</function> function variants on that

   170  * @queue. Otherwise it will deadlock.

   171  **/

   172 EXPORT_C void

   173 g_async_queue_lock (GAsyncQueue *queue)

   174 {

   175   g_return_if_fail (queue);

   176   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   177 

   178   g_mutex_lock (queue->mutex);

   179 }

   180 

   181 /**

   182  * g_async_queue_unlock:

   183  * @queue: a #GAsyncQueue.

   184  * 

   185  * Releases the queue's lock.

   186  **/

   187 EXPORT_C void 

   188 g_async_queue_unlock (GAsyncQueue *queue)

   189 {

   190   g_return_if_fail (queue);

   191   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   192 

   193   g_mutex_unlock (queue->mutex);

   194 }

   195 

   196 /**

   197  * g_async_queue_push:

   198  * @queue: a #GAsyncQueue.

   199  * @data: @data to push into the @queue.

   200  *

   201  * Pushes the @data into the @queue. @data must not be %NULL.

   202  **/

   203 EXPORT_C void

   204 g_async_queue_push (GAsyncQueue* queue, gpointer data)

   205 {

   206   g_return_if_fail (queue);

   207   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   208   g_return_if_fail (data);

   209 

   210   g_mutex_lock (queue->mutex);

   211   g_async_queue_push_unlocked (queue, data);

   212   g_mutex_unlock (queue->mutex);

   213 }

   214 

   215 /**

   216  * g_async_queue_push_unlocked:

   217  * @queue: a #GAsyncQueue.

   218  * @data: @data to push into the @queue.

   219  * 

   220  * Pushes the @data into the @queue. @data must not be %NULL. This

   221  * function must be called while holding the @queue's lock.

   222  **/

   223 EXPORT_C void

   224 g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)

   225 {

   226   g_return_if_fail (queue);

   227   g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);

   228   g_return_if_fail (data);

   229 

   230   g_queue_push_head (queue->queue, data);

   231   if (queue->waiting_threads > 0)

   232     g_cond_signal (queue->cond);

   233 }

   234 

   235 /**

   236  * g_async_queue_push_sorted:

   237  * @queue: a #GAsyncQueue

   238  * @data: the @data to push into the @queue

   239  * @func: the #GCompareDataFunc is used to sort @queue. This function

   240  *     is passed two elements of the @queue. The function should return

   241  *     0 if they are equal, a negative value if the first element

   242  *     should be higher in the @queue or a positive value if the first

   243  *     element should be lower in the @queue than the second element.

   244  * @user_data: user data passed to @func.

   245  * 

   246  * Inserts @data into @queue using @func to determine the new

   247  * position. 

   248  * 

   249  * This function requires that the @queue is sorted before pushing on

   250  * new elements.

   251  * 

   252  * This function will lock @queue before it sorts the queue and unlock

   253  * it when it is finished.

   254  * 

   255  * For an example of @func see g_async_queue_sort(). 

   256  *

   257  * Since: 2.10

   258  **/

   259 EXPORT_C void

   260 g_async_queue_push_sorted (GAsyncQueue      *queue,

   261 			   gpointer          data,

   262 			   GCompareDataFunc  func,

   263 			   gpointer          user_data)

   264 {

   265   g_return_if_fail (queue != NULL);

   266 

   267   g_mutex_lock (queue->mutex);

   268   g_async_queue_push_sorted_unlocked (queue, data, func, user_data);

   269   g_mutex_unlock (queue->mutex);

   270 }

   271 

   272 static gint 

   273 g_async_queue_invert_compare (gpointer  v1, 

   274 			      gpointer  v2, 

   275 			      SortData *sd)

   276 {

   277   return -sd->func (v1, v2, sd->user_data);

   278 }

   279 

   280 /**

   281  * g_async_queue_push_sorted_unlocked:

   282  * @queue: a #GAsyncQueue

   283  * @data: the @data to push into the @queue

   284  * @func: the #GCompareDataFunc is used to sort @queue. This function

   285  *     is passed two elements of the @queue. The function should return

   286  *     0 if they are equal, a negative value if the first element

   287  *     should be higher in the @queue or a positive value if the first

   288  *     element should be lower in the @queue than the second element.

   289  * @user_data: user data passed to @func.

   290  * 

   291  * Inserts @data into @queue using @func to determine the new

   292  * position.

   293  * 

   294  * This function requires that the @queue is sorted before pushing on

   295  * new elements.

   296  * 

   297  * This function is called while holding the @queue's lock.

   298  * 

   299  * For an example of @func see g_async_queue_sort(). 

   300  *

   301  * Since: 2.10

   302  **/

   303 EXPORT_C void

   304 g_async_queue_push_sorted_unlocked (GAsyncQueue      *queue,

   305 				    gpointer          data,

   306 				    GCompareDataFunc  func,

   307 				    gpointer          user_data)

   308 {

   309   SortData sd;

   310   

   311   g_return_if_fail (queue != NULL);

   312 

   313   sd.func = func;

   314   sd.user_data = user_data;

   315 

   316   g_queue_insert_sorted (queue->queue, 

   317 			 data, 

   318 			 (GCompareDataFunc)g_async_queue_invert_compare, 

   319 			 &sd);

   320   if (queue->waiting_threads > 0)

   321     g_cond_signal (queue->cond);

   322 }

   323 

   324 static gpointer

   325 g_async_queue_pop_intern_unlocked (GAsyncQueue *queue, 

   326 				   gboolean     try, 

   327 				   GTimeVal    *end_time)

   328 {

   329   gpointer retval;

   330 

   331   if (!g_queue_peek_tail_link (queue->queue))

   332     {

   333       if (try)

   334 	return NULL;

   335       

   336       if (!queue->cond)

   337 	queue->cond = g_cond_new ();

   338 

   339       if (!end_time)

   340         {

   341           queue->waiting_threads++;

   342 	  while (!g_queue_peek_tail_link (queue->queue))

   343             g_cond_wait (queue->cond, queue->mutex);

   344           queue->waiting_threads--;

   345         }

   346       else

   347         {

   348           queue->waiting_threads++;

   349           while (!g_queue_peek_tail_link (queue->queue))

   350             if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))

   351               break;

   352           queue->waiting_threads--;

   353           if (!g_queue_peek_tail_link (queue->queue))

   354 	    return NULL;

   355         }

   356     }

   357 

   358   retval = g_queue_pop_tail (queue->queue);

   359 

   360   g_assert (retval);

   361 

   362   return retval;

   363 }

   364 

   365 /**

   366  * g_async_queue_pop:

   367  * @queue: a #GAsyncQueue.

   368  * 

   369  * Pops data from the @queue. This function blocks until data become

   370  * available.

   371  *

   372  * Return value: data from the queue.

   373  **/

   374 EXPORT_C gpointer

   375 g_async_queue_pop (GAsyncQueue* queue)

   376 {

   377   gpointer retval;

   378 

   379   g_return_val_if_fail (queue, NULL);

   380   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   381 

   382   g_mutex_lock (queue->mutex);

   383   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);

   384   g_mutex_unlock (queue->mutex);

   385 

   386   return retval;

   387 }

   388 

   389 /**

   390  * g_async_queue_pop_unlocked:

   391  * @queue: a #GAsyncQueue.

   392  * 

   393  * Pops data from the @queue. This function blocks until data become

   394  * available. This function must be called while holding the @queue's

   395  * lock.

   396  *

   397  * Return value: data from the queue.

   398  **/

   399 EXPORT_C gpointer

   400 g_async_queue_pop_unlocked (GAsyncQueue* queue)

   401 {

   402   g_return_val_if_fail (queue, NULL);

   403   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   404 

   405   return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);

   406 }

   407 

   408 /**

   409  * g_async_queue_try_pop:

   410  * @queue: a #GAsyncQueue.

   411  * 

   412  * Tries to pop data from the @queue. If no data is available, %NULL is

   413  * returned.

   414  *

   415  * Return value: data from the queue or %NULL, when no data is

   416  * available immediately.

   417  **/

   418 EXPORT_C gpointer

   419 g_async_queue_try_pop (GAsyncQueue* queue)

   420 {

   421   gpointer retval;

   422 

   423   g_return_val_if_fail (queue, NULL);

   424   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   425 

   426   g_mutex_lock (queue->mutex);

   427   retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);

   428   g_mutex_unlock (queue->mutex);

   429 

   430   return retval;

   431 }

   432 

   433 /**

   434  * g_async_queue_try_pop_unlocked:

   435  * @queue: a #GAsyncQueue.

   436  * 

   437  * Tries to pop data from the @queue. If no data is available, %NULL is

   438  * returned. This function must be called while holding the @queue's

   439  * lock.

   440  *

   441  * Return value: data from the queue or %NULL, when no data is

   442  * available immediately.

   443  **/

   444 EXPORT_C gpointer

   445 g_async_queue_try_pop_unlocked (GAsyncQueue* queue)

   446 {

   447   g_return_val_if_fail (queue, NULL);

   448   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   449 

   450   return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);

   451 }

   452 

   453 /**

   454  * g_async_queue_timed_pop:

   455  * @queue: a #GAsyncQueue.

   456  * @end_time: a #GTimeVal, determining the final time.

   457  *

   458  * Pops data from the @queue. If no data is received before @end_time,

   459  * %NULL is returned.

   460  *

   461  * To easily calculate @end_time a combination of g_get_current_time()

   462  * and g_time_val_add() can be used.

   463  *

   464  * Return value: data from the queue or %NULL, when no data is

   465  * received before @end_time.

   466  **/

   467 EXPORT_C gpointer

   468 g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)

   469 {

   470   gpointer retval;

   471 

   472   g_return_val_if_fail (queue, NULL);

   473   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   474 

   475   g_mutex_lock (queue->mutex);

   476   retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);

   477   g_mutex_unlock (queue->mutex);

   478 

   479   return retval;  

   480 }

   481 

   482 /**

   483  * g_async_queue_timed_pop_unlocked:

   484  * @queue: a #GAsyncQueue.

   485  * @end_time: a #GTimeVal, determining the final time.

   486  *

   487  * Pops data from the @queue. If no data is received before @end_time,

   488  * %NULL is returned. This function must be called while holding the

   489  * @queue's lock.

   490  *

   491  * To easily calculate @end_time a combination of g_get_current_time()

   492  * and g_time_val_add() can be used.

   493  *

   494  * Return value: data from the queue or %NULL, when no data is

   495  * received before @end_time.

   496  **/

   497 EXPORT_C gpointer

   498 g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)

   499 {

   500   g_return_val_if_fail (queue, NULL);

   501   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   502 

   503   return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);

   504 }

   505 

   506 /**

   507  * g_async_queue_length:

   508  * @queue: a #GAsyncQueue.

   509  * 

   510  * Returns the length of the queue, negative values mean waiting

   511  * threads, positive values mean available entries in the

   512  * @queue. Actually this function returns the number of data items in

   513  * the queue minus the number of waiting threads. Thus a return value

   514  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.

   515  * That can happen due to locking of the queue or due to

   516  * scheduling.  

   517  *

   518  * Return value: the length of the @queue.

   519  **/

   520 EXPORT_C gint

   521 g_async_queue_length (GAsyncQueue* queue)

   522 {

   523   gint retval;

   524 

   525   g_return_val_if_fail (queue, 0);

   526   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);

   527 

   528   g_mutex_lock (queue->mutex);

   529   retval = queue->queue->length - queue->waiting_threads;

   530   g_mutex_unlock (queue->mutex);

   531 

   532   return retval;

   533 }

   534 

   535 /**

   536  * g_async_queue_length_unlocked:

   537  * @queue: a #GAsyncQueue.

   538  * 

   539  * Returns the length of the queue, negative values mean waiting

   540  * threads, positive values mean available entries in the

   541  * @queue. Actually this function returns the number of data items in

   542  * the queue minus the number of waiting threads. Thus a return value

   543  * of 0 could mean 'n' entries in the queue and 'n' thread waiting.

   544  * That can happen due to locking of the queue or due to

   545  * scheduling. This function must be called while holding the @queue's

   546  * lock.

   547  *

   548  * Return value: the length of the @queue.

   549  **/

   550 EXPORT_C gint

   551 g_async_queue_length_unlocked (GAsyncQueue* queue)

   552 {

   553   g_return_val_if_fail (queue, 0);

   554   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);

   555 

   556   return queue->queue->length - queue->waiting_threads;

   557 }

   558 

   559 /**

   560  * g_async_queue_sort:

   561  * @queue: a #GAsyncQueue

   562  * @func: the #GCompareDataFunc is used to sort @queue. This

   563  *     function is passed two elements of the @queue. The function

   564  *     should return 0 if they are equal, a negative value if the

   565  *     first element should be higher in the @queue or a positive

   566  *     value if the first element should be lower in the @queue than

   567  *     the second element. 

   568  * @user_data: user data passed to @func

   569  *

   570  * Sorts @queue using @func. 

   571  *

   572  * This function will lock @queue before it sorts the queue and unlock

   573  * it when it is finished.

   574  *

   575  * If you were sorting a list of priority numbers to make sure the

   576  * lowest priority would be at the top of the queue, you could use:

   577  * <informalexample><programlisting> 

   578  *  gint32 id1;

   579  *  gint32 id2;

   580  *   

   581  *  id1 = GPOINTER_TO_INT (element1);

   582  *  id2 = GPOINTER_TO_INT (element2);

   583  *   

   584  *  return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);

   585  * </programlisting></informalexample>

   586  *

   587  * Since: 2.10

   588  **/

   589 EXPORT_C void

   590 g_async_queue_sort (GAsyncQueue      *queue,

   591 		    GCompareDataFunc  func,

   592 		    gpointer          user_data)

   593 {

   594   g_return_if_fail (queue != NULL);

   595   g_return_if_fail (func != NULL);

   596 

   597   g_mutex_lock (queue->mutex);

   598   g_async_queue_sort_unlocked (queue, func, user_data);

   599   g_mutex_unlock (queue->mutex);

   600 }

   601 

   602 /**

   603  * g_async_queue_sort_unlocked:

   604  * @queue: a #GAsyncQueue

   605  * @func: the #GCompareDataFunc is used to sort @queue. This

   606  *     function is passed two elements of the @queue. The function

   607  *     should return 0 if they are equal, a negative value if the

   608  *     first element should be higher in the @queue or a positive

   609  *     value if the first element should be lower in the @queue than

   610  *     the second element. 

   611  * @user_data: user data passed to @func

   612  *

   613  * Sorts @queue using @func. 

   614  *

   615  * This function is called while holding the @queue's lock.

   616  * 

   617  * Since: 2.10

   618  **/

   619 EXPORT_C void

   620 g_async_queue_sort_unlocked (GAsyncQueue      *queue,

   621 			     GCompareDataFunc  func,

   622 			     gpointer          user_data)

   623 {

   624   SortData sd;

   625 

   626   g_return_if_fail (queue != NULL);

   627   g_return_if_fail (func != NULL);

   628 

   629   sd.func = func;

   630   sd.user_data = user_data;

   631 

   632   g_queue_sort (queue->queue, 

   633 		(GCompareDataFunc)g_async_queue_invert_compare, 

   634 		&sd);

   635 }

   636 

   637 /*

   638  * Private API

   639  */

   640 

   641 GMutex*

   642 _g_async_queue_get_mutex (GAsyncQueue* queue)

   643 {

   644   g_return_val_if_fail (queue, NULL);

   645   g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);

   646 

   647   return queue->mutex;

   648 }

   649 

   650 #define __G_ASYNCQUEUE_C__

   651 #include "galiasdef.c"