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 void
 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_get (g_thread_self());
 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
 184 void
 185 io_pipe (GIOChannel **channels)
 186 {
 187   gint fds[2];
 188
 189   if (pipe(fds) < 0)
 190     {
 191       g_warning ("Cannot create pipe %s\n", g_strerror (errno));
 192       exit (1);
 193     }
 194
 195   channels[0] = g_io_channel_unix_new (fds[0]);
 196   channels[1] = g_io_channel_unix_new (fds[1]);
 197 }
 198
 199 void
 200 do_add (GIOChannel *in, gint a, gint b)
 201 {
 202   char buf1[32];
 203   char buf2[32];
 204
 205   sprintf (buf1, "%d", a);
 206   sprintf (buf2, "%d", b);
 207
 208   write_all (in, buf1, 32);
 209   write_all (in, buf2, 32);
 210 }
 211
 212 gboolean
 213 adder_response (GIOChannel   *source,
 214                 GIOCondition  condition,
 215                 gpointer      data)
 216 {
 217   char result[32];
 218   TestData *test_data = data;
 219
 220   if (!read_all (source, result, 32))
 221     return FALSE;
 222
 223   test_data->current_val = atoi (result);
 224   test_data->iters--;
 225
 226   if (test_data->iters == 0)
 227     {
 228       if (test_data->current_val != ITERS * INCREMENT)
 229         {
 230           g_print ("Addition failed: %d != %d\n",
 231                    test_data->current_val, ITERS * INCREMENT);
 232           exit (1);
 233         }
 234
 235       g_io_channel_close (source);
 236       g_io_channel_close (test_data->in);
 237
 238       g_io_channel_unref (source);
 239       g_io_channel_unref (test_data->in);
 240
 241       return FALSE;
 242     }
 243
 244   do_add (test_data->in, test_data->current_val, INCREMENT);
 245
 246   return TRUE;
 247 }
 248
 249 void
 250 create_adder_thread (void)
 251 {
 252   GError *err = NULL;
 253   TestData *test_data;
 254
 255   GIOChannel *in_channels[2];
 256   GIOChannel *out_channels[2];
 257
 258   GIOChannel **sub_channels;
 259
 260   sub_channels = g_new (GIOChannel *, 2);
 261
 262   io_pipe (in_channels);
 263   io_pipe (out_channels);
 264
 265   sub_channels[0] = in_channels[0];
 266   sub_channels[1] = out_channels[1];
 267
 268   g_thread_create (adder_thread, sub_channels, 0,
 269                    FALSE, TRUE, G_THREAD_PRIORITY_NORMAL, &err);
 270
 271   if (err)
 272     {
 273       g_warning ("Cannot create thread: %s", err->message);
 274       exit (1);
 275     }
 276
 277   test_data = g_new (TestData, 1);
 278   test_data->in = in_channels[1];
 279   test_data->current_val = 0;
 280   test_data->iters = ITERS;
 281
 282   g_io_add_watch (out_channels[0], G_IO_IN | G_IO_HUP,
 283                   adder_response, test_data);
 284
 285   do_add (test_data->in, test_data->current_val, INCREMENT);
 286 }
 287
 288 static void create_crawler (void);
 289
 290 static void
 291 remove_crawler (void)
 292 {
 293   GSource *other_source;
 294
 295   if (crawler_array->len > 0)
 296     {
 297       other_source = crawler_array->pdata[g_random_int_range (0, crawler_array->len)];
 298       g_source_destroy (other_source);
 299       g_assert (g_ptr_array_remove_fast (crawler_array, other_source));
 300     }
 301 }
 302
 303 static gint
 304 crawler_callback (gpointer data)
 305 {
 306   GSource *source = data;
 307
 308   G_LOCK (crawler_array_lock);
 309
 310   if (!g_ptr_array_remove_fast (crawler_array, source))
 311     remove_crawler();
 312
 313   remove_crawler();
 314   G_UNLOCK (crawler_array_lock);
 315
 316   create_crawler();
 317   create_crawler();
 318
 319   return FALSE;
 320 }
 321
 322 static void
 323 create_crawler (void)
 324 {
 325   GSource *source = g_timeout_source_new (g_random_int_range (0, CRAWLER_TIMEOUT_RANGE));
 326   g_source_set_callback (source, (GSourceFunc)crawler_callback, source, NULL);
 327
 328   g_mutex_lock (context_array_mutex);
 329   g_source_attach (source, context_array->pdata[g_random_int_range (0, context_array->len)]);
 330   g_source_unref (source);
 331   g_mutex_unlock (context_array_mutex);
 332
 333   G_LOCK (crawler_array_lock);
 334   g_ptr_array_add (crawler_array, source);
 335   G_UNLOCK (crawler_array_lock);
 336 }
 337
 338 static void
 339 cleanup_crawlers (GMainContext *context)
 340 {
 341   gint i;
 342
 343   G_LOCK (crawler_array_lock);
 344   for (i=0; i < crawler_array->len; i++)
 345     {
 346       if (g_source_get_context (crawler_array->pdata[i]) == context)
 347         {
 348           g_source_destroy (g_ptr_array_remove_index (crawler_array, i));
 349           i--;
 350         }
 351     }
 352   G_UNLOCK (crawler_array_lock);
 353 }
 354
 355 static gboolean
 356 recurser_idle (gpointer data)
 357 {
 358   GMainContext *context = data;
 359   gint i;
 360
 361   for (i = 0; i < 10; i++)
 362     g_main_context_iteration (context, TRUE);
 363
 364   return FALSE;
 365 }
 366
 367 static gboolean
 368 recurser_start (gpointer data)
 369 {
 370   GMainContext *context;
 371   GSource *source;
 372
 373   g_mutex_lock (context_array_mutex);
 374   context = context_array->pdata[g_random_int_range (0, context_array->len)];
 375   source = g_idle_source_new ();
 376   g_source_set_callback (source, recurser_idle, context, NULL);
 377   g_source_attach (source, context);
 378   g_source_unref (source);
 379   g_mutex_unlock (context_array_mutex);
 380
 381   return TRUE;
 382 }
 383
 384 int
 385 main (int   argc,
 386       char *argv[])
 387 {
 388   /* Only run the test, if threads are enabled and a default thread
 389      implementation is available */
 390 #if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE)
 391   gint i;
 392
 393   g_thread_init (NULL);
 394
 395   context_array = g_ptr_array_new ();
 396   context_array_mutex = g_mutex_new ();
 397   context_array_cond = g_cond_new ();
 398
 399   crawler_array = g_ptr_array_new ();
 400
 401   main_loop = g_main_loop_new (NULL, FALSE);
 402
 403   for (i = 0; i < NTHREADS; i++)
 404     create_adder_thread ();
 405
 406   /* Wait for all threads to start
 407    */
 408   g_mutex_lock (context_array_mutex);
 409
 410   if (context_array->len < NTHREADS)
 411     g_cond_wait (context_array_cond, context_array_mutex);
 412
 413   g_mutex_unlock (context_array_mutex);
 414
 415   for (i = 0; i < NCRAWLERS; i++)
 416     create_crawler ();
 417
 418   g_timeout_add (RECURSER_TIMEOUT, recurser_start, NULL);
 419
 420   g_main_loop_run (main_loop);
 421   g_main_loop_unref (main_loop);
 422
 423 #endif
 424   return 0;
 425 }