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