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