tests/mainloop-test.c

   1 #include <errno.h>
   2 #include <glib.h>
   3 #ifdef G_OS_UNIX
   4 #include <unistd.h>
   5 #endif
   6 #include <stdio.h>
   7 #include <stdlib.h>
   8
   9 #ifdef G_OS_WIN32
  10 #include <fcntl.h>              /* For _O_BINARY used by pipe() macro */
  11 #include <io.h>                 /* for _pipe() */
  12 #endif
  13
  14 #define ITERS 10000
  15 #define INCREMENT 10
  16 #define NTHREADS 4
  17 #define NCRAWLERS 4
  18 #define CRAWLER_TIMEOUT_RANGE 40
  19 #define RECURSER_TIMEOUT 50
  20
  21 /* The partial ordering between the context array mutex and
  22  * crawler array mutex is that the crawler array mutex cannot
  23  * be locked while the context array mutex is locked
  24  */
  25 GPtrArray *context_array;
  26 GMutex *context_array_mutex;
  27 GCond *context_array_cond;
  28
  29 GMainLoop *main_loop;
  30
  31 G_LOCK_DEFINE_STATIC (crawler_array_lock);
  32 GPtrArray *crawler_array;
  33
  34 typedef struct _AddrData AddrData;
  35 typedef struct _TestData TestData;
  36
  37 struct _AddrData
  38 {
  39   GMainLoop *loop;
  40   GIOChannel *dest;
  41   gint count;
  42 };
  43
  44 struct _TestData
  45 {
  46   gint current_val;
  47   gint iters;
  48   GIOChannel *in;
  49 };
  50
  51 static void cleanup_crawlers (GMainContext *context);
  52
  53 gboolean
  54 read_all (GIOChannel *channel, char *buf, int len)
  55 {
  56   int bytes_read = 0;
  57   int count;
  58   GIOError err;
  59
  60   while (bytes_read < len)
  61     {
  62       err = g_io_channel_read (channel, buf + bytes_read, len - bytes_read, &count);
  63       if (err)
  64         {
  65           if (err != G_IO_ERROR_AGAIN)
  66             return FALSE;
  67         }
  68       else if (count == 0)
  69         return FALSE;
  70
  71       bytes_read += count;
  72     }
  73
  74   return TRUE;
  75 }
  76
  77 gboolean
  78 write_all (GIOChannel *channel, char *buf, int len)
  79 {
  80   int bytes_written = 0;
  81   int count;
  82   GIOError err;
  83
  84   while (bytes_written < len)
  85     {
  86       err = g_io_channel_write (channel, buf + bytes_written, len - bytes_written, &count);
  87       if (err && err != G_IO_ERROR_AGAIN)
  88         return FALSE;
  89
  90       bytes_written += count;
  91     }
  92
  93   return TRUE;
  94 }
  95
  96 gboolean
  97 adder_callback (GIOChannel   *source,
  98                 GIOCondition  condition,
  99                 gpointer      data)
 100 {
 101   char buf1[32];
 102   char buf2[32];
 103
 104   char result[32];
 105
 106   AddrData *addr_data = data;
 107
 108   if (!read_all (source, buf1, 32) ||
 109       !read_all (source, buf2, 32))
 110     {
 111       g_main_loop_quit (addr_data->loop);
 112       return FALSE;
 113     }
 114
 115   sprintf (result, "%d", atoi(buf1) + atoi(buf2));
 116   write_all (addr_data->dest, result, 32);
 117
 118   return TRUE;
 119 }
 120
 121 gboolean
 122 timeout_callback (gpointer data)
 123 {
 124   AddrData *addr_data = data;
 125
 126   addr_data->count++;
 127
 128   return TRUE;
 129 }
 130
 131 gpointer
 132 adder_thread (gpointer data)
 133 {
 134   GMainContext *context;
 135   GSource *adder_source;
 136   GSource *timeout_source;
 137
 138   GIOChannel **channels = data;
 139   AddrData addr_data;
 140
 141   context = g_main_context_new ();
 142
 143   g_mutex_lock (context_array_mutex);
 144
 145   g_ptr_array_add (context_array, context);
 146
 147   if (context_array->len == NTHREADS)
 148     g_cond_broadcast (context_array_cond);
 149
 150   g_mutex_unlock (context_array_mutex);
 151
 152   addr_data.dest = channels[1];
 153   addr_data.loop = g_main_loop_new (context, FALSE);
 154   addr_data.count = 0;
 155
 156   adder_source = g_io_create_watch (channels[0], G_IO_IN | G_IO_HUP);
 157   g_source_set_callback (adder_source, (GSourceFunc)adder_callback, &addr_data, NULL);
 158   g_source_attach (adder_source, context);
 159   g_source_unref (adder_source);
 160
 161   timeout_source = g_timeout_source_new (10);
 162   g_source_set_callback (timeout_source, (GSourceFunc)timeout_callback, &addr_data, NULL);
 163   g_source_set_priority (timeout_source, G_PRIORITY_HIGH);
 164   g_source_attach (timeout_source, context);
 165   g_source_unref (timeout_source);
 166
 167   g_main_run (addr_data.loop);
 168
 169   g_io_channel_close (channels[0]);
 170   g_io_channel_close (channels[1]);
 171   g_io_channel_unref (channels[0]);
 172   g_io_channel_unref (channels[1]);
 173
 174   g_free (channels);
 175
 176   g_main_loop_unref (addr_data.loop);
 177
 178   g_print ("Timeout run %d times\n", addr_data.count);
 179
 180   g_mutex_lock (context_array_mutex);
 181   g_ptr_array_remove (context_array, context);
 182   if (context_array->len == 0)
 183     g_main_loop_quit (main_loop);
 184   g_mutex_unlock (context_array_mutex);
 185
 186   cleanup_crawlers (context);
 187
 188   return NULL;
 189 }
 190
 191 void
 192 io_pipe (GIOChannel **channels)
 193 {
 194   gint fds[2];
 195
 196   if (pipe(fds) < 0)
 197     {
 198       g_warning ("Cannot create pipe %s\n", g_strerror (errno));
 199       exit (1);
 200     }
 201
 202   channels[0] = g_io_channel_unix_new (fds[0]);
 203   channels[1] = g_io_channel_unix_new (fds[1]);
 204 }
 205
 206 void
 207 do_add (GIOChannel *in, gint a, gint b)
 208 {
 209   char buf1[32];
 210   char buf2[32];
 211
 212   sprintf (buf1, "%d", a);
 213   sprintf (buf2, "%d", b);
 214
 215   write_all (in, buf1, 32);
 216   write_all (in, buf2, 32);
 217 }
 218
 219 gboolean
 220 adder_response (GIOChannel   *source,
 221                 GIOCondition  condition,
 222                 gpointer      data)
 223 {
 224   char result[32];
 225   TestData *test_data = data;
 226
 227   if (!read_all (source, result, 32))
 228     return FALSE;
 229
 230   test_data->current_val = atoi (result);
 231   test_data->iters--;
 232
 233   if (test_data->iters == 0)
 234     {
 235       if (test_data->current_val != ITERS * INCREMENT)
 236         {
 237           g_print ("Addition failed: %d != %d\n",
 238                    test_data->current_val, ITERS * INCREMENT);
 239           exit (1);
 240         }
 241
 242       g_io_channel_close (source);
 243       g_io_channel_close (test_data->in);
 244
 245       g_io_channel_unref (source);
 246       g_io_channel_unref (test_data->in);
 247
 248       return FALSE;
 249     }
 250
 251   do_add (test_data->in, test_data->current_val, INCREMENT);
 252
 253   return TRUE;
 254 }
 255
 256 void
 257 create_adder_thread (void)
 258 {
 259   GError *err = NULL;
 260   TestData *test_data;
 261
 262   GIOChannel *in_channels[2];
 263   GIOChannel *out_channels[2];
 264
 265   GIOChannel **sub_channels;
 266
 267   sub_channels = g_new (GIOChannel *, 2);
 268
 269   io_pipe (in_channels);
 270   io_pipe (out_channels);
 271
 272   sub_channels[0] = in_channels[0];
 273   sub_channels[1] = out_channels[1];
 274
 275   g_thread_create (adder_thread, sub_channels, FALSE, &err);
 276
 277   if (err)
 278     {
 279       g_warning ("Cannot create thread: %s", err->message);
 280       exit (1);
 281     }
 282
 283   test_data = g_new (TestData, 1);
 284   test_data->in = in_channels[1];
 285   test_data->current_val = 0;
 286   test_data->iters = ITERS;
 287
 288   g_io_add_watch (out_channels[0], G_IO_IN | G_IO_HUP,
 289                   adder_response, test_data);
 290
 291   do_add (test_data->in, test_data->current_val, INCREMENT);
 292 }
 293
 294 static void create_crawler (void);
 295
 296 static void
 297 remove_crawler (void)
 298 {
 299   GSource *other_source;
 300
 301   if (crawler_array->len > 0)
 302     {
 303       other_source = crawler_array->pdata[g_random_int_range (0, crawler_array->len)];
 304       g_source_destroy (other_source);
 305       g_assert (g_ptr_array_remove_fast (crawler_array, other_source));
 306     }
 307 }
 308
 309 static gint
 310 crawler_callback (gpointer data)
 311 {
 312   GSource *source = data;
 313
 314   G_LOCK (crawler_array_lock);
 315
 316   if (!g_ptr_array_remove_fast (crawler_array, source))
 317     remove_crawler();
 318
 319   remove_crawler();
 320   G_UNLOCK (crawler_array_lock);
 321
 322   create_crawler();
 323   create_crawler();
 324
 325   return FALSE;
 326 }
 327
 328 static void
 329 create_crawler (void)
 330 {
 331   GSource *source = g_timeout_source_new (g_random_int_range (0, CRAWLER_TIMEOUT_RANGE));
 332   g_source_set_callback (source, (GSourceFunc)crawler_callback, source, NULL);
 333
 334   G_LOCK (crawler_array_lock);
 335   g_ptr_array_add (crawler_array, source);
 336
 337   g_mutex_lock (context_array_mutex);
 338   g_source_attach (source, context_array->pdata[g_random_int_range (0, context_array->len)]);
 339   g_source_unref (source);
 340   g_mutex_unlock (context_array_mutex);
 341
 342   G_UNLOCK (crawler_array_lock);
 343 }
 344
 345 static void
 346 cleanup_crawlers (GMainContext *context)
 347 {
 348   gint i;
 349
 350   G_LOCK (crawler_array_lock);
 351   for (i=0; i < crawler_array->len; i++)
 352     {
 353       if (g_source_get_context (crawler_array->pdata[i]) == context)
 354         {
 355           g_source_destroy (g_ptr_array_remove_index (crawler_array, i));
 356           i--;
 357         }
 358     }
 359   G_UNLOCK (crawler_array_lock);
 360 }
 361
 362 static gboolean
 363 recurser_idle (gpointer data)
 364 {
 365   GMainContext *context = data;
 366   gint i;
 367
 368   for (i = 0; i < 10; i++)
 369     g_main_context_iteration (context, FALSE);
 370
 371   return FALSE;
 372 }
 373
 374 static gboolean
 375 recurser_start (gpointer data)
 376 {
 377   GMainContext *context;
 378   GSource *source;
 379
 380   g_mutex_lock (context_array_mutex);
 381   context = context_array->pdata[g_random_int_range (0, context_array->len)];
 382   source = g_idle_source_new ();
 383   g_source_set_callback (source, recurser_idle, context, NULL);
 384   g_source_attach (source, context);
 385   g_source_unref (source);
 386   g_mutex_unlock (context_array_mutex);
 387
 388   return TRUE;
 389 }
 390
 391 int
 392 main (int   argc,
 393       char *argv[])
 394 {
 395   /* Only run the test, if threads are enabled and a default thread
 396      implementation is available */
 397 #if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE)
 398   gint i;
 399
 400   g_thread_init (NULL);
 401
 402   context_array = g_ptr_array_new ();
 403   context_array_mutex = g_mutex_new ();
 404   context_array_cond = g_cond_new ();
 405
 406   crawler_array = g_ptr_array_new ();
 407
 408   main_loop = g_main_loop_new (NULL, FALSE);
 409
 410   for (i = 0; i < NTHREADS; i++)
 411     create_adder_thread ();
 412
 413   /* Wait for all threads to start
 414    */
 415   g_mutex_lock (context_array_mutex);
 416
 417   if (context_array->len < NTHREADS)
 418     g_cond_wait (context_array_cond, context_array_mutex);
 419
 420   g_mutex_unlock (context_array_mutex);
 421
 422   for (i = 0; i < NCRAWLERS; i++)
 423     create_crawler ();
 424
 425   g_timeout_add (RECURSER_TIMEOUT, recurser_start, NULL);
 426
 427   g_main_loop_run (main_loop);
 428   g_main_loop_unref (main_loop);
 429
 430 #endif
 431   return 0;
 432 }