text: /* 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 <unistd.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) getpid ());
  
    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) getpid ());
  
    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) getpid ());
  
    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();
  }
  