tests/mainloop-test.c

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