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