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 continously (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    g_assert (test_count_threads () == limit);
 125
 126    DEBUG_MSG (("[unused] stopping unused threads"));
 127
 128    g_thread_pool_stop_unused_threads ();
 129
 130    DEBUG_MSG (("[unused] waiting ONE second for threads to die"));
 131
 132    /* Some time for threads to die. */
 133    g_usleep (G_USEC_PER_SEC);
 134
 135    DEBUG_MSG (("[unused] stopped idle threads, %d remain, %d threads still exist",
 136                g_thread_pool_get_num_unused_threads (),
 137                test_count_threads ()));
 138
 139    g_assert (g_thread_pool_get_num_unused_threads () == test_count_threads ());
 140    g_assert (g_thread_pool_get_num_unused_threads () == 0);
 141
 142    g_thread_pool_set_max_unused_threads (MAX_THREADS);
 143
 144    DEBUG_MSG (("[unused] cleaning up thread pool"));
 145    g_thread_pool_free (pool, FALSE, TRUE);
 146 }
 147
 148 static void
 149 test_thread_pools_entry_func (gpointer data, gpointer user_data)
 150 {
 151   guint id = 0;
 152
 153   id = GPOINTER_TO_UINT (data);
 154
 155   DEBUG_MSG (("[pool] ---> [%3.3d] entered thread.", id));
 156
 157   G_LOCK (thread_counter_pools);
 158   abs_thread_counter++;
 159   running_thread_counter++;
 160   G_UNLOCK (thread_counter_pools);
 161
 162   g_usleep (g_random_int_range (0, 4000));
 163
 164   G_LOCK (thread_counter_pools);
 165   running_thread_counter--;
 166   leftover_task_counter--;
 167
 168   DEBUG_MSG (("[pool] ---> [%3.3d] exiting thread (abs count:%ld, "
 169               "running count:%ld, left over:%ld)",
 170               id, abs_thread_counter,
 171               running_thread_counter, leftover_task_counter));
 172   G_UNLOCK (thread_counter_pools);
 173 }
 174
 175 static void
 176 test_thread_pools (void)
 177 {
 178   GThreadPool *pool1, *pool2, *pool3;
 179   guint runs;
 180   guint i;
 181
 182   pool1 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 3, FALSE, NULL);
 183   pool2 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 5, TRUE, NULL);
 184   pool3 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 7, TRUE, NULL);
 185
 186   runs = 300;
 187   for (i = 0; i < runs; i++)
 188     {
 189       g_thread_pool_push (pool1, GUINT_TO_POINTER (i + 1), NULL);
 190       g_thread_pool_push (pool2, GUINT_TO_POINTER (i + 1), NULL);
 191       g_thread_pool_push (pool3, GUINT_TO_POINTER (i + 1), NULL);
 192       leftover_task_counter += 3;
 193     }
 194
 195   g_thread_pool_free (pool1, TRUE, TRUE);
 196   g_thread_pool_free (pool2, FALSE, TRUE);
 197   g_thread_pool_free (pool3, FALSE, TRUE);
 198
 199   g_assert (runs * 3 == abs_thread_counter + leftover_task_counter);
 200   g_assert (running_thread_counter == 0);
 201 }
 202
 203 static gint
 204 test_thread_sort_compare_func (gconstpointer a, gconstpointer b, gpointer user_data)
 205 {
 206   guint32 id1, id2;
 207
 208   id1 = GPOINTER_TO_UINT (a);
 209   id2 = GPOINTER_TO_UINT (b);
 210
 211   return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);
 212 }
 213
 214 static void
 215 test_thread_sort_entry_func (gpointer data, gpointer user_data)
 216 {
 217   guint thread_id;
 218   gboolean is_sorted;
 219
 220   G_LOCK (last_thread);
 221
 222   thread_id = GPOINTER_TO_UINT (data);
 223   is_sorted = GPOINTER_TO_INT (user_data);
 224
 225   DEBUG_MSG (("%s ---> entered thread:%2.2d, last thread:%2.2d",
 226               is_sorted ? "[  sorted]" : "[unsorted]",
 227               thread_id, last_thread_id));
 228
 229   if (is_sorted) {
 230     static gboolean last_failed = FALSE;
 231
 232     if (last_thread_id > thread_id) {
 233       if (last_failed) {
 234         g_assert (last_thread_id <= thread_id);
 235       }
 236
 237       /* Here we remember one fail and if it concurrently fails, it
 238        * can not be sorted. the last thread id might be < this thread
 239        * id if something is added to the queue since threads were
 240        * created
 241        */
 242       last_failed = TRUE;
 243     } else {
 244       last_failed = FALSE;
 245     }
 246
 247     last_thread_id = thread_id;
 248   }
 249
 250   G_UNLOCK (last_thread);
 251
 252   g_usleep (WAIT * 1000);
 253 }
 254
 255 static void
 256 test_thread_sort (gboolean sort)
 257 {
 258   GThreadPool *pool;
 259   guint limit;
 260   guint max_threads;
 261   gint i;
 262
 263   limit = MAX_THREADS * 10;
 264
 265   if (sort) {
 266     max_threads = 1;
 267   } else {
 268     max_threads = MAX_THREADS;
 269   }
 270
 271   /* It is important that we only have a maximum of 1 thread for this
 272    * test since the results can not be guranteed to be sorted if > 1.
 273    *
 274    * Threads are scheduled by the operating system and are executed at
 275    * random. It cannot be assumed that threads are executed in the
 276    * order they are created. This was discussed in bug #334943.
 277    */
 278
 279   pool = g_thread_pool_new (test_thread_sort_entry_func,
 280                             GINT_TO_POINTER (sort),
 281                             max_threads,
 282                             FALSE,
 283                             NULL);
 284
 285   g_thread_pool_set_max_unused_threads (MAX_UNUSED_THREADS);
 286
 287   if (sort) {
 288     g_thread_pool_set_sort_function (pool,
 289                                      test_thread_sort_compare_func,
 290                                      GUINT_TO_POINTER (69));
 291   }
 292
 293   for (i = 0; i < limit; i++) {
 294     guint id;
 295
 296     id = g_random_int_range (1, limit) + 1;
 297     g_thread_pool_push (pool, GUINT_TO_POINTER (id), NULL);
 298     DEBUG_MSG (("%s ===> pushed new thread with id:%d, number "
 299                 "of threads:%d, unprocessed:%d",
 300                 sort ? "[  sorted]" : "[unsorted]",
 301                 id,
 302                 g_thread_pool_get_num_threads (pool),
 303                 g_thread_pool_unprocessed (pool)));
 304   }
 305
 306   g_assert (g_thread_pool_get_max_threads (pool) == max_threads);
 307   g_assert (g_thread_pool_get_num_threads (pool) == g_thread_pool_get_max_threads (pool));
 308 }
 309
 310 static void
 311 test_thread_idle_time_entry_func (gpointer data, gpointer user_data)
 312 {
 313   guint thread_id;
 314
 315   thread_id = GPOINTER_TO_UINT (data);
 316
 317   DEBUG_MSG (("[idle] ---> entered thread:%2.2d", thread_id));
 318
 319   g_usleep (WAIT * 1000);
 320
 321   DEBUG_MSG (("[idle] <--- exiting thread:%2.2d", thread_id));
 322 }
 323
 324 static gboolean
 325 test_thread_idle_timeout (gpointer data)
 326 {
 327   guint interval;
 328   gint i;
 329
 330   interval = GPOINTER_TO_UINT (data);
 331
 332   for (i = 0; i < 2; i++) {
 333     g_thread_pool_push (idle_pool, GUINT_TO_POINTER (100 + i), NULL);
 334     DEBUG_MSG (("[idle] ===> pushed new thread with id:%d, number "
 335                 "of threads:%d, unprocessed:%d",
 336                 100 + i,
 337                 g_thread_pool_get_num_threads (idle_pool),
 338                 g_thread_pool_unprocessed (idle_pool)));
 339   }
 340
 341
 342   return FALSE;
 343 }
 344
 345 static void
 346 test_thread_idle_time ()
 347 {
 348   guint limit = 50;
 349   guint interval = 10000;
 350   gint i;
 351
 352   idle_pool = g_thread_pool_new (test_thread_idle_time_entry_func,
 353                                  NULL,
 354                                  MAX_THREADS,
 355                                  FALSE,
 356                                  NULL);
 357
 358   g_thread_pool_set_max_unused_threads (MAX_UNUSED_THREADS);
 359   g_thread_pool_set_max_idle_time (interval);
 360
 361   g_assert (g_thread_pool_get_max_unused_threads () == MAX_UNUSED_THREADS);
 362   g_assert (g_thread_pool_get_max_idle_time () == interval);
 363
 364   for (i = 0; i < limit; i++) {
 365     g_thread_pool_push (idle_pool, GUINT_TO_POINTER (i + 1), NULL);
 366     DEBUG_MSG (("[idle] ===> pushed new thread with id:%d, "
 367                 "number of threads:%d, unprocessed:%d",
 368                 i,
 369                 g_thread_pool_get_num_threads (idle_pool),
 370                 g_thread_pool_unprocessed (idle_pool)));
 371   }
 372
 373   g_timeout_add ((interval - 1000),
 374                  test_thread_idle_timeout,
 375                  GUINT_TO_POINTER (interval));
 376 }
 377
 378 static gboolean
 379 test_check_start_and_stop (gpointer user_data)
 380 {
 381   static guint test_number = 0;
 382   static gboolean run_next = FALSE;
 383   gboolean continue_timeout = TRUE;
 384   gboolean quit = TRUE;
 385
 386   if (test_number == 0) {
 387     run_next = TRUE;
 388     DEBUG_MSG (("***** RUNNING TEST %2.2d *****", test_number));
 389   }
 390
 391   if (run_next) {
 392     test_number++;
 393
 394     switch (test_number) {
 395     case 1:
 396       test_thread_functions ();
 397       break;
 398     case 2:
 399       test_thread_stop_unused ();
 400       break;
 401     case 3:
 402       test_thread_pools ();
 403       break;
 404     case 4:
 405       test_thread_sort (FALSE);
 406       break;
 407     case 5:
 408       test_thread_sort (TRUE);
 409       break;
 410     case 6:
 411       test_thread_idle_time ();
 412       break;
 413     default:
 414       DEBUG_MSG (("***** END OF TESTS *****"));
 415       g_main_loop_quit (main_loop);
 416       continue_timeout = FALSE;
 417       break;
 418     }
 419
 420     run_next = FALSE;
 421     return TRUE;
 422   }
 423
 424   if (test_number == 3) {
 425     G_LOCK (thread_counter_pools);
 426     quit &= running_thread_counter <= 0;
 427     DEBUG_MSG (("***** POOL RUNNING THREAD COUNT:%ld",
 428                 running_thread_counter));
 429     G_UNLOCK (thread_counter_pools);
 430   }
 431
 432   if (test_number == 4 || test_number == 5) {
 433     G_LOCK (thread_counter_sort);
 434     quit &= sort_thread_counter <= 0;
 435     DEBUG_MSG (("***** POOL SORT THREAD COUNT:%ld",
 436                 sort_thread_counter));
 437     G_UNLOCK (thread_counter_sort);
 438   }
 439
 440   if (test_number == 6) {
 441     guint idle;
 442
 443     idle = g_thread_pool_get_num_unused_threads ();
 444     quit &= idle < 1;
 445     DEBUG_MSG (("***** POOL IDLE THREAD COUNT:%d, UNPROCESSED JOBS:%d",
 446                 idle, g_thread_pool_unprocessed (idle_pool)));
 447   }
 448
 449   if (quit) {
 450     run_next = TRUE;
 451   }
 452
 453   return continue_timeout;
 454 }
 455
 456 int
 457 main (int argc, char *argv[])
 458 {
 459   /* Only run the test, if threads are enabled and a default thread
 460      implementation is available */
 461
 462 #if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE)
 463   g_thread_init (NULL);
 464
 465   DEBUG_MSG (("Starting... (in one second)"));
 466   g_timeout_add (1000, test_check_start_and_stop, NULL);
 467
 468   main_loop = g_main_loop_new (NULL, FALSE);
 469   g_main_loop_run (main_loop);
 470 #endif
 471
 472   return 0;
 473 }