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, see <http://www.gnu.org/licenses/>.
  17  *
  18  * Author: David Zeuthen <davidz@redhat.com>
  19  */
  20
  21 #include <gio/gio.h>
  22 #include <unistd.h>
  23 #include <string.h>
  24
  25 #include "gdbus-tests.h"
  26
  27 /* all tests rely on a global connection */
  28 static GDBusConnection *c = NULL;
  29
  30 /* ---------------------------------------------------------------------------------------------------- */
  31 /* Ensure that signal and method replies are delivered in the right thread */
  32 /* ---------------------------------------------------------------------------------------------------- */
  33
  34 typedef struct {
  35   GThread *thread;
  36   GMainLoop *thread_loop;
  37   guint signal_count;
  38 } DeliveryData;
  39
  40 static void
  41 msg_cb_expect_success (GDBusConnection *connection,
  42                        GAsyncResult    *res,
  43                        gpointer         user_data)
  44 {
  45   DeliveryData *data = user_data;
  46   GError *error;
  47   GVariant *result;
  48
  49   error = NULL;
  50   result = g_dbus_connection_call_finish (connection,
  51                                           res,
  52                                           &error);
  53   g_assert_no_error (error);
  54   g_assert (result != NULL);
  55   g_variant_unref (result);
  56
  57   g_assert (g_thread_self () == data->thread);
  58
  59   g_main_loop_quit (data->thread_loop);
  60 }
  61
  62 static void
  63 msg_cb_expect_error_cancelled (GDBusConnection *connection,
  64                                GAsyncResult    *res,
  65                                gpointer         user_data)
  66 {
  67   DeliveryData *data = user_data;
  68   GError *error;
  69   GVariant *result;
  70
  71   error = NULL;
  72   result = g_dbus_connection_call_finish (connection,
  73                                           res,
  74                                           &error);
  75   g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
  76   g_assert (!g_dbus_error_is_remote_error (error));
  77   g_error_free (error);
  78   g_assert (result == NULL);
  79
  80   g_assert (g_thread_self () == data->thread);
  81
  82   g_main_loop_quit (data->thread_loop);
  83 }
  84
  85 static void
  86 signal_handler (GDBusConnection *connection,
  87                 const gchar      *sender_name,
  88                 const gchar      *object_path,
  89                 const gchar      *interface_name,
  90                 const gchar      *signal_name,
  91                 GVariant         *parameters,
  92                 gpointer         user_data)
  93 {
  94   DeliveryData *data = user_data;
  95
  96   g_assert (g_thread_self () == data->thread);
  97
  98   data->signal_count++;
  99
 100   g_main_loop_quit (data->thread_loop);
 101 }
 102
 103 static gpointer
 104 test_delivery_in_thread_func (gpointer _data)
 105 {
 106   GMainLoop *thread_loop;
 107   GMainContext *thread_context;
 108   DeliveryData data;
 109   GCancellable *ca;
 110   guint subscription_id;
 111   GDBusConnection *priv_c;
 112   GError *error;
 113
 114   error = NULL;
 115
 116   thread_context = g_main_context_new ();
 117   thread_loop = g_main_loop_new (thread_context, FALSE);
 118   g_main_context_push_thread_default (thread_context);
 119
 120   data.thread = g_thread_self ();
 121   data.thread_loop = thread_loop;
 122   data.signal_count = 0;
 123
 124   /* ---------------------------------------------------------------------------------------------------- */
 125
 126   /*
 127    * Check that we get a reply to the GetId() method call.
 128    */
 129   g_dbus_connection_call (c,
 130                           "org.freedesktop.DBus",  /* bus_name */
 131                           "/org/freedesktop/DBus", /* object path */
 132                           "org.freedesktop.DBus",  /* interface name */
 133                           "GetId",                 /* method name */
 134                           NULL, NULL,
 135                           G_DBUS_CALL_FLAGS_NONE,
 136                           -1,
 137                           NULL,
 138                           (GAsyncReadyCallback) msg_cb_expect_success,
 139                           &data);
 140   g_main_loop_run (thread_loop);
 141
 142   /*
 143    * Check that we never actually send a message if the GCancellable
 144    * is already cancelled - i.e.  we should get #G_IO_ERROR_CANCELLED
 145    * when the actual connection is not up.
 146    */
 147   ca = g_cancellable_new ();
 148   g_cancellable_cancel (ca);
 149   g_dbus_connection_call (c,
 150                           "org.freedesktop.DBus",  /* bus_name */
 151                           "/org/freedesktop/DBus", /* object path */
 152                           "org.freedesktop.DBus",  /* interface name */
 153                           "GetId",                 /* method name */
 154                           NULL, NULL,
 155                           G_DBUS_CALL_FLAGS_NONE,
 156                           -1,
 157                           ca,
 158                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 159                           &data);
 160   g_main_loop_run (thread_loop);
 161   g_object_unref (ca);
 162
 163   /*
 164    * Check that cancellation works when the message is already in flight.
 165    */
 166   ca = g_cancellable_new ();
 167   g_dbus_connection_call (c,
 168                           "org.freedesktop.DBus",  /* bus_name */
 169                           "/org/freedesktop/DBus", /* object path */
 170                           "org.freedesktop.DBus",  /* interface name */
 171                           "GetId",                 /* method name */
 172                           NULL, NULL,
 173                           G_DBUS_CALL_FLAGS_NONE,
 174                           -1,
 175                           ca,
 176                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 177                           &data);
 178   g_cancellable_cancel (ca);
 179   g_main_loop_run (thread_loop);
 180   g_object_unref (ca);
 181
 182   /*
 183    * Check that signals are delivered to the correct thread.
 184    *
 185    * First we subscribe to the signal, then we create a a private
 186    * connection. This should cause a NameOwnerChanged message from
 187    * the message bus.
 188    */
 189   subscription_id = g_dbus_connection_signal_subscribe (c,
 190                                                         "org.freedesktop.DBus",  /* sender */
 191                                                         "org.freedesktop.DBus",  /* interface */
 192                                                         "NameOwnerChanged",      /* member */
 193                                                         "/org/freedesktop/DBus", /* path */
 194                                                         NULL,
 195                                                         G_DBUS_SIGNAL_FLAGS_NONE,
 196                                                         signal_handler,
 197                                                         &data,
 198                                                         NULL);
 199   g_assert (subscription_id != 0);
 200   g_assert (data.signal_count == 0);
 201
 202   priv_c = _g_bus_get_priv (G_BUS_TYPE_SESSION, NULL, &error);
 203   g_assert_no_error (error);
 204   g_assert (priv_c != NULL);
 205
 206   g_main_loop_run (thread_loop);
 207   g_assert (data.signal_count == 1);
 208
 209   g_object_unref (priv_c);
 210
 211   g_dbus_connection_signal_unsubscribe (c, subscription_id);
 212
 213   /* ---------------------------------------------------------------------------------------------------- */
 214
 215   g_main_context_pop_thread_default (thread_context);
 216   g_main_loop_unref (thread_loop);
 217   g_main_context_unref (thread_context);
 218
 219   return NULL;
 220 }
 221
 222 static void
 223 test_delivery_in_thread (void)
 224 {
 225   GThread *thread;
 226
 227   thread = g_thread_new ("deliver",
 228                          test_delivery_in_thread_func,
 229                          NULL);
 230
 231   g_thread_join (thread);
 232 }
 233
 234 /* ---------------------------------------------------------------------------------------------------- */
 235
 236 typedef struct {
 237   GDBusProxy *proxy;
 238   gint msec;
 239   guint num;
 240   gboolean async;
 241
 242   GMainLoop *thread_loop;
 243   GThread *thread;
 244 } SyncThreadData;
 245
 246 static void
 247 sleep_cb (GDBusProxy   *proxy,
 248           GAsyncResult *res,
 249           gpointer      user_data)
 250 {
 251   SyncThreadData *data = user_data;
 252   GError *error;
 253   GVariant *result;
 254
 255   error = NULL;
 256   result = g_dbus_proxy_call_finish (proxy,
 257                                      res,
 258                                      &error);
 259   g_assert_no_error (error);
 260   g_assert (result != NULL);
 261   g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
 262   g_variant_unref (result);
 263
 264   g_assert (data->thread == g_thread_self ());
 265
 266   g_main_loop_quit (data->thread_loop);
 267
 268   //g_debug ("async cb (%p)", g_thread_self ());
 269 }
 270
 271 static gpointer
 272 test_sleep_in_thread_func (gpointer _data)
 273 {
 274   SyncThreadData *data = _data;
 275   GMainContext *thread_context;
 276   guint n;
 277
 278   thread_context = g_main_context_new ();
 279   data->thread_loop = g_main_loop_new (thread_context, FALSE);
 280   g_main_context_push_thread_default (thread_context);
 281
 282   data->thread = g_thread_self ();
 283
 284   for (n = 0; n < data->num; n++)
 285     {
 286       if (data->async)
 287         {
 288           //g_debug ("invoking async (%p)", g_thread_self ());
 289           g_dbus_proxy_call (data->proxy,
 290                              "Sleep",
 291                              g_variant_new ("(i)", data->msec),
 292                              G_DBUS_CALL_FLAGS_NONE,
 293                              -1,
 294                              NULL,
 295                              (GAsyncReadyCallback) sleep_cb,
 296                              data);
 297           g_main_loop_run (data->thread_loop);
 298           if (g_test_verbose ())
 299             g_printerr ("A");
 300           //g_debug ("done invoking async (%p)", g_thread_self ());
 301         }
 302       else
 303         {
 304           GError *error;
 305           GVariant *result;
 306
 307           error = NULL;
 308           //g_debug ("invoking sync (%p)", g_thread_self ());
 309           result = g_dbus_proxy_call_sync (data->proxy,
 310                                            "Sleep",
 311                                            g_variant_new ("(i)", data->msec),
 312                                            G_DBUS_CALL_FLAGS_NONE,
 313                                            -1,
 314                                            NULL,
 315                                            &error);
 316           if (g_test_verbose ())
 317             g_printerr ("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       if (g_test_verbose ())
 412         g_printerr (" ");
 413     }
 414 }
 415
 416 static void
 417 test_method_calls_in_thread (void)
 418 {
 419   GDBusProxy *proxy;
 420   GDBusConnection *connection;
 421   GError *error;
 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   test_method_calls_on_proxy (proxy);
 440
 441   g_object_unref (proxy);
 442   g_object_unref (connection);
 443
 444   if (g_test_verbose ())
 445     g_printerr ("\n");
 446 }
 447
 448 #define SLEEP_MIN_USEC 1
 449 #define SLEEP_MAX_USEC 10
 450
 451 /* Can run in any thread */
 452 static void
 453 ensure_connection_works (GDBusConnection *conn)
 454 {
 455   GVariant *v;
 456   GError *error = NULL;
 457
 458   v = g_dbus_connection_call_sync (conn, "org.freedesktop.DBus",
 459       "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetId", NULL, NULL, 0, -1,
 460       NULL, &error);
 461   g_assert_no_error (error);
 462   g_assert (v != NULL);
 463   g_assert (g_variant_is_of_type (v, G_VARIANT_TYPE ("(s)")));
 464   g_variant_unref (v);
 465 }
 466
 467 /**
 468  * get_sync_in_thread:
 469  * @data: (type guint): delay in microseconds
 470  *
 471  * Sleep for a short time, then get a session bus connection and call
 472  * a method on it.
 473  *
 474  * Runs in a non-main thread.
 475  *
 476  * Returns: (transfer full): the connection
 477  */
 478 static gpointer
 479 get_sync_in_thread (gpointer data)
 480 {
 481   guint delay = GPOINTER_TO_UINT (data);
 482   GError *error = NULL;
 483   GDBusConnection *conn;
 484
 485   g_usleep (delay);
 486
 487   conn = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 488   g_assert_no_error (error);
 489
 490   ensure_connection_works (conn);
 491
 492   return conn;
 493 }
 494
 495 static void
 496 test_threaded_singleton (void)
 497 {
 498   guint i, n;
 499   guint unref_wins = 0;
 500   guint get_wins = 0;
 501
 502   if (g_test_thorough ())
 503     n = 100000;
 504   else
 505     n = 5000;
 506
 507   for (i = 0; i < n; i++)
 508     {
 509       GThread *thread;
 510       guint j;
 511       guint unref_delay, get_delay;
 512       GDBusConnection *new_conn;
 513
 514       /* We want to be the last ref, so let it finish setting up */
 515       for (j = 0; j < 100; j++)
 516         {
 517           guint r = g_atomic_int_get (&G_OBJECT (c)->ref_count);
 518
 519           if (r == 1)
 520             break;
 521
 522           g_debug ("run %u: refcount is %u, sleeping", i, r);
 523           g_usleep (1000);
 524         }
 525
 526       if (j == 100)
 527         g_error ("connection had too many refs");
 528
 529       if (g_test_verbose () && (i % (n/50)) == 0)
 530         g_printerr ("%u%%\n", ((i * 100) / n));
 531
 532       /* Delay for a random time on each side of the race, to perturb the
 533        * timing. Ideally, we want each side to win half the races; these
 534        * timings are about right on smcv's laptop.
 535        */
 536       unref_delay = g_random_int_range (SLEEP_MIN_USEC, SLEEP_MAX_USEC);
 537       get_delay = g_random_int_range (SLEEP_MIN_USEC / 2, SLEEP_MAX_USEC / 2);
 538
 539       /* One half of the race is to call g_bus_get_sync... */
 540       thread = g_thread_new ("get_sync_in_thread", get_sync_in_thread,
 541           GUINT_TO_POINTER (get_delay));
 542
 543       /* ... and the other half is to unref the shared connection, which must
 544        * have exactly one ref at this point
 545        */
 546       g_usleep (unref_delay);
 547       g_object_unref (c);
 548
 549       /* Wait for the thread to run; see what it got */
 550       new_conn = g_thread_join (thread);
 551
 552       /* If the thread won the race, it will have kept the same connection,
 553        * and it'll have one ref
 554        */
 555       if (new_conn == c)
 556         {
 557           get_wins++;
 558         }
 559       else
 560         {
 561           unref_wins++;
 562           /* c is invalid now, but new_conn is suitable for the
 563            * next round
 564            */
 565           c = new_conn;
 566         }
 567
 568       ensure_connection_works (c);
 569     }
 570
 571   if (g_test_verbose ())
 572     g_printerr ("Unref won %u races; Get won %u races\n", unref_wins, get_wins);
 573 }
 574
 575 /* ---------------------------------------------------------------------------------------------------- */
 576
 577 int
 578 main (int   argc,
 579       char *argv[])
 580 {
 581   GError *error;
 582   gint ret;
 583   gchar *path;
 584
 585   g_test_init (&argc, &argv, NULL);
 586
 587   session_bus_up ();
 588
 589   /* this is safe; testserver will exit once the bus goes away */
 590   path = g_test_build_filename (G_TEST_BUILT, "gdbus-testserver", NULL);
 591   g_assert (g_spawn_command_line_async (path, NULL));
 592   g_free (path);
 593
 594   ensure_gdbus_testserver_up ();
 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 }