glib/gio/tests/unix-streams.c

/* GLib testing framework examples and tests
 * Copyright (C) 2008 Red Hat, Inc
 *
 * This work is provided "as is"; redistribution and modification
 * in whole or in part, in any medium, physical or electronic is
 * permitted without restriction.
 *
 * This work 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.
 *
 * In no event shall the authors or contributors be liable for any
 * direct, indirect, incidental, special, exemplary, or consequential
 * damages (including, but not limited to, procurement of substitute
 * goods or services; loss of use, data, or profits; or business
 * interruption) however caused and on any theory of liability, whether
 * in contract, strict liability, or tort (including negligence or
 * otherwise) arising in any way out of the use of this software, even
 * if advised of the possibility of such damage.
 */

#include <glib/glib.h>
#include <gio/gio.h>
#include <gio/gunixinputstream.h>
#include <gio/gunixoutputstream.h>
#include <signal.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define DATA "abcdefghijklmnopqrstuvwxyz"

int writer_pipe[2], reader_pipe[2];
GCancellable *writer_cancel, *reader_cancel, *main_cancel;
GMainLoop *loop;

static gboolean
cancel_main (gpointer data)
{
  GCancellable *main_cancel = data;

  g_cancellable_cancel (main_cancel);

  return FALSE;
}


static gpointer
writer_thread (gpointer user_data)
{
  GOutputStream *out;
  gssize nwrote, offset;
  GError *err = NULL;

  out = g_unix_output_stream_new (writer_pipe[1], TRUE);

  do
    {
      g_usleep (10);

      offset = 0;
      while (offset < (gssize) sizeof (DATA))
	{
	  nwrote = g_output_stream_write (out, DATA + offset,
					  sizeof (DATA) - offset,
					  writer_cancel, &err);
	  if (nwrote <= 0 || err != NULL)
	    break;
	  offset += nwrote;
	}

      g_assert (nwrote > 0 || err != NULL);
    }
  while (err == NULL);

  if (g_cancellable_is_cancelled (writer_cancel))
    {
      /* FIXME: directly calling g_cancellable_cancel (main_cancel) here
       * leads to sporadic deadlock, because it will try to wake up the
       * main context, for which it needs to acquire the main context lock.
       * This lock may be held by the main loop running in the main thread,
       * and it may be held while the main thread is blocking in
       * fd_source_finalize -> g_cancellable_disconnect
       * until the ::cancelled callbacks have run.
       *
       * Work around by deferring the cancellation to a timeout.
       */
      g_timeout_add (0, cancel_main, main_cancel);
      g_object_unref (out);
      return NULL;
    }

  g_warning ("writer: %s", err->message);
  g_assert_not_reached ();
}

static gpointer
reader_thread (gpointer user_data)
{
  GInputStream *in;
  gssize nread = 0, total;
  GError *err = NULL;
  char buf[sizeof (DATA)];

  in = g_unix_input_stream_new (reader_pipe[0], TRUE);

  do
    {
      total = 0;
      while (total < (gssize) sizeof (DATA))
	{
	  nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
				       reader_cancel, &err);
	  if (nread <= 0 || err != NULL)
	    break;
	  total += nread;
	}

      if (err)
	break;

      if (nread == 0)
	{
	  g_assert (err == NULL);
	  /* pipe closed */
	  g_object_unref (in);
	  return NULL;
	}

      g_assert_cmpstr (buf, ==, DATA);
      g_assert (!g_cancellable_is_cancelled (reader_cancel));
    }
  while (err == NULL);

  g_warning ("reader: %s", err->message);
  g_assert_not_reached ();
}

char main_buf[sizeof (DATA)];
gssize main_len, main_offset;

static void readable (GObject *source, GAsyncResult *res, gpointer user_data);
static void writable (GObject *source, GAsyncResult *res, gpointer user_data);

static void
do_main_cancel (GOutputStream *out)
{
  g_output_stream_close (out, NULL, NULL);
  g_main_loop_quit (loop);
}

static void
readable (GObject *source, GAsyncResult *res, gpointer user_data)
{
  GInputStream *in = G_INPUT_STREAM (source);
  GOutputStream *out = user_data;
  GError *err = NULL;

  main_len = g_input_stream_read_finish (in, res, &err);

  if (g_cancellable_is_cancelled (main_cancel))
    {
      do_main_cancel (out);
      return;
    }

  g_assert (err == NULL);

  main_offset = 0;
  g_output_stream_write_async (out, main_buf, main_len,
			       G_PRIORITY_DEFAULT, main_cancel,
			       writable, in);
}

static void
writable (GObject *source, GAsyncResult *res, gpointer user_data)
{
  GOutputStream *out = G_OUTPUT_STREAM (source);
  GInputStream *in = user_data;
  GError *err = NULL;
  gssize nwrote;

  nwrote = g_output_stream_write_finish (out, res, &err);

  if (g_cancellable_is_cancelled (main_cancel))
    {
      do_main_cancel (out);
      return;
    }

  g_assert (err == NULL);
  g_assert_cmpint (nwrote, <=, main_len - main_offset);

  main_offset += nwrote;
  if (main_offset == main_len)
    {
      g_input_stream_read_async (in, main_buf, sizeof (main_buf),
				 G_PRIORITY_DEFAULT, main_cancel,
				 readable, out);
    }
  else
    {
      g_output_stream_write_async (out, main_buf + main_offset,
				   main_len - main_offset,
				   G_PRIORITY_DEFAULT, main_cancel,
				   writable, in);
    }
}

static gboolean
timeout (gpointer cancellable)
{
  g_cancellable_cancel (cancellable);
  return FALSE;
}

static void
test_pipe_io (void)
{
  GThread *writer, *reader;
  GInputStream *in;
  GOutputStream *out;

  /* Split off two (additional) threads, a reader and a writer. From
   * the writer thread, write data synchronously in small chunks,
   * which gets read asynchronously by the main thread and then
   * written asynchronously to the reader thread, which reads it
   * synchronously. Eventually a timeout in the main thread will cause
   * it to cancel the writer thread, which will in turn cancel the
   * read op in the main thread, which will then close the pipe to
   * the reader thread, causing the read op to fail.
   */

  g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);

  writer_cancel = g_cancellable_new ();
  reader_cancel = g_cancellable_new ();
  main_cancel = g_cancellable_new ();

  writer = g_thread_create (writer_thread, NULL, TRUE, NULL);
  reader = g_thread_create (reader_thread, NULL, TRUE, NULL);

  in = g_unix_input_stream_new (writer_pipe[0], TRUE);
  out = g_unix_output_stream_new (reader_pipe[1], TRUE);

  g_input_stream_read_async (in, main_buf, sizeof (main_buf),
			     G_PRIORITY_DEFAULT, main_cancel,
			     readable, out);

  g_timeout_add (500, timeout, writer_cancel);

  loop = g_main_loop_new (NULL, TRUE);
  g_main_loop_run (loop);
  g_main_loop_unref (loop);

  g_thread_join (reader);
  g_thread_join (writer);

  g_object_unref (main_cancel);
  g_object_unref (reader_cancel);
  g_object_unref (writer_cancel);
  g_object_unref (in);
  g_object_unref (out);
}

int
main (int   argc,
      char *argv[])
{
  g_thread_init (NULL);
  g_type_init ();
  g_test_init (&argc, &argv, NULL);

  g_test_add_func ("/unix-streams/pipe-io-test", test_pipe_io);

  return g_test_run();
}