tests/asyncqueue-test.c

   1 #undef G_DISABLE_ASSERT
   2 #undef G_LOG_DOMAIN
   3
   4 #include <time.h>
   5 #include <stdlib.h>
   6
   7 #include <glib.h>
   8
   9 #define DEBUG_MSG(args)
  10 /* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n");  */
  11 #define PRINT_MSG(args)
  12 /* #define PRINT_MSG(args) g_print args ; g_print ("\n"); */
  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   DEBUG_MSG (("comparing #1:%d and #2:%d, returning %d",
  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   static gpointer last_p = NULL;
  57   gpointer        p;
  58   gboolean        can_quit = FALSE;
  59   gint            sort_multiplier;
  60   gint            len;
  61   gint            i;
  62
  63   sort_multiplier = GPOINTER_TO_INT (user_data);
  64
  65   if (SORT_QUEUE_AFTER) {
  66     PRINT_MSG (("sorting async queue..."));
  67     g_async_queue_sort (async_queue, sort_compare, NULL);
  68
  69     sorts++;
  70
  71     if (sorts >= sort_multiplier) {
  72       can_quit = TRUE;
  73     }
  74
  75     g_async_queue_sort (async_queue, sort_compare, NULL);
  76     len = g_async_queue_length (async_queue);
  77
  78     PRINT_MSG (("sorted queue (for %d/%d times, size:%d)...", sorts, MAX_SORTS, len));
  79   } else {
  80     can_quit = TRUE;
  81     len = g_async_queue_length (async_queue);
  82     DEBUG_MSG (("printing queue (size:%d)...", len));
  83   }
  84
  85   for (i = 0, last_p = NULL; i < len; i++) {
  86     p = g_async_queue_pop (async_queue);
  87     DEBUG_MSG (("item %d ---> %d", i, GPOINTER_TO_INT (p)));
  88
  89     if (last_p) {
  90       g_assert (GPOINTER_TO_INT (last_p) <= GPOINTER_TO_INT (p));
  91     }
  92
  93     last_p = p;
  94   }
  95
  96   if (can_quit && QUIT_WHEN_DONE) {
  97     g_main_loop_quit (main_loop);
  98   }
  99
 100   return !can_quit;
 101 }
 102
 103 static void
 104 enter_thread (gpointer data, gpointer user_data)
 105 {
 106   gint   len;
 107   gint   id;
 108   gulong ms;
 109
 110   id = GPOINTER_TO_INT (data);
 111
 112   ms = g_random_int_range (MIN_TIME * 1000, MAX_TIME * 1000);
 113   DEBUG_MSG (("entered thread with id:%d, adding to queue in:%ld ms", id, ms));
 114
 115   g_usleep (ms * 1000);
 116
 117   if (SORT_QUEUE_ON_PUSH) {
 118     g_async_queue_push_sorted (async_queue, GINT_TO_POINTER (id), sort_compare, NULL);
 119   } else {
 120     g_async_queue_push (async_queue, GINT_TO_POINTER (id));
 121   }
 122
 123   len = g_async_queue_length (async_queue);
 124
 125   DEBUG_MSG (("thread id:%d added to async queue (size:%d)",
 126              id, len));
 127 }
 128
 129 int
 130 main (int argc, char *argv[])
 131 {
 132 #if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE)
 133   gint   i;
 134   gint   max_threads = MAX_THREADS;
 135   gint   max_unused_threads = MAX_THREADS;
 136   gint   sort_multiplier = MAX_SORTS;
 137   gint   sort_interval;
 138   gchar *msg;
 139
 140   g_thread_init (NULL);
 141
 142   PRINT_MSG (("creating async queue..."));
 143   async_queue = g_async_queue_new ();
 144
 145   g_return_val_if_fail (async_queue != NULL, EXIT_FAILURE);
 146
 147   PRINT_MSG (("creating thread pool with max threads:%d, max unused threads:%d...",
 148              max_threads, max_unused_threads));
 149   thread_pool = g_thread_pool_new (enter_thread,
 150                                    async_queue,
 151                                    max_threads,
 152                                    FALSE,
 153                                    NULL);
 154
 155   g_return_val_if_fail (thread_pool != NULL, EXIT_FAILURE);
 156
 157   g_thread_pool_set_max_unused_threads (max_unused_threads);
 158
 159   PRINT_MSG (("creating threads..."));
 160   for (i = 1; i <= max_threads; i++) {
 161     GError *error = NULL;
 162
 163     g_thread_pool_push (thread_pool, GINT_TO_POINTER (i), &error);
 164
 165     g_assert_no_error (error);
 166   }
 167
 168   if (!SORT_QUEUE_AFTER) {
 169     sort_multiplier = 1;
 170   }
 171
 172   sort_interval = ((MAX_TIME / sort_multiplier) + 2)  * 1000;
 173   g_timeout_add (sort_interval, sort_queue, GINT_TO_POINTER (sort_multiplier));
 174
 175   if (SORT_QUEUE_ON_PUSH) {
 176     msg = "sorting when pushing into the queue, checking queue is sorted";
 177   } else {
 178     msg = "sorting";
 179   }
 180
 181   PRINT_MSG (("%s %d %s %d ms",
 182               msg,
 183               sort_multiplier,
 184               sort_multiplier == 1 ? "time in" : "times, once every",
 185               sort_interval));
 186
 187   DEBUG_MSG (("entering main event loop"));
 188
 189   main_loop = g_main_loop_new (NULL, FALSE);
 190   g_main_loop_run (main_loop);
 191 #endif
 192
 193   return EXIT_SUCCESS;
 194 }