gio/tests/win32-streams.c - third_party/glib - Git at Google

 /* 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/gwin32inputstream.h>
 #include <gio/gwin32outputstream.h>
 #include <stdlib.h>
 #include <string.h>
 #include <fcntl.h>
 #include <io.h>

 #include <windows.h>

 #define DATA "abcdefghijklmnopqrstuvwxyz"

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

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

   g_assert (DuplicateHandle (GetCurrentProcess (),
 			     (HANDLE) (gintptr) _get_osfhandle (writer_pipe[1]),
 			     GetCurrentProcess (),
 			     &out_handle,
 			     0, FALSE,
 			     DUPLICATE_SAME_ACCESS));
   close (writer_pipe[1]);

   out = g_win32_output_stream_new (out_handle, 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))
     {
       g_cancellable_cancel (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)];
   HANDLE in_handle;

   g_assert (DuplicateHandle (GetCurrentProcess (),
 			     (HANDLE) (gintptr) _get_osfhandle (reader_pipe[0]),
 			     GetCurrentProcess (),
 			     &in_handle,
 			     0, FALSE,
 			     DUPLICATE_SAME_ACCESS));
   close (reader_pipe[0]);

   in = g_win32_input_stream_new (in_handle, 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;
   HANDLE in_handle, out_handle;

   /* 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, 10, _O_BINARY) == 0 && _pipe (reader_pipe, 10, _O_BINARY) == 0);

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

   writer = g_thread_new ("writer", writer_thread, NULL);
   reader = g_thread_new ("reader", reader_thread, NULL);

   g_assert (DuplicateHandle (GetCurrentProcess (),
 			     (HANDLE) (gintptr) _get_osfhandle (writer_pipe[0]),
 			     GetCurrentProcess (),
 			     &in_handle,
 			     0, FALSE,
 			     DUPLICATE_SAME_ACCESS));
   close (writer_pipe[0]);

   g_assert (DuplicateHandle (GetCurrentProcess (),
 			     (HANDLE) (gintptr) _get_osfhandle (reader_pipe[1]),
 			     GetCurrentProcess (),
 			     &out_handle,
 			     0, FALSE,
 			     DUPLICATE_SAME_ACCESS));
   close (reader_pipe[1]);

   in = g_win32_input_stream_new (in_handle, TRUE);
   out = g_win32_output_stream_new (out_handle, 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);
 }

 typedef struct _PipeIOOverlapReader
 {
   char buf[sizeof (DATA)];
   GInputStream *in;
   GThread *thread;
   GCancellable *cancellable;
   gboolean success;
 } PipeIOOverlapReader;

 #define TEST_PIPE_IO_OVERLAP (1024 * 4)

 static gpointer
 pipe_io_overlap_reader_thread (gpointer user_data)
 {
   PipeIOOverlapReader *p = user_data;
   GError *err = NULL;
   gsize read;
   guint i;

   for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
     memset (p->buf, 0, sizeof (p->buf));
     g_input_stream_read_all (p->in, p->buf, sizeof (p->buf),
                              &read, NULL, &err);

     g_assert_cmpuint (read, ==, sizeof (p->buf));
     g_assert_no_error (err);
     g_assert_cmpstr (p->buf, ==, DATA);
   }

   return NULL;
 }

 static gpointer
 pipe_io_overlap_writer_thread (gpointer user_data)
 {
   GOutputStream *out = user_data;
   GError *err = NULL;
   gsize bytes_written;
   guint i;

   for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
     g_output_stream_write_all (out, DATA, sizeof (DATA),
                                &bytes_written, NULL, &err);

     g_assert_cmpuint (bytes_written, ==, sizeof (DATA));
     g_assert_no_error (err);
   }

   return NULL;
 }

 static void
 test_pipe_io_overlap (void)
 {
   GOutputStream *out_server, *out_client;
   GThread *writer_server, *writer_client;
   PipeIOOverlapReader rs, rc;
   HANDLE server, client;
   gchar name[256];

   g_snprintf (name, sizeof (name),
               "\\\\.\\pipe\\gtest-io-overlap-%u", (guint) GetCurrentProcessId ());

   server = CreateNamedPipe (name,
                             PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
                             PIPE_READMODE_BYTE | PIPE_WAIT,
                             1, 0, 0, 0, NULL);
   g_assert (server != INVALID_HANDLE_VALUE);

   client = CreateFile (name, GENERIC_WRITE | GENERIC_READ, 0, NULL, OPEN_EXISTING, 0, NULL);
   g_assert (client != INVALID_HANDLE_VALUE);

   out_server = g_win32_output_stream_new (server, TRUE);
   writer_server = g_thread_new ("writer_server", pipe_io_overlap_writer_thread, out_server);
   rs.in = g_win32_input_stream_new (server, TRUE);
   rs.thread = g_thread_new ("reader_server", pipe_io_overlap_reader_thread, &rs);

   out_client = g_win32_output_stream_new (client, TRUE);
   writer_client = g_thread_new ("writer_client", pipe_io_overlap_writer_thread, out_client);
   rc.in = g_win32_input_stream_new (client, TRUE);
   rc.thread = g_thread_new ("reader_client", pipe_io_overlap_reader_thread, &rc);

   g_thread_join (writer_client);
   g_thread_join (writer_server);
   g_thread_join (rc.thread);
   g_thread_join (rs.thread);

   g_object_unref (rs.in);
   g_object_unref (rc.in);
   g_object_unref (out_server);
   g_object_unref (out_client);
 }

 static gpointer
 pipe_io_concurrent_writer_thread (gpointer user_data)
 {
   GOutputStream *out = user_data;
   GError *err = NULL;
   gsize bytes_written;

   g_output_stream_write_all (out, DATA, 1, &bytes_written, NULL, &err);

   g_assert_cmpuint (bytes_written, ==, 1);
   g_assert_no_error (err);

   return NULL;
 }

 static gpointer
 pipe_io_concurrent_reader_thread (gpointer user_data)
 {
   PipeIOOverlapReader *p = user_data;
   GError *err = NULL;
   gsize read;

   memset (p->buf, 0, sizeof (p->buf));
   p->success = g_input_stream_read_all (p->in, p->buf, 1, &read, p->cancellable, &err);

   /* only one thread will succeed, the other will be cancelled */
   if (p->success)
     {
       /* continue the main thread */
       write (writer_pipe[1], "", 1);
       g_assert_cmpuint (read, ==, 1);
       g_assert_no_error (err);
     }

   return NULL;
 }

 static void
 test_pipe_io_concurrent (void)
 {
   GOutputStream *out_server;
   GThread *writer_server;
   PipeIOOverlapReader rc1, rc2;
   HANDLE server, client;
   gchar name[256], c;

   g_snprintf (name, sizeof (name),
               "\\\\.\\pipe\\gtest-io-concurrent-%u", (guint) GetCurrentProcessId ());

   server = CreateNamedPipe (name,
                             PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
                             PIPE_READMODE_BYTE | PIPE_WAIT,
                             1, 0, 0, 0, NULL);
   g_assert (server != INVALID_HANDLE_VALUE);
   g_assert (_pipe (writer_pipe, 10, _O_BINARY) == 0);

   client = CreateFile (name, GENERIC_WRITE | GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
   g_assert (client != INVALID_HANDLE_VALUE);

   rc1.in = g_win32_input_stream_new (client, TRUE);
   rc1.success = FALSE;
   rc1.cancellable = g_cancellable_new ();
   rc1.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc1);

   rc2.in = g_win32_input_stream_new (client, TRUE);
   rc2.success = FALSE;
   rc2.cancellable = g_cancellable_new ();
   rc2.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc2);

   /* FIXME: how to synchronize on both reader thread waiting in read,
      before starting the writer thread? */
   g_usleep (G_USEC_PER_SEC / 10);

   out_server = g_win32_output_stream_new (server, TRUE);
   writer_server = g_thread_new ("writer_server", pipe_io_concurrent_writer_thread, out_server);

   read (writer_pipe[0], &c, 1);

   g_assert (rc1.success ^ rc2.success);

   g_cancellable_cancel (rc1.cancellable);
   g_cancellable_cancel (rc2.cancellable);

   g_thread_join (writer_server);
   g_thread_join (rc1.thread);
   g_thread_join (rc2.thread);

   g_object_unref (rc1.in);
   g_object_unref (rc2.in);
   g_object_unref (out_server);

   close (writer_pipe[0]);
   close (writer_pipe[1]);
 }

 static void
 readable_cancel (GObject *source, GAsyncResult *res, gpointer user_data)
 {
   GInputStream *in = G_INPUT_STREAM (source);
   GError *err = NULL;
   gssize len;

   len = g_input_stream_read_finish (in, res, &err);
   g_assert_cmpint (len, ==, -1);
   g_assert_error (err, G_IO_ERROR, G_IO_ERROR_CANCELLED);
   g_error_free (err);

   g_main_loop_quit (loop);
 }

 static void
 test_pipe_io_cancel (void)
 {
   GInputStream *in;
   GOutputStream *out;
   HANDLE in_handle, out_handle;
   gchar name[256];

   g_snprintf (name, sizeof (name),
               "\\\\.\\pipe\\gtest-io-cancel-%u", (guint) GetCurrentProcessId ());

   in_handle = CreateNamedPipe (name,
                                PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED,
                                PIPE_READMODE_BYTE | PIPE_WAIT,
                                1, 0, 0, 0, NULL);
   g_assert (in_handle != INVALID_HANDLE_VALUE);

   out_handle = CreateFile (name, GENERIC_WRITE, 0, NULL, OPEN_EXISTING, 0, NULL);
   g_assert (out_handle != INVALID_HANDLE_VALUE);

   in = g_win32_input_stream_new (in_handle, TRUE);
   out = g_win32_output_stream_new (out_handle, TRUE);

   reader_cancel = g_cancellable_new ();
   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
                              G_PRIORITY_DEFAULT, reader_cancel,
                              readable_cancel, out);

   g_timeout_add (500, timeout, reader_cancel);

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

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

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

   g_test_add_func ("/win32-streams/pipe-io-test", test_pipe_io);
   g_test_add_func ("/win32-streams/pipe-io-cancel-test", test_pipe_io_cancel);
   g_test_add_func ("/win32-streams/pipe-io-overlap-test", test_pipe_io_overlap);
   g_test_add_func ("/win32-streams/pipe-io-concurrent-test", test_pipe_io_concurrent);

   return g_test_run();
 }
	/* 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/gwin32inputstream.h>
	#include <gio/gwin32outputstream.h>
	#include <stdlib.h>
	#include <string.h>
	#include <fcntl.h>
	#include <io.h>

	#include <windows.h>

	#define DATA "abcdefghijklmnopqrstuvwxyz"

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

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

	g_assert (DuplicateHandle (GetCurrentProcess (),
	(HANDLE) (gintptr) _get_osfhandle (writer_pipe[1]),
	GetCurrentProcess (),
	&out_handle,
	0, FALSE,
	DUPLICATE_SAME_ACCESS));
	close (writer_pipe[1]);

	out = g_win32_output_stream_new (out_handle, 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))
	{
	g_cancellable_cancel (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)];
	HANDLE in_handle;

	g_assert (DuplicateHandle (GetCurrentProcess (),
	(HANDLE) (gintptr) _get_osfhandle (reader_pipe[0]),
	GetCurrentProcess (),
	&in_handle,
	0, FALSE,
	DUPLICATE_SAME_ACCESS));
	close (reader_pipe[0]);

	in = g_win32_input_stream_new (in_handle, 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;
	HANDLE in_handle, out_handle;

	/* 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, 10, _O_BINARY) == 0 && _pipe (reader_pipe, 10, _O_BINARY) == 0);

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

	writer = g_thread_new ("writer", writer_thread, NULL);
	reader = g_thread_new ("reader", reader_thread, NULL);

	g_assert (DuplicateHandle (GetCurrentProcess (),
	(HANDLE) (gintptr) _get_osfhandle (writer_pipe[0]),
	GetCurrentProcess (),
	&in_handle,
	0, FALSE,
	DUPLICATE_SAME_ACCESS));
	close (writer_pipe[0]);

	g_assert (DuplicateHandle (GetCurrentProcess (),
	(HANDLE) (gintptr) _get_osfhandle (reader_pipe[1]),
	GetCurrentProcess (),
	&out_handle,
	0, FALSE,
	DUPLICATE_SAME_ACCESS));
	close (reader_pipe[1]);

	in = g_win32_input_stream_new (in_handle, TRUE);
	out = g_win32_output_stream_new (out_handle, 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);
	}

	typedef struct _PipeIOOverlapReader
	{
	char buf[sizeof (DATA)];
	GInputStream *in;
	GThread *thread;
	GCancellable *cancellable;
	gboolean success;
	} PipeIOOverlapReader;

	#define TEST_PIPE_IO_OVERLAP (1024 * 4)

	static gpointer
	pipe_io_overlap_reader_thread (gpointer user_data)
	{
	PipeIOOverlapReader *p = user_data;
	GError *err = NULL;
	gsize read;
	guint i;

	for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
	memset (p->buf, 0, sizeof (p->buf));
	g_input_stream_read_all (p->in, p->buf, sizeof (p->buf),
	&read, NULL, &err);

	g_assert_cmpuint (read, ==, sizeof (p->buf));
	g_assert_no_error (err);
	g_assert_cmpstr (p->buf, ==, DATA);
	}

	return NULL;
	}

	static gpointer
	pipe_io_overlap_writer_thread (gpointer user_data)
	{
	GOutputStream *out = user_data;
	GError *err = NULL;
	gsize bytes_written;
	guint i;

	for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
	g_output_stream_write_all (out, DATA, sizeof (DATA),
	&bytes_written, NULL, &err);

	g_assert_cmpuint (bytes_written, ==, sizeof (DATA));
	g_assert_no_error (err);
	}

	return NULL;
	}

	static void
	test_pipe_io_overlap (void)
	{
	GOutputStream out_server, out_client;
	GThread writer_server, writer_client;
	PipeIOOverlapReader rs, rc;
	HANDLE server, client;
	gchar name[256];

	g_snprintf (name, sizeof (name),
	"\\\\.\\pipe\\gtest-io-overlap-%u", (guint) GetCurrentProcessId ());

	server = CreateNamedPipe (name,
	PIPE_ACCESS_DUPLEX \| FILE_FLAG_OVERLAPPED,
	PIPE_READMODE_BYTE \| PIPE_WAIT,
	1, 0, 0, 0, NULL);
	g_assert (server != INVALID_HANDLE_VALUE);

	client = CreateFile (name, GENERIC_WRITE \| GENERIC_READ, 0, NULL, OPEN_EXISTING, 0, NULL);
	g_assert (client != INVALID_HANDLE_VALUE);

	out_server = g_win32_output_stream_new (server, TRUE);
	writer_server = g_thread_new ("writer_server", pipe_io_overlap_writer_thread, out_server);
	rs.in = g_win32_input_stream_new (server, TRUE);
	rs.thread = g_thread_new ("reader_server", pipe_io_overlap_reader_thread, &rs);

	out_client = g_win32_output_stream_new (client, TRUE);
	writer_client = g_thread_new ("writer_client", pipe_io_overlap_writer_thread, out_client);
	rc.in = g_win32_input_stream_new (client, TRUE);
	rc.thread = g_thread_new ("reader_client", pipe_io_overlap_reader_thread, &rc);

	g_thread_join (writer_client);
	g_thread_join (writer_server);
	g_thread_join (rc.thread);
	g_thread_join (rs.thread);

	g_object_unref (rs.in);
	g_object_unref (rc.in);
	g_object_unref (out_server);
	g_object_unref (out_client);
	}

	static gpointer
	pipe_io_concurrent_writer_thread (gpointer user_data)
	{
	GOutputStream *out = user_data;
	GError *err = NULL;
	gsize bytes_written;

	g_output_stream_write_all (out, DATA, 1, &bytes_written, NULL, &err);

	g_assert_cmpuint (bytes_written, ==, 1);
	g_assert_no_error (err);

	return NULL;
	}

	static gpointer
	pipe_io_concurrent_reader_thread (gpointer user_data)
	{
	PipeIOOverlapReader *p = user_data;
	GError *err = NULL;
	gsize read;

	memset (p->buf, 0, sizeof (p->buf));
	p->success = g_input_stream_read_all (p->in, p->buf, 1, &read, p->cancellable, &err);

	/* only one thread will succeed, the other will be cancelled */
	if (p->success)
	{
	/* continue the main thread */
	write (writer_pipe[1], "", 1);
	g_assert_cmpuint (read, ==, 1);
	g_assert_no_error (err);
	}

	return NULL;
	}

	static void
	test_pipe_io_concurrent (void)
	{
	GOutputStream *out_server;
	GThread *writer_server;
	PipeIOOverlapReader rc1, rc2;
	HANDLE server, client;
	gchar name[256], c;

	g_snprintf (name, sizeof (name),
	"\\\\.\\pipe\\gtest-io-concurrent-%u", (guint) GetCurrentProcessId ());

	server = CreateNamedPipe (name,
	PIPE_ACCESS_DUPLEX \| FILE_FLAG_OVERLAPPED,
	PIPE_READMODE_BYTE \| PIPE_WAIT,
	1, 0, 0, 0, NULL);
	g_assert (server != INVALID_HANDLE_VALUE);
	g_assert (_pipe (writer_pipe, 10, _O_BINARY) == 0);

	client = CreateFile (name, GENERIC_WRITE \| GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
	g_assert (client != INVALID_HANDLE_VALUE);

	rc1.in = g_win32_input_stream_new (client, TRUE);
	rc1.success = FALSE;
	rc1.cancellable = g_cancellable_new ();
	rc1.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc1);

	rc2.in = g_win32_input_stream_new (client, TRUE);
	rc2.success = FALSE;
	rc2.cancellable = g_cancellable_new ();
	rc2.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc2);

	/* FIXME: how to synchronize on both reader thread waiting in read,
	before starting the writer thread? */
	g_usleep (G_USEC_PER_SEC / 10);

	out_server = g_win32_output_stream_new (server, TRUE);
	writer_server = g_thread_new ("writer_server", pipe_io_concurrent_writer_thread, out_server);

	read (writer_pipe[0], &c, 1);

	g_assert (rc1.success ^ rc2.success);

	g_cancellable_cancel (rc1.cancellable);
	g_cancellable_cancel (rc2.cancellable);

	g_thread_join (writer_server);
	g_thread_join (rc1.thread);
	g_thread_join (rc2.thread);

	g_object_unref (rc1.in);
	g_object_unref (rc2.in);
	g_object_unref (out_server);

	close (writer_pipe[0]);
	close (writer_pipe[1]);
	}

	static void
	readable_cancel (GObject source, GAsyncResult res, gpointer user_data)
	{
	GInputStream *in = G_INPUT_STREAM (source);
	GError *err = NULL;
	gssize len;

	len = g_input_stream_read_finish (in, res, &err);
	g_assert_cmpint (len, ==, -1);
	g_assert_error (err, G_IO_ERROR, G_IO_ERROR_CANCELLED);
	g_error_free (err);

	g_main_loop_quit (loop);
	}

	static void
	test_pipe_io_cancel (void)
	{
	GInputStream *in;
	GOutputStream *out;
	HANDLE in_handle, out_handle;
	gchar name[256];

	g_snprintf (name, sizeof (name),
	"\\\\.\\pipe\\gtest-io-cancel-%u", (guint) GetCurrentProcessId ());

	in_handle = CreateNamedPipe (name,
	PIPE_ACCESS_INBOUND \| FILE_FLAG_OVERLAPPED,
	PIPE_READMODE_BYTE \| PIPE_WAIT,
	1, 0, 0, 0, NULL);
	g_assert (in_handle != INVALID_HANDLE_VALUE);

	out_handle = CreateFile (name, GENERIC_WRITE, 0, NULL, OPEN_EXISTING, 0, NULL);
	g_assert (out_handle != INVALID_HANDLE_VALUE);

	in = g_win32_input_stream_new (in_handle, TRUE);
	out = g_win32_output_stream_new (out_handle, TRUE);

	reader_cancel = g_cancellable_new ();
	g_input_stream_read_async (in, main_buf, sizeof (main_buf),
	G_PRIORITY_DEFAULT, reader_cancel,
	readable_cancel, out);

	g_timeout_add (500, timeout, reader_cancel);

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

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

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

	g_test_add_func ("/win32-streams/pipe-io-test", test_pipe_io);
	g_test_add_func ("/win32-streams/pipe-io-cancel-test", test_pipe_io_cancel);
	g_test_add_func ("/win32-streams/pipe-io-overlap-test", test_pipe_io_overlap);
	g_test_add_func ("/win32-streams/pipe-io-concurrent-test", test_pipe_io_concurrent);

	return g_test_run();
	}