gio/tests/gdbus-connection-flush.c

   1 /* Test case for GNOME #662395
   2  *
   3  * Copyright (C) 2008-2010 Red Hat, Inc.
   4  * Copyright (C) 2011 Nokia Corporation
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2.1 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General
  17  * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
  18  *
  19  * Author: Simon McVittie <simon.mcvittie@collabora.co.uk>
  20  */
  21
  22 #include <config.h>
  23
  24 #include <unistd.h>
  25 #include <string.h>
  26
  27 #include <gio/gio.h>
  28
  29 #include "test-io-stream.h"
  30 #include "test-pipe-unix.h"
  31
  32 #define MY_TYPE_OUTPUT_STREAM \
  33         (my_output_stream_get_type ())
  34 #define MY_OUTPUT_STREAM(o) \
  35         (G_TYPE_CHECK_INSTANCE_CAST ((o), \
  36                                      MY_TYPE_OUTPUT_STREAM, \
  37                                      MyOutputStream))
  38 #define MY_IS_OUTPUT_STREAM(o) \
  39         (G_TYPE_CHECK_INSTANCE_TYPE ((o), MY_TYPE_OUTPUT_STREAM))
  40
  41 G_LOCK_DEFINE_STATIC (write);
  42
  43 typedef struct {
  44     GFilterOutputStream parent;
  45
  46     volatile gint started;
  47     volatile gint finished;
  48     volatile gint flushed;
  49
  50     GOutputStream *real_output;
  51 } MyOutputStream;
  52
  53 typedef struct {
  54     GFilterOutputStreamClass parent;
  55 } MyOutputStreamClass;
  56
  57 static GType my_output_stream_get_type (void) G_GNUC_CONST;
  58
  59 G_DEFINE_TYPE (MyOutputStream, my_output_stream, G_TYPE_FILTER_OUTPUT_STREAM)
  60
  61 /* Called from GDBusWorker thread */
  62 static gssize
  63 my_output_stream_write (GOutputStream  *os,
  64                         const void     *buffer,
  65                         gsize           count,
  66                         GCancellable   *cancellable,
  67                         GError        **error)
  68 {
  69   MyOutputStream *self = MY_OUTPUT_STREAM (os);
  70   GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
  71   GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
  72   gssize ret;
  73
  74   g_atomic_int_add (&self->started, count);
  75   /* Other threads can make writing block forever by taking this lock */
  76   G_LOCK (write);
  77   ret = g_output_stream_write (real, buffer, count, cancellable, error);
  78   G_UNLOCK (write);
  79   g_atomic_int_add (&self->finished, count);
  80   return ret;
  81 }
  82
  83 /* Called from GDBusWorker thread */
  84 static gboolean
  85 my_output_stream_flush (GOutputStream             *os,
  86                         GCancellable              *cancellable,
  87                         GError                   **error)
  88 {
  89   MyOutputStream *self = MY_OUTPUT_STREAM (os);
  90   GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
  91   GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
  92   gint started, finished;
  93   gboolean ret;
  94
  95   /* These should be equal because you're not allowed to flush with a
  96    * write pending, and GOutputStream enforces that for its subclasses
  97    */
  98   started = g_atomic_int_get (&self->started);
  99   finished = g_atomic_int_get (&self->finished);
 100   g_assert_cmpint (started, ==, finished);
 101
 102   ret = g_output_stream_flush (real, cancellable, error);
 103
 104   /* As above, this shouldn't have changed during the flush */
 105   finished = g_atomic_int_get (&self->finished);
 106   g_assert_cmpint (started, ==, finished);
 107
 108   /* Checkpoint reached */
 109   g_atomic_int_set (&self->flushed, finished);
 110   return ret;
 111 }
 112
 113 /* Called from any thread; thread-safe */
 114 static gint
 115 my_output_stream_get_bytes_started (GOutputStream *os)
 116 {
 117   MyOutputStream *self = MY_OUTPUT_STREAM (os);
 118
 119   return g_atomic_int_get (&self->started);
 120 }
 121
 122 /* Called from any thread; thread-safe */
 123 static gint
 124 my_output_stream_get_bytes_finished (GOutputStream *os)
 125 {
 126   MyOutputStream *self = MY_OUTPUT_STREAM (os);
 127
 128   return g_atomic_int_get (&self->finished);
 129 }
 130
 131 /* Called from any thread; thread-safe */
 132 static gint
 133 my_output_stream_get_bytes_flushed (GOutputStream *os)
 134 {
 135   MyOutputStream *self = MY_OUTPUT_STREAM (os);
 136
 137   return g_atomic_int_get (&self->flushed);
 138 }
 139
 140 static void
 141 my_output_stream_init (MyOutputStream *self)
 142 {
 143 }
 144
 145 static void
 146 my_output_stream_class_init (MyOutputStreamClass *cls)
 147 {
 148   GOutputStreamClass *ostream_class = (GOutputStreamClass *) cls;
 149
 150   ostream_class->write_fn = my_output_stream_write;
 151   ostream_class->flush = my_output_stream_flush;
 152 }
 153
 154 /* ---------------------------------------------------------------------------------------------------- */
 155
 156 typedef struct {
 157     GError *error;
 158     gchar *guid;
 159     gboolean flushed;
 160
 161     GIOStream *client_stream;
 162     GInputStream *client_istream;
 163     GOutputStream *client_ostream;
 164     GOutputStream *client_real_ostream;
 165     GDBusConnection *client_conn;
 166
 167     GIOStream *server_stream;
 168     GInputStream *server_istream;
 169     GOutputStream *server_ostream;
 170     GDBusConnection *server_conn;
 171 } Fixture;
 172
 173 static void
 174 setup_client_cb (GObject      *source,
 175                  GAsyncResult *res,
 176                  gpointer      user_data)
 177 {
 178   Fixture *f = user_data;
 179
 180   f->client_conn = g_dbus_connection_new_finish (res, &f->error);
 181   g_assert_no_error (f->error);
 182   g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
 183   g_assert (f->client_conn == G_DBUS_CONNECTION (source));
 184 }
 185
 186 static void
 187 setup_server_cb (GObject      *source,
 188                  GAsyncResult *res,
 189                  gpointer      user_data)
 190 {
 191   Fixture *f = user_data;
 192
 193   f->server_conn = g_dbus_connection_new_finish (res, &f->error);
 194   g_assert_no_error (f->error);
 195   g_assert (G_IS_DBUS_CONNECTION (f->server_conn));
 196   g_assert (f->server_conn == G_DBUS_CONNECTION (source));
 197 }
 198
 199 static void
 200 setup (Fixture       *f,
 201        gconstpointer  test_data G_GNUC_UNUSED)
 202 {
 203   gboolean ok;
 204
 205   f->guid = g_dbus_generate_guid ();
 206
 207   ok = test_pipe (&f->server_istream, &f->client_real_ostream, &f->error);
 208   g_assert_no_error (f->error);
 209   g_assert (G_IS_OUTPUT_STREAM (f->client_real_ostream));
 210   g_assert (G_IS_INPUT_STREAM (f->server_istream));
 211   g_assert (ok);
 212
 213   f->client_ostream = g_object_new (MY_TYPE_OUTPUT_STREAM,
 214                                     "base-stream", f->client_real_ostream,
 215                                     "close-base-stream", TRUE,
 216                                     NULL);
 217   g_assert (G_IS_OUTPUT_STREAM (f->client_ostream));
 218
 219   ok = test_pipe (&f->client_istream, &f->server_ostream, &f->error);
 220   g_assert_no_error (f->error);
 221   g_assert (G_IS_OUTPUT_STREAM (f->server_ostream));
 222   g_assert (G_IS_INPUT_STREAM (f->client_istream));
 223   g_assert (ok);
 224
 225   f->client_stream = test_io_stream_new (f->client_istream, f->client_ostream);
 226   f->server_stream = test_io_stream_new (f->server_istream, f->server_ostream);
 227
 228   g_dbus_connection_new (f->client_stream, NULL,
 229                          G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_CLIENT,
 230                          NULL, NULL, setup_client_cb, f);
 231   g_dbus_connection_new (f->server_stream, f->guid,
 232                          G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_SERVER,
 233                          NULL, NULL, setup_server_cb, f);
 234
 235   while (f->client_conn == NULL || f->server_conn == NULL)
 236     g_main_context_iteration (NULL, TRUE);
 237 }
 238
 239 static void
 240 flush_cb (GObject      *source,
 241           GAsyncResult *res,
 242           gpointer      user_data)
 243 {
 244   Fixture *f = user_data;
 245   gboolean ok;
 246
 247   g_assert (G_IS_DBUS_CONNECTION (source));
 248   g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
 249   g_assert_cmpuint ((guintptr) f->client_conn, ==, (guintptr) G_DBUS_CONNECTION (source));
 250
 251   ok = g_dbus_connection_flush_finish (f->client_conn, res, &f->error);
 252   g_assert_no_error (f->error);
 253   g_assert (ok);
 254
 255   f->flushed = TRUE;
 256 }
 257
 258 static void
 259 test_flush_busy (Fixture       *f,
 260                  gconstpointer  test_data G_GNUC_UNUSED)
 261 {
 262   gint initial, started;
 263   gboolean ok;
 264
 265   initial = my_output_stream_get_bytes_started (f->client_ostream);
 266   /* make sure the actual write will block */
 267   G_LOCK (write);
 268
 269   ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
 270                                       "com.example.Foo", "SomeSignal", NULL,
 271                                       &f->error);
 272   g_assert_no_error (f->error);
 273   g_assert (ok);
 274
 275   /* wait for at least part of the message to have started writing -
 276    * the write will block indefinitely in the worker thread
 277    */
 278   do {
 279     started = my_output_stream_get_bytes_started (f->client_ostream);
 280     g_thread_yield ();
 281   } while (initial >= started);
 282
 283   /* we haven't flushed anything */
 284   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
 285                    <=, initial);
 286
 287   /* start to flush: it can't happen til the write finishes */
 288   g_dbus_connection_flush (f->client_conn, NULL, flush_cb, f);
 289
 290   /* we still haven't actually flushed anything */
 291   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
 292                    <=, initial);
 293
 294   /* let the write finish */
 295   G_UNLOCK (write);
 296
 297   /* wait for the flush to happen */
 298   while (!f->flushed)
 299     g_main_context_iteration (NULL, TRUE);
 300
 301   /* now we have flushed at least what we'd written - but before fixing
 302    * GNOME#662395 this assertion would fail
 303    */
 304   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
 305                    >=, started);
 306 }
 307
 308 static void
 309 test_flush_idle (Fixture       *f,
 310                  gconstpointer  test_data G_GNUC_UNUSED)
 311 {
 312   gint initial, finished;
 313   gboolean ok;
 314
 315   initial = my_output_stream_get_bytes_finished (f->client_ostream);
 316
 317   ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
 318                                       "com.example.Foo", "SomeSignal", NULL,
 319                                       &f->error);
 320   g_assert_no_error (f->error);
 321   g_assert (ok);
 322
 323   /* wait for at least part of the message to have been written */
 324   do {
 325     finished = my_output_stream_get_bytes_finished (f->client_ostream);
 326     g_thread_yield ();
 327   } while (initial >= finished);
 328
 329   /* we haven't flushed anything */
 330   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
 331                    <=, initial);
 332
 333   /* flush with fully-written, but unflushed, messages */
 334   ok = g_dbus_connection_flush_sync (f->client_conn, NULL, &f->error);
 335
 336   /* now we have flushed at least what we'd written - but before fixing
 337    * GNOME#662395 this assertion would fail
 338    */
 339   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
 340                    >=, finished);
 341 }
 342
 343 static void
 344 teardown (Fixture       *f,
 345           gconstpointer  test_data G_GNUC_UNUSED)
 346 {
 347   g_clear_error (&f->error);
 348
 349   g_clear_object (&f->client_stream);
 350   g_clear_object (&f->client_istream);
 351   g_clear_object (&f->client_ostream);
 352   g_clear_object (&f->client_real_ostream);
 353   g_clear_object (&f->client_conn);
 354
 355   g_clear_object (&f->server_stream);
 356   g_clear_object (&f->server_istream);
 357   g_clear_object (&f->server_ostream);
 358   g_clear_object (&f->server_conn);
 359
 360   g_free (f->guid);
 361 }
 362
 363 /* ---------------------------------------------------------------------------------------------------- */
 364
 365 int
 366 main (int   argc,
 367       char *argv[])
 368 {
 369   gint ret;
 370
 371   g_test_init (&argc, &argv, NULL);
 372
 373   g_test_add ("/gdbus/connection/flush/busy", Fixture, NULL,
 374               setup, test_flush_busy, teardown);
 375   g_test_add ("/gdbus/connection/flush/idle", Fixture, NULL,
 376               setup, test_flush_idle, teardown);
 377
 378   ret = g_test_run();
 379
 380   return ret;
 381 }