gio/tests/gdbus-threading.c

   1 /* GLib testing framework examples and tests
   2  *
   3  * Copyright (C) 2008-2010 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  *
  20  * Author: David Zeuthen <davidz@redhat.com>
  21  */
  22
  23 #include <gio/gio.h>
  24 #include <unistd.h>
  25 #include <string.h>
  26
  27 #include "gdbus-tests.h"
  28
  29 /* all tests rely on a global connection */
  30 static GDBusConnection *c = NULL;
  31
  32 /* ---------------------------------------------------------------------------------------------------- */
  33 /* Ensure that signal and method replies are delivered in the right thread */
  34 /* ---------------------------------------------------------------------------------------------------- */
  35
  36 typedef struct {
  37   GThread *thread;
  38   GMainLoop *thread_loop;
  39   guint signal_count;
  40 } DeliveryData;
  41
  42 static void
  43 msg_cb_expect_success (GDBusConnection *connection,
  44                        GAsyncResult    *res,
  45                        gpointer         user_data)
  46 {
  47   DeliveryData *data = user_data;
  48   GError *error;
  49   GVariant *result;
  50
  51   error = NULL;
  52   result = g_dbus_connection_call_finish (connection,
  53                                           res,
  54                                           &error);
  55   g_assert_no_error (error);
  56   g_assert (result != NULL);
  57   g_variant_unref (result);
  58
  59   g_assert (g_thread_self () == data->thread);
  60
  61   g_main_loop_quit (data->thread_loop);
  62 }
  63
  64 static void
  65 msg_cb_expect_error_cancelled (GDBusConnection *connection,
  66                                GAsyncResult    *res,
  67                                gpointer         user_data)
  68 {
  69   DeliveryData *data = user_data;
  70   GError *error;
  71   GVariant *result;
  72
  73   error = NULL;
  74   result = g_dbus_connection_call_finish (connection,
  75                                           res,
  76                                           &error);
  77   g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
  78   g_assert (!g_dbus_error_is_remote_error (error));
  79   g_error_free (error);
  80   g_assert (result == NULL);
  81
  82   g_assert (g_thread_self () == data->thread);
  83
  84   g_main_loop_quit (data->thread_loop);
  85 }
  86
  87 static void
  88 signal_handler (GDBusConnection *connection,
  89                 const gchar      *sender_name,
  90                 const gchar      *object_path,
  91                 const gchar      *interface_name,
  92                 const gchar      *signal_name,
  93                 GVariant         *parameters,
  94                 gpointer         user_data)
  95 {
  96   DeliveryData *data = user_data;
  97
  98   g_assert (g_thread_self () == data->thread);
  99
 100   data->signal_count++;
 101
 102   g_main_loop_quit (data->thread_loop);
 103 }
 104
 105 static gpointer
 106 test_delivery_in_thread_func (gpointer _data)
 107 {
 108   GMainLoop *thread_loop;
 109   GMainContext *thread_context;
 110   DeliveryData data;
 111   GCancellable *ca;
 112   guint subscription_id;
 113   GDBusConnection *priv_c;
 114   GError *error;
 115
 116   error = NULL;
 117
 118   thread_context = g_main_context_new ();
 119   thread_loop = g_main_loop_new (thread_context, FALSE);
 120   g_main_context_push_thread_default (thread_context);
 121
 122   data.thread = g_thread_self ();
 123   data.thread_loop = thread_loop;
 124   data.signal_count = 0;
 125
 126   /* ---------------------------------------------------------------------------------------------------- */
 127
 128   /*
 129    * Check that we get a reply to the GetId() method call.
 130    */
 131   g_dbus_connection_call (c,
 132                           "org.freedesktop.DBus",  /* bus_name */
 133                           "/org/freedesktop/DBus", /* object path */
 134                           "org.freedesktop.DBus",  /* interface name */
 135                           "GetId",                 /* method name */
 136                           NULL, NULL,
 137                           G_DBUS_CALL_FLAGS_NONE,
 138                           -1,
 139                           NULL,
 140                           (GAsyncReadyCallback) msg_cb_expect_success,
 141                           &data);
 142   g_main_loop_run (thread_loop);
 143
 144   /*
 145    * Check that we never actually send a message if the GCancellable
 146    * is already cancelled - i.e.  we should get #G_IO_ERROR_CANCELLED
 147    * when the actual connection is not up.
 148    */
 149   ca = g_cancellable_new ();
 150   g_cancellable_cancel (ca);
 151   g_dbus_connection_call (c,
 152                           "org.freedesktop.DBus",  /* bus_name */
 153                           "/org/freedesktop/DBus", /* object path */
 154                           "org.freedesktop.DBus",  /* interface name */
 155                           "GetId",                 /* method name */
 156                           NULL, NULL,
 157                           G_DBUS_CALL_FLAGS_NONE,
 158                           -1,
 159                           ca,
 160                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 161                           &data);
 162   g_main_loop_run (thread_loop);
 163   g_object_unref (ca);
 164
 165   /*
 166    * Check that cancellation works when the message is already in flight.
 167    */
 168   ca = g_cancellable_new ();
 169   g_dbus_connection_call (c,
 170                           "org.freedesktop.DBus",  /* bus_name */
 171                           "/org/freedesktop/DBus", /* object path */
 172                           "org.freedesktop.DBus",  /* interface name */
 173                           "GetId",                 /* method name */
 174                           NULL, NULL,
 175                           G_DBUS_CALL_FLAGS_NONE,
 176                           -1,
 177                           ca,
 178                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 179                           &data);
 180   g_cancellable_cancel (ca);
 181   g_main_loop_run (thread_loop);
 182   g_object_unref (ca);
 183
 184   /*
 185    * Check that signals are delivered to the correct thread.
 186    *
 187    * First we subscribe to the signal, then we create a a private
 188    * connection. This should cause a NameOwnerChanged message from
 189    * the message bus.
 190    */
 191   subscription_id = g_dbus_connection_signal_subscribe (c,
 192                                                         "org.freedesktop.DBus",  /* sender */
 193                                                         "org.freedesktop.DBus",  /* interface */
 194                                                         "NameOwnerChanged",      /* member */
 195                                                         "/org/freedesktop/DBus", /* path */
 196                                                         NULL,
 197                                                         G_DBUS_SIGNAL_FLAGS_NONE,
 198                                                         signal_handler,
 199                                                         &data,
 200                                                         NULL);
 201   g_assert (subscription_id != 0);
 202   g_assert (data.signal_count == 0);
 203
 204   priv_c = _g_bus_get_priv (G_BUS_TYPE_SESSION, NULL, &error);
 205   g_assert_no_error (error);
 206   g_assert (priv_c != NULL);
 207
 208   g_main_loop_run (thread_loop);
 209   g_assert (data.signal_count == 1);
 210
 211   g_object_unref (priv_c);
 212
 213   g_dbus_connection_signal_unsubscribe (c, subscription_id);
 214
 215   /* ---------------------------------------------------------------------------------------------------- */
 216
 217   g_main_context_pop_thread_default (thread_context);
 218   g_main_loop_unref (thread_loop);
 219   g_main_context_unref (thread_context);
 220
 221   return NULL;
 222 }
 223
 224 static void
 225 test_delivery_in_thread (void)
 226 {
 227   GThread *thread;
 228
 229   thread = g_thread_new ("deliver",
 230                          test_delivery_in_thread_func,
 231                          NULL);
 232
 233   g_thread_join (thread);
 234 }
 235
 236 /* ---------------------------------------------------------------------------------------------------- */
 237
 238 typedef struct {
 239   GDBusProxy *proxy;
 240   gint msec;
 241   guint num;
 242   gboolean async;
 243
 244   GMainLoop *thread_loop;
 245   GThread *thread;
 246 } SyncThreadData;
 247
 248 static void
 249 sleep_cb (GDBusProxy   *proxy,
 250           GAsyncResult *res,
 251           gpointer      user_data)
 252 {
 253   SyncThreadData *data = user_data;
 254   GError *error;
 255   GVariant *result;
 256
 257   error = NULL;
 258   result = g_dbus_proxy_call_finish (proxy,
 259                                      res,
 260                                      &error);
 261   g_assert_no_error (error);
 262   g_assert (result != NULL);
 263   g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
 264   g_variant_unref (result);
 265
 266   g_assert (data->thread == g_thread_self ());
 267
 268   g_main_loop_quit (data->thread_loop);
 269
 270   //g_debug ("async cb (%p)", g_thread_self ());
 271 }
 272
 273 static gpointer
 274 test_sleep_in_thread_func (gpointer _data)
 275 {
 276   SyncThreadData *data = _data;
 277   GMainContext *thread_context;
 278   guint n;
 279
 280   thread_context = g_main_context_new ();
 281   data->thread_loop = g_main_loop_new (thread_context, FALSE);
 282   g_main_context_push_thread_default (thread_context);
 283
 284   data->thread = g_thread_self ();
 285
 286   for (n = 0; n < data->num; n++)
 287     {
 288       if (data->async)
 289         {
 290           //g_debug ("invoking async (%p)", g_thread_self ());
 291           g_dbus_proxy_call (data->proxy,
 292                              "Sleep",
 293                              g_variant_new ("(i)", data->msec),
 294                              G_DBUS_CALL_FLAGS_NONE,
 295                              -1,
 296                              NULL,
 297                              (GAsyncReadyCallback) sleep_cb,
 298                              data);
 299           g_main_loop_run (data->thread_loop);
 300           g_print ("A");
 301           //g_debug ("done invoking async (%p)", g_thread_self ());
 302         }
 303       else
 304         {
 305           GError *error;
 306           GVariant *result;
 307
 308           error = NULL;
 309           //g_debug ("invoking sync (%p)", g_thread_self ());
 310           result = g_dbus_proxy_call_sync (data->proxy,
 311                                            "Sleep",
 312                                            g_variant_new ("(i)", data->msec),
 313                                            G_DBUS_CALL_FLAGS_NONE,
 314                                            -1,
 315                                            NULL,
 316                                            &error);
 317           g_print ("S");
 318           //g_debug ("done invoking sync (%p)", g_thread_self ());
 319           g_assert_no_error (error);
 320           g_assert (result != NULL);
 321           g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
 322           g_variant_unref (result);
 323         }
 324     }
 325
 326   g_main_context_pop_thread_default (thread_context);
 327   g_main_loop_unref (data->thread_loop);
 328   g_main_context_unref (thread_context);
 329
 330   return NULL;
 331 }
 332
 333 static void
 334 test_method_calls_on_proxy (GDBusProxy *proxy)
 335 {
 336   guint n;
 337
 338   /*
 339    * Check that multiple threads can do calls without interferring with
 340    * each other. We do this by creating three threads that call the
 341    * Sleep() method on the server (which handles it asynchronously, e.g.
 342    * it won't block other requests) with different sleep durations and
 343    * a number of times. We do this so each set of calls add up to 4000
 344    * milliseconds.
 345    *
 346    * The dbus test server that this code calls into uses glib timeouts
 347    * to do the sleeping which have only a granularity of 1ms.  It is
 348    * therefore possible to lose as much as 40ms; the test could finish
 349    * in slightly less than 4 seconds.
 350    *
 351    * We run this test twice - first with async calls in each thread, then
 352    * again with sync calls
 353    */
 354
 355   for (n = 0; n < 2; n++)
 356     {
 357       gboolean do_async;
 358       GThread *thread1;
 359       GThread *thread2;
 360       GThread *thread3;
 361       SyncThreadData data1;
 362       SyncThreadData data2;
 363       SyncThreadData data3;
 364       GTimeVal start_time;
 365       GTimeVal end_time;
 366       guint elapsed_msec;
 367
 368       do_async = (n == 0);
 369
 370       g_get_current_time (&start_time);
 371
 372       data1.proxy = proxy;
 373       data1.msec = 40;
 374       data1.num = 100;
 375       data1.async = do_async;
 376       thread1 = g_thread_new ("sleep",
 377                               test_sleep_in_thread_func,
 378                               &data1);
 379
 380       data2.proxy = proxy;
 381       data2.msec = 20;
 382       data2.num = 200;
 383       data2.async = do_async;
 384       thread2 = g_thread_new ("sleep2",
 385                               test_sleep_in_thread_func,
 386                               &data2);
 387
 388       data3.proxy = proxy;
 389       data3.msec = 100;
 390       data3.num = 40;
 391       data3.async = do_async;
 392       thread3 = g_thread_new ("sleep3",
 393                               test_sleep_in_thread_func,
 394                               &data3);
 395
 396       g_thread_join (thread1);
 397       g_thread_join (thread2);
 398       g_thread_join (thread3);
 399
 400       g_get_current_time (&end_time);
 401
 402       elapsed_msec = ((end_time.tv_sec * G_USEC_PER_SEC + end_time.tv_usec) -
 403                       (start_time.tv_sec * G_USEC_PER_SEC + start_time.tv_usec)) / 1000;
 404
 405       //g_debug ("Elapsed time for %s = %d msec", n == 0 ? "async" : "sync", elapsed_msec);
 406
 407       /* elapsed_msec should be 4000 msec +/- change for overhead/inaccuracy */
 408       g_assert_cmpint (elapsed_msec, >=, 3950);
 409       g_assert_cmpint (elapsed_msec,  <, 8000);
 410
 411       g_print (" ");
 412     }
 413 }
 414
 415 static void
 416 test_method_calls_in_thread (void)
 417 {
 418   GDBusProxy *proxy;
 419   GDBusConnection *connection;
 420   GError *error;
 421   gchar *name_owner;
 422
 423   error = NULL;
 424   connection = g_bus_get_sync (G_BUS_TYPE_SESSION,
 425                                NULL,
 426                                &error);
 427   g_assert_no_error (error);
 428   error = NULL;
 429   proxy = g_dbus_proxy_new_sync (connection,
 430                                  G_DBUS_PROXY_FLAGS_NONE,
 431                                  NULL,                      /* GDBusInterfaceInfo */
 432                                  "com.example.TestService", /* name */
 433                                  "/com/example/TestObject", /* object path */
 434                                  "com.example.Frob",        /* interface */
 435                                  NULL, /* GCancellable */
 436                                  &error);
 437   g_assert_no_error (error);
 438
 439   name_owner = g_dbus_proxy_get_name_owner (proxy);
 440   g_assert_cmpstr (name_owner, !=, NULL);
 441   g_free (name_owner);
 442
 443   test_method_calls_on_proxy (proxy);
 444
 445   g_object_unref (proxy);
 446   g_object_unref (connection);
 447 }
 448
 449 #define SLEEP_MIN_USEC 1
 450 #define SLEEP_MAX_USEC 10
 451
 452 /* Can run in any thread */
 453 static void
 454 ensure_connection_works (GDBusConnection *conn)
 455 {
 456   GVariant *v;
 457   GError *error = NULL;
 458
 459   v = g_dbus_connection_call_sync (conn, "org.freedesktop.DBus",
 460       "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetId", NULL, NULL, 0, -1,
 461       NULL, &error);
 462   g_assert_no_error (error);
 463   g_assert (v != NULL);
 464   g_assert (g_variant_is_of_type (v, G_VARIANT_TYPE ("(s)")));
 465   g_variant_unref (v);
 466 }
 467
 468 /*
 469  * get_sync_in_thread:
 470  * @data: (type guint): delay in microseconds
 471  *
 472  * Sleep for a short time, then get a session bus connection and call
 473  * a method on it.
 474  *
 475  * Runs in a non-main thread.
 476  *
 477  * Returns: (transfer full): the connection
 478  */
 479 static gpointer
 480 get_sync_in_thread (gpointer data)
 481 {
 482   guint delay = GPOINTER_TO_UINT (data);
 483   GError *error = NULL;
 484   GDBusConnection *conn;
 485
 486   g_usleep (delay);
 487
 488   conn = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 489   g_assert_no_error (error);
 490
 491   ensure_connection_works (conn);
 492
 493   return conn;
 494 }
 495
 496 static void
 497 test_threaded_singleton (void)
 498 {
 499   guint i, n;
 500   guint unref_wins = 0;
 501   guint get_wins = 0;
 502
 503   if (g_test_thorough ())
 504     n = 100000;
 505   else
 506     n = 5000;
 507
 508   for (i = 0; i < n; i++)
 509     {
 510       GThread *thread;
 511       guint j;
 512       guint unref_delay, get_delay;
 513       GDBusConnection *new_conn;
 514
 515       /* We want to be the last ref, so let it finish setting up */
 516       for (j = 0; j < 100; j++)
 517         {
 518           guint r = g_atomic_int_get (&G_OBJECT (c)->ref_count);
 519
 520           if (r == 1)
 521             break;
 522
 523           g_debug ("run %u: refcount is %u, sleeping", i, r);
 524           g_usleep (1000);
 525         }
 526
 527       if (j == 100)
 528         g_error ("connection had too many refs");
 529
 530       if (g_test_verbose () && (i % (n/50)) == 0)
 531         g_print ("%u%%\n", ((i * 100) / n));
 532
 533       /* Delay for a random time on each side of the race, to perturb the
 534        * timing. Ideally, we want each side to win half the races; these
 535        * timings are about right on smcv's laptop.
 536        */
 537       unref_delay = g_random_int_range (SLEEP_MIN_USEC, SLEEP_MAX_USEC);
 538       get_delay = g_random_int_range (SLEEP_MIN_USEC / 2, SLEEP_MAX_USEC / 2);
 539
 540       /* One half of the race is to call g_bus_get_sync... */
 541       thread = g_thread_new ("get_sync_in_thread", get_sync_in_thread,
 542           GUINT_TO_POINTER (get_delay));
 543
 544       /* ... and the other half is to unref the shared connection, which must
 545        * have exactly one ref at this point
 546        */
 547       g_usleep (unref_delay);
 548       g_object_unref (c);
 549
 550       /* Wait for the thread to run; see what it got */
 551       new_conn = g_thread_join (thread);
 552
 553       /* If the thread won the race, it will have kept the same connection,
 554        * and it'll have one ref
 555        */
 556       if (new_conn == c)
 557         {
 558           get_wins++;
 559         }
 560       else
 561         {
 562           unref_wins++;
 563           /* c is invalid now, but new_conn is suitable for the
 564            * next round
 565            */
 566           c = new_conn;
 567         }
 568
 569       ensure_connection_works (c);
 570     }
 571
 572   if (g_test_verbose ())
 573     g_print ("Unref won %u races; Get won %u races\n", unref_wins, get_wins);
 574 }
 575
 576 /* ---------------------------------------------------------------------------------------------------- */
 577
 578 int
 579 main (int   argc,
 580       char *argv[])
 581 {
 582   GError *error;
 583   gint ret;
 584
 585   g_type_init ();
 586   g_test_init (&argc, &argv, NULL);
 587
 588   session_bus_up ();
 589
 590   /* this is safe; testserver will exit once the bus goes away */
 591   g_assert (g_spawn_command_line_async (SRCDIR "/gdbus-testserver.py", NULL));
 592
 593   /* wait for the service to come up */
 594   usleep (500 * 1000);
 595
 596   /* Create the connection in the main thread */
 597   error = NULL;
 598   c = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 599   g_assert_no_error (error);
 600   g_assert (c != NULL);
 601
 602   g_test_add_func ("/gdbus/delivery-in-thread", test_delivery_in_thread);
 603   g_test_add_func ("/gdbus/method-calls-in-thread", test_method_calls_in_thread);
 604   g_test_add_func ("/gdbus/threaded-singleton", test_threaded_singleton);
 605
 606   ret = g_test_run();
 607
 608   g_object_unref (c);
 609
 610   /* tear down bus */
 611   session_bus_down ();
 612
 613   return ret;
 614 }