Implement GOutputStream::writev() and GPollableOutputStream::writev_nonblocking() GUnixOutputStream

This commit is contained in:
Sebastian Dröge 2018-09-13 21:23:56 +03:00
parent c39264d35b
commit 90d9e4ab72

View File

@ -26,6 +26,7 @@
#include <errno.h>
#include <stdio.h>
#include <fcntl.h>
#include <sys/uio.h>
#include <glib.h>
#include <glib/gstdio.h>
@ -91,6 +92,12 @@ static gssize g_unix_output_stream_write (GOutputStream *stream,
gsize count,
GCancellable *cancellable,
GError **error);
static gboolean g_unix_output_stream_writev (GOutputStream *stream,
const GOutputVector *vectors,
gsize n_vectors,
gsize *bytes_written,
GCancellable *cancellable,
GError **error);
static gboolean g_unix_output_stream_close (GOutputStream *stream,
GCancellable *cancellable,
GError **error);
@ -107,6 +114,11 @@ static gboolean g_unix_output_stream_pollable_can_poll (GPollableOutputStre
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 GPollableReturn g_unix_output_stream_pollable_writev_nonblocking (GPollableOutputStream *stream,
const GOutputVector *vectors,
gsize n_vectors,
gsize *bytes_written,
GError **error);
static void
g_unix_output_stream_class_init (GUnixOutputStreamClass *klass)
@ -118,6 +130,7 @@ g_unix_output_stream_class_init (GUnixOutputStreamClass *klass)
gobject_class->set_property = g_unix_output_stream_set_property;
stream_class->write_fn = g_unix_output_stream_write;
stream_class->writev_fn = g_unix_output_stream_writev;
stream_class->close_fn = g_unix_output_stream_close;
stream_class->close_async = g_unix_output_stream_close_async;
stream_class->close_finish = g_unix_output_stream_close_finish;
@ -159,6 +172,7 @@ g_unix_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface)
iface->can_poll = g_unix_output_stream_pollable_can_poll;
iface->is_writable = g_unix_output_stream_pollable_is_writable;
iface->create_source = g_unix_output_stream_pollable_create_source;
iface->writev_nonblocking = g_unix_output_stream_pollable_writev_nonblocking;
}
static void
@ -387,6 +401,116 @@ g_unix_output_stream_write (GOutputStream *stream,
return res;
}
/* Macro to check if struct iovec and GOutputVector have the same ABI */
#define G_OUTPUT_VECTOR_IS_IOVEC (sizeof (struct iovec) == sizeof (GOutputVector) && \
sizeof ((struct iovec *) 0)->iov_base == sizeof ((GOutputVector *) 0)->buffer && \
G_STRUCT_OFFSET (struct iovec, iov_base) == G_STRUCT_OFFSET (GOutputVector, buffer) && \
sizeof ((struct iovec *) 0)->iov_len == sizeof((GOutputVector *) 0)->size && \
G_STRUCT_OFFSET (struct iovec, iov_len) == G_STRUCT_OFFSET (GOutputVector, size))
static gboolean
g_unix_output_stream_writev (GOutputStream *stream,
const GOutputVector *vectors,
gsize n_vectors,
gsize *bytes_written,
GCancellable *cancellable,
GError **error)
{
GUnixOutputStream *unix_stream;
gssize res = -1;
GPollFD poll_fds[2];
int nfds = 0;
int poll_ret;
struct iovec *iov;
if (bytes_written)
*bytes_written = 0;
/* Clamp to G_MAXINT as writev() takes an integer for the number of vectors.
* We handle this like a short write in this case
*/
if (n_vectors > G_MAXINT)
n_vectors = G_MAXINT;
unix_stream = G_UNIX_OUTPUT_STREAM (stream);
if (G_OUTPUT_VECTOR_IS_IOVEC)
{
/* ABI is compatible */
iov = (struct iovec *) vectors;
}
else
{
gsize i;
/* ABI is incompatible */
iov = g_newa (struct iovec, n_vectors);
for (i = 0; i < n_vectors; i++)
{
iov[i].iov_base = (void *)vectors[i].buffer;
iov[i].iov_len = vectors[i].size;
}
}
poll_fds[0].fd = unix_stream->priv->fd;
poll_fds[0].events = G_IO_OUT;
nfds++;
if (unix_stream->priv->is_pipe_or_socket &&
g_cancellable_make_pollfd (cancellable, &poll_fds[1]))
nfds++;
while (1)
{
int errsv;
poll_fds[0].revents = poll_fds[1].revents = 0;
do
{
poll_ret = g_poll (poll_fds, nfds, -1);
errsv = errno;
}
while (poll_ret == -1 && errsv == EINTR);
if (poll_ret == -1)
{
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
_("Error writing to file descriptor: %s"),
g_strerror (errsv));
break;
}
if (g_cancellable_set_error_if_cancelled (cancellable, error))
break;
if (!poll_fds[0].revents)
continue;
res = writev (unix_stream->priv->fd, iov, n_vectors);
errsv = errno;
if (res == -1)
{
if (errsv == EINTR || errsv == EAGAIN)
continue;
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
_("Error writing to file descriptor: %s"),
g_strerror (errsv));
}
if (bytes_written)
*bytes_written = res;
break;
}
if (nfds == 2)
g_cancellable_release_fd (cancellable);
return res != -1;
}
static gboolean
g_unix_output_stream_close (GOutputStream *stream,
GCancellable *cancellable,
@ -494,3 +618,70 @@ g_unix_output_stream_pollable_create_source (GPollableOutputStream *stream,
return pollable_source;
}
static GPollableReturn
g_unix_output_stream_pollable_writev_nonblocking (GPollableOutputStream *stream,
const GOutputVector *vectors,
gsize n_vectors,
gsize *bytes_written,
GError **error)
{
GUnixOutputStream *unix_stream = G_UNIX_OUTPUT_STREAM (stream);
struct iovec *iov;
gssize res = -1;
if (!g_pollable_output_stream_is_writable (stream))
{
*bytes_written = 0;
return G_POLLABLE_RETURN_WOULD_BLOCK;
}
/* Clamp to G_MAXINT as writev() takes an integer for the number of vectors.
* We handle this like a short write in this case
*/
if (n_vectors > G_MAXINT)
n_vectors = G_MAXINT;
if (G_OUTPUT_VECTOR_IS_IOVEC)
{
/* ABI is compatible */
iov = (struct iovec *) vectors;
}
else
{
gsize i;
/* ABI is incompatible */
iov = g_newa (struct iovec, n_vectors);
for (i = 0; i < n_vectors; i++)
{
iov[i].iov_base = (void *)vectors[i].buffer;
iov[i].iov_len = vectors[i].size;
}
}
while (1)
{
int errsv;
res = writev (unix_stream->priv->fd, iov, n_vectors);
errsv = errno;
if (res == -1)
{
if (errsv == EINTR)
continue;
g_set_error (error, G_IO_ERROR,
g_io_error_from_errno (errsv),
_("Error writing to file descriptor: %s"),
g_strerror (errsv));
}
if (bytes_written)
*bytes_written = res;
break;
}
return res != -1 ? G_POLLABLE_RETURN_OK : G_POLLABLE_RETURN_FAILED;
}