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_quiet ())
 299             g_print ("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_quiet ())
 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       if (!g_test_quiet ())
 412         g_print (" ");
 413     }
 414 }
 415
 416 static void
 417 test_method_calls_in_thread (void)
 418 {
 419   GDBusProxy *proxy;
 420   GDBusConnection *connection;
 421   GError *error;
 422   gchar *name_owner;
 423
 424   error = NULL;
 425   connection = g_bus_get_sync (G_BUS_TYPE_SESSION,
 426                                NULL,
 427                                &error);
 428   g_assert_no_error (error);
 429   error = NULL;
 430   proxy = g_dbus_proxy_new_sync (connection,
 431                                  G_DBUS_PROXY_FLAGS_NONE,
 432                                  NULL,                      /* GDBusInterfaceInfo */
 433                                  "com.example.TestService", /* name */
 434                                  "/com/example/TestObject", /* object path */
 435                                  "com.example.Frob",        /* interface */
 436                                  NULL, /* GCancellable */
 437                                  &error);
 438   g_assert_no_error (error);
 439
 440   name_owner = g_dbus_proxy_get_name_owner (proxy);
 441   g_assert_cmpstr (name_owner, !=, NULL);
 442   g_free (name_owner);
 443
 444   test_method_calls_on_proxy (proxy);
 445
 446   g_object_unref (proxy);
 447   g_object_unref (connection);
 448
 449   if (!g_test_quiet ())
 450     g_print ("\n");
 451 }
 452
 453 #define SLEEP_MIN_USEC 1
 454 #define SLEEP_MAX_USEC 10
 455
 456 /* Can run in any thread */
 457 static void
 458 ensure_connection_works (GDBusConnection *conn)
 459 {
 460   GVariant *v;
 461   GError *error = NULL;
 462
 463   v = g_dbus_connection_call_sync (conn, "org.freedesktop.DBus",
 464       "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetId", NULL, NULL, 0, -1,
 465       NULL, &error);
 466   g_assert_no_error (error);
 467   g_assert (v != NULL);
 468   g_assert (g_variant_is_of_type (v, G_VARIANT_TYPE ("(s)")));
 469   g_variant_unref (v);
 470 }
 471
 472 /**
 473  * get_sync_in_thread:
 474  * @data: (type guint): delay in microseconds
 475  *
 476  * Sleep for a short time, then get a session bus connection and call
 477  * a method on it.
 478  *
 479  * Runs in a non-main thread.
 480  *
 481  * Returns: (transfer full): the connection
 482  */
 483 static gpointer
 484 get_sync_in_thread (gpointer data)
 485 {
 486   guint delay = GPOINTER_TO_UINT (data);
 487   GError *error = NULL;
 488   GDBusConnection *conn;
 489
 490   g_usleep (delay);
 491
 492   conn = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 493   g_assert_no_error (error);
 494
 495   ensure_connection_works (conn);
 496
 497   return conn;
 498 }
 499
 500 static void
 501 test_threaded_singleton (void)
 502 {
 503   guint i, n;
 504   guint unref_wins = 0;
 505   guint get_wins = 0;
 506
 507   if (g_test_thorough ())
 508     n = 100000;
 509   else
 510     n = 5000;
 511
 512   for (i = 0; i < n; i++)
 513     {
 514       GThread *thread;
 515       guint j;
 516       guint unref_delay, get_delay;
 517       GDBusConnection *new_conn;
 518
 519       /* We want to be the last ref, so let it finish setting up */
 520       for (j = 0; j < 100; j++)
 521         {
 522           guint r = g_atomic_int_get (&G_OBJECT (c)->ref_count);
 523
 524           if (r == 1)
 525             break;
 526
 527           g_debug ("run %u: refcount is %u, sleeping", i, r);
 528           g_usleep (1000);
 529         }
 530
 531       if (j == 100)
 532         g_error ("connection had too many refs");
 533
 534       if (g_test_verbose () && (i % (n/50)) == 0)
 535         g_print ("%u%%\n", ((i * 100) / n));
 536
 537       /* Delay for a random time on each side of the race, to perturb the
 538        * timing. Ideally, we want each side to win half the races; these
 539        * timings are about right on smcv's laptop.
 540        */
 541       unref_delay = g_random_int_range (SLEEP_MIN_USEC, SLEEP_MAX_USEC);
 542       get_delay = g_random_int_range (SLEEP_MIN_USEC / 2, SLEEP_MAX_USEC / 2);
 543
 544       /* One half of the race is to call g_bus_get_sync... */
 545       thread = g_thread_new ("get_sync_in_thread", get_sync_in_thread,
 546           GUINT_TO_POINTER (get_delay));
 547
 548       /* ... and the other half is to unref the shared connection, which must
 549        * have exactly one ref at this point
 550        */
 551       g_usleep (unref_delay);
 552       g_object_unref (c);
 553
 554       /* Wait for the thread to run; see what it got */
 555       new_conn = g_thread_join (thread);
 556
 557       /* If the thread won the race, it will have kept the same connection,
 558        * and it'll have one ref
 559        */
 560       if (new_conn == c)
 561         {
 562           get_wins++;
 563         }
 564       else
 565         {
 566           unref_wins++;
 567           /* c is invalid now, but new_conn is suitable for the
 568            * next round
 569            */
 570           c = new_conn;
 571         }
 572
 573       ensure_connection_works (c);
 574     }
 575
 576   if (g_test_verbose ())
 577     g_print ("Unref won %u races; Get won %u races\n", unref_wins, get_wins);
 578 }
 579
 580 /* ---------------------------------------------------------------------------------------------------- */
 581
 582 int
 583 main (int   argc,
 584       char *argv[])
 585 {
 586   GError *error;
 587   gint ret;
 588   gchar *path;
 589
 590   g_test_init (&argc, &argv, NULL);
 591
 592   session_bus_up ();
 593
 594   /* this is safe; testserver will exit once the bus goes away */
 595   path = g_test_build_filename (G_TEST_BUILT, "gdbus-testserver", NULL);
 596   g_assert (g_spawn_command_line_async (path, NULL));
 597   g_free (path);
 598
 599   /* wait for the service to come up */
 600   usleep (500 * 1000);
 601
 602   /* Create the connection in the main thread */
 603   error = NULL;
 604   c = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 605   g_assert_no_error (error);
 606   g_assert (c != NULL);
 607
 608   g_test_add_func ("/gdbus/delivery-in-thread", test_delivery_in_thread);
 609   g_test_add_func ("/gdbus/method-calls-in-thread", test_method_calls_in_thread);
 610   g_test_add_func ("/gdbus/threaded-singleton", test_threaded_singleton);
 611
 612   ret = g_test_run();
 613
 614   g_object_unref (c);
 615
 616   /* tear down bus */
 617   session_bus_down ();
 618
 619   return ret;
 620 }