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