gio/tests/win32-streams.c

   1 /* GLib testing framework examples and tests
   2  * Copyright (C) 2008 Red Hat, Inc
   3  *
   4  * This work is provided "as is"; redistribution and modification
   5  * in whole or in part, in any medium, physical or electronic is
   6  * permitted without restriction.
   7  *
   8  * This work is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  11  *
  12  * In no event shall the authors or contributors be liable for any
  13  * direct, indirect, incidental, special, exemplary, or consequential
  14  * damages (including, but not limited to, procurement of substitute
  15  * goods or services; loss of use, data, or profits; or business
  16  * interruption) however caused and on any theory of liability, whether
  17  * in contract, strict liability, or tort (including negligence or
  18  * otherwise) arising in any way out of the use of this software, even
  19  * if advised of the possibility of such damage.
  20  */
  21
  22 #include <glib/glib.h>
  23 #include <gio/gio.h>
  24 #include <gio/gwin32inputstream.h>
  25 #include <gio/gwin32outputstream.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <fcntl.h>
  29 #include <io.h>
  30
  31 #include <windows.h>
  32
  33 #define DATA "abcdefghijklmnopqrstuvwxyz"
  34
  35 int writer_pipe[2], reader_pipe[2];
  36 GCancellable *writer_cancel, *reader_cancel, *main_cancel;
  37 GMainLoop *loop;
  38
  39 static gpointer
  40 writer_thread (gpointer user_data)
  41 {
  42   GOutputStream *out;
  43   gssize nwrote, offset;
  44   GError *err = NULL;
  45   HANDLE out_handle;
  46
  47   g_assert (DuplicateHandle (GetCurrentProcess (),
  48                              (HANDLE) (gintptr) _get_osfhandle (writer_pipe[1]),
  49                              GetCurrentProcess (),
  50                              &out_handle,
  51                              0, FALSE,
  52                              DUPLICATE_SAME_ACCESS));
  53   close (writer_pipe[1]);
  54
  55   out = g_win32_output_stream_new (out_handle, TRUE);
  56   do
  57     {
  58       g_usleep (10);
  59
  60       offset = 0;
  61       while (offset < (gssize) sizeof (DATA))
  62         {
  63           nwrote = g_output_stream_write (out, DATA + offset,
  64                                           sizeof (DATA) - offset,
  65                                           writer_cancel, &err);
  66           if (nwrote <= 0 || err != NULL)
  67             break;
  68           offset += nwrote;
  69         }
  70
  71       g_assert (nwrote > 0 || err != NULL);
  72     }
  73   while (err == NULL);
  74
  75   if (g_cancellable_is_cancelled (writer_cancel))
  76     {
  77       g_cancellable_cancel (main_cancel);
  78       g_object_unref (out);
  79       return NULL;
  80     }
  81
  82   g_warning ("writer: %s", err->message);
  83   g_assert_not_reached ();
  84 }
  85
  86 static gpointer
  87 reader_thread (gpointer user_data)
  88 {
  89   GInputStream *in;
  90   gssize nread = 0, total;
  91   GError *err = NULL;
  92   char buf[sizeof (DATA)];
  93   HANDLE in_handle;
  94
  95   g_assert (DuplicateHandle (GetCurrentProcess (),
  96                              (HANDLE) (gintptr) _get_osfhandle (reader_pipe[0]),
  97                              GetCurrentProcess (),
  98                              &in_handle,
  99                              0, FALSE,
 100                              DUPLICATE_SAME_ACCESS));
 101   close (reader_pipe[0]);
 102
 103   in = g_win32_input_stream_new (in_handle, TRUE);
 104
 105   do
 106     {
 107       total = 0;
 108       while (total < (gssize) sizeof (DATA))
 109         {
 110           nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
 111                                        reader_cancel, &err);
 112           if (nread <= 0 || err != NULL)
 113             break;
 114           total += nread;
 115         }
 116
 117       if (err)
 118         break;
 119
 120       if (nread == 0)
 121         {
 122           g_assert (err == NULL);
 123           /* pipe closed */
 124           g_object_unref (in);
 125           return NULL;
 126         }
 127
 128       g_assert_cmpstr (buf, ==, DATA);
 129       g_assert (!g_cancellable_is_cancelled (reader_cancel));
 130     }
 131   while (err == NULL);
 132
 133   g_warning ("reader: %s", err->message);
 134   g_assert_not_reached ();
 135 }
 136
 137 char main_buf[sizeof (DATA)];
 138 gssize main_len, main_offset;
 139
 140 static void readable (GObject *source, GAsyncResult *res, gpointer user_data);
 141 static void writable (GObject *source, GAsyncResult *res, gpointer user_data);
 142
 143 static void
 144 do_main_cancel (GOutputStream *out)
 145 {
 146   g_output_stream_close (out, NULL, NULL);
 147   g_main_loop_quit (loop);
 148 }
 149
 150 static void
 151 readable (GObject *source, GAsyncResult *res, gpointer user_data)
 152 {
 153   GInputStream *in = G_INPUT_STREAM (source);
 154   GOutputStream *out = user_data;
 155   GError *err = NULL;
 156
 157   main_len = g_input_stream_read_finish (in, res, &err);
 158
 159   if (g_cancellable_is_cancelled (main_cancel))
 160     {
 161       do_main_cancel (out);
 162       return;
 163     }
 164
 165   g_assert (err == NULL);
 166
 167   main_offset = 0;
 168   g_output_stream_write_async (out, main_buf, main_len,
 169                                G_PRIORITY_DEFAULT, main_cancel,
 170                                writable, in);
 171 }
 172
 173 static void
 174 writable (GObject *source, GAsyncResult *res, gpointer user_data)
 175 {
 176   GOutputStream *out = G_OUTPUT_STREAM (source);
 177   GInputStream *in = user_data;
 178   GError *err = NULL;
 179   gssize nwrote;
 180
 181   nwrote = g_output_stream_write_finish (out, res, &err);
 182
 183   if (g_cancellable_is_cancelled (main_cancel))
 184     {
 185       do_main_cancel (out);
 186       return;
 187     }
 188
 189   g_assert (err == NULL);
 190   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 191
 192   main_offset += nwrote;
 193   if (main_offset == main_len)
 194     {
 195       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 196                                  G_PRIORITY_DEFAULT, main_cancel,
 197                                  readable, out);
 198     }
 199   else
 200     {
 201       g_output_stream_write_async (out, main_buf + main_offset,
 202                                    main_len - main_offset,
 203                                    G_PRIORITY_DEFAULT, main_cancel,
 204                                    writable, in);
 205     }
 206 }
 207
 208 static gboolean
 209 timeout (gpointer cancellable)
 210 {
 211   g_cancellable_cancel (cancellable);
 212   return FALSE;
 213 }
 214
 215 static void
 216 test_pipe_io (void)
 217 {
 218   GThread *writer, *reader;
 219   GInputStream *in;
 220   GOutputStream *out;
 221   HANDLE in_handle, out_handle;
 222
 223   /* Split off two (additional) threads, a reader and a writer. From
 224    * the writer thread, write data synchronously in small chunks,
 225    * which gets read asynchronously by the main thread and then
 226    * written asynchronously to the reader thread, which reads it
 227    * synchronously. Eventually a timeout in the main thread will cause
 228    * it to cancel the writer thread, which will in turn cancel the
 229    * read op in the main thread, which will then close the pipe to
 230    * the reader thread, causing the read op to fail.
 231    */
 232
 233   g_assert (_pipe (writer_pipe, 10, _O_BINARY) == 0 && _pipe (reader_pipe, 10, _O_BINARY) == 0);
 234
 235   writer_cancel = g_cancellable_new ();
 236   reader_cancel = g_cancellable_new ();
 237   main_cancel = g_cancellable_new ();
 238
 239   writer = g_thread_new ("writer", writer_thread, NULL);
 240   reader = g_thread_new ("reader", reader_thread, NULL);
 241
 242   g_assert (DuplicateHandle (GetCurrentProcess (),
 243                              (HANDLE) (gintptr) _get_osfhandle (writer_pipe[0]),
 244                              GetCurrentProcess (),
 245                              &in_handle,
 246                              0, FALSE,
 247                              DUPLICATE_SAME_ACCESS));
 248   close (writer_pipe[0]);
 249
 250   g_assert (DuplicateHandle (GetCurrentProcess (),
 251                              (HANDLE) (gintptr) _get_osfhandle (reader_pipe[1]),
 252                              GetCurrentProcess (),
 253                              &out_handle,
 254                              0, FALSE,
 255                              DUPLICATE_SAME_ACCESS));
 256   close (reader_pipe[1]);
 257
 258   in = g_win32_input_stream_new (in_handle, TRUE);
 259   out = g_win32_output_stream_new (out_handle, TRUE);
 260
 261   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 262                              G_PRIORITY_DEFAULT, main_cancel,
 263                              readable, out);
 264
 265   g_timeout_add (500, timeout, writer_cancel);
 266
 267   loop = g_main_loop_new (NULL, TRUE);
 268   g_main_loop_run (loop);
 269   g_main_loop_unref (loop);
 270
 271   g_thread_join (reader);
 272   g_thread_join (writer);
 273
 274   g_object_unref (main_cancel);
 275   g_object_unref (reader_cancel);
 276   g_object_unref (writer_cancel);
 277   g_object_unref (in);
 278   g_object_unref (out);
 279 }
 280
 281 typedef struct _PipeIOOverlapReader
 282 {
 283   char buf[sizeof (DATA)];
 284   GInputStream *in;
 285   GThread *thread;
 286   GCancellable *cancellable;
 287   gboolean success;
 288 } PipeIOOverlapReader;
 289
 290 #define TEST_PIPE_IO_OVERLAP (1024 * 4)
 291
 292 static gpointer
 293 pipe_io_overlap_reader_thread (gpointer user_data)
 294 {
 295   PipeIOOverlapReader *p = user_data;
 296   GError *err = NULL;
 297   gsize read;
 298   guint i;
 299
 300   for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
 301     memset (p->buf, 0, sizeof (p->buf));
 302     g_input_stream_read_all (p->in, p->buf, sizeof (p->buf),
 303                              &read, NULL, &err);
 304
 305     g_assert_cmpuint (read, ==, sizeof (p->buf));
 306     g_assert_no_error (err);
 307     g_assert_cmpstr (p->buf, ==, DATA);
 308   }
 309
 310   return NULL;
 311 }
 312
 313 static gpointer
 314 pipe_io_overlap_writer_thread (gpointer user_data)
 315 {
 316   GOutputStream *out = user_data;
 317   GError *err = NULL;
 318   gsize bytes_written;
 319   guint i;
 320
 321   for (i = 0; i < TEST_PIPE_IO_OVERLAP; ++i) {
 322     g_output_stream_write_all (out, DATA, sizeof (DATA),
 323                                &bytes_written, NULL, &err);
 324
 325     g_assert_cmpuint (bytes_written, ==, sizeof (DATA));
 326     g_assert_no_error (err);
 327   }
 328
 329   return NULL;
 330 }
 331
 332 static void
 333 test_pipe_io_overlap (void)
 334 {
 335   GOutputStream *out_server, *out_client;
 336   GThread *writer_server, *writer_client;
 337   PipeIOOverlapReader rs, rc;
 338   HANDLE server, client;
 339   gchar name[256];
 340
 341   g_snprintf (name, sizeof (name),
 342               "\\\\.\\pipe\\gtest-io-overlap-%u", (guint) GetCurrentProcessId ());
 343
 344   server = CreateNamedPipe (name,
 345                             PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
 346                             PIPE_READMODE_BYTE | PIPE_WAIT,
 347                             1, 0, 0, 0, NULL);
 348   g_assert (server != INVALID_HANDLE_VALUE);
 349
 350   client = CreateFile (name, GENERIC_WRITE | GENERIC_READ, 0, NULL, OPEN_EXISTING, 0, NULL);
 351   g_assert (client != INVALID_HANDLE_VALUE);
 352
 353   out_server = g_win32_output_stream_new (server, TRUE);
 354   writer_server = g_thread_new ("writer_server", pipe_io_overlap_writer_thread, out_server);
 355   rs.in = g_win32_input_stream_new (server, TRUE);
 356   rs.thread = g_thread_new ("reader_server", pipe_io_overlap_reader_thread, &rs);
 357
 358   out_client = g_win32_output_stream_new (client, TRUE);
 359   writer_client = g_thread_new ("writer_client", pipe_io_overlap_writer_thread, out_client);
 360   rc.in = g_win32_input_stream_new (client, TRUE);
 361   rc.thread = g_thread_new ("reader_client", pipe_io_overlap_reader_thread, &rc);
 362
 363   g_thread_join (writer_client);
 364   g_thread_join (writer_server);
 365   g_thread_join (rc.thread);
 366   g_thread_join (rs.thread);
 367
 368   g_object_unref (rs.in);
 369   g_object_unref (rc.in);
 370   g_object_unref (out_server);
 371   g_object_unref (out_client);
 372 }
 373
 374 static gpointer
 375 pipe_io_concurrent_writer_thread (gpointer user_data)
 376 {
 377   GOutputStream *out = user_data;
 378   GError *err = NULL;
 379   gsize bytes_written;
 380
 381   g_output_stream_write_all (out, DATA, 1, &bytes_written, NULL, &err);
 382
 383   g_assert_cmpuint (bytes_written, ==, 1);
 384   g_assert_no_error (err);
 385
 386   return NULL;
 387 }
 388
 389 static gpointer
 390 pipe_io_concurrent_reader_thread (gpointer user_data)
 391 {
 392   PipeIOOverlapReader *p = user_data;
 393   GError *err = NULL;
 394   gsize read;
 395
 396   memset (p->buf, 0, sizeof (p->buf));
 397   p->success = g_input_stream_read_all (p->in, p->buf, 1, &read, p->cancellable, &err);
 398
 399   /* only one thread will succeed, the other will be cancelled */
 400   if (p->success)
 401     {
 402       /* continue the main thread */
 403       write (writer_pipe[1], "", 1);
 404       g_assert_cmpuint (read, ==, 1);
 405       g_assert_no_error (err);
 406     }
 407
 408   return NULL;
 409 }
 410
 411 static void
 412 test_pipe_io_concurrent (void)
 413 {
 414   GOutputStream *out_server;
 415   GThread *writer_server;
 416   PipeIOOverlapReader rc1, rc2;
 417   HANDLE server, client;
 418   gchar name[256], c;
 419
 420   g_snprintf (name, sizeof (name),
 421               "\\\\.\\pipe\\gtest-io-concurrent-%u", (guint) GetCurrentProcessId ());
 422
 423   server = CreateNamedPipe (name,
 424                             PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
 425                             PIPE_READMODE_BYTE | PIPE_WAIT,
 426                             1, 0, 0, 0, NULL);
 427   g_assert (server != INVALID_HANDLE_VALUE);
 428   g_assert (_pipe (writer_pipe, 10, _O_BINARY) == 0);
 429
 430   client = CreateFile (name, GENERIC_WRITE | GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
 431   g_assert (client != INVALID_HANDLE_VALUE);
 432
 433   rc1.in = g_win32_input_stream_new (client, TRUE);
 434   rc1.success = FALSE;
 435   rc1.cancellable = g_cancellable_new ();
 436   rc1.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc1);
 437
 438   rc2.in = g_win32_input_stream_new (client, TRUE);
 439   rc2.success = FALSE;
 440   rc2.cancellable = g_cancellable_new ();
 441   rc2.thread = g_thread_new ("reader_client", pipe_io_concurrent_reader_thread, &rc2);
 442
 443   /* FIXME: how to synchronize on both reader thread waiting in read,
 444      before starting the writer thread? */
 445   g_usleep (G_USEC_PER_SEC / 10);
 446
 447   out_server = g_win32_output_stream_new (server, TRUE);
 448   writer_server = g_thread_new ("writer_server", pipe_io_concurrent_writer_thread, out_server);
 449
 450   read (writer_pipe[0], &c, 1);
 451
 452   g_assert (rc1.success ^ rc2.success);
 453
 454   g_cancellable_cancel (rc1.cancellable);
 455   g_cancellable_cancel (rc2.cancellable);
 456
 457   g_thread_join (writer_server);
 458   g_thread_join (rc1.thread);
 459   g_thread_join (rc2.thread);
 460
 461   g_object_unref (rc1.in);
 462   g_object_unref (rc2.in);
 463   g_object_unref (out_server);
 464
 465   close (writer_pipe[0]);
 466   close (writer_pipe[1]);
 467 }
 468
 469 static void
 470 readable_cancel (GObject *source, GAsyncResult *res, gpointer user_data)
 471 {
 472   GInputStream *in = G_INPUT_STREAM (source);
 473   GError *err = NULL;
 474   gssize len;
 475
 476   len = g_input_stream_read_finish (in, res, &err);
 477   g_assert_cmpint (len, ==, -1);
 478   g_assert_error (err, G_IO_ERROR, G_IO_ERROR_CANCELLED);
 479   g_error_free (err);
 480
 481   g_main_loop_quit (loop);
 482 }
 483
 484 static void
 485 test_pipe_io_cancel (void)
 486 {
 487   GInputStream *in;
 488   GOutputStream *out;
 489   HANDLE in_handle, out_handle;
 490   gchar name[256];
 491
 492   g_snprintf (name, sizeof (name),
 493               "\\\\.\\pipe\\gtest-io-cancel-%u", (guint) GetCurrentProcessId ());
 494
 495   in_handle = CreateNamedPipe (name,
 496                                PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED,
 497                                PIPE_READMODE_BYTE | PIPE_WAIT,
 498                                1, 0, 0, 0, NULL);
 499   g_assert (in_handle != INVALID_HANDLE_VALUE);
 500
 501   out_handle = CreateFile (name, GENERIC_WRITE, 0, NULL, OPEN_EXISTING, 0, NULL);
 502   g_assert (out_handle != INVALID_HANDLE_VALUE);
 503
 504   in = g_win32_input_stream_new (in_handle, TRUE);
 505   out = g_win32_output_stream_new (out_handle, TRUE);
 506
 507   reader_cancel = g_cancellable_new ();
 508   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 509                              G_PRIORITY_DEFAULT, reader_cancel,
 510                              readable_cancel, out);
 511
 512   g_timeout_add (500, timeout, reader_cancel);
 513
 514   loop = g_main_loop_new (NULL, TRUE);
 515   g_main_loop_run (loop);
 516   g_main_loop_unref (loop);
 517
 518   g_object_unref (reader_cancel);
 519   g_object_unref (in);
 520   g_object_unref (out);
 521 }
 522
 523 int
 524 main (int   argc,
 525       char *argv[])
 526 {
 527   g_test_init (&argc, &argv, NULL);
 528
 529   g_test_add_func ("/win32-streams/pipe-io-test", test_pipe_io);
 530   g_test_add_func ("/win32-streams/pipe-io-cancel-test", test_pipe_io_cancel);
 531   g_test_add_func ("/win32-streams/pipe-io-overlap-test", test_pipe_io_overlap);
 532   g_test_add_func ("/win32-streams/pipe-io-concurrent-test", test_pipe_io_concurrent);
 533
 534   return g_test_run();
 535 }