glib/gio/tests/gdbus-connection-flush.c

/* Test case for GNOME #662395
 *
 * Copyright (C) 2008-2010 Red Hat, Inc.
 * Copyright (C) 2011 Nokia Corporation
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General
 * Public License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307, USA.
 *
 * Author: Simon McVittie <simon.mcvittie@collabora.co.uk>
 */

#include <config.h>

#include <unistd.h>
#include <string.h>

#include <gio/gio.h>

#include "test-io-stream.h"
#include "test-pipe-unix.h"

#define MY_TYPE_OUTPUT_STREAM \
        (my_output_stream_get_type ())
#define MY_OUTPUT_STREAM(o) \
        (G_TYPE_CHECK_INSTANCE_CAST ((o), \
                                     MY_TYPE_OUTPUT_STREAM, \
                                     MyOutputStream))
#define MY_IS_OUTPUT_STREAM(o) \
        (G_TYPE_CHECK_INSTANCE_TYPE ((o), MY_TYPE_OUTPUT_STREAM))

G_LOCK_DEFINE_STATIC (write);

typedef struct {
    GFilterOutputStream parent;

    volatile gint started;
    volatile gint finished;
    volatile gint flushed;

    GOutputStream *real_output;
} MyOutputStream;

typedef struct {
    GFilterOutputStreamClass parent;
} MyOutputStreamClass;

static GType my_output_stream_get_type (void) G_GNUC_CONST;

G_DEFINE_TYPE (MyOutputStream, my_output_stream, G_TYPE_FILTER_OUTPUT_STREAM)

/* Called from GDBusWorker thread */
static gssize
my_output_stream_write (GOutputStream  *os,
                        const void     *buffer,
                        gsize           count,
                        GCancellable   *cancellable,
                        GError        **error)
{
  MyOutputStream *self = MY_OUTPUT_STREAM (os);
  GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
  GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
  gssize ret;

  g_atomic_int_add (&self->started, count);
  /* Other threads can make writing block forever by taking this lock */
  G_LOCK (write);
  ret = g_output_stream_write (real, buffer, count, cancellable, error);
  G_UNLOCK (write);
  g_atomic_int_add (&self->finished, count);
  return ret;
}

/* Called from GDBusWorker thread */
static gboolean
my_output_stream_flush (GOutputStream             *os,
                        GCancellable              *cancellable,
                        GError                   **error)
{
  MyOutputStream *self = MY_OUTPUT_STREAM (os);
  GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
  GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
  gint started, finished;
  gboolean ret;

  /* These should be equal because you're not allowed to flush with a
   * write pending, and GOutputStream enforces that for its subclasses
   */
  started = g_atomic_int_get (&self->started);
  finished = g_atomic_int_get (&self->finished);
  g_assert_cmpint (started, ==, finished);

  ret = g_output_stream_flush (real, cancellable, error);

  /* As above, this shouldn't have changed during the flush */
  finished = g_atomic_int_get (&self->finished);
  g_assert_cmpint (started, ==, finished);

  /* Checkpoint reached */
  g_atomic_int_set (&self->flushed, finished);
  return ret;
}

/* Called from any thread; thread-safe */
static gint
my_output_stream_get_bytes_started (GOutputStream *os)
{
  MyOutputStream *self = MY_OUTPUT_STREAM (os);

  return g_atomic_int_get (&self->started);
}

/* Called from any thread; thread-safe */
static gint
my_output_stream_get_bytes_finished (GOutputStream *os)
{
  MyOutputStream *self = MY_OUTPUT_STREAM (os);

  return g_atomic_int_get (&self->finished);
}

/* Called from any thread; thread-safe */
static gint
my_output_stream_get_bytes_flushed (GOutputStream *os)
{
  MyOutputStream *self = MY_OUTPUT_STREAM (os);

  return g_atomic_int_get (&self->flushed);
}

static void
my_output_stream_init (MyOutputStream *self)
{
}

static void
my_output_stream_class_init (MyOutputStreamClass *cls)
{
  GOutputStreamClass *ostream_class = (GOutputStreamClass *) cls;

  ostream_class->write_fn = my_output_stream_write;
  ostream_class->flush = my_output_stream_flush;
}

/* ---------------------------------------------------------------------------------------------------- */

typedef struct {
    GError *error;
    gchar *guid;
    gboolean flushed;

    GIOStream *client_stream;
    GInputStream *client_istream;
    GOutputStream *client_ostream;
    GOutputStream *client_real_ostream;
    GDBusConnection *client_conn;

    GIOStream *server_stream;
    GInputStream *server_istream;
    GOutputStream *server_ostream;
    GDBusConnection *server_conn;
} Fixture;

static void
setup_client_cb (GObject      *source,
                 GAsyncResult *res,
                 gpointer      user_data)
{
  Fixture *f = user_data;

  f->client_conn = g_dbus_connection_new_finish (res, &f->error);
  g_assert_no_error (f->error);
  g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
  g_assert (f->client_conn == G_DBUS_CONNECTION (source));
}

static void
setup_server_cb (GObject      *source,
                 GAsyncResult *res,
                 gpointer      user_data)
{
  Fixture *f = user_data;

  f->server_conn = g_dbus_connection_new_finish (res, &f->error);
  g_assert_no_error (f->error);
  g_assert (G_IS_DBUS_CONNECTION (f->server_conn));
  g_assert (f->server_conn == G_DBUS_CONNECTION (source));
}

static void
setup (Fixture       *f,
       gconstpointer  test_data G_GNUC_UNUSED)
{
  gboolean ok;

  f->guid = g_dbus_generate_guid ();

  ok = test_pipe (&f->server_istream, &f->client_real_ostream, &f->error);
  g_assert_no_error (f->error);
  g_assert (G_IS_OUTPUT_STREAM (f->client_real_ostream));
  g_assert (G_IS_INPUT_STREAM (f->server_istream));
  g_assert (ok);

  f->client_ostream = g_object_new (MY_TYPE_OUTPUT_STREAM,
                                    "base-stream", f->client_real_ostream,
                                    "close-base-stream", TRUE,
                                    NULL);
  g_assert (G_IS_OUTPUT_STREAM (f->client_ostream));

  ok = test_pipe (&f->client_istream, &f->server_ostream, &f->error);
  g_assert_no_error (f->error);
  g_assert (G_IS_OUTPUT_STREAM (f->server_ostream));
  g_assert (G_IS_INPUT_STREAM (f->client_istream));
  g_assert (ok);

  f->client_stream = test_io_stream_new (f->client_istream, f->client_ostream);
  f->server_stream = test_io_stream_new (f->server_istream, f->server_ostream);

  g_dbus_connection_new (f->client_stream, NULL,
                         G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_CLIENT,
                         NULL, NULL, setup_client_cb, f);
  g_dbus_connection_new (f->server_stream, f->guid,
                         G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_SERVER,
                         NULL, NULL, setup_server_cb, f);

  while (f->client_conn == NULL || f->server_conn == NULL)
    g_main_context_iteration (NULL, TRUE);
}

static void
flush_cb (GObject      *source,
          GAsyncResult *res,
          gpointer      user_data)
{
  Fixture *f = user_data;
  gboolean ok;

  g_assert (G_IS_DBUS_CONNECTION (source));
  g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
  g_assert_cmpuint ((guintptr) f->client_conn, ==, (guintptr) G_DBUS_CONNECTION (source));

  ok = g_dbus_connection_flush_finish (f->client_conn, res, &f->error);
  g_assert_no_error (f->error);
  g_assert (ok);

  f->flushed = TRUE;
}

static void
test_flush_busy (Fixture       *f,
                 gconstpointer  test_data G_GNUC_UNUSED)
{
  gint initial, started;
  gboolean ok;

  initial = my_output_stream_get_bytes_started (f->client_ostream);
  /* make sure the actual write will block */
  G_LOCK (write);

  ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
                                      "com.example.Foo", "SomeSignal", NULL,
                                      &f->error);
  g_assert_no_error (f->error);
  g_assert (ok);

  /* wait for at least part of the message to have started writing -
   * the write will block indefinitely in the worker thread
   */
  do {
    started = my_output_stream_get_bytes_started (f->client_ostream);
    g_thread_yield ();
  } while (initial >= started);

  /* we haven't flushed anything */
  g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
                   <=, initial);

  /* start to flush: it can't happen til the write finishes */
  g_dbus_connection_flush (f->client_conn, NULL, flush_cb, f);

  /* we still haven't actually flushed anything */
  g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
                   <=, initial);

  /* let the write finish */
  G_UNLOCK (write);

  /* wait for the flush to happen */
  while (!f->flushed)
    g_main_context_iteration (NULL, TRUE);

  /* now we have flushed at least what we'd written - but before fixing
   * GNOME#662395 this assertion would fail
   */
  g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
                   >=, started);
}

static void
test_flush_idle (Fixture       *f,
                 gconstpointer  test_data G_GNUC_UNUSED)
{
  gint initial, finished;
  gboolean ok;

  initial = my_output_stream_get_bytes_finished (f->client_ostream);

  ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
                                      "com.example.Foo", "SomeSignal", NULL,
                                      &f->error);
  g_assert_no_error (f->error);
  g_assert (ok);

  /* wait for at least part of the message to have been written */
  do {
    finished = my_output_stream_get_bytes_finished (f->client_ostream);
    g_thread_yield ();
  } while (initial >= finished);

  /* we haven't flushed anything */
  g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
                   <=, initial);

  /* flush with fully-written, but unflushed, messages */
  ok = g_dbus_connection_flush_sync (f->client_conn, NULL, &f->error);

  /* now we have flushed at least what we'd written - but before fixing
   * GNOME#662395 this assertion would fail
   */
  g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
                   >=, finished);
}

static void
teardown (Fixture       *f,
          gconstpointer  test_data G_GNUC_UNUSED)
{
  g_clear_error (&f->error);

  g_clear_object (&f->client_stream);
  g_clear_object (&f->client_istream);
  g_clear_object (&f->client_ostream);
  g_clear_object (&f->client_real_ostream);
  g_clear_object (&f->client_conn);

  g_clear_object (&f->server_stream);
  g_clear_object (&f->server_istream);
  g_clear_object (&f->server_ostream);
  g_clear_object (&f->server_conn);

  g_free (f->guid);
}

/* ---------------------------------------------------------------------------------------------------- */

int
main (int   argc,
      char *argv[])
{
  gint ret;

  g_test_init (&argc, &argv, NULL);

  g_test_add ("/gdbus/connection/flush/busy", Fixture, NULL,
              setup, test_flush_busy, teardown);
  g_test_add ("/gdbus/connection/flush/idle", Fixture, NULL,
              setup, test_flush_idle, teardown);

  ret = g_test_run();

  return ret;
}