#include "gcancellable.h"
#include "gsimpleasyncresult.h"
#include "gasynchelper.h"
+#include "gfiledescriptorbased.h"
#include "glibintl.h"
-#include "gioalias.h"
/**
* SECTION:gunixoutputstream
- * @short_description: Streaming output operations for Unix file descriptors
+ * @short_description: Streaming output operations for UNIX file descriptors
* @include: gio/gunixoutputstream.h
* @see_also: #GOutputStream
*
- * #GUnixOutputStream implements #GOutputStream for writing to a
- * unix file descriptor, including asynchronous operations. The file
- * descriptor must be selectable, so it doesn't work with opened files.
+ * #GUnixOutputStream implements #GOutputStream for writing to a UNIX
+ * file descriptor, including asynchronous operations. (If the file
+ * descriptor refers to a socket or pipe, this will use poll() to do
+ * asynchronous I/O. If it refers to a regular file, it will fall back
+ * to doing asynchronous I/O in another thread.)
*
- * Note that <filename><gio/gunixoutputstream.h></filename> belongs
+ * Note that <filename><gio/gunixoutputstream.h></filename> belongs
* to the UNIX-specific GIO interfaces, thus you have to use the
* <filename>gio-unix-2.0.pc</filename> pkg-config file when using it.
*/
PROP_CLOSE_FD
};
-G_DEFINE_TYPE (GUnixOutputStream, g_unix_output_stream, G_TYPE_OUTPUT_STREAM);
+static void g_unix_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface);
+static void g_unix_output_stream_file_descriptor_based_iface_init (GFileDescriptorBasedIface *iface);
+G_DEFINE_TYPE_WITH_CODE (GUnixOutputStream, g_unix_output_stream, G_TYPE_OUTPUT_STREAM,
+ G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_OUTPUT_STREAM,
+ g_unix_output_stream_pollable_iface_init)
+ G_IMPLEMENT_INTERFACE (G_TYPE_FILE_DESCRIPTOR_BASED,
+ g_unix_output_stream_file_descriptor_based_iface_init)
+ )
struct _GUnixOutputStreamPrivate {
int fd;
- gboolean close_fd;
+ guint close_fd : 1;
+ guint is_pipe_or_socket : 1;
};
static void g_unix_output_stream_set_property (GObject *object,
GAsyncResult *result,
GError **error);
+static gboolean g_unix_output_stream_pollable_is_writable (GPollableOutputStream *stream);
+static GSource *g_unix_output_stream_pollable_create_source (GPollableOutputStream *stream,
+ GCancellable *cancellable);
static void
g_unix_output_stream_finalize (GObject *object)
}
static void
+g_unix_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface)
+{
+ iface->is_writable = g_unix_output_stream_pollable_is_writable;
+ iface->create_source = g_unix_output_stream_pollable_create_source;
+}
+
+static void
+g_unix_output_stream_file_descriptor_based_iface_init (GFileDescriptorBasedIface *iface)
+{
+ iface->get_fd = (int (*) (GFileDescriptorBased *))g_unix_output_stream_get_fd;
+}
+
+static void
g_unix_output_stream_set_property (GObject *object,
guint prop_id,
const GValue *value,
{
case PROP_FD:
unix_stream->priv->fd = g_value_get_int (value);
+ if (lseek (unix_stream->priv->fd, 0, SEEK_CUR) == -1 && errno == ESPIPE)
+ unix_stream->priv->is_pipe_or_socket = TRUE;
+ else
+ unix_stream->priv->is_pipe_or_socket = FALSE;
break;
case PROP_CLOSE_FD:
unix_stream->priv->close_fd = g_value_get_boolean (value);
GError **error)
{
GUnixOutputStream *unix_stream;
- gssize res;
+ gssize res = -1;
GPollFD poll_fds[2];
+ int nfds;
int poll_ret;
unix_stream = G_UNIX_OUTPUT_STREAM (stream);
- if (g_cancellable_make_pollfd (cancellable, &poll_fds[1]))
+ poll_fds[0].fd = unix_stream->priv->fd;
+ poll_fds[0].events = G_IO_OUT;
+
+ if (unix_stream->priv->is_pipe_or_socket &&
+ g_cancellable_make_pollfd (cancellable, &poll_fds[1]))
+ nfds = 2;
+ else
+ nfds = 1;
+
+ while (1)
{
- poll_fds[0].fd = unix_stream->priv->fd;
- poll_fds[0].events = G_IO_OUT;
+ poll_fds[0].revents = poll_fds[1].revents = 0;
do
- poll_ret = g_poll (poll_fds, 2, -1);
+ poll_ret = g_poll (poll_fds, nfds, -1);
while (poll_ret == -1 && errno == EINTR);
- g_cancellable_release_fd (cancellable);
-
+
if (poll_ret == -1)
{
int errsv = errno;
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
- _("Error writing to unix: %s"),
+ _("Error writing to file descriptor: %s"),
g_strerror (errsv));
- return -1;
+ break;
}
- }
-
- while (1)
- {
+
if (g_cancellable_set_error_if_cancelled (cancellable, error))
- return -1;
+ break;
+
+ if (!poll_fds[0].revents)
+ continue;
res = write (unix_stream->priv->fd, buffer, count);
if (res == -1)
{
int errsv = errno;
- if (errsv == EINTR)
+ if (errsv == EINTR || errsv == EAGAIN)
continue;
-
+
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
- _("Error writing to unix: %s"),
+ _("Error writing to file descriptor: %s"),
g_strerror (errsv));
}
-
+
break;
}
-
+
+ if (nfds == 2)
+ g_cancellable_release_fd (cancellable);
return res;
}
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
- _("Error closing unix: %s"),
+ _("Error closing file descriptor: %s"),
g_strerror (errsv));
}
break;
} WriteAsyncData;
static gboolean
-write_async_cb (WriteAsyncData *data,
+write_async_cb (int fd,
GIOCondition condition,
- int fd)
+ WriteAsyncData *data)
{
GSimpleAsyncResult *simple;
GError *error = NULL;
{
int errsv = errno;
- if (errsv == EINTR)
- continue;
+ if (errsv == EINTR || errsv == EAGAIN)
+ return TRUE;
g_set_error (&error, G_IO_ERROR,
g_io_error_from_errno (errsv),
- _("Error reading from unix: %s"),
+ _("Error writing to file descriptor: %s"),
g_strerror (errsv));
}
break;
g_simple_async_result_set_op_res_gssize (simple, count_written);
if (count_written == -1)
- {
- g_simple_async_result_set_from_error (simple, error);
- g_error_free (error);
- }
+ g_simple_async_result_take_error (simple, error);
/* Complete immediately, not in idle, since we're already in a mainloop callout */
g_simple_async_result_complete (simple);
unix_stream = G_UNIX_OUTPUT_STREAM (stream);
+ if (!unix_stream->priv->is_pipe_or_socket)
+ {
+ G_OUTPUT_STREAM_CLASS (g_unix_output_stream_parent_class)->
+ write_async (stream, buffer, count, io_priority,
+ cancellable, callback, user_data);
+ return;
+ }
+
data = g_new0 (WriteAsyncData, 1);
data->count = count;
data->buffer = buffer;
source = _g_fd_source_new (unix_stream->priv->fd,
G_IO_OUT,
cancellable);
+ g_source_set_name (source, "GUnixOutputStream");
g_source_set_callback (source, (GSourceFunc)write_async_cb, data, g_free);
g_source_attach (source, g_main_context_get_thread_default ());
GAsyncResult *result,
GError **error)
{
+ GUnixOutputStream *unix_stream = G_UNIX_OUTPUT_STREAM (stream);
GSimpleAsyncResult *simple;
gssize nwritten;
+ if (!unix_stream->priv->is_pipe_or_socket)
+ {
+ return G_OUTPUT_STREAM_CLASS (g_unix_output_stream_parent_class)->
+ write_finish (stream, result, error);
+ }
+
simple = G_SIMPLE_ASYNC_RESULT (result);
g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_unix_output_stream_write_async);
g_set_error (&error, G_IO_ERROR,
g_io_error_from_errno (errsv),
- _("Error closing unix: %s"),
+ _("Error closing file descriptor: %s"),
g_strerror (errsv));
}
break;
g_unix_output_stream_close_async);
if (!result)
- {
- g_simple_async_result_set_from_error (simple, error);
- g_error_free (error);
- }
+ g_simple_async_result_take_error (simple, error);
/* Complete immediately, not in idle, since we're already in a mainloop callout */
g_simple_async_result_complete (simple);
return TRUE;
}
-#define __G_UNIX_OUTPUT_STREAM_C__
-#include "gioaliasdef.c"
+static gboolean
+g_unix_output_stream_pollable_is_writable (GPollableOutputStream *stream)
+{
+ GUnixOutputStream *unix_stream = G_UNIX_OUTPUT_STREAM (stream);
+ GPollFD poll_fd;
+ gint result;
+
+ poll_fd.fd = unix_stream->priv->fd;
+ poll_fd.events = G_IO_OUT;
+
+ do
+ result = g_poll (&poll_fd, 1, 0);
+ while (result == -1 && errno == EINTR);
+
+ return poll_fd.revents != 0;
+}
+
+static GSource *
+g_unix_output_stream_pollable_create_source (GPollableOutputStream *stream,
+ GCancellable *cancellable)
+{
+ GUnixOutputStream *unix_stream = G_UNIX_OUTPUT_STREAM (stream);
+ GSource *inner_source, *pollable_source;
+
+ pollable_source = g_pollable_source_new (G_OBJECT (stream));
+
+ inner_source = _g_fd_source_new (unix_stream->priv->fd, G_IO_OUT, cancellable);
+ g_source_set_dummy_callback (inner_source);
+ g_source_add_child_source (pollable_source, inner_source);
+ g_source_unref (inner_source);
+
+ return pollable_source;
+}
projects / platform / upstream / glib2.0.git / blobdiff