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 <glib/glib.h>
  23 #include <gio/gio.h>
  24 #include <gio/gunixinputstream.h>
  25 #include <gio/gunixoutputstream.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 static gboolean
  38 cancel_main (gpointer data)
  39 {
  40   GCancellable *main_cancel = data;
  41
  42   g_cancellable_cancel (main_cancel);
  43
  44   return FALSE;
  45 }
  46
  47
  48 static gpointer
  49 writer_thread (gpointer user_data)
  50 {
  51   GOutputStream *out;
  52   gssize nwrote, offset;
  53   GError *err = NULL;
  54
  55   out = g_unix_output_stream_new (writer_pipe[1], TRUE);
  56
  57   do
  58     {
  59       g_usleep (10);
  60
  61       offset = 0;
  62       while (offset < (gssize) sizeof (DATA))
  63         {
  64           nwrote = g_output_stream_write (out, DATA + offset,
  65                                           sizeof (DATA) - offset,
  66                                           writer_cancel, &err);
  67           if (nwrote <= 0 || err != NULL)
  68             break;
  69           offset += nwrote;
  70         }
  71
  72       g_assert (nwrote > 0 || err != NULL);
  73     }
  74   while (err == NULL);
  75
  76   if (g_cancellable_is_cancelled (writer_cancel))
  77     {
  78       /* FIXME: directly calling g_cancellable_cancel (main_cancel) here
  79        * leads to sporadic deadlock, because it will try to wake up the
  80        * main context, for which it needs to acquire the main context lock.
  81        * This lock may be held by the main loop running in the main thread,
  82        * and it may be held while the main thread is blocking in
  83        * fd_source_finalize -> g_cancellable_disconnect
  84        * until the ::cancelled callbacks have run.
  85        *
  86        * Work around by deferring the cancellation to a timeout.
  87        */
  88       g_timeout_add (0, cancel_main, main_cancel);
  89       g_object_unref (out);
  90       return NULL;
  91     }
  92
  93   g_warning ("writer: %s", err->message);
  94   g_assert_not_reached ();
  95 }
  96
  97 static gpointer
  98 reader_thread (gpointer user_data)
  99 {
 100   GInputStream *in;
 101   gssize nread = 0, total;
 102   GError *err = NULL;
 103   char buf[sizeof (DATA)];
 104
 105   in = g_unix_input_stream_new (reader_pipe[0], TRUE);
 106
 107   do
 108     {
 109       total = 0;
 110       while (total < (gssize) sizeof (DATA))
 111         {
 112           nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
 113                                        reader_cancel, &err);
 114           if (nread <= 0 || err != NULL)
 115             break;
 116           total += nread;
 117         }
 118
 119       if (err)
 120         break;
 121
 122       if (nread == 0)
 123         {
 124           g_assert (err == NULL);
 125           /* pipe closed */
 126           g_object_unref (in);
 127           return NULL;
 128         }
 129
 130       g_assert_cmpstr (buf, ==, DATA);
 131       g_assert (!g_cancellable_is_cancelled (reader_cancel));
 132     }
 133   while (err == NULL);
 134
 135   g_warning ("reader: %s", err->message);
 136   g_assert_not_reached ();
 137 }
 138
 139 char main_buf[sizeof (DATA)];
 140 gssize main_len, main_offset;
 141
 142 static void readable (GObject *source, GAsyncResult *res, gpointer user_data);
 143 static void writable (GObject *source, GAsyncResult *res, gpointer user_data);
 144
 145 static void
 146 do_main_cancel (GOutputStream *out)
 147 {
 148   g_output_stream_close (out, NULL, NULL);
 149   g_main_loop_quit (loop);
 150 }
 151
 152 static void
 153 readable (GObject *source, GAsyncResult *res, gpointer user_data)
 154 {
 155   GInputStream *in = G_INPUT_STREAM (source);
 156   GOutputStream *out = user_data;
 157   GError *err = NULL;
 158
 159   main_len = g_input_stream_read_finish (in, res, &err);
 160
 161   if (g_cancellable_is_cancelled (main_cancel))
 162     {
 163       do_main_cancel (out);
 164       return;
 165     }
 166
 167   g_assert (err == NULL);
 168
 169   main_offset = 0;
 170   g_output_stream_write_async (out, main_buf, main_len,
 171                                G_PRIORITY_DEFAULT, main_cancel,
 172                                writable, in);
 173 }
 174
 175 static void
 176 writable (GObject *source, GAsyncResult *res, gpointer user_data)
 177 {
 178   GOutputStream *out = G_OUTPUT_STREAM (source);
 179   GInputStream *in = user_data;
 180   GError *err = NULL;
 181   gssize nwrote;
 182
 183   nwrote = g_output_stream_write_finish (out, res, &err);
 184
 185   if (g_cancellable_is_cancelled (main_cancel))
 186     {
 187       do_main_cancel (out);
 188       return;
 189     }
 190
 191   g_assert (err == NULL);
 192   g_assert_cmpint (nwrote, <=, main_len - main_offset);
 193
 194   main_offset += nwrote;
 195   if (main_offset == main_len)
 196     {
 197       g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 198                                  G_PRIORITY_DEFAULT, main_cancel,
 199                                  readable, out);
 200     }
 201   else
 202     {
 203       g_output_stream_write_async (out, main_buf + main_offset,
 204                                    main_len - main_offset,
 205                                    G_PRIORITY_DEFAULT, main_cancel,
 206                                    writable, in);
 207     }
 208 }
 209
 210 static gboolean
 211 timeout (gpointer cancellable)
 212 {
 213   g_cancellable_cancel (cancellable);
 214   return FALSE;
 215 }
 216
 217 static void
 218 test_pipe_io (void)
 219 {
 220   GThread *writer, *reader;
 221   GInputStream *in;
 222   GOutputStream *out;
 223
 224   /* Split off two (additional) threads, a reader and a writer. From
 225    * the writer thread, write data synchronously in small chunks,
 226    * which gets read asynchronously by the main thread and then
 227    * written asynchronously to the reader thread, which reads it
 228    * synchronously. Eventually a timeout in the main thread will cause
 229    * it to cancel the writer thread, which will in turn cancel the
 230    * read op in the main thread, which will then close the pipe to
 231    * the reader thread, causing the read op to fail.
 232    */
 233
 234   g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
 235
 236   writer_cancel = g_cancellable_new ();
 237   reader_cancel = g_cancellable_new ();
 238   main_cancel = g_cancellable_new ();
 239
 240   writer = g_thread_create (writer_thread, NULL, TRUE, NULL);
 241   reader = g_thread_create (reader_thread, NULL, TRUE, NULL);
 242
 243   in = g_unix_input_stream_new (writer_pipe[0], TRUE);
 244   out = g_unix_output_stream_new (reader_pipe[1], TRUE);
 245
 246   g_input_stream_read_async (in, main_buf, sizeof (main_buf),
 247                              G_PRIORITY_DEFAULT, main_cancel,
 248                              readable, out);
 249
 250   g_timeout_add (500, timeout, writer_cancel);
 251
 252   loop = g_main_loop_new (NULL, TRUE);
 253   g_main_loop_run (loop);
 254   g_main_loop_unref (loop);
 255
 256   g_thread_join (reader);
 257   g_thread_join (writer);
 258
 259   g_object_unref (main_cancel);
 260   g_object_unref (reader_cancel);
 261   g_object_unref (writer_cancel);
 262   g_object_unref (in);
 263   g_object_unref (out);
 264 }
 265
 266 int
 267 main (int   argc,
 268       char *argv[])
 269 {
 270   g_thread_init (NULL);
 271   g_type_init ();
 272   g_test_init (&argc, &argv, NULL);
 273
 274   g_test_add_func ("/unix-streams/pipe-io-test", test_pipe_io);
 275
 276   return g_test_run();
 277 }