tests/threadpool-test.c

   1 #undef G_DISABLE_ASSERT
   2 #undef G_LOG_DOMAIN
   3
   4 #include "config.h"
   5
   6 #include <glib.h>
   7
   8 #define DEBUG_MSG(x)
   9 /* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n");  */
  10
  11 #define WAIT                5    /* seconds */
  12 #define MAX_THREADS         10
  13
  14 /* if > 0 the test will run continuously (since the test ends when
  15  * thread count is 0), if -1 it means no limit to unused threads
  16  * if 0 then no unused threads are possible */
  17 #define MAX_UNUSED_THREADS -1
  18
  19 G_LOCK_DEFINE_STATIC (thread_counter_pools);
  20
  21 static gulong abs_thread_counter = 0;
  22 static gulong running_thread_counter = 0;
  23 static gulong leftover_task_counter = 0;
  24
  25 G_LOCK_DEFINE_STATIC (last_thread);
  26
  27 static guint last_thread_id = 0;
  28
  29 G_LOCK_DEFINE_STATIC (thread_counter_sort);
  30
  31 static gulong sort_thread_counter = 0;
  32
  33 static GThreadPool *idle_pool = NULL;
  34
  35 static GMainLoop *main_loop = NULL;
  36
  37 static void
  38 test_thread_functions (void)
  39 {
  40   gint max_unused_threads;
  41   guint max_idle_time;
  42
  43   /* This function attempts to call functions which don't need a
  44    * threadpool to operate to make sure no uninitialised pointers
  45    * accessed and no errors occur.
  46    */
  47
  48   max_unused_threads = 3;
  49
  50   DEBUG_MSG (("[funcs] Setting max unused threads to %d",
  51               max_unused_threads));
  52   g_thread_pool_set_max_unused_threads (max_unused_threads);
  53
  54   DEBUG_MSG (("[funcs] Getting max unused threads = %d",
  55              g_thread_pool_get_max_unused_threads ()));
  56   g_assert (g_thread_pool_get_max_unused_threads() == max_unused_threads);
  57
  58   DEBUG_MSG (("[funcs] Getting num unused threads = %d",
  59              g_thread_pool_get_num_unused_threads ()));
  60   g_assert (g_thread_pool_get_num_unused_threads () == 0);
  61
  62   DEBUG_MSG (("[funcs] Stopping unused threads"));
  63   g_thread_pool_stop_unused_threads ();
  64
  65   max_idle_time = 10 * G_USEC_PER_SEC;
  66
  67   DEBUG_MSG (("[funcs] Setting max idle time to %d",
  68               max_idle_time));
  69   g_thread_pool_set_max_idle_time (max_idle_time);
  70
  71   DEBUG_MSG (("[funcs] Getting max idle time = %d",
  72              g_thread_pool_get_max_idle_time ()));
  73   g_assert (g_thread_pool_get_max_idle_time () == max_idle_time);
  74
  75   DEBUG_MSG (("[funcs] Setting max idle time to 0"));
  76   g_thread_pool_set_max_idle_time (0);
  77
  78   DEBUG_MSG (("[funcs] Getting max idle time = %d",
  79              g_thread_pool_get_max_idle_time ()));
  80   g_assert (g_thread_pool_get_max_idle_time () == 0);
  81 }
  82
  83 static void
  84 test_count_threads_foreach (GThread *thread,
  85                             guint   *count)
  86 {
  87    ++*count;
  88 }
  89
  90 static guint
  91 test_count_threads (void)
  92 {
  93   guint count = 0;
  94
  95   g_thread_foreach ((GFunc) test_count_threads_foreach, &count);
  96
  97   /* Exclude main thread */
  98   return count - 1;
  99 }
 100
 101 static void
 102 test_thread_stop_unused (void)
 103 {
 104    GThreadPool *pool;
 105    guint i;
 106    guint limit = 100;
 107
 108    /* Spawn a few threads. */
 109    g_thread_pool_set_max_unused_threads (-1);
 110    pool = g_thread_pool_new ((GFunc) g_usleep, NULL, -1, FALSE, NULL);
 111
 112    for (i = 0; i < limit; i++)
 113      g_thread_pool_push (pool, GUINT_TO_POINTER (1000), NULL);
 114
 115    DEBUG_MSG (("[unused] ===> pushed %d threads onto the idle pool",
 116                limit));
 117
 118    /* Wait for the threads to migrate. */
 119    g_usleep (G_USEC_PER_SEC);
 120
 121    DEBUG_MSG (("[unused] current threads %d",
 122                test_count_threads()));
 123
 124    DEBUG_MSG (("[unused] stopping unused threads"));
 125    g_thread_pool_stop_unused_threads ();
 126
 127    for (i = 0; i < 5; i++)
 128      {
 129        if (g_thread_pool_get_num_unused_threads () == 0 &&
 130            test_count_threads () == 0)
 131          break;
 132
 133        DEBUG_MSG (("[unused] waiting ONE second for threads to die"));
 134
 135        /* Some time for threads to die. */
 136        g_usleep (G_USEC_PER_SEC);
 137      }
 138
 139    DEBUG_MSG (("[unused] stopped idle threads, %d remain, %d threads still exist",
 140                g_thread_pool_get_num_unused_threads (),
 141                test_count_threads ()));
 142
 143    g_assert (g_thread_pool_get_num_unused_threads () == test_count_threads ());
 144    g_assert (g_thread_pool_get_num_unused_threads () == 0);
 145
 146    g_thread_pool_set_max_unused_threads (MAX_THREADS);
 147
 148    DEBUG_MSG (("[unused] cleaning up thread pool"));
 149    g_thread_pool_free (pool, FALSE, TRUE);
 150 }
 151
 152 static void
 153 test_thread_pools_entry_func (gpointer data, gpointer user_data)
 154 {
 155   guint id = 0;
 156
 157   id = GPOINTER_TO_UINT (data);
 158
 159   DEBUG_MSG (("[pool] ---> [%3.3d] entered thread.", id));
 160
 161   G_LOCK (thread_counter_pools);
 162   abs_thread_counter++;
 163   running_thread_counter++;
 164   G_UNLOCK (thread_counter_pools);
 165
 166   g_usleep (g_random_int_range (0, 4000));
 167
 168   G_LOCK (thread_counter_pools);
 169   running_thread_counter--;
 170   leftover_task_counter--;
 171
 172   DEBUG_MSG (("[pool] ---> [%3.3d] exiting thread (abs count:%ld, "
 173               "running count:%ld, left over:%ld)",
 174               id, abs_thread_counter,
 175               running_thread_counter, leftover_task_counter));
 176   G_UNLOCK (thread_counter_pools);
 177 }
 178
 179 static void
 180 test_thread_pools (void)
 181 {
 182   GThreadPool *pool1, *pool2, *pool3;
 183   guint runs;
 184   guint i;
 185
 186   pool1 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 3, FALSE, NULL);
 187   pool2 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 5, TRUE, NULL);
 188   pool3 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 7, TRUE, NULL);
 189
 190   runs = 300;
 191   for (i = 0; i < runs; i++)
 192     {
 193       g_thread_pool_push (pool1, GUINT_TO_POINTER (i + 1), NULL);
 194       g_thread_pool_push (pool2, GUINT_TO_POINTER (i + 1), NULL);
 195       g_thread_pool_push (pool3, GUINT_TO_POINTER (i + 1), NULL);
 196
 197       G_LOCK (thread_counter_pools);
 198       leftover_task_counter += 3;
 199       G_UNLOCK (thread_counter_pools);
 200     }
 201
 202   g_thread_pool_free (pool1, TRUE, TRUE);
 203   g_thread_pool_free (pool2, FALSE, TRUE);
 204   g_thread_pool_free (pool3, FALSE, TRUE);
 205
 206   g_assert (runs * 3 == abs_thread_counter + leftover_task_counter);
 207   g_assert (running_thread_counter == 0);
 208 }
 209
 210 static gint
 211 test_thread_sort_compare_func (gconstpointer a, gconstpointer b, gpointer user_data)
 212 {
 213   guint32 id1, id2;
 214
 215   id1 = GPOINTER_TO_UINT (a);
 216   id2 = GPOINTER_TO_UINT (b);
 217
 218   return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 219 }
 220
 221 static void
 222 test_thread_sort_entry_func (gpointer data, gpointer user_data)
 223 {
 224   guint thread_id;
 225   gboolean is_sorted;
 226
 227   G_LOCK (last_thread);
 228
 229   thread_id = GPOINTER_TO_UINT (data);
 230   is_sorted = GPOINTER_TO_INT (user_data);
 231
 232   DEBUG_MSG (("%s ---> entered thread:%2.2d, last thread:%2.2d",
 233               is_sorted ? "[  sorted]" : "[unsorted]",
 234               thread_id, last_thread_id));
 235
 236   if (is_sorted) {
 237     static gboolean last_failed = FALSE;
 238
 239     if (last_thread_id > thread_id) {
 240       if (last_failed) {
 241         g_assert (last_thread_id <= thread_id);
 242       }
 243
 244       /* Here we remember one fail and if it concurrently fails, it
 245        * can not be sorted. the last thread id might be < this thread
 246        * id if something is added to the queue since threads were
 247        * created
 248        */
 249       last_failed = TRUE;
 250     } else {
 251       last_failed = FALSE;
 252     }
 253
 254     last_thread_id = thread_id;
 255   }
 256
 257   G_UNLOCK (last_thread);
 258
 259   g_usleep (WAIT * 1000);
 260 }
 261
 262 static void
 263 test_thread_sort (gboolean sort)
 264 {
 265   GThreadPool *pool;
 266   guint limit;
 267   guint max_threads;
 268   gint i;
 269
 270   limit = MAX_THREADS * 10;
 271
 272   if (sort) {
 273     max_threads = 1;
 274   } else {
 275     max_threads = MAX_THREADS;
 276   }
 277
 278   /* It is important that we only have a maximum of 1 thread for this
 279    * test since the results can not be guaranteed to be sorted if > 1.
 280    *
 281    * Threads are scheduled by the operating system and are executed at
 282    * random. It cannot be assumed that threads are executed in the
 283    * order they are created. This was discussed in bug #334943.
 284    */
 285
 286   pool = g_thread_pool_new (test_thread_sort_entry_func,
 287                             GINT_TO_POINTER (sort),
 288                             max_threads,
 289                             FALSE,
 290                             NULL);
 291
 292   g_thread_pool_set_max_unused_threads (MAX_UNUSED_THREADS);
 293
 294   if (sort) {
 295     g_thread_pool_set_sort_function (pool,
 296                                      test_thread_sort_compare_func,
 297                                      GUINT_TO_POINTER (69));
 298   }
 299
 300   for (i = 0; i < limit; i++) {
 301     guint id;
 302
 303     id = g_random_int_range (1, limit) + 1;
 304     g_thread_pool_push (pool, GUINT_TO_POINTER (id), NULL);
 305     DEBUG_MSG (("%s ===> pushed new thread with id:%d, number "
 306                 "of threads:%d, unprocessed:%d",
 307                 sort ? "[  sorted]" : "[unsorted]",
 308                 id,
 309                 g_thread_pool_get_num_threads (pool),
 310                 g_thread_pool_unprocessed (pool)));
 311   }
 312
 313   g_assert (g_thread_pool_get_max_threads (pool) == max_threads);
 314   g_assert (g_thread_pool_get_num_threads (pool) == g_thread_pool_get_max_threads (pool));
 315 }
 316
 317 static void
 318 test_thread_idle_time_entry_func (gpointer data, gpointer user_data)
 319 {
 320   guint thread_id;
 321
 322   thread_id = GPOINTER_TO_UINT (data);
 323
 324   DEBUG_MSG (("[idle] ---> entered thread:%2.2d", thread_id));
 325
 326   g_usleep (WAIT * 1000);
 327
 328   DEBUG_MSG (("[idle] <--- exiting thread:%2.2d", thread_id));
 329 }
 330
 331 static gboolean
 332 test_thread_idle_timeout (gpointer data)
 333 {
 334   guint interval;
 335   gint i;
 336
 337   interval = GPOINTER_TO_UINT (data);
 338
 339   for (i = 0; i < 2; i++) {
 340     g_thread_pool_push (idle_pool, GUINT_TO_POINTER (100 + i), NULL);
 341     DEBUG_MSG (("[idle] ===> pushed new thread with id:%d, number "
 342                 "of threads:%d, unprocessed:%d",
 343                 100 + i,
 344                 g_thread_pool_get_num_threads (idle_pool),
 345                 g_thread_pool_unprocessed (idle_pool)));
 346   }
 347
 348
 349   return FALSE;
 350 }
 351
 352 static void
 353 test_thread_idle_time ()
 354 {
 355   guint limit = 50;
 356   guint interval = 10000;
 357   gint i;
 358
 359   idle_pool = g_thread_pool_new (test_thread_idle_time_entry_func,
 360                                  NULL,
 361                                  MAX_THREADS,
 362                                  FALSE,
 363                                  NULL);
 364
 365   g_thread_pool_set_max_unused_threads (MAX_UNUSED_THREADS);
 366   g_thread_pool_set_max_idle_time (interval);
 367
 368   g_assert (g_thread_pool_get_max_unused_threads () == MAX_UNUSED_THREADS);
 369   g_assert (g_thread_pool_get_max_idle_time () == interval);
 370
 371   for (i = 0; i < limit; i++) {
 372     g_thread_pool_push (idle_pool, GUINT_TO_POINTER (i + 1), NULL);
 373     DEBUG_MSG (("[idle] ===> pushed new thread with id:%d, "
 374                 "number of threads:%d, unprocessed:%d",
 375                 i,
 376                 g_thread_pool_get_num_threads (idle_pool),
 377                 g_thread_pool_unprocessed (idle_pool)));
 378   }
 379
 380   g_timeout_add ((interval - 1000),
 381                  test_thread_idle_timeout,
 382                  GUINT_TO_POINTER (interval));
 383 }
 384
 385 static gboolean
 386 test_check_start_and_stop (gpointer user_data)
 387 {
 388   static guint test_number = 0;
 389   static gboolean run_next = FALSE;
 390   gboolean continue_timeout = TRUE;
 391   gboolean quit = TRUE;
 392
 393   if (test_number == 0) {
 394     run_next = TRUE;
 395     DEBUG_MSG (("***** RUNNING TEST %2.2d *****", test_number));
 396   }
 397
 398   if (run_next) {
 399     test_number++;
 400
 401     switch (test_number) {
 402     case 1:
 403       test_thread_functions ();
 404       break;
 405     case 2:
 406       test_thread_stop_unused ();
 407       break;
 408     case 3:
 409       test_thread_pools ();
 410       break;
 411     case 4:
 412       test_thread_sort (FALSE);
 413       break;
 414     case 5:
 415       test_thread_sort (TRUE);
 416       break;
 417     case 6:
 418       test_thread_stop_unused ();
 419       break;
 420     case 7:
 421       test_thread_idle_time ();
 422       break;
 423     default:
 424       DEBUG_MSG (("***** END OF TESTS *****"));
 425       g_main_loop_quit (main_loop);
 426       continue_timeout = FALSE;
 427       break;
 428     }
 429
 430     run_next = FALSE;
 431     return TRUE;
 432   }
 433
 434   if (test_number == 3) {
 435     G_LOCK (thread_counter_pools);
 436     quit &= running_thread_counter <= 0;
 437     DEBUG_MSG (("***** POOL RUNNING THREAD COUNT:%ld",
 438                 running_thread_counter));
 439     G_UNLOCK (thread_counter_pools);
 440   }
 441
 442   if (test_number == 4 || test_number == 5) {
 443     G_LOCK (thread_counter_sort);
 444     quit &= sort_thread_counter <= 0;
 445     DEBUG_MSG (("***** POOL SORT THREAD COUNT:%ld",
 446                 sort_thread_counter));
 447     G_UNLOCK (thread_counter_sort);
 448   }
 449
 450   if (test_number == 7) {
 451     guint idle;
 452
 453     idle = g_thread_pool_get_num_unused_threads ();
 454     quit &= idle < 1;
 455     DEBUG_MSG (("***** POOL IDLE THREAD COUNT:%d, UNPROCESSED JOBS:%d",
 456                 idle, g_thread_pool_unprocessed (idle_pool)));
 457   }
 458
 459   if (quit) {
 460     run_next = TRUE;
 461   }
 462
 463   return continue_timeout;
 464 }
 465
 466 int
 467 main (int argc, char *argv[])
 468 {
 469   g_thread_init (NULL);
 470
 471   DEBUG_MSG (("Starting... (in one second)"));
 472   g_timeout_add (1000, test_check_start_and_stop, NULL);
 473
 474   main_loop = g_main_loop_new (NULL, FALSE);
 475   g_main_loop_run (main_loop);
 476
 477   return 0;
 478 }