tests/mainloop-test.c

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