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 readable (GObject *source, GAsyncResult *res, gpointer user_data);
 124 static void writable (GObject *source, GAsyncResult *res, gpointer user_data);
 125
 126 static void
 127 do_main_cancel (GOutputStream *out)
 128 {
 129   g_output_stream_close (out, NULL, NULL);
 130   g_main_loop_quit (loop);
 131 }
 132
 133 static void
 134 readable (GObject *source, GAsyncResult *res, gpointer user_data)
 135 {
 136   GInputStream *in = G_INPUT_STREAM (source);
 137   GOutputStream *out = user_data;
 138   GError *err = NULL;
 139
 140   main_len = g_input_stream_read_finish (in, res, &err);
 141
 142   if (g_cancellable_is_cancelled (main_cancel))
 143     {
 144       do_main_cancel (out);
 145       return;
 146     }
 147
 148   g_assert (err == NULL);
 149
 150   main_offset = 0;
 151   g_output_stream_write_async (out, main_buf, main_len,
 152                                G_PRIORITY_DEFAULT, main_cancel,
 153                                writable, in);
 154 }
 155
 156 static void
 157 writable (GObject *source, GAsyncResult *res, gpointer user_data)
 158 {
 159   GOutputStream *out = G_OUTPUT_STREAM (source);
 160   GInputStream *in = user_data;
 161   GError *err = NULL;
 162   gssize nwrote;
 163
 164   nwrote = g_output_stream_write_finish (out, res, &err);
 165
 166   if (g_cancellable_is_cancelled (main_cancel))
 167     {
 168       do_main_cancel (out);
 169       return;
 170     }
 171
 172   g_assert (err == NULL);
 173   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 174
 175   main_offset += nwrote;
 176   if (main_offset == main_len)
 177     {
 178       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 179                                  G_PRIORITY_DEFAULT, main_cancel,
 180                                  readable, out);
 181     }
 182   else
 183     {
 184       g_output_stream_write_async (out, main_buf + main_offset,
 185                                    main_len - main_offset,
 186                                    G_PRIORITY_DEFAULT, main_cancel,
 187                                    writable, in);
 188     }
 189 }
 190
 191 static gboolean
 192 timeout (gpointer cancellable)
 193 {
 194   g_cancellable_cancel (cancellable);
 195   return FALSE;
 196 }
 197
 198 static void
 199 test_pipe_io (gconstpointer nonblocking)
 200 {
 201   GThread *writer, *reader;
 202   GInputStream *in;
 203   GOutputStream *out;
 204
 205   /* Split off two (additional) threads, a reader and a writer. From
 206    * the writer thread, write data synchronously in small chunks,
 207    * which gets read asynchronously by the main thread and then
 208    * written asynchronously to the reader thread, which reads it
 209    * synchronously. Eventually a timeout in the main thread will cause
 210    * it to cancel the writer thread, which will in turn cancel the
 211    * read op in the main thread, which will then close the pipe to
 212    * the reader thread, causing the read op to fail.
 213    */
 214
 215   g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
 216
 217   if (nonblocking)
 218     {
 219       GError *error = NULL;
 220
 221       g_unix_set_fd_nonblocking (writer_pipe[0], TRUE, &error);
 222       g_assert_no_error (error);
 223       g_unix_set_fd_nonblocking (writer_pipe[1], TRUE, &error);
 224       g_assert_no_error (error);
 225       g_unix_set_fd_nonblocking (reader_pipe[0], TRUE, &error);
 226       g_assert_no_error (error);
 227       g_unix_set_fd_nonblocking (reader_pipe[1], TRUE, &error);
 228       g_assert_no_error (error);
 229     }
 230
 231   writer_cancel = g_cancellable_new ();
 232   reader_cancel = g_cancellable_new ();
 233   main_cancel = g_cancellable_new ();
 234
 235   writer = g_thread_new ("writer", writer_thread, NULL);
 236   reader = g_thread_new ("reader", reader_thread, NULL);
 237
 238   in = g_unix_input_stream_new (writer_pipe[0], TRUE);
 239   out = g_unix_output_stream_new (reader_pipe[1], TRUE);
 240
 241   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 242                              G_PRIORITY_DEFAULT, main_cancel,
 243                              readable, out);
 244
 245   g_timeout_add (500, timeout, writer_cancel);
 246
 247   loop = g_main_loop_new (NULL, TRUE);
 248   g_main_loop_run (loop);
 249   g_main_loop_unref (loop);
 250
 251   g_thread_join (reader);
 252   g_thread_join (writer);
 253
 254   g_object_unref (main_cancel);
 255   g_object_unref (reader_cancel);
 256   g_object_unref (writer_cancel);
 257   g_object_unref (in);
 258   g_object_unref (out);
 259 }
 260
 261 static void
 262 test_basic (void)
 263 {
 264   GUnixInputStream *is;
 265   GUnixOutputStream *os;
 266   gint fd;
 267   gboolean close_fd;
 268
 269   is = G_UNIX_INPUT_STREAM (g_unix_input_stream_new (0, TRUE));
 270   g_object_get (is,
 271                 "fd", &fd,
 272                 "close-fd", &close_fd,
 273                 NULL);
 274   g_assert_cmpint (fd, ==, 0);
 275   g_assert (close_fd);
 276
 277   g_unix_input_stream_set_close_fd (is, FALSE);
 278   g_assert (!g_unix_input_stream_get_close_fd (is));
 279   g_assert_cmpint (g_unix_input_stream_get_fd (is), ==, 0);
 280
 281   g_object_unref (is);
 282
 283   os = G_UNIX_OUTPUT_STREAM (g_unix_output_stream_new (1, TRUE));
 284   g_object_get (os,
 285                 "fd", &fd,
 286                 "close-fd", &close_fd,
 287                 NULL);
 288   g_assert_cmpint (fd, ==, 1);
 289   g_assert (close_fd);
 290
 291   g_unix_output_stream_set_close_fd (os, FALSE);
 292   g_assert (!g_unix_output_stream_get_close_fd (os));
 293   g_assert_cmpint (g_unix_output_stream_get_fd (os), ==, 1);
 294
 295   g_object_unref (os);
 296 }
 297
 298 int
 299 main (int   argc,
 300       char *argv[])
 301 {
 302   g_type_init ();
 303   g_test_init (&argc, &argv, NULL);
 304
 305   g_test_add_func ("/unix-streams/basic", test_basic);
 306   g_test_add_data_func ("/unix-streams/pipe-io-test",
 307                         GINT_TO_POINTER (FALSE),
 308                         test_pipe_io);
 309   g_test_add_data_func ("/unix-streams/nonblocking-io-test",
 310                         GINT_TO_POINTER (TRUE),
 311                         test_pipe_io);
 312
 313   return g_test_run();
 314 }