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