gio/tests/unix-streams.c

   1 /* GLib testing framework examples and tests
   2  * Copyright (C) 2008 Red Hat, Inc
   3  *
   4  * SPDX-License-Identifier: LicenseRef-old-glib-tests
   5  *
   6  * This work is provided "as is"; redistribution and modification
   7  * in whole or in part, in any medium, physical or electronic is
   8  * permitted without restriction.
   9  *
  10  * This work is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  13  *
  14  * In no event shall the authors or contributors be liable for any
  15  * direct, indirect, incidental, special, exemplary, or consequential
  16  * damages (including, but not limited to, procurement of substitute
  17  * goods or services; loss of use, data, or profits; or business
  18  * interruption) however caused and on any theory of liability, whether
  19  * in contract, strict liability, or tort (including negligence or
  20  * otherwise) arising in any way out of the use of this software, even
  21  * if advised of the possibility of such damage.
  22  */
  23
  24 #include <gio/gio.h>
  25 #include <gio/gunixinputstream.h>
  26 #include <gio/gunixoutputstream.h>
  27 #include <glib.h>
  28 #include <glib/glib-unix.h>
  29 #include <signal.h>
  30 #include <stdlib.h>
  31 #include <string.h>
  32 #include <unistd.h>
  33 #include <fcntl.h>
  34
  35 /* sizeof(DATA) will give the number of bytes in the array, plus the terminating nul */
  36 static const gchar DATA[] = "abcdefghijklmnopqrstuvwxyz";
  37
  38 int writer_pipe[2], reader_pipe[2];
  39 GCancellable *writer_cancel, *reader_cancel, *main_cancel;
  40 GMainLoop *loop;
  41
  42
  43 static gpointer
  44 writer_thread (gpointer user_data)
  45 {
  46   GOutputStream *out;
  47   gssize nwrote, offset;
  48   GError *err = NULL;
  49
  50   out = g_unix_output_stream_new (writer_pipe[1], TRUE);
  51
  52   do
  53     {
  54       g_usleep (10);
  55
  56       offset = 0;
  57       while (offset < (gssize) sizeof (DATA))
  58         {
  59           nwrote = g_output_stream_write (out, DATA + offset,
  60                                           sizeof (DATA) - offset,
  61                                           writer_cancel, &err);
  62           if (nwrote <= 0 || err != NULL)
  63             break;
  64           offset += nwrote;
  65         }
  66
  67       g_assert_true (nwrote > 0 || err != NULL);
  68     }
  69   while (err == NULL);
  70
  71   if (g_cancellable_is_cancelled (writer_cancel))
  72     {
  73       g_clear_error (&err);
  74       g_cancellable_cancel (main_cancel);
  75       g_object_unref (out);
  76       return NULL;
  77     }
  78
  79   g_warning ("writer: %s", err->message);
  80   g_assert_not_reached ();
  81 }
  82
  83 static gpointer
  84 reader_thread (gpointer user_data)
  85 {
  86   GInputStream *in;
  87   gssize nread = 0, total;
  88   GError *err = NULL;
  89   char buf[sizeof (DATA)];
  90
  91   in = g_unix_input_stream_new (reader_pipe[0], TRUE);
  92
  93   do
  94     {
  95       total = 0;
  96       while (total < (gssize) sizeof (DATA))
  97         {
  98           nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
  99                                        reader_cancel, &err);
 100           if (nread <= 0 || err != NULL)
 101             break;
 102           total += nread;
 103         }
 104
 105       if (err)
 106         break;
 107
 108       if (nread == 0)
 109         {
 110           g_assert_no_error (err);
 111           /* pipe closed */
 112           g_object_unref (in);
 113           return NULL;
 114         }
 115
 116       g_assert_cmpstr (buf, ==, DATA);
 117       g_assert_false (g_cancellable_is_cancelled (reader_cancel));
 118     }
 119   while (err == NULL);
 120
 121   g_warning ("reader: %s", err->message);
 122   g_assert_not_reached ();
 123 }
 124
 125 static char main_buf[sizeof (DATA)];
 126 static gssize main_len, main_offset;
 127
 128 static void main_thread_read (GObject *source, GAsyncResult *res, gpointer user_data);
 129 static void main_thread_skipped (GObject *source, GAsyncResult *res, gpointer user_data);
 130 static void main_thread_wrote (GObject *source, GAsyncResult *res, gpointer user_data);
 131
 132 static void
 133 do_main_cancel (GOutputStream *out)
 134 {
 135   g_output_stream_close (out, NULL, NULL);
 136   g_main_loop_quit (loop);
 137 }
 138
 139 static void
 140 main_thread_skipped (GObject *source, GAsyncResult *res, gpointer user_data)
 141 {
 142   GInputStream *in = G_INPUT_STREAM (source);
 143   GOutputStream *out = user_data;
 144   GError *err = NULL;
 145   gssize nskipped;
 146
 147   nskipped = g_input_stream_skip_finish (in, res, &err);
 148
 149   if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 150     {
 151       g_assert_true (g_cancellable_is_cancelled (main_cancel));
 152       do_main_cancel (out);
 153       g_clear_error (&err);
 154       return;
 155     }
 156
 157   g_assert_no_error (err);
 158
 159   main_offset += nskipped;
 160   if (main_offset == main_len)
 161     {
 162       main_offset = 0;
 163       g_output_stream_write_async (out, main_buf, main_len,
 164                                    G_PRIORITY_DEFAULT, main_cancel,
 165                                    main_thread_wrote, in);
 166     }
 167   else
 168     {
 169       g_input_stream_skip_async (in, main_len - main_offset,
 170                                  G_PRIORITY_DEFAULT, main_cancel,
 171                                  main_thread_skipped, out);
 172     }
 173 }
 174
 175 static void
 176 main_thread_read (GObject *source, GAsyncResult *res, gpointer user_data)
 177 {
 178   GInputStream *in = G_INPUT_STREAM (source);
 179   GOutputStream *out = user_data;
 180   GError *err = NULL;
 181   gssize nread;
 182
 183   nread = g_input_stream_read_finish (in, res, &err);
 184
 185   if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 186     {
 187       g_assert_true (g_cancellable_is_cancelled (main_cancel));
 188       do_main_cancel (out);
 189       g_clear_error (&err);
 190       return;
 191     }
 192
 193   g_assert_no_error (err);
 194
 195   main_offset += nread;
 196   if (main_offset == sizeof (DATA))
 197     {
 198       main_len = main_offset;
 199       main_offset = 0;
 200       /* Now skip the same amount */
 201       g_input_stream_skip_async (in, main_len,
 202                                  G_PRIORITY_DEFAULT, main_cancel,
 203                                  main_thread_skipped, out);
 204     }
 205   else
 206     {
 207       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 208                                  G_PRIORITY_DEFAULT, main_cancel,
 209                                  main_thread_read, out);
 210     }
 211 }
 212
 213 static void
 214 main_thread_wrote (GObject *source, GAsyncResult *res, gpointer user_data)
 215 {
 216   GOutputStream *out = G_OUTPUT_STREAM (source);
 217   GInputStream *in = user_data;
 218   GError *err = NULL;
 219   gssize nwrote;
 220
 221   nwrote = g_output_stream_write_finish (out, res, &err);
 222
 223   if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 224     {
 225       g_assert_true (g_cancellable_is_cancelled (main_cancel));
 226       do_main_cancel (out);
 227       g_clear_error (&err);
 228       return;
 229     }
 230
 231   g_assert_no_error (err);
 232   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 233
 234   main_offset += nwrote;
 235   if (main_offset == main_len)
 236     {
 237       main_offset = 0;
 238       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 239                                  G_PRIORITY_DEFAULT, main_cancel,
 240                                  main_thread_read, out);
 241     }
 242   else
 243     {
 244       g_output_stream_write_async (out, main_buf + main_offset,
 245                                    main_len - main_offset,
 246                                    G_PRIORITY_DEFAULT, main_cancel,
 247                                    main_thread_wrote, in);
 248     }
 249 }
 250
 251 static gboolean
 252 timeout (gpointer cancellable)
 253 {
 254   g_cancellable_cancel (cancellable);
 255   return FALSE;
 256 }
 257
 258 static void
 259 test_pipe_io (gconstpointer nonblocking)
 260 {
 261   GThread *writer, *reader;
 262   GInputStream *in;
 263   GOutputStream *out;
 264
 265   /* Split off two (additional) threads, a reader and a writer. From
 266    * the writer thread, write data synchronously in small chunks,
 267    * which gets alternately read and skipped asynchronously by the
 268    * main thread and then (if not skipped) written asynchronously to
 269    * the reader thread, which reads it synchronously. Eventually a
 270    * timeout in the main thread will cause it to cancel the writer
 271    * thread, which will in turn cancel the read op in the main thread,
 272    * which will then close the pipe to the reader thread, causing the
 273    * read op to fail.
 274    */
 275
 276   g_assert_true (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
 277
 278   if (nonblocking)
 279     {
 280       GError *error = NULL;
 281
 282       g_unix_set_fd_nonblocking (writer_pipe[0], TRUE, &error);
 283       g_assert_no_error (error);
 284       g_unix_set_fd_nonblocking (writer_pipe[1], TRUE, &error);
 285       g_assert_no_error (error);
 286       g_unix_set_fd_nonblocking (reader_pipe[0], TRUE, &error);
 287       g_assert_no_error (error);
 288       g_unix_set_fd_nonblocking (reader_pipe[1], TRUE, &error);
 289       g_assert_no_error (error);
 290     }
 291
 292   writer_cancel = g_cancellable_new ();
 293   reader_cancel = g_cancellable_new ();
 294   main_cancel = g_cancellable_new ();
 295
 296   writer = g_thread_new ("writer", writer_thread, NULL);
 297   reader = g_thread_new ("reader", reader_thread, NULL);
 298
 299   in = g_unix_input_stream_new (writer_pipe[0], TRUE);
 300   out = g_unix_output_stream_new (reader_pipe[1], TRUE);
 301
 302   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 303                              G_PRIORITY_DEFAULT, main_cancel,
 304                              main_thread_read, out);
 305
 306   g_timeout_add (500, timeout, writer_cancel);
 307
 308   loop = g_main_loop_new (NULL, TRUE);
 309   g_main_loop_run (loop);
 310   g_main_loop_unref (loop);
 311
 312   g_thread_join (reader);
 313   g_thread_join (writer);
 314
 315   g_object_unref (main_cancel);
 316   g_object_unref (reader_cancel);
 317   g_object_unref (writer_cancel);
 318   g_object_unref (in);
 319   g_object_unref (out);
 320 }
 321
 322 static void
 323 test_basic (void)
 324 {
 325   GUnixInputStream *is;
 326   GUnixOutputStream *os;
 327   gint fd;
 328   gboolean close_fd;
 329
 330   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (0, TRUE));
 331   g_object_get (is,
 332                 "fd", &fd,
 333                 "close-fd", &close_fd,
 334                 NULL);
 335   g_assert_cmpint (fd, ==, 0);
 336   g_assert_true (close_fd);
 337
 338   g_unix_input_stream_set_close_fd (is, FALSE);
 339   g_assert_false (g_unix_input_stream_get_close_fd (is));
 340   g_assert_cmpint (g_unix_input_stream_get_fd (is), ==, 0);
 341
 342   g_assert_false (g_input_stream_has_pending (G_INPUT_STREAM (is)));
 343
 344   g_object_unref (is);
 345
 346   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (1, TRUE));
 347   g_object_get (os,
 348                 "fd", &fd,
 349                 "close-fd", &close_fd,
 350                 NULL);
 351   g_assert_cmpint (fd, ==, 1);
 352   g_assert_true (close_fd);
 353
 354   g_unix_output_stream_set_close_fd (os, FALSE);
 355   g_assert_false (g_unix_output_stream_get_close_fd (os));
 356   g_assert_cmpint (g_unix_output_stream_get_fd (os), ==, 1);
 357
 358   g_assert_false (g_output_stream_has_pending (G_OUTPUT_STREAM (os)));
 359
 360   g_object_unref (os);
 361 }
 362
 363 typedef struct {
 364   GInputStream *is;
 365   GOutputStream *os;
 366   const guint8 *write_data;
 367   guint8 *read_data;
 368 } TestReadWriteData;
 369
 370 static gpointer
 371 test_read_write_write_thread (gpointer user_data)
 372 {
 373   TestReadWriteData *data = user_data;
 374   gsize bytes_written;
 375   GError *error = NULL;
 376   gboolean res;
 377
 378   res = g_output_stream_write_all (data->os, data->write_data, 1024, &bytes_written, NULL, &error);
 379   g_assert_true (res);
 380   g_assert_no_error (error);
 381   g_assert_cmpuint (bytes_written, ==, 1024);
 382
 383   return NULL;
 384 }
 385
 386 static gpointer
 387 test_read_write_read_thread (gpointer user_data)
 388 {
 389   TestReadWriteData *data = user_data;
 390   gsize bytes_read;
 391   GError *error = NULL;
 392   gboolean res;
 393
 394   res = g_input_stream_read_all (data->is, data->read_data, 1024, &bytes_read, NULL, &error);
 395   g_assert_true (res);
 396   g_assert_no_error (error);
 397   g_assert_cmpuint (bytes_read, ==, 1024);
 398
 399   return NULL;
 400 }
 401
 402 static gpointer
 403 test_read_write_writev_thread (gpointer user_data)
 404 {
 405   TestReadWriteData *data = user_data;
 406   gsize bytes_written;
 407   GError *error = NULL;
 408   gboolean res;
 409   GOutputVector vectors[3];
 410
 411   vectors[0].buffer = data->write_data;
 412   vectors[0].size = 256;
 413   vectors[1].buffer = data->write_data + 256;
 414   vectors[1].size = 256;
 415   vectors[2].buffer = data->write_data + 512;
 416   vectors[2].size = 512;
 417
 418   res = g_output_stream_writev_all (data->os, vectors, G_N_ELEMENTS (vectors), &bytes_written, NULL, &error);
 419   g_assert_true (res);
 420   g_assert_no_error (error);
 421   g_assert_cmpuint (bytes_written, ==, 1024);
 422
 423   return NULL;
 424 }
 425
 426 /* test if normal writing/reading from a pipe works */
 427 static void
 428 test_read_write (gconstpointer user_data)
 429 {
 430   gboolean writev = GPOINTER_TO_INT (user_data);
 431   GUnixInputStream *is;
 432   GUnixOutputStream *os;
 433   gint fd[2];
 434   guint8 data_write[1024], data_read[1024];
 435   guint i;
 436   GThread *write_thread, *read_thread;
 437   TestReadWriteData data;
 438
 439   for (i = 0; i < sizeof (data_write); i++)
 440     data_write[i] = i;
 441
 442   g_assert_cmpint (pipe (fd), ==, 0);
 443
 444   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (fd[0], TRUE));
 445   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (fd[1], TRUE));
 446
 447   data.is = G_INPUT_STREAM (is);
 448   data.os = G_OUTPUT_STREAM (os);
 449   data.read_data = data_read;
 450   data.write_data = data_write;
 451
 452   if (writev)
 453     write_thread = g_thread_new ("writer", test_read_write_writev_thread, &data);
 454   else
 455     write_thread = g_thread_new ("writer", test_read_write_write_thread, &data);
 456   read_thread = g_thread_new ("reader", test_read_write_read_thread, &data);
 457
 458   g_thread_join (write_thread);
 459   g_thread_join (read_thread);
 460
 461   g_assert_cmpmem (data_write, sizeof data_write, data_read, sizeof data_read);
 462
 463   g_object_unref (os);
 464   g_object_unref (is);
 465 }
 466
 467 /* test if g_pollable_output_stream_write_nonblocking() and
 468  * g_pollable_output_stream_read_nonblocking() correctly return WOULD_BLOCK
 469  * and correctly reset their status afterwards again, and all data that is
 470  * written can also be read again.
 471  */
 472 static void
 473 test_write_wouldblock (void)
 474 {
 475 #ifndef F_GETPIPE_SZ
 476   g_test_skip ("F_GETPIPE_SZ not defined");
 477 #else  /* if F_GETPIPE_SZ */
 478   GUnixInputStream *is;
 479   GUnixOutputStream *os;
 480   gint fd[2];
 481   GError *err = NULL;
 482   guint8 data_write[1024], data_read[1024];
 483   gsize i;
 484   int retval;
 485   gsize pipe_capacity;
 486
 487   for (i = 0; i < sizeof (data_write); i++)
 488     data_write[i] = i;
 489
 490   g_assert_cmpint (pipe (fd), ==, 0);
 491
 492   g_assert_cmpint (fcntl (fd[0], F_SETPIPE_SZ, 4096, NULL), !=, 0);
 493   retval = fcntl (fd[0], F_GETPIPE_SZ);
 494   g_assert_cmpint (retval, >=, 0);
 495   pipe_capacity = (gsize) retval;
 496   g_assert_cmpint (pipe_capacity, >=, 4096);
 497   g_assert_cmpint (pipe_capacity % 1024, >=, 0);
 498
 499   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (fd[0], TRUE));
 500   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (fd[1], TRUE));
 501
 502   /* Run the whole thing three times to make sure that the streams
 503    * reset the writability/readability state again */
 504   for (i = 0; i < 3; i++) {
 505     gssize written = 0, written_complete = 0;
 506     gssize read = 0, read_complete = 0;
 507
 508     do
 509       {
 510         written_complete += written;
 511         written = g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM (os),
 512                                                               data_write,
 513                                                               sizeof (data_write),
 514                                                               NULL,
 515                                                               &err);
 516       }
 517     while (written > 0);
 518
 519     g_assert_cmpuint (written_complete, >, 0);
 520     g_assert_nonnull (err);
 521     g_assert_error (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
 522     g_clear_error (&err);
 523
 524     do
 525       {
 526         read_complete += read;
 527         read = g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (is),
 528                                                          data_read,
 529                                                          sizeof (data_read),
 530                                                          NULL,
 531                                                          &err);
 532         if (read > 0)
 533           g_assert_cmpmem (data_read, read, data_write, sizeof (data_write));
 534       }
 535     while (read > 0);
 536
 537     g_assert_cmpuint (read_complete, ==, written_complete);
 538     g_assert_nonnull (err);
 539     g_assert_error (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
 540     g_clear_error (&err);
 541   }
 542
 543   g_object_unref (os);
 544   g_object_unref (is);
 545 #endif  /* if F_GETPIPE_SZ */
 546 }
 547
 548 /* test if g_pollable_output_stream_writev_nonblocking() and
 549  * g_pollable_output_stream_read_nonblocking() correctly return WOULD_BLOCK
 550  * and correctly reset their status afterwards again, and all data that is
 551  * written can also be read again.
 552  */
 553 static void
 554 test_writev_wouldblock (void)
 555 {
 556 #ifndef F_GETPIPE_SZ
 557   g_test_skip ("F_GETPIPE_SZ not defined");
 558 #else  /* if F_GETPIPE_SZ */
 559   GUnixInputStream *is;
 560   GUnixOutputStream *os;
 561   gint fd[2];
 562   GError *err = NULL;
 563   guint8 data_write[1024], data_read[1024];
 564   gsize i;
 565   int retval;
 566   gsize pipe_capacity;
 567   GOutputVector vectors[4];
 568   GPollableReturn res;
 569
 570   for (i = 0; i < sizeof (data_write); i++)
 571     data_write[i] = i;
 572
 573   g_assert_cmpint (pipe (fd), ==, 0);
 574
 575   g_assert_cmpint (fcntl (fd[0], F_SETPIPE_SZ, 4096, NULL), !=, 0);
 576   retval = fcntl (fd[0], F_GETPIPE_SZ);
 577   g_assert_cmpint (retval, >=, 0);
 578   pipe_capacity = (gsize) retval;
 579   g_assert_cmpint (pipe_capacity, >=, 4096);
 580   g_assert_cmpint (pipe_capacity % 1024, >=, 0);
 581
 582   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (fd[0], TRUE));
 583   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (fd[1], TRUE));
 584
 585   /* Run the whole thing three times to make sure that the streams
 586    * reset the writability/readability state again */
 587   for (i = 0; i < 3; i++) {
 588     gsize written = 0, written_complete = 0;
 589     gssize read = 0, read_complete = 0;
 590
 591     do
 592     {
 593         written_complete += written;
 594
 595         vectors[0].buffer = data_write;
 596         vectors[0].size = 256;
 597         vectors[1].buffer = data_write + 256;
 598         vectors[1].size = 256;
 599         vectors[2].buffer = data_write + 512;
 600         vectors[2].size = 256;
 601         vectors[3].buffer = data_write + 768;
 602         vectors[3].size = 256;
 603
 604         res = g_pollable_output_stream_writev_nonblocking (G_POLLABLE_OUTPUT_STREAM (os),
 605                                                            vectors,
 606                                                            G_N_ELEMENTS (vectors),
 607                                                            &written,
 608                                                            NULL,
 609                                                            &err);
 610       }
 611     while (res == G_POLLABLE_RETURN_OK);
 612
 613     g_assert_cmpuint (written_complete, >, 0);
 614     g_assert_null (err);
 615     g_assert_cmpint (res, ==, G_POLLABLE_RETURN_WOULD_BLOCK);
 616     /* writev() on UNIX streams either succeeds fully or not at all */
 617     g_assert_cmpuint (written, ==, 0);
 618
 619     do
 620       {
 621         read_complete += read;
 622         read = g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (is),
 623                                                          data_read,
 624                                                          sizeof (data_read),
 625                                                          NULL,
 626                                                          &err);
 627         if (read > 0)
 628           g_assert_cmpmem (data_read, read, data_write, sizeof (data_write));
 629       }
 630     while (read > 0);
 631
 632     g_assert_cmpuint (read_complete, ==, written_complete);
 633     g_assert_nonnull (err);
 634     g_assert_error (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
 635     g_clear_error (&err);
 636   }
 637
 638   g_object_unref (os);
 639   g_object_unref (is);
 640 #endif  /* if F_GETPIPE_SZ */
 641 }
 642
 643 #ifdef F_GETPIPE_SZ
 644 static void
 645 write_async_wouldblock_cb (GUnixOutputStream *os,
 646                            GAsyncResult      *result,
 647                            gpointer           user_data)
 648 {
 649   gsize *bytes_written = user_data;
 650   GError *err = NULL;
 651
 652   g_output_stream_write_all_finish (G_OUTPUT_STREAM (os), result, bytes_written, &err);
 653   g_assert_no_error (err);
 654 }
 655
 656 static void
 657 read_async_wouldblock_cb (GUnixInputStream  *is,
 658                           GAsyncResult      *result,
 659                           gpointer           user_data)
 660 {
 661   gsize *bytes_read = user_data;
 662   GError *err = NULL;
 663
 664   g_input_stream_read_all_finish (G_INPUT_STREAM (is), result, bytes_read, &err);
 665   g_assert_no_error (err);
 666 }
 667 #endif  /* if F_GETPIPE_SZ */
 668
 669 /* test if the async implementation of write_all() and read_all() in G*Stream
 670  * around the GPollable*Stream API is working correctly.
 671  */
 672 static void
 673 test_write_async_wouldblock (void)
 674 {
 675 #ifndef F_GETPIPE_SZ
 676   g_test_skip ("F_GETPIPE_SZ not defined");
 677 #else  /* if F_GETPIPE_SZ */
 678   GUnixInputStream *is;
 679   GUnixOutputStream *os;
 680   gint fd[2];
 681   guint8 *data, *data_read;
 682   gsize i;
 683   int retval;
 684   gsize pipe_capacity;
 685   gsize bytes_written = 0, bytes_read = 0;
 686
 687   g_assert_cmpint (pipe (fd), ==, 0);
 688
 689   /* FIXME: These should not be needed but otherwise
 690    * g_unix_output_stream_write() will block because
 691    *   a) the fd is writable
 692    *   b) writing 4x capacity will block because writes are atomic
 693    *   c) the fd is blocking
 694    *
 695    * See https://gitlab.gnome.org/GNOME/glib/issues/1654
 696    */
 697   g_unix_set_fd_nonblocking (fd[0], TRUE, NULL);
 698   g_unix_set_fd_nonblocking (fd[1], TRUE, NULL);
 699
 700   g_assert_cmpint (fcntl (fd[0], F_SETPIPE_SZ, 4096, NULL), !=, 0);
 701   retval = fcntl (fd[0], F_GETPIPE_SZ);
 702   g_assert_cmpint (retval, >=, 0);
 703   pipe_capacity = (gsize) retval;
 704   g_assert_cmpint (pipe_capacity, >=, 4096);
 705
 706   data = g_new (guint8, 4 * pipe_capacity);
 707   for (i = 0; i < 4 * pipe_capacity; i++)
 708     data[i] = i;
 709   data_read = g_new (guint8, 4 * pipe_capacity);
 710
 711   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (fd[0], TRUE));
 712   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (fd[1], TRUE));
 713
 714   g_output_stream_write_all_async (G_OUTPUT_STREAM (os),
 715                                    data,
 716                                    4 * pipe_capacity,
 717                                    G_PRIORITY_DEFAULT,
 718                                    NULL,
 719                                    (GAsyncReadyCallback) write_async_wouldblock_cb,
 720                                    &bytes_written);
 721
 722   g_input_stream_read_all_async (G_INPUT_STREAM (is),
 723                                  data_read,
 724                                  4 * pipe_capacity,
 725                                  G_PRIORITY_DEFAULT,
 726                                  NULL,
 727                                  (GAsyncReadyCallback) read_async_wouldblock_cb,
 728                                  &bytes_read);
 729
 730   while (bytes_written == 0 && bytes_read == 0)
 731     g_main_context_iteration (NULL, TRUE);
 732
 733   g_assert_cmpuint (bytes_written, ==, 4 * pipe_capacity);
 734   g_assert_cmpuint (bytes_read, ==, 4 * pipe_capacity);
 735   g_assert_cmpmem (data_read, bytes_read, data, bytes_written);
 736
 737   g_free (data);
 738   g_free (data_read);
 739
 740   g_object_unref (os);
 741   g_object_unref (is);
 742 #endif  /* if F_GETPIPE_SZ */
 743 }
 744
 745 #ifdef F_GETPIPE_SZ
 746 static void
 747 writev_async_wouldblock_cb (GUnixOutputStream *os,
 748                             GAsyncResult      *result,
 749                             gpointer           user_data)
 750 {
 751   gsize *bytes_written = user_data;
 752   GError *err = NULL;
 753
 754   g_output_stream_writev_all_finish (G_OUTPUT_STREAM (os), result, bytes_written, &err);
 755   g_assert_no_error (err);
 756 }
 757 #endif  /* if F_GETPIPE_SZ */
 758
 759 /* test if the async implementation of writev_all() and read_all() in G*Stream
 760  * around the GPollable*Stream API is working correctly.
 761  */
 762 static void
 763 test_writev_async_wouldblock (void)
 764 {
 765 #ifndef F_GETPIPE_SZ
 766   g_test_skip ("F_GETPIPE_SZ not defined");
 767 #else  /* if F_GETPIPE_SZ */
 768   GUnixInputStream *is;
 769   GUnixOutputStream *os;
 770   gint fd[2];
 771   guint8 *data, *data_read;
 772   gsize i;
 773   int retval;
 774   gsize pipe_capacity;
 775   gsize bytes_written = 0, bytes_read = 0;
 776   GOutputVector vectors[4];
 777
 778   g_assert_cmpint (pipe (fd), ==, 0);
 779
 780   /* FIXME: These should not be needed but otherwise
 781    * g_unix_output_stream_writev() will block because
 782    *   a) the fd is writable
 783    *   b) writing 4x capacity will block because writes are atomic
 784    *   c) the fd is blocking
 785    *
 786    * See https://gitlab.gnome.org/GNOME/glib/issues/1654
 787    */
 788   g_unix_set_fd_nonblocking (fd[0], TRUE, NULL);
 789   g_unix_set_fd_nonblocking (fd[1], TRUE, NULL);
 790
 791   g_assert_cmpint (fcntl (fd[0], F_SETPIPE_SZ, 4096, NULL), !=, 0);
 792   retval = fcntl (fd[0], F_GETPIPE_SZ);
 793   g_assert_cmpint (retval, >=, 0);
 794   pipe_capacity = (gsize) retval;
 795   g_assert_cmpint (pipe_capacity, >=, 4096);
 796
 797   data = g_new (guint8, 4 * pipe_capacity);
 798   for (i = 0; i < 4 * pipe_capacity; i++)
 799     data[i] = i;
 800   data_read = g_new (guint8, 4 * pipe_capacity);
 801
 802   vectors[0].buffer = data;
 803   vectors[0].size = 1024;
 804   vectors[1].buffer = data + 1024;
 805   vectors[1].size = 1024;
 806   vectors[2].buffer = data + 2048;
 807   vectors[2].size = 1024;
 808   vectors[3].buffer = data + 3072;
 809   vectors[3].size = 4 * pipe_capacity - 3072;
 810
 811   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (fd[0], TRUE));
 812   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (fd[1], TRUE));
 813
 814   g_output_stream_writev_all_async (G_OUTPUT_STREAM (os),
 815                                     vectors,
 816                                     G_N_ELEMENTS (vectors),
 817                                     G_PRIORITY_DEFAULT,
 818                                     NULL,
 819                                     (GAsyncReadyCallback) writev_async_wouldblock_cb,
 820                                     &bytes_written);
 821
 822   g_input_stream_read_all_async (G_INPUT_STREAM (is),
 823                                  data_read,
 824                                  4 * pipe_capacity,
 825                                  G_PRIORITY_DEFAULT,
 826                                  NULL,
 827                                  (GAsyncReadyCallback) read_async_wouldblock_cb,
 828                                  &bytes_read);
 829
 830   while (bytes_written == 0 && bytes_read == 0)
 831     g_main_context_iteration (NULL, TRUE);
 832
 833   g_assert_cmpuint (bytes_written, ==, 4 * pipe_capacity);
 834   g_assert_cmpuint (bytes_read, ==, 4 * pipe_capacity);
 835   g_assert_cmpmem (data_read, bytes_read, data, bytes_written);
 836
 837   g_free (data);
 838   g_free (data_read);
 839
 840   g_object_unref (os);
 841   g_object_unref (is);
 842 #endif  /* F_GETPIPE_SZ */
 843 }
 844
 845 int
 846 main (int   argc,
 847       char *argv[])
 848 {
 849   g_test_init (&argc, &argv, NULL);
 850
 851   g_test_add_func ("/unix-streams/basic", test_basic);
 852   g_test_add_data_func ("/unix-streams/pipe-io-test",
 853                         GINT_TO_POINTER (FALSE),
 854                         test_pipe_io);
 855   g_test_add_data_func ("/unix-streams/nonblocking-io-test",
 856                         GINT_TO_POINTER (TRUE),
 857                         test_pipe_io);
 858
 859   g_test_add_data_func ("/unix-streams/read_write",
 860                         GINT_TO_POINTER (FALSE),
 861                         test_read_write);
 862
 863   g_test_add_data_func ("/unix-streams/read_writev",
 864                         GINT_TO_POINTER (TRUE),
 865                         test_read_write);
 866
 867   g_test_add_func ("/unix-streams/write-wouldblock",
 868                    test_write_wouldblock);
 869   g_test_add_func ("/unix-streams/writev-wouldblock",
 870                    test_writev_wouldblock);
 871
 872   g_test_add_func ("/unix-streams/write-async-wouldblock",
 873                    test_write_async_wouldblock);
 874   g_test_add_func ("/unix-streams/writev-async-wouldblock",
 875                    test_writev_async_wouldblock);
 876
 877   return g_test_run();
 878 }