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       return;
 182     }
 183
 184   g_assert_no_error (err);
 185
 186   main_offset += nread;
 187   if (main_offset == sizeof (DATA))
 188     {
 189       main_len = main_offset;
 190       main_offset = 0;
 191       /* Now skip the same amount */
 192       g_input_stream_skip_async (in, main_len,
 193                                  G_PRIORITY_DEFAULT, main_cancel,
 194                                  main_thread_skipped, out);
 195     }
 196   else
 197     {
 198       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 199                                  G_PRIORITY_DEFAULT, main_cancel,
 200                                  main_thread_read, out);
 201     }
 202 }
 203
 204 static void
 205 main_thread_wrote (GObject *source, GAsyncResult *res, gpointer user_data)
 206 {
 207   GOutputStream *out = G_OUTPUT_STREAM (source);
 208   GInputStream *in = user_data;
 209   GError *err = NULL;
 210   gssize nwrote;
 211
 212   nwrote = g_output_stream_write_finish (out, res, &err);
 213
 214   if (g_cancellable_is_cancelled (main_cancel))
 215     {
 216       do_main_cancel (out);
 217       return;
 218     }
 219
 220   g_assert_no_error (err);
 221   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 222
 223   main_offset += nwrote;
 224   if (main_offset == main_len)
 225     {
 226       main_offset = 0;
 227       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 228                                  G_PRIORITY_DEFAULT, main_cancel,
 229                                  main_thread_read, out);
 230     }
 231   else
 232     {
 233       g_output_stream_write_async (out, main_buf + main_offset,
 234                                    main_len - main_offset,
 235                                    G_PRIORITY_DEFAULT, main_cancel,
 236                                    main_thread_wrote, in);
 237     }
 238 }
 239
 240 static gboolean
 241 timeout (gpointer cancellable)
 242 {
 243   g_cancellable_cancel (cancellable);
 244   return FALSE;
 245 }
 246
 247 static void
 248 test_pipe_io (gconstpointer nonblocking)
 249 {
 250   GThread *writer, *reader;
 251   GInputStream *in;
 252   GOutputStream *out;
 253
 254   /* Split off two (additional) threads, a reader and a writer. From
 255    * the writer thread, write data synchronously in small chunks,
 256    * which gets alternately read and skipped asynchronously by the
 257    * main thread and then (if not skipped) written asynchronously to
 258    * the reader thread, which reads it synchronously. Eventually a
 259    * timeout in the main thread will cause it to cancel the writer
 260    * thread, which will in turn cancel the read op in the main thread,
 261    * which will then close the pipe to the reader thread, causing the
 262    * read op to fail.
 263    */
 264
 265   g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
 266
 267   if (nonblocking)
 268     {
 269       GError *error = NULL;
 270
 271       g_unix_set_fd_nonblocking (writer_pipe[0], TRUE, &error);
 272       g_assert_no_error (error);
 273       g_unix_set_fd_nonblocking (writer_pipe[1], TRUE, &error);
 274       g_assert_no_error (error);
 275       g_unix_set_fd_nonblocking (reader_pipe[0], TRUE, &error);
 276       g_assert_no_error (error);
 277       g_unix_set_fd_nonblocking (reader_pipe[1], TRUE, &error);
 278       g_assert_no_error (error);
 279     }
 280
 281   writer_cancel = g_cancellable_new ();
 282   reader_cancel = g_cancellable_new ();
 283   main_cancel = g_cancellable_new ();
 284
 285   writer = g_thread_new ("writer", writer_thread, NULL);
 286   reader = g_thread_new ("reader", reader_thread, NULL);
 287
 288   in = g_unix_input_stream_new (writer_pipe[0], TRUE);
 289   out = g_unix_output_stream_new (reader_pipe[1], TRUE);
 290
 291   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 292                              G_PRIORITY_DEFAULT, main_cancel,
 293                              main_thread_read, out);
 294
 295   g_timeout_add (500, timeout, writer_cancel);
 296
 297   loop = g_main_loop_new (NULL, TRUE);
 298   g_main_loop_run (loop);
 299   g_main_loop_unref (loop);
 300
 301   g_thread_join (reader);
 302   g_thread_join (writer);
 303
 304   g_object_unref (main_cancel);
 305   g_object_unref (reader_cancel);
 306   g_object_unref (writer_cancel);
 307   g_object_unref (in);
 308   g_object_unref (out);
 309 }
 310
 311 static void
 312 test_basic (void)
 313 {
 314   GUnixInputStream *is;
 315   GUnixOutputStream *os;
 316   gint fd;
 317   gboolean close_fd;
 318
 319   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (0, TRUE));
 320   g_object_get (is,
 321                 "fd", &fd,
 322                 "close-fd", &close_fd,
 323                 NULL);
 324   g_assert_cmpint (fd, ==, 0);
 325   g_assert (close_fd);
 326
 327   g_unix_input_stream_set_close_fd (is, FALSE);
 328   g_assert (!g_unix_input_stream_get_close_fd (is));
 329   g_assert_cmpint (g_unix_input_stream_get_fd (is), ==, 0);
 330
 331   g_assert (!g_input_stream_has_pending (G_INPUT_STREAM (is)));
 332
 333   g_object_unref (is);
 334
 335   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (1, TRUE));
 336   g_object_get (os,
 337                 "fd", &fd,
 338                 "close-fd", &close_fd,
 339                 NULL);
 340   g_assert_cmpint (fd, ==, 1);
 341   g_assert (close_fd);
 342
 343   g_unix_output_stream_set_close_fd (os, FALSE);
 344   g_assert (!g_unix_output_stream_get_close_fd (os));
 345   g_assert_cmpint (g_unix_output_stream_get_fd (os), ==, 1);
 346
 347   g_assert (!g_output_stream_has_pending (G_OUTPUT_STREAM (os)));
 348
 349   g_object_unref (os);
 350 }
 351
 352 int
 353 main (int   argc,
 354       char *argv[])
 355 {
 356   g_test_init (&argc, &argv, NULL);
 357
 358   g_test_add_func ("/unix-streams/basic", test_basic);
 359   g_test_add_data_func ("/unix-streams/pipe-io-test",
 360                         GINT_TO_POINTER (FALSE),
 361                         test_pipe_io);
 362   g_test_add_data_func ("/unix-streams/nonblocking-io-test",
 363                         GINT_TO_POINTER (TRUE),
 364                         test_pipe_io);
 365
 366   return g_test_run();
 367 }