guint cancelled : 1;
guint allocated_pipe : 1;
int cancel_pipe[2];
+
+#ifdef G_OS_WIN32
+ GIOChannel *read_channel;
+#endif
};
static guint signals[LAST_SIGNAL] = { 0 };
if (cancellable->cancel_pipe[1] != -1)
close (cancellable->cancel_pipe[1]);
+#ifdef G_OS_WIN32
+ if (cancellable->read_channel)
+ g_io_channel_unref (cancellable->read_channel);
+#endif
+
G_OBJECT_CLASS (g_cancellable_parent_class)->finalize (object);
}
if (cancellable->cancelled)
{
char ch;
+#ifdef G_OS_WIN32
+ if (cancellable->read_channel)
+ {
+ gsize bytes_read;
+ g_io_channel_read_chars (cancellable->read_channel, &ch, 1,
+ &bytes_read, NULL);
+ }
+ else
+#endif
if (cancellable->cancel_pipe[0] != -1)
read (cancellable->cancel_pipe[0], &ch, 1);
cancellable->cancelled = FALSE;
* Gets the file descriptor for a cancellable job. This can be used to
* implement cancellable operations on Unix systems. The returned fd will
* turn readable when @cancellable is cancelled.
- *
+ *
+ * See also g_cancellable_make_pollfd().
+ *
* Returns: A valid file descriptor. %-1 if the file descriptor
* is not supported, or on errors.
**/
}
/**
+ * g_cancellable_make_pollfd:
+ * @cancellable: a #GCancellable.
+ * @pollfd: a pointer to a #GPollFD
+ *
+ * Creates a #GPollFD corresponding to @cancellable; this can be passed
+ * to g_poll() and used to poll for cancellation.
+ **/
+void
+g_cancellable_make_pollfd (GCancellable *cancellable, GPollFD *pollfd)
+{
+ g_return_if_fail (G_IS_CANCELLABLE (cancellable));
+ g_return_if_fail (pollfd != NULL);
+
+#ifdef G_OS_WIN32
+ if (!cancellable->read_channel)
+ {
+ int fd = g_cancellable_get_fd (cancellable);
+ cancellable->read_channel = g_io_channel_win32_new_fd (fd);
+ g_io_channel_set_buffered (cancellable->read_channel, FALSE);
+ g_io_channel_set_flags (cancellable->read_channel,
+ G_IO_FLAG_NONBLOCK, NULL);
+ g_io_channel_set_encoding (cancellable->read_channel, NULL, NULL);
+ }
+ g_io_channel_win32_make_pollfd (cancellable->read_channel, G_IO_IN, pollfd);
+ /* (We need to keep cancellable->read_channel around, because it's
+ * keeping track of state related to the pollfd.)
+ */
+#else /* !G_OS_WIN32 */
+ pollfd->fd = g_cancellable_get_fd (cancellable);
+ pollfd->events = G_IO_IN;
+#endif /* G_OS_WIN32 */
+ pollfd->revents = 0;
+}
+
+/**
* g_cancellable_cancel:
* @cancellable: a #GCancellable object.
*
--- /dev/null
+/* GLib testing framework examples and tests
+ * Copyright (C) 2008 Red Hat, Inc
+ *
+ * This work is provided "as is"; redistribution and modification
+ * in whole or in part, in any medium, physical or electronic is
+ * permitted without restriction.
+ *
+ * This work is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * In no event shall the authors or contributors be liable for any
+ * direct, indirect, incidental, special, exemplary, or consequential
+ * damages (including, but not limited to, procurement of substitute
+ * goods or services; loss of use, data, or profits; or business
+ * interruption) however caused and on any theory of liability, whether
+ * in contract, strict liability, or tort (including negligence or
+ * otherwise) arising in any way out of the use of this software, even
+ * if advised of the possibility of such damage.
+ */
+
+#include <glib/glib.h>
+#include <gio/gio.h>
+#include <gio/gunixinputstream.h>
+#include <gio/gunixoutputstream.h>
+#include <signal.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define DATA "abcdefghijklmnopqrstuvwxyz"
+
+int writer_pipe[2], reader_pipe[2];
+GCancellable *writer_cancel, *reader_cancel, *main_cancel;
+GMainLoop *loop;
+
+static gpointer
+writer_thread (gpointer user_data)
+{
+ GOutputStream *out;
+ gssize nwrote, offset;
+ GError *err = NULL;
+
+ out = g_unix_output_stream_new (writer_pipe[1], TRUE);
+
+ do
+ {
+ g_usleep (10);
+
+ offset = 0;
+ while (offset < sizeof (DATA))
+ {
+ nwrote = g_output_stream_write (out, DATA + offset,
+ sizeof (DATA) - offset,
+ writer_cancel, &err);
+ if (nwrote <= 0 || err != NULL)
+ break;
+ offset += nwrote;
+ }
+
+ g_assert (nwrote > 0 || err != NULL);
+ }
+ while (err == NULL);
+
+ if (g_cancellable_is_cancelled (writer_cancel))
+ {
+ g_cancellable_cancel (main_cancel);
+ g_object_unref (out);
+ return NULL;
+ }
+
+ g_warning ("writer: %s", err->message);
+ g_assert_not_reached ();
+}
+
+static gpointer
+reader_thread (gpointer user_data)
+{
+ GInputStream *in;
+ gssize nread, total;
+ GError *err = NULL;
+ char buf[sizeof (DATA)];
+
+ in = g_unix_input_stream_new (reader_pipe[0], TRUE);
+
+ do
+ {
+ total = 0;
+ while (total < sizeof (DATA))
+ {
+ nread = g_input_stream_read (in, buf + total, sizeof (buf) - total,
+ reader_cancel, &err);
+ if (nread <= 0 || err != NULL)
+ break;
+ total += nread;
+ }
+
+ if (err)
+ break;
+
+ if (nread == 0)
+ {
+ g_assert (err == NULL);
+ /* pipe closed */
+ g_object_unref (in);
+ return NULL;
+ }
+
+ g_assert_cmpstr (buf, ==, DATA);
+ g_assert (!g_cancellable_is_cancelled (reader_cancel));
+ }
+ while (err == NULL);
+
+ g_warning ("reader: %s", err->message);
+ g_assert_not_reached ();
+}
+
+char main_buf[sizeof (DATA)];
+gssize main_len, main_offset;
+
+static void readable (GObject *source, GAsyncResult *res, gpointer user_data);
+static void writable (GObject *source, GAsyncResult *res, gpointer user_data);
+
+static void
+do_main_cancel (GOutputStream *out)
+{
+ g_output_stream_close (out, NULL, NULL);
+ g_main_loop_quit (loop);
+}
+
+static void
+readable (GObject *source, GAsyncResult *res, gpointer user_data)
+{
+ GInputStream *in = G_INPUT_STREAM (source);
+ GOutputStream *out = user_data;
+ GError *err = NULL;
+
+ main_len = g_input_stream_read_finish (in, res, &err);
+
+ if (g_cancellable_is_cancelled (main_cancel))
+ {
+ do_main_cancel (out);
+ return;
+ }
+
+ g_assert (err == NULL);
+
+ main_offset = 0;
+ g_output_stream_write_async (out, main_buf, main_len,
+ G_PRIORITY_DEFAULT, main_cancel,
+ writable, in);
+}
+
+static void
+writable (GObject *source, GAsyncResult *res, gpointer user_data)
+{
+ GOutputStream *out = G_OUTPUT_STREAM (source);
+ GInputStream *in = user_data;
+ GError *err = NULL;
+ gssize nwrote;
+
+ nwrote = g_output_stream_write_finish (out, res, &err);
+
+ if (g_cancellable_is_cancelled (main_cancel))
+ {
+ do_main_cancel (out);
+ return;
+ }
+
+ g_assert (err == NULL);
+ g_assert_cmpint (nwrote, <=, main_len - main_offset);
+
+ main_offset += nwrote;
+ if (main_offset == main_len)
+ {
+ g_input_stream_read_async (in, main_buf, sizeof (main_buf),
+ G_PRIORITY_DEFAULT, main_cancel,
+ readable, out);
+ }
+ else
+ {
+ g_output_stream_write_async (out, main_buf + main_offset,
+ main_len - main_offset,
+ G_PRIORITY_DEFAULT, main_cancel,
+ writable, in);
+ }
+}
+
+static gboolean
+timeout (gpointer cancellable)
+{
+ g_cancellable_cancel (cancellable);
+ return FALSE;
+}
+
+static void
+test_pipe_io (void)
+{
+ GThread *writer, *reader;
+ GInputStream *in;
+ GOutputStream *out;
+
+ /* Split off two (additional) threads, a reader and a writer. From
+ * the writer thread, write data synchronously in small chunks,
+ * which gets read asynchronously by the main thread and then
+ * written asynchronously to the reader thread, which reads it
+ * synchronously. Eventually a timeout in the main thread will cause
+ * it to cancel the writer thread, which will in turn cancel the
+ * read op in the main thread, which will then close the pipe to
+ * the reader thread, causing the read op to fail.
+ */
+
+ g_assert (pipe (writer_pipe) == 0 && pipe (reader_pipe) == 0);
+
+ writer_cancel = g_cancellable_new ();
+ reader_cancel = g_cancellable_new ();
+ main_cancel = g_cancellable_new ();
+
+ writer = g_thread_create (writer_thread, NULL, TRUE, NULL);
+ reader = g_thread_create (reader_thread, NULL, TRUE, NULL);
+
+ in = g_unix_input_stream_new (writer_pipe[0], TRUE);
+ out = g_unix_output_stream_new (reader_pipe[1], TRUE);
+
+ g_input_stream_read_async (in, main_buf, sizeof (main_buf),
+ G_PRIORITY_DEFAULT, main_cancel,
+ readable, out);
+
+ g_timeout_add (500, timeout, writer_cancel);
+
+ loop = g_main_loop_new (NULL, TRUE);
+ g_main_loop_run (loop);
+ g_main_loop_unref (loop);
+
+ g_thread_join (reader);
+ g_thread_join (writer);
+
+ g_object_unref (main_cancel);
+ g_object_unref (reader_cancel);
+ g_object_unref (writer_cancel);
+ g_object_unref (in);
+ g_object_unref (out);
+}
+
+int
+main (int argc,
+ char *argv[])
+{
+ g_thread_init (NULL);
+ g_type_init ();
+ g_test_init (&argc, &argv, NULL);
+
+ g_test_add_func ("/unix-streams/pipe-io-test", test_pipe_io);
+
+ return g_test_run();
+}