test/thread-blocking.c

   1 /* -*- mode: C; c-file-style: "gnu"; indent-tabs-mode: nil; -*- */
   2 /*
   3  * Regression test for fd.o #102839: blocking on pending calls in threads
   4  *
   5  * Copyright © 2018 KPIT Technologies Ltd.
   6  * Copyright © 2018 Manish Narang <manrock007@gmail.com>
   7  * Copyright © 2018 Collabora Ltd.
   8  *
   9  * Licensed under the Academic Free License version 2.1
  10  *
  11  * This program is free software; you can redistribute it and/or modify
  12  * it under the terms of the GNU General Public License as published by
  13  * the Free Software Foundation; either version 2 of the License, or
  14  * (at your option) any later version.
  15  *
  16  * This program is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  * GNU General Public License for more details.
  20  */
  21
  22 #include <config.h>
  23
  24 #include <dbus/dbus.h>
  25
  26 #include <glib.h>
  27 #include <gio/gio.h>
  28
  29 #include "test-utils-glib.h"
  30
  31 typedef struct
  32 {
  33   DBusError e;
  34   GError *ge;
  35   GPid daemon_pid;
  36   gchar *address;
  37
  38   DBusConnection *service_conn;
  39   GThread *service_thread;
  40   const gchar *service_name;
  41
  42   GThread *client_dispatcher_thread;
  43   GThread **client_caller_threads;
  44   DBusConnection *client_conn;
  45   GMutex callers_remaining_mutex;
  46   GCond callers_remaining_cond;
  47   gsize callers_remaining;
  48
  49   gsize n_caller_threads;
  50   gsize calls_per_thread;
  51 } Fixture;
  52
  53 static DBusHandlerResult
  54 echo_filter (DBusConnection *connection,
  55              DBusMessage *message,
  56              void *user_data)
  57 {
  58   DBusMessage *reply;
  59
  60   if (dbus_message_get_type (message) != DBUS_MESSAGE_TYPE_METHOD_CALL)
  61     return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
  62
  63   reply = dbus_message_new_method_return (message);
  64
  65   if (reply == NULL ||
  66       !dbus_connection_send (connection, reply, NULL))
  67     g_error ("OOM");
  68
  69   dbus_clear_message (&reply);
  70
  71   return DBUS_HANDLER_RESULT_HANDLED;
  72 }
  73
  74 /*
  75  * Thread function to dispatch the service connection. This function runs
  76  * in its own thread, dispatching the service connection and replying to
  77  * method calls from the callers, until the service connection is
  78  * disconnected by the main thread.
  79  */
  80 static gpointer
  81 service_thread_cb (gpointer user_data)
  82 {
  83   Fixture *f = user_data;
  84
  85   /* In principle we should be able to wait indefinitely (-1) but that
  86    * seems to prevent closing the connection from the main thread from
  87    * having its desired effect */
  88   while (dbus_connection_read_write_dispatch (f->service_conn, 1000))
  89     {
  90       /* Disconnected message not received; keep processing */
  91     }
  92
  93   return NULL;
  94 }
  95
  96 /*
  97  * Much like service_thread_cb, but for the client connection. In a real
  98  * program this would often be the main thread, running an event loop
  99  * like the one provided by GLib.
 100  */
 101 static gpointer
 102 client_dispatcher_thread_cb (gpointer user_data)
 103 {
 104   Fixture *f = user_data;
 105
 106   /* This thread needs to yield occasionally, otherwise the caller
 107    * threads won't ever get a chance to send their messages.
 108    * This is not a recommended I/O pattern, but in principle it was
 109    * always meant to work... */
 110   while (dbus_connection_read_write_dispatch (f->client_conn, 10))
 111     {
 112       /* Disconnected message not received; keep processing */
 113     }
 114
 115   return NULL;
 116 }
 117
 118 /*
 119  * Worker thread representing some background task. Some programs
 120  * dispatch a DBusConnection in one thread (in this test that's the
 121  * "client dispatcher thread"), and expect to be able to block on
 122  * pending calls in other threads. This represents one of those other
 123  * threads.
 124  */
 125 static gpointer
 126 client_caller_thread_cb (gpointer user_data)
 127 {
 128   Fixture *f = user_data;
 129   gsize i;
 130
 131   for (i = 0; i < f->calls_per_thread; i++)
 132     {
 133       DBusMessage *call, *reply;
 134       DBusError error = DBUS_ERROR_INIT;
 135       gint64 time_before, time_after;
 136
 137       /* This deliberately isn't g_test_message() to avoid buffering
 138        * issues: stderr is line-buffered */
 139       if (i % 100 == 0)
 140         g_printerr ("# thread %p: %" G_GSIZE_FORMAT "/%" G_GSIZE_FORMAT "\n",
 141                         g_thread_self (), i, f->calls_per_thread);
 142
 143       call = dbus_message_new_method_call (f->service_name, "/",
 144                                            "com.example.Echo", "Echo");
 145       g_assert_nonnull (call);
 146
 147       time_before = g_get_monotonic_time ();
 148       reply = dbus_connection_send_with_reply_and_block (f->client_conn, call,
 149                                                          30000, &error);
 150       time_after = g_get_monotonic_time ();
 151       test_assert_no_error (&error);
 152       g_assert_nonnull (reply);
 153       /* We don't expect it to have taken long - a few seconds max, even
 154        * with all the other threads contending with us. If we were
 155        * anywhere near the 30 second timeout then that's a failure. */
 156       g_assert_cmpint (time_after - time_before, <=, 10 * G_USEC_PER_SEC);
 157       dbus_clear_message (&reply);
 158       dbus_clear_message (&call);
 159     }
 160
 161   g_printerr ("# thread %p: finishing\n", g_thread_self ());
 162   g_mutex_lock (&f->callers_remaining_mutex);
 163   f->callers_remaining--;
 164   g_cond_signal (&f->callers_remaining_cond);
 165   g_mutex_unlock (&f->callers_remaining_mutex);
 166
 167   return NULL;
 168 }
 169
 170 static void
 171 setup (Fixture *f,
 172        gconstpointer context)
 173 {
 174   f->ge = NULL;
 175   dbus_error_init (&f->e);
 176
 177   f->address = test_get_dbus_daemon (NULL, TEST_USER_ME, NULL, &f->daemon_pid);
 178   g_assert_nonnull (f->address);
 179
 180   f->service_conn = test_connect_to_bus (NULL, f->address);
 181   f->service_name = dbus_bus_get_unique_name (f->service_conn);
 182   f->client_conn = test_connect_to_bus (NULL, f->address);
 183
 184   if (!dbus_connection_add_filter (f->service_conn, echo_filter, f, NULL))
 185     g_error ("OOM");
 186
 187   f->service_thread = g_thread_new ("service", service_thread_cb, f);
 188   f->client_dispatcher_thread = g_thread_new ("client dispatcher",
 189                                               client_dispatcher_thread_cb, f);
 190 }
 191
 192 /*
 193  * Assert that in the following situation:
 194  *
 195  * - one thread dispatches the server connection (in real life this would
 196  *   normally be a separate process, of course)
 197  * - one thread dispatches the client connection
 198  * - many threads make blocking method calls on the same client connection
 199  *
 200  * the caller threads are regularly dispatched, and never get blocked
 201  * until their method call timeout.
 202  */
 203 static void
 204 test_threads (Fixture *f,
 205               gconstpointer context)
 206 {
 207   gsize i;
 208
 209   g_test_bug ("102839");
 210
 211   if (g_test_slow ())
 212     {
 213       test_timeout_reset (10);
 214       f->n_caller_threads = 20;
 215       f->calls_per_thread = 10000;
 216     }
 217   else
 218     {
 219       test_timeout_reset (1);
 220       f->n_caller_threads = 4;
 221       f->calls_per_thread = 1000;
 222     }
 223
 224   f->client_caller_threads = g_new0 (GThread *, f->n_caller_threads);
 225   f->callers_remaining = f->n_caller_threads;
 226
 227   /* Start the caller threads off */
 228
 229   for (i = 0; i < f->n_caller_threads; i++)
 230     {
 231       gchar *name = g_strdup_printf ("client %" G_GSIZE_FORMAT, i);
 232
 233       f->client_caller_threads[i] = g_thread_new (name,
 234                                                   client_caller_thread_cb,
 235                                                   f);
 236     }
 237
 238   /* Wait for all caller threads to exit */
 239
 240   g_mutex_lock (&f->callers_remaining_mutex);
 241
 242   while (f->callers_remaining > 0)
 243     g_cond_wait (&f->callers_remaining_cond, &f->callers_remaining_mutex);
 244
 245   g_mutex_unlock (&f->callers_remaining_mutex);
 246
 247   for (i = 0; i < f->n_caller_threads; i++)
 248     g_thread_join (g_steal_pointer (&f->client_caller_threads[i]));
 249
 250   /* If we haven't crashed out yet, then we're good! */
 251 }
 252
 253 static void
 254 teardown (Fixture *f,
 255           gconstpointer context G_GNUC_UNUSED)
 256 {
 257   dbus_error_free (&f->e);
 258   g_clear_error (&f->ge);
 259
 260   if (f->client_conn != NULL)
 261     dbus_connection_close (f->client_conn);
 262
 263   if (f->service_conn != NULL)
 264     {
 265       dbus_connection_remove_filter (f->service_conn, echo_filter, f);
 266       dbus_connection_close (f->service_conn);
 267     }
 268
 269   /* Now that the connections have been closed, the threads will exit */
 270   if (f->service_thread != NULL)
 271     g_thread_join (g_steal_pointer (&f->service_thread));
 272
 273   if (f->client_dispatcher_thread != NULL)
 274     g_thread_join (g_steal_pointer (&f->client_dispatcher_thread));
 275
 276   dbus_clear_connection (&f->service_conn);
 277   dbus_clear_connection (&f->client_conn);
 278
 279   if (f->daemon_pid != 0)
 280     {
 281       test_kill_pid (f->daemon_pid);
 282       g_spawn_close_pid (f->daemon_pid);
 283       f->daemon_pid = 0;
 284     }
 285
 286   g_free (f->address);
 287 }
 288
 289 int
 290 main (int argc,
 291     char **argv)
 292 {
 293   test_init (&argc, &argv);
 294
 295   g_test_add ("/thread-blocking", Fixture, NULL, setup, test_threads,
 296               teardown);
 297
 298   return g_test_run ();
 299 }