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 /* all tests rely on a shared mainloop */
  33 static GMainLoop *loop = NULL;
  34
  35 /* ---------------------------------------------------------------------------------------------------- */
  36 /* Ensure that signal and method replies are delivered in the right thread */
  37 /* ---------------------------------------------------------------------------------------------------- */
  38
  39 typedef struct {
  40   GThread *thread;
  41   GMainLoop *thread_loop;
  42   guint signal_count;
  43 } DeliveryData;
  44
  45 static void
  46 msg_cb_expect_success (GDBusConnection *connection,
  47                        GAsyncResult    *res,
  48                        gpointer         user_data)
  49 {
  50   DeliveryData *data = user_data;
  51   GError *error;
  52   GVariant *result;
  53
  54   error = NULL;
  55   result = g_dbus_connection_call_finish (connection,
  56                                           res,
  57                                           &error);
  58   g_assert_no_error (error);
  59   g_assert (result != NULL);
  60   g_variant_unref (result);
  61
  62   g_assert (g_thread_self () == data->thread);
  63
  64   g_main_loop_quit (data->thread_loop);
  65 }
  66
  67 static void
  68 msg_cb_expect_error_cancelled (GDBusConnection *connection,
  69                                GAsyncResult    *res,
  70                                gpointer         user_data)
  71 {
  72   DeliveryData *data = user_data;
  73   GError *error;
  74   GVariant *result;
  75
  76   error = NULL;
  77   result = g_dbus_connection_call_finish (connection,
  78                                           res,
  79                                           &error);
  80   g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
  81   g_assert (!g_dbus_error_is_remote_error (error));
  82   g_error_free (error);
  83   g_assert (result == NULL);
  84
  85   g_assert (g_thread_self () == data->thread);
  86
  87   g_main_loop_quit (data->thread_loop);
  88 }
  89
  90 static void
  91 signal_handler (GDBusConnection *connection,
  92                 const gchar      *sender_name,
  93                 const gchar      *object_path,
  94                 const gchar      *interface_name,
  95                 const gchar      *signal_name,
  96                 GVariant         *parameters,
  97                 gpointer         user_data)
  98 {
  99   DeliveryData *data = user_data;
 100
 101   g_assert (g_thread_self () == data->thread);
 102
 103   data->signal_count++;
 104
 105   g_main_loop_quit (data->thread_loop);
 106 }
 107
 108 static gpointer
 109 test_delivery_in_thread_func (gpointer _data)
 110 {
 111   GMainLoop *thread_loop;
 112   GMainContext *thread_context;
 113   DeliveryData data;
 114   GCancellable *ca;
 115   guint subscription_id;
 116   GDBusConnection *priv_c;
 117   GError *error;
 118
 119   error = NULL;
 120
 121   thread_context = g_main_context_new ();
 122   thread_loop = g_main_loop_new (thread_context, FALSE);
 123   g_main_context_push_thread_default (thread_context);
 124
 125   data.thread = g_thread_self ();
 126   data.thread_loop = thread_loop;
 127   data.signal_count = 0;
 128
 129   /* ---------------------------------------------------------------------------------------------------- */
 130
 131   /*
 132    * Check that we get a reply to the GetId() method call.
 133    */
 134   g_dbus_connection_call (c,
 135                           "org.freedesktop.DBus",  /* bus_name */
 136                           "/org/freedesktop/DBus", /* object path */
 137                           "org.freedesktop.DBus",  /* interface name */
 138                           "GetId",                 /* method name */
 139                           NULL, NULL,
 140                           G_DBUS_CALL_FLAGS_NONE,
 141                           -1,
 142                           NULL,
 143                           (GAsyncReadyCallback) msg_cb_expect_success,
 144                           &data);
 145   g_main_loop_run (thread_loop);
 146
 147   /*
 148    * Check that we never actually send a message if the GCancellable
 149    * is already cancelled - i.e.  we should get #G_IO_ERROR_CANCELLED
 150    * when the actual connection is not up.
 151    */
 152   ca = g_cancellable_new ();
 153   g_cancellable_cancel (ca);
 154   g_dbus_connection_call (c,
 155                           "org.freedesktop.DBus",  /* bus_name */
 156                           "/org/freedesktop/DBus", /* object path */
 157                           "org.freedesktop.DBus",  /* interface name */
 158                           "GetId",                 /* method name */
 159                           NULL, NULL,
 160                           G_DBUS_CALL_FLAGS_NONE,
 161                           -1,
 162                           ca,
 163                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 164                           &data);
 165   g_main_loop_run (thread_loop);
 166   g_object_unref (ca);
 167
 168   /*
 169    * Check that cancellation works when the message is already in flight.
 170    */
 171   ca = g_cancellable_new ();
 172   g_dbus_connection_call (c,
 173                           "org.freedesktop.DBus",  /* bus_name */
 174                           "/org/freedesktop/DBus", /* object path */
 175                           "org.freedesktop.DBus",  /* interface name */
 176                           "GetId",                 /* method name */
 177                           NULL, NULL,
 178                           G_DBUS_CALL_FLAGS_NONE,
 179                           -1,
 180                           ca,
 181                           (GAsyncReadyCallback) msg_cb_expect_error_cancelled,
 182                           &data);
 183   g_cancellable_cancel (ca);
 184   g_main_loop_run (thread_loop);
 185   g_object_unref (ca);
 186
 187   /*
 188    * Check that signals are delivered to the correct thread.
 189    *
 190    * First we subscribe to the signal, then we create a a private
 191    * connection. This should cause a NameOwnerChanged message from
 192    * the message bus.
 193    */
 194   subscription_id = g_dbus_connection_signal_subscribe (c,
 195                                                         "org.freedesktop.DBus",  /* sender */
 196                                                         "org.freedesktop.DBus",  /* interface */
 197                                                         "NameOwnerChanged",      /* member */
 198                                                         "/org/freedesktop/DBus", /* path */
 199                                                         NULL,
 200                                                         G_DBUS_SIGNAL_FLAGS_NONE,
 201                                                         signal_handler,
 202                                                         &data,
 203                                                         NULL);
 204   g_assert (subscription_id != 0);
 205   g_assert (data.signal_count == 0);
 206
 207   priv_c = _g_bus_get_priv (G_BUS_TYPE_SESSION, NULL, &error);
 208   g_assert_no_error (error);
 209   g_assert (priv_c != NULL);
 210
 211   g_main_loop_run (thread_loop);
 212   g_assert (data.signal_count == 1);
 213
 214   g_object_unref (priv_c);
 215
 216   g_dbus_connection_signal_unsubscribe (c, subscription_id);
 217
 218   /* ---------------------------------------------------------------------------------------------------- */
 219
 220   g_main_context_pop_thread_default (thread_context);
 221   g_main_loop_unref (thread_loop);
 222   g_main_context_unref (thread_context);
 223
 224   g_main_loop_quit (loop);
 225
 226   return NULL;
 227 }
 228
 229 static void
 230 test_delivery_in_thread (void)
 231 {
 232   GError *error;
 233   GThread *thread;
 234
 235   error = NULL;
 236   thread = g_thread_new ("deliver",
 237                          test_delivery_in_thread_func,
 238                          NULL,
 239                          TRUE,
 240                          &error);
 241   g_assert_no_error (error);
 242   g_assert (thread != NULL);
 243
 244   /* run the event loop - it is needed to dispatch D-Bus messages */
 245   g_main_loop_run (loop);
 246
 247   g_thread_join (thread);
 248 }
 249
 250 /* ---------------------------------------------------------------------------------------------------- */
 251
 252 typedef struct {
 253   GDBusProxy *proxy;
 254   gint msec;
 255   guint num;
 256   gboolean async;
 257
 258   GMainLoop *thread_loop;
 259   GThread *thread;
 260
 261   gboolean done;
 262 } SyncThreadData;
 263
 264 static void
 265 sleep_cb (GDBusProxy   *proxy,
 266           GAsyncResult *res,
 267           gpointer      user_data)
 268 {
 269   SyncThreadData *data = user_data;
 270   GError *error;
 271   GVariant *result;
 272
 273   error = NULL;
 274   result = g_dbus_proxy_call_finish (proxy,
 275                                      res,
 276                                      &error);
 277   g_assert_no_error (error);
 278   g_assert (result != NULL);
 279   g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
 280   g_variant_unref (result);
 281
 282   g_assert (data->thread == g_thread_self ());
 283
 284   g_main_loop_quit (data->thread_loop);
 285
 286   //g_debug ("async cb (%p)", g_thread_self ());
 287 }
 288
 289 static gpointer
 290 test_sleep_in_thread_func (gpointer _data)
 291 {
 292   SyncThreadData *data = _data;
 293   GMainContext *thread_context;
 294   guint n;
 295
 296   thread_context = g_main_context_new ();
 297   data->thread_loop = g_main_loop_new (thread_context, FALSE);
 298   g_main_context_push_thread_default (thread_context);
 299
 300   data->thread = g_thread_self ();
 301
 302   for (n = 0; n < data->num; n++)
 303     {
 304       if (data->async)
 305         {
 306           //g_debug ("invoking async (%p)", g_thread_self ());
 307           g_dbus_proxy_call (data->proxy,
 308                              "Sleep",
 309                              g_variant_new ("(i)", data->msec),
 310                              G_DBUS_CALL_FLAGS_NONE,
 311                              -1,
 312                              NULL,
 313                              (GAsyncReadyCallback) sleep_cb,
 314                              data);
 315           g_main_loop_run (data->thread_loop);
 316           g_print ("A");
 317           //g_debug ("done invoking async (%p)", g_thread_self ());
 318         }
 319       else
 320         {
 321           GError *error;
 322           GVariant *result;
 323
 324           error = NULL;
 325           //g_debug ("invoking sync (%p)", g_thread_self ());
 326           result = g_dbus_proxy_call_sync (data->proxy,
 327                                            "Sleep",
 328                                            g_variant_new ("(i)", data->msec),
 329                                            G_DBUS_CALL_FLAGS_NONE,
 330                                            -1,
 331                                            NULL,
 332                                            &error);
 333           g_print ("S");
 334           //g_debug ("done invoking sync (%p)", g_thread_self ());
 335           g_assert_no_error (error);
 336           g_assert (result != NULL);
 337           g_assert_cmpstr (g_variant_get_type_string (result), ==, "()");
 338           g_variant_unref (result);
 339         }
 340     }
 341
 342   g_main_context_pop_thread_default (thread_context);
 343   g_main_loop_unref (data->thread_loop);
 344   g_main_context_unref (thread_context);
 345
 346   data->done = TRUE;
 347   g_main_loop_quit (loop);
 348
 349   return NULL;
 350 }
 351
 352 static void
 353 test_method_calls_on_proxy (GDBusProxy *proxy)
 354 {
 355   guint n;
 356
 357   /*
 358    * Check that multiple threads can do calls without interferring with
 359    * each other. We do this by creating three threads that call the
 360    * Sleep() method on the server (which handles it asynchronously, e.g.
 361    * it won't block other requests) with different sleep durations and
 362    * a number of times. We do this so each set of calls add up to 4000
 363    * milliseconds.
 364    *
 365    * The dbus test server that this code calls into uses glib timeouts
 366    * to do the sleeping which have only a granularity of 1ms.  It is
 367    * therefore possible to lose as much as 40ms; the test could finish
 368    * in slightly less than 4 seconds.
 369    *
 370    * We run this test twice - first with async calls in each thread, then
 371    * again with sync calls
 372    */
 373
 374   for (n = 0; n < 2; n++)
 375     {
 376       gboolean do_async;
 377       GThread *thread1;
 378       GThread *thread2;
 379       GThread *thread3;
 380       SyncThreadData data1;
 381       SyncThreadData data2;
 382       SyncThreadData data3;
 383       GError *error;
 384       GTimeVal start_time;
 385       GTimeVal end_time;
 386       guint elapsed_msec;
 387
 388       error = NULL;
 389       do_async = (n == 0);
 390
 391       g_get_current_time (&start_time);
 392
 393       data1.proxy = proxy;
 394       data1.msec = 40;
 395       data1.num = 100;
 396       data1.async = do_async;
 397       data1.done = FALSE;
 398       thread1 = g_thread_new ("sleep",
 399                               test_sleep_in_thread_func,
 400                               &data1,
 401                               TRUE,
 402                               &error);
 403       g_assert_no_error (error);
 404       g_assert (thread1 != NULL);
 405
 406       data2.proxy = proxy;
 407       data2.msec = 20;
 408       data2.num = 200;
 409       data2.async = do_async;
 410       data2.done = FALSE;
 411       thread2 = g_thread_new ("sleep2",
 412                               test_sleep_in_thread_func,
 413                               &data2,
 414                               TRUE,
 415                               &error);
 416       g_assert_no_error (error);
 417       g_assert (thread2 != NULL);
 418
 419       data3.proxy = proxy;
 420       data3.msec = 100;
 421       data3.num = 40;
 422       data3.async = do_async;
 423       data3.done = FALSE;
 424       thread3 = g_thread_new ("sleep3",
 425                               test_sleep_in_thread_func,
 426                               &data3,
 427                               TRUE,
 428                               &error);
 429       g_assert_no_error (error);
 430       g_assert (thread3 != NULL);
 431
 432       /* we handle messages in the main loop - threads will quit it when they are done */
 433       while (!(data1.done && data2.done && data3.done))
 434         g_main_loop_run (loop);
 435
 436       g_thread_join (thread1);
 437       g_thread_join (thread2);
 438       g_thread_join (thread3);
 439
 440       g_get_current_time (&end_time);
 441
 442       elapsed_msec = ((end_time.tv_sec * G_USEC_PER_SEC + end_time.tv_usec) -
 443                       (start_time.tv_sec * G_USEC_PER_SEC + start_time.tv_usec)) / 1000;
 444
 445       //g_debug ("Elapsed time for %s = %d msec", n == 0 ? "async" : "sync", elapsed_msec);
 446
 447       /* elapsed_msec should be 4000 msec +/- change for overhead/inaccuracy */
 448       g_assert_cmpint (elapsed_msec, >=, 3950);
 449       g_assert_cmpint (elapsed_msec,  <, 6000);
 450
 451       g_print (" ");
 452     }
 453
 454   g_main_loop_quit (loop);
 455 }
 456
 457 static void
 458 test_method_calls_in_thread (void)
 459 {
 460   GDBusProxy *proxy;
 461   GDBusConnection *connection;
 462   GError *error;
 463   gchar *name_owner;
 464
 465   error = NULL;
 466   connection = g_bus_get_sync (G_BUS_TYPE_SESSION,
 467                                NULL,
 468                                &error);
 469   g_assert_no_error (error);
 470   error = NULL;
 471   proxy = g_dbus_proxy_new_sync (connection,
 472                                  G_DBUS_PROXY_FLAGS_NONE,
 473                                  NULL,                      /* GDBusInterfaceInfo */
 474                                  "com.example.TestService", /* name */
 475                                  "/com/example/TestObject", /* object path */
 476                                  "com.example.Frob",        /* interface */
 477                                  NULL, /* GCancellable */
 478                                  &error);
 479   g_assert_no_error (error);
 480
 481   name_owner = g_dbus_proxy_get_name_owner (proxy);
 482   g_assert_cmpstr (name_owner, !=, NULL);
 483   g_free (name_owner);
 484
 485   test_method_calls_on_proxy (proxy);
 486
 487   g_object_unref (proxy);
 488   g_object_unref (connection);
 489 }
 490
 491 /* ---------------------------------------------------------------------------------------------------- */
 492
 493 int
 494 main (int   argc,
 495       char *argv[])
 496 {
 497   GError *error;
 498   gint ret;
 499
 500   g_type_init ();
 501   g_test_init (&argc, &argv, NULL);
 502
 503   /* all the tests rely on a shared main loop */
 504   loop = g_main_loop_new (NULL, FALSE);
 505
 506   /* all the tests use a session bus with a well-known address that we can bring up and down
 507    * using session_bus_up() and session_bus_down().
 508    */
 509   g_unsetenv ("DISPLAY");
 510   g_setenv ("DBUS_SESSION_BUS_ADDRESS", session_bus_get_temporary_address (), TRUE);
 511
 512   session_bus_up ();
 513
 514   /* TODO: wait a bit for the bus to come up.. ideally session_bus_up() won't return
 515    * until one can connect to the bus but that's not how things work right now
 516    */
 517   usleep (500 * 1000);
 518
 519   /* this is safe; testserver will exit once the bus goes away */
 520   g_assert (g_spawn_command_line_async (SRCDIR "/gdbus-testserver.py", NULL));
 521
 522   /* wait for the service to come up */
 523   usleep (500 * 1000);
 524
 525   /* Create the connection in the main thread */
 526   error = NULL;
 527   c = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
 528   g_assert_no_error (error);
 529   g_assert (c != NULL);
 530
 531   g_test_add_func ("/gdbus/delivery-in-thread", test_delivery_in_thread);
 532   g_test_add_func ("/gdbus/method-calls-in-thread", test_method_calls_in_thread);
 533
 534   ret = g_test_run();
 535
 536   g_object_unref (c);
 537
 538   /* tear down bus */
 539   session_bus_down ();
 540
 541   return ret;
 542 }