tests/asyncqueue-test.c

   1 #undef G_DISABLE_ASSERT
   2 #undef G_LOG_DOMAIN
   3
   4 #ifdef HAVE_CONFIG_H
   5 #  include <config.h>
   6 #endif
   7 #include <glib.h>
   8
   9 #include <time.h>
  10 #include <stdlib.h>
  11
  12 #define d(x) x
  13
  14 #define MAX_THREADS            50
  15 #define MAX_SORTS              5    /* only applies if
  16                                        ASYC_QUEUE_DO_SORT is set to 1 */
  17 #define MAX_TIME               20   /* seconds */
  18 #define MIN_TIME               5    /* seconds */
  19
  20 #define SORT_QUEUE_AFTER       1
  21 #define SORT_QUEUE_ON_PUSH     1    /* if this is done, the
  22                                        SORT_QUEUE_AFTER is ignored */
  23 #define QUIT_WHEN_DONE         1
  24
  25
  26 #if SORT_QUEUE_ON_PUSH == 1
  27 #  undef SORT_QUEUE_AFTER
  28 #  define SORT_QUEUE_AFTER     0
  29 #endif
  30
  31
  32 static GMainLoop *main_loop = NULL;
  33 static GThreadPool *thread_pool = NULL;
  34 static GAsyncQueue *async_queue = NULL;
  35
  36
  37 static gint
  38 sort_compare (gconstpointer p1, gconstpointer p2, gpointer user_data)
  39 {
  40   gint32 id1;
  41   gint32 id2;
  42
  43   id1 = GPOINTER_TO_INT (p1);
  44   id2 = GPOINTER_TO_INT (p2);
  45
  46   d(g_print ("comparing #1:%d and #2:%d, returning %d\n",
  47              id1, id2, (id1 > id2 ? +1 : id1 == id2 ? 0 : -1)));
  48
  49   return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
  50 }
  51
  52 static gboolean
  53 sort_queue (gpointer user_data)
  54 {
  55   static gint sorts = 0;
  56   gboolean can_quit = FALSE;
  57   gint sort_multiplier;
  58   gint len;
  59   gint i;
  60
  61   sort_multiplier = GPOINTER_TO_INT (user_data);
  62
  63   if (SORT_QUEUE_AFTER) {
  64     d(g_print ("sorting async queue...\n"));
  65     g_async_queue_sort (async_queue, sort_compare, NULL);
  66
  67     sorts++;
  68
  69     if (sorts >= sort_multiplier) {
  70       can_quit = TRUE;
  71     }
  72
  73     g_async_queue_sort (async_queue, sort_compare, NULL);
  74     len = g_async_queue_length (async_queue);
  75
  76     d(g_print ("sorted queue (for %d/%d times, size:%d)...\n", sorts, MAX_SORTS, len));
  77   } else {
  78     can_quit = TRUE;
  79     len = g_async_queue_length (async_queue);
  80     d(g_print ("printing queue (size:%d)...\n", len));
  81   }
  82
  83   for (i = 0; i < len; i++) {
  84     gpointer p;
  85
  86     p = g_async_queue_pop (async_queue);
  87     d(g_print ("item %d ---> %d\n", i, GPOINTER_TO_INT (p)));
  88   }
  89
  90   if (can_quit && QUIT_WHEN_DONE) {
  91     g_main_loop_quit (main_loop);
  92   }
  93
  94   return !can_quit;
  95 }
  96
  97 static void
  98 enter_thread (gpointer data, gpointer user_data)
  99 {
 100   gint   len;
 101   gint   id;
 102   gulong ms;
 103
 104   id = GPOINTER_TO_INT (data);
 105
 106   ms = g_random_int_range (MIN_TIME * 1000, MAX_TIME * 1000);
 107   d(g_print ("entered thread with id:%d, adding to queue in:%ld ms\n", id, ms));
 108
 109   g_usleep (ms * 1000);
 110
 111   if (SORT_QUEUE_ON_PUSH) {
 112     g_async_queue_push_sorted (async_queue, GINT_TO_POINTER (id), sort_compare, NULL);
 113   } else {
 114     g_async_queue_push (async_queue, GINT_TO_POINTER (id));
 115   }
 116
 117   len = g_async_queue_length (async_queue);
 118
 119   d(g_print ("thread id:%d added to async queue (size:%d)\n",
 120              id, len));
 121 }
 122
 123 int main (int argc, char *argv[])
 124 {
 125 #if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE)
 126   gint i;
 127   gint max_threads = MAX_THREADS;
 128   gint max_unused_threads = MAX_THREADS;
 129   gint sort_multiplier = MAX_SORTS;
 130   gint sort_interval;
 131
 132   g_thread_init (NULL);
 133
 134   d(g_print ("creating async queue...\n"));
 135   async_queue = g_async_queue_new ();
 136
 137   g_return_val_if_fail (async_queue != NULL, EXIT_FAILURE);
 138
 139   d(g_print ("creating thread pool with max threads:%d, max unused threads:%d...\n",
 140              max_threads, max_unused_threads));
 141   thread_pool = g_thread_pool_new (enter_thread,
 142                                    async_queue,
 143                                    max_threads,
 144                                    FALSE,
 145                                    NULL);
 146
 147   g_return_val_if_fail (thread_pool != NULL, EXIT_FAILURE);
 148
 149   g_thread_pool_set_max_unused_threads (max_unused_threads);
 150
 151   d(g_print ("creating threads...\n"));
 152   for (i = 0; i <= max_threads; i++) {
 153     GError *error = NULL;
 154
 155     g_thread_pool_push (thread_pool, GINT_TO_POINTER (i), &error);
 156
 157     g_assert (error == NULL);
 158   }
 159
 160   if (!SORT_QUEUE_AFTER) {
 161     sort_multiplier = 1;
 162   }
 163
 164   sort_interval = ((MAX_TIME / sort_multiplier) + 2)  * 1000;
 165   d(g_print ("adding timeout of %d seconds to sort %d times\n",
 166              sort_interval, sort_multiplier));
 167   g_timeout_add (sort_interval, sort_queue, GINT_TO_POINTER (sort_multiplier));
 168
 169   if (SORT_QUEUE_ON_PUSH) {
 170     d(g_print ("sorting when pushing into the queue...\n"));
 171   }
 172
 173   d(g_print ("entering main event loop\n"));
 174
 175   main_loop = g_main_loop_new (NULL, FALSE);
 176   g_main_loop_run (main_loop);
 177 #endif
 178
 179   return EXIT_SUCCESS;
 180 }