gio/tests/unix-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 <gio/gio.h>
  23 #include <gio/gunixinputstream.h>
  24 #include <gio/gunixoutputstream.h>
  25 #include <glib/glib-unix.h>
  26 #include <signal.h>
  27 #include <stdlib.h>
  28 #include <string.h>
  29 #include <unistd.h>
  30
  31 #define DATA "abcdefghijklmnopqrstuvwxyz"
  32
  33 int writer_pipe[2], reader_pipe[2];
  34 GCancellable *writer_cancel, *reader_cancel, *main_cancel;
  35 GMainLoop *loop;
  36
  37
  38 static gpointer
  39 writer_thread (gpointer user_data)
  40 {
  41   GOutputStream *out;
  42   gssize nwrote, offset;
  43   GError *err = NULL;
  44
  45   out = g_unix_output_stream_new (writer_pipe[1], TRUE);
  46
  47   do
  48     {
  49       g_usleep (10);
  50
  51       offset = 0;
  52       while (offset < (gssize) sizeof (DATA))
  53         {
  54           nwrote = g_output_stream_write (out, DATA + offset,
  55                                           sizeof (DATA) - offset,
  56                                           writer_cancel, &err);
  57           if (nwrote <= 0 || err != NULL)
  58             break;
  59           offset += nwrote;
  60         }
  61
  62       g_assert (nwrote > 0 || err != NULL);
  63     }
  64   while (err == NULL);
  65
  66   if (g_cancellable_is_cancelled (writer_cancel))
  67     {
  68       g_clear_error (&err);
  69       g_cancellable_cancel (main_cancel);
  70       g_object_unref (out);
  71       return NULL;
  72     }
  73
  74   g_warning ("writer: %s", err->message);
  75   g_assert_not_reached ();
  76 }
  77
  78 static gpointer
  79 reader_thread (gpointer user_data)
  80 {
  81   GInputStream *in;
  82   gssize nread = 0, total;
  83   GError *err = NULL;
  84   char buf[sizeof (DATA)];
  85
  86   in = g_unix_input_stream_new (reader_pipe[0], TRUE);
  87
  88   do
  89     {
  90       total = 0;
  91       while (total < (gssize) sizeof (DATA))
  92         {
  93           nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
  94                                        reader_cancel, &err);
  95           if (nread <= 0 || err != NULL)
  96             break;
  97           total += nread;
  98         }
  99
 100       if (err)
 101         break;
 102
 103       if (nread == 0)
 104         {
 105           g_assert (err == NULL);
 106           /* pipe closed */
 107           g_object_unref (in);
 108           return NULL;
 109         }
 110
 111       g_assert_cmpstr (buf, ==, DATA);
 112       g_assert (!g_cancellable_is_cancelled (reader_cancel));
 113     }
 114   while (err == NULL);
 115
 116   g_warning ("reader: %s", err->message);
 117   g_assert_not_reached ();
 118 }
 119
 120 char main_buf[sizeof (DATA)];
 121 gssize main_len, main_offset;
 122
 123 static void main_thread_read (GObject *source, GAsyncResult *res, gpointer user_data);
 124 static void main_thread_skipped (GObject *source, GAsyncResult *res, gpointer user_data);
 125 static void main_thread_wrote (GObject *source, GAsyncResult *res, gpointer user_data);
 126
 127 static void
 128 do_main_cancel (GOutputStream *out)
 129 {
 130   g_output_stream_close (out, NULL, NULL);
 131   g_main_loop_quit (loop);
 132 }
 133
 134 static void
 135 main_thread_skipped (GObject *source, GAsyncResult *res, gpointer user_data)
 136 {
 137   GInputStream *in = G_INPUT_STREAM (source);
 138   GOutputStream *out = user_data;
 139   GError *err = NULL;
 140   gssize nskipped;
 141
 142   nskipped = g_input_stream_skip_finish (in, res, &err);
 143
 144   if (g_cancellable_is_cancelled (main_cancel))
 145     {
 146       do_main_cancel (out);
 147       return;
 148     }
 149
 150   g_assert_no_error (err);
 151
 152   main_offset += nskipped;
 153   if (main_offset == main_len)
 154     {
 155       main_offset = 0;
 156       g_output_stream_write_async (out, main_buf, main_len,
 157                                    G_PRIORITY_DEFAULT, main_cancel,
 158                                    main_thread_wrote, in);
 159     }
 160   else
 161     {
 162       g_input_stream_skip_async (in, main_len - main_offset,
 163                                  G_PRIORITY_DEFAULT, main_cancel,
 164                                  main_thread_skipped, out);
 165     }
 166 }
 167
 168 static void
 169 main_thread_read (GObject *source, GAsyncResult *res, gpointer user_data)
 170 {
 171   GInputStream *in = G_INPUT_STREAM (source);
 172   GOutputStream *out = user_data;
 173   GError *err = NULL;
 174   gssize nread;
 175
 176   nread = g_input_stream_read_finish (in, res, &err);
 177
 178   if (g_cancellable_is_cancelled (main_cancel))
 179     {
 180       do_main_cancel (out);
 181       g_clear_error (&err);
 182       return;
 183     }
 184
 185   g_assert_no_error (err);
 186
 187   main_offset += nread;
 188   if (main_offset == sizeof (DATA))
 189     {
 190       main_len = main_offset;
 191       main_offset = 0;
 192       /* Now skip the same amount */
 193       g_input_stream_skip_async (in, main_len,
 194                                  G_PRIORITY_DEFAULT, main_cancel,
 195                                  main_thread_skipped, out);
 196     }
 197   else
 198     {
 199       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 200                                  G_PRIORITY_DEFAULT, main_cancel,
 201                                  main_thread_read, out);
 202     }
 203 }
 204
 205 static void
 206 main_thread_wrote (GObject *source, GAsyncResult *res, gpointer user_data)
 207 {
 208   GOutputStream *out = G_OUTPUT_STREAM (source);
 209   GInputStream *in = user_data;
 210   GError *err = NULL;
 211   gssize nwrote;
 212
 213   nwrote = g_output_stream_write_finish (out, res, &err);
 214
 215   if (g_cancellable_is_cancelled (main_cancel))
 216     {
 217       do_main_cancel (out);
 218       g_clear_error (&err);
 219       return;
 220     }
 221
 222   g_assert_no_error (err);
 223   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 224
 225   main_offset += nwrote;
 226   if (main_offset == main_len)
 227     {
 228       main_offset = 0;
 229       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 230                                  G_PRIORITY_DEFAULT, main_cancel,
 231                                  main_thread_read, out);
 232     }
 233   else
 234     {
 235       g_output_stream_write_async (out, main_buf + main_offset,
 236                                    main_len - main_offset,
 237                                    G_PRIORITY_DEFAULT, main_cancel,
 238                                    main_thread_wrote, in);
 239     }
 240 }
 241
 242 static gboolean
 243 timeout (gpointer cancellable)
 244 {
 245   g_cancellable_cancel (cancellable);
 246   return FALSE;
 247 }
 248
 249 static void
 250 test_pipe_io (gconstpointer nonblocking)
 251 {
 252   GThread *writer, *reader;
 253   GInputStream *in;
 254   GOutputStream *out;
 255
 256   /* Split off two (additional) threads, a reader and a writer. From
 257    * the writer thread, write data synchronously in small chunks,
 258    * which gets alternately read and skipped asynchronously by the
 259    * main thread and then (if not skipped) written asynchronously to
 260    * the reader thread, which reads it synchronously. Eventually a
 261    * timeout in the main thread will cause it to cancel the writer
 262    * thread, which will in turn cancel the read op in the main thread,
 263    * which will then close the pipe to the reader thread, causing the
 264    * read op to fail.
 265    */
 266
 267   g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
 268
 269   if (nonblocking)
 270     {
 271       GError *error = NULL;
 272
 273       g_unix_set_fd_nonblocking (writer_pipe[0], TRUE, &error);
 274       g_assert_no_error (error);
 275       g_unix_set_fd_nonblocking (writer_pipe[1], TRUE, &error);
 276       g_assert_no_error (error);
 277       g_unix_set_fd_nonblocking (reader_pipe[0], TRUE, &error);
 278       g_assert_no_error (error);
 279       g_unix_set_fd_nonblocking (reader_pipe[1], TRUE, &error);
 280       g_assert_no_error (error);
 281     }
 282
 283   writer_cancel = g_cancellable_new ();
 284   reader_cancel = g_cancellable_new ();
 285   main_cancel = g_cancellable_new ();
 286
 287   writer = g_thread_new ("writer", writer_thread, NULL);
 288   reader = g_thread_new ("reader", reader_thread, NULL);
 289
 290   in = g_unix_input_stream_new (writer_pipe[0], TRUE);
 291   out = g_unix_output_stream_new (reader_pipe[1], TRUE);
 292
 293   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 294                              G_PRIORITY_DEFAULT, main_cancel,
 295                              main_thread_read, out);
 296
 297   g_timeout_add (500, timeout, writer_cancel);
 298
 299   loop = g_main_loop_new (NULL, TRUE);
 300   g_main_loop_run (loop);
 301   g_main_loop_unref (loop);
 302
 303   g_thread_join (reader);
 304   g_thread_join (writer);
 305
 306   g_object_unref (main_cancel);
 307   g_object_unref (reader_cancel);
 308   g_object_unref (writer_cancel);
 309   g_object_unref (in);
 310   g_object_unref (out);
 311 }
 312
 313 static void
 314 test_basic (void)
 315 {
 316   GUnixInputStream *is;
 317   GUnixOutputStream *os;
 318   gint fd;
 319   gboolean close_fd;
 320
 321   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (0, TRUE));
 322   g_object_get (is,
 323                 "fd", &fd,
 324                 "close-fd", &close_fd,
 325                 NULL);
 326   g_assert_cmpint (fd, ==, 0);
 327   g_assert (close_fd);
 328
 329   g_unix_input_stream_set_close_fd (is, FALSE);
 330   g_assert (!g_unix_input_stream_get_close_fd (is));
 331   g_assert_cmpint (g_unix_input_stream_get_fd (is), ==, 0);
 332
 333   g_assert (!g_input_stream_has_pending (G_INPUT_STREAM (is)));
 334
 335   g_object_unref (is);
 336
 337   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (1, TRUE));
 338   g_object_get (os,
 339                 "fd", &fd,
 340                 "close-fd", &close_fd,
 341                 NULL);
 342   g_assert_cmpint (fd, ==, 1);
 343   g_assert (close_fd);
 344
 345   g_unix_output_stream_set_close_fd (os, FALSE);
 346   g_assert (!g_unix_output_stream_get_close_fd (os));
 347   g_assert_cmpint (g_unix_output_stream_get_fd (os), ==, 1);
 348
 349   g_assert (!g_output_stream_has_pending (G_OUTPUT_STREAM (os)));
 350
 351   g_object_unref (os);
 352 }
 353
 354 int
 355 main (int   argc,
 356       char *argv[])
 357 {
 358   g_test_init (&argc, &argv, NULL);
 359
 360   g_test_add_func ("/unix-streams/basic", test_basic);
 361   g_test_add_data_func ("/unix-streams/pipe-io-test",
 362                         GINT_TO_POINTER (FALSE),
 363                         test_pipe_io);
 364   g_test_add_data_func ("/unix-streams/nonblocking-io-test",
 365                         GINT_TO_POINTER (TRUE),
 366                         test_pipe_io);
 367
 368   return g_test_run();
 369 }