[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [PULL v2 4/9] io: add QIOChannelSocket class
|
From: |
Daniel P. Berrange |
|
Subject: |
[Qemu-devel] [PULL v2 4/9] io: add QIOChannelSocket class |
|
Date: |
Thu, 17 Dec 2015 15:40:04 +0000 |
Implement a QIOChannel subclass that supports sockets I/O.
The implementation is able to manage a single socket file
descriptor, whether a TCP/UNIX listener, TCP/UNIX connection,
or a UDP datagram. It provides APIs which can listen and
connect either asynchronously or synchronously. Since there
is no asynchronous DNS lookup API available, it uses the
QIOTask helper for spawning a background thread to ensure
non-blocking operation.
Signed-off-by: Daniel P. Berrange <address@hidden>
---
configure | 11 +
include/io/channel-socket.h | 244 ++++++++++++++
include/qemu/sockets.h | 19 ++
io/Makefile.objs | 1 +
io/channel-socket.c | 741 +++++++++++++++++++++++++++++++++++++++++
scripts/create_config | 9 +
tests/.gitignore | 1 +
tests/Makefile | 3 +
tests/io-channel-helpers.c | 246 ++++++++++++++
tests/io-channel-helpers.h | 42 +++
tests/test-io-channel-socket.c | 399 ++++++++++++++++++++++
trace-events | 19 ++
util/qemu-sockets.c | 2 +-
13 files changed, 1736 insertions(+), 1 deletion(-)
create mode 100644 include/io/channel-socket.h
create mode 100644 io/channel-socket.c
create mode 100644 tests/io-channel-helpers.c
create mode 100644 tests/io-channel-helpers.h
create mode 100644 tests/test-io-channel-socket.c
diff --git a/configure b/configure
index b9552fd..375f103 100755
--- a/configure
+++ b/configure
@@ -2427,6 +2427,14 @@ fi
##########################################
+# getifaddrs (for tests/test-io-channel-socket )
+
+have_ifaddrs_h=yes
+if ! check_include "ifaddrs.h" ; then
+ have_ifaddrs_h=no
+fi
+
+##########################################
# VTE probe
if test "$vte" != "no"; then
@@ -5137,6 +5145,9 @@ fi
if test "$tasn1" = "yes" ; then
echo "CONFIG_TASN1=y" >> $config_host_mak
fi
+if test "$have_ifaddrs_h" = "yes" ; then
+ echo "HAVE_IFADDRS_H=y" >> $config_host_mak
+fi
if test "$vte" = "yes" ; then
echo "CONFIG_VTE=y" >> $config_host_mak
echo "VTE_CFLAGS=$vte_cflags" >> $config_host_mak
diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
new file mode 100644
index 0000000..0719757
--- /dev/null
+++ b/include/io/channel-socket.h
@@ -0,0 +1,244 @@
+/*
+ * QEMU I/O channels sockets driver
+ *
+ * Copyright (c) 2015 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#ifndef QIO_CHANNEL_SOCKET_H__
+#define QIO_CHANNEL_SOCKET_H__
+
+#include "io/channel.h"
+#include "io/task.h"
+#include "qemu/sockets.h"
+
+#define TYPE_QIO_CHANNEL_SOCKET "qio-channel-socket"
+#define QIO_CHANNEL_SOCKET(obj) \
+ OBJECT_CHECK(QIOChannelSocket, (obj), TYPE_QIO_CHANNEL_SOCKET)
+
+typedef struct QIOChannelSocket QIOChannelSocket;
+
+/**
+ * QIOChannelSocket:
+ *
+ * The QIOChannelSocket class provides a channel implementation
+ * that can transport data over a UNIX socket or TCP socket.
+ * Beyond the core channel API, it also provides functionality
+ * for accepting client connections, tuning some socket
+ * parameters and getting socket address strings.
+ */
+
+struct QIOChannelSocket {
+ QIOChannel parent;
+ int fd;
+ struct sockaddr_storage localAddr;
+ socklen_t localAddrLen;
+ struct sockaddr_storage remoteAddr;
+ socklen_t remoteAddrLen;
+};
+
+
+/**
+ * qio_channel_socket_new:
+ *
+ * Create a channel for performing I/O on a socket
+ * connection, that is initially closed. After
+ * creating the socket, it must be setup as a client
+ * connection or server.
+ *
+ * Returns: the socket channel object
+ */
+QIOChannelSocket *
+qio_channel_socket_new(void);
+
+/**
+ * qio_channel_socket_new_fd:
+ * @fd: the socket file descriptor
+ * @errp: pointer to an uninitialized error object
+ *
+ * Create a channel for performing I/O on the socket
+ * connection represented by the file descriptor @fd.
+ *
+ * Returns: the socket channel object, or NULL on error
+ */
+QIOChannelSocket *
+qio_channel_socket_new_fd(int fd,
+ Error **errp);
+
+
+/**
+ * qio_channel_socket_connect_sync:
+ * @ioc: the socket channel object
+ * @addr: the address to connect to
+ * @errp: pointer to an uninitialized error object
+ *
+ * Attempt to connect to the address @addr. This method
+ * will run in the foreground so the caller will not regain
+ * execution control until the connection is established or
+ * an error occurs.
+ */
+int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ Error **errp);
+
+/**
+ * qio_channel_socket_connect_async:
+ * @ioc: the socket channel object
+ * @addr: the address to connect to
+ * @callback: the function to invoke on completion
+ * @opaque: user data to pass to @callback
+ * @destroy: the function to free @opaque
+ *
+ * Attempt to connect to the address @addr. This method
+ * will run in the background so the caller will regain
+ * execution control immediately. The function @callback
+ * will be invoked on completion or failure.
+ */
+void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy);
+
+
+/**
+ * qio_channel_socket_listen_sync:
+ * @ioc: the socket channel object
+ * @addr: the address to listen to
+ * @errp: pointer to an uninitialized error object
+ *
+ * Attempt to listen to the address @addr. This method
+ * will run in the foreground so the caller will not regain
+ * execution control until the connection is established or
+ * an error occurs.
+ */
+int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ Error **errp);
+
+/**
+ * qio_channel_socket_listen_async:
+ * @ioc: the socket channel object
+ * @addr: the address to listen to
+ * @callback: the function to invoke on completion
+ * @opaque: user data to pass to @callback
+ * @destroy: the function to free @opaque
+ *
+ * Attempt to listen to the address @addr. This method
+ * will run in the background so the caller will regain
+ * execution control immediately. The function @callback
+ * will be invoked on completion or failure.
+ */
+void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy);
+
+
+/**
+ * qio_channel_socket_dgram_sync:
+ * @ioc: the socket channel object
+ * @localAddr: the address to local bind address
+ * @remoteAddr: the address to remote peer address
+ * @errp: pointer to an uninitialized error object
+ *
+ * Attempt to initialize a datagram socket bound to
+ * @localAddr and communicating with peer @remoteAddr.
+ * This method will run in the foreground so the caller
+ * will not regain execution control until the socket
+ * is established or an error occurs.
+ */
+int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
+ SocketAddress *localAddr,
+ SocketAddress *remoteAddr,
+ Error **errp);
+
+/**
+ * qio_channel_socket_dgram_async:
+ * @ioc: the socket channel object
+ * @localAddr: the address to local bind address
+ * @remoteAddr: the address to remote peer address
+ * @callback: the function to invoke on completion
+ * @opaque: user data to pass to @callback
+ * @destroy: the function to free @opaque
+ *
+ * Attempt to initialize a datagram socket bound to
+ * @localAddr and communicating with peer @remoteAddr.
+ * This method will run in the background so the caller
+ * will regain execution control immediately. The function
+ * @callback will be invoked on completion or failure.
+ */
+void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
+ SocketAddress *localAddr,
+ SocketAddress *remoteAddr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy);
+
+
+/**
+ * qio_channel_socket_get_local_address:
+ * @ioc: the socket channel object
+ * @errp: pointer to an uninitialized error object
+ *
+ * Get the string representation of the local socket
+ * address. A pointer to the allocated address information
+ * struct will be returned, which the caller is required to
+ * release with a call qapi_free_SocketAddress when no
+ * longer required.
+ *
+ * Returns: 0 on success, -1 on error
+ */
+SocketAddress *
+qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
+ Error **errp);
+
+/**
+ * qio_channel_socket_get_remote_address:
+ * @ioc: the socket channel object
+ * @errp: pointer to an uninitialized error object
+ *
+ * Get the string representation of the local socket
+ * address. A pointer to the allocated address information
+ * struct will be returned, which the caller is required to
+ * release with a call qapi_free_SocketAddress when no
+ * longer required.
+ *
+ * Returns: the socket address struct, or NULL on error
+ */
+SocketAddress *
+qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
+ Error **errp);
+
+
+/**
+ * qio_channel_socket_accept:
+ * @ioc: the socket channel object
+ * @errp: pointer to an uninitialized error object
+ *
+ * If the socket represents a server, then this accepts
+ * a new client connection. The returned channel will
+ * represent the connected client socket.
+ *
+ * Returns: the new client channel, or NULL on error
+ */
+QIOChannelSocket *
+qio_channel_socket_accept(QIOChannelSocket *ioc,
+ Error **errp);
+
+
+#endif /* QIO_CHANNEL_SOCKET_H__ */
diff --git a/include/qemu/sockets.h b/include/qemu/sockets.h
index 5a183c5..74c692d 100644
--- a/include/qemu/sockets.h
+++ b/include/qemu/sockets.h
@@ -89,6 +89,25 @@ int parse_host_port(struct sockaddr_in *saddr, const char
*str);
int socket_init(void);
/**
+ * socket_sockaddr_to_address:
+ * @sa: socket address struct
+ * @salen: size of @sa struct
+ * @errp: pointer to uninitialized error object
+ *
+ * Get the string representation of the socket
+ * address. A pointer to the allocated address information
+ * struct will be returned, which the caller is required to
+ * release with a call qapi_free_SocketAddress when no
+ * longer required.
+ *
+ * Returns: the socket address struct, or NULL on error
+ */
+SocketAddress *
+socket_sockaddr_to_address(struct sockaddr_storage *sa,
+ socklen_t salen,
+ Error **errp);
+
+/**
* socket_local_address:
* @fd: the socket file handle
* @errp: pointer to uninitialized error object
diff --git a/io/Makefile.objs b/io/Makefile.objs
index 503b95c..e9d77aa 100644
--- a/io/Makefile.objs
+++ b/io/Makefile.objs
@@ -1,3 +1,4 @@
io-obj-y = channel.o
+io-obj-y += channel-socket.o
io-obj-y += channel-watch.o
io-obj-y += task.o
diff --git a/io/channel-socket.c b/io/channel-socket.c
new file mode 100644
index 0000000..90b3c73
--- /dev/null
+++ b/io/channel-socket.c
@@ -0,0 +1,741 @@
+/*
+ * QEMU I/O channels sockets driver
+ *
+ * Copyright (c) 2015 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "io/channel-socket.h"
+#include "io/channel-watch.h"
+#include "trace.h"
+
+#define SOCKET_MAX_FDS 16
+
+SocketAddress *
+qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
+ Error **errp)
+{
+ return socket_sockaddr_to_address(&ioc->localAddr,
+ ioc->localAddrLen,
+ errp);
+}
+
+SocketAddress *
+qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
+ Error **errp)
+{
+ return socket_sockaddr_to_address(&ioc->remoteAddr,
+ ioc->remoteAddrLen,
+ errp);
+}
+
+QIOChannelSocket *
+qio_channel_socket_new(void)
+{
+ QIOChannelSocket *sioc;
+ QIOChannel *ioc;
+
+ sioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
+ sioc->fd = -1;
+
+ ioc = QIO_CHANNEL(sioc);
+ ioc->features |= (1 << QIO_CHANNEL_FEATURE_SHUTDOWN);
+
+ trace_qio_channel_socket_new(sioc);
+
+ return sioc;
+}
+
+
+static int
+qio_channel_socket_set_fd(QIOChannelSocket *sioc,
+ int fd,
+ Error **errp)
+{
+ if (sioc->fd != -1) {
+ error_setg(errp, "Socket is already open");
+ return -1;
+ }
+
+ sioc->fd = fd;
+ sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
+ sioc->localAddrLen = sizeof(sioc->localAddr);
+
+
+ if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
+ &sioc->remoteAddrLen) < 0) {
+ if (socket_error() == ENOTCONN) {
+ memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
+ sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
+ } else {
+ error_setg_errno(errp, socket_error(),
+ "Unable to query remote socket address");
+ goto error;
+ }
+ }
+
+ if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
+ &sioc->localAddrLen) < 0) {
+ error_setg_errno(errp, socket_error(),
+ "Unable to query local socket address");
+ goto error;
+ }
+
+#ifndef WIN32
+ if (sioc->localAddr.ss_family == AF_UNIX) {
+ QIOChannel *ioc = QIO_CHANNEL(sioc);
+ ioc->features |= (1 << QIO_CHANNEL_FEATURE_FD_PASS);
+ }
+#endif /* WIN32 */
+
+ return 0;
+
+ error:
+ sioc->fd = -1; /* Let the caller close FD on failure */
+ return -1;
+}
+
+QIOChannelSocket *
+qio_channel_socket_new_fd(int fd,
+ Error **errp)
+{
+ QIOChannelSocket *ioc;
+
+ ioc = qio_channel_socket_new();
+ if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
+ object_unref(OBJECT(ioc));
+ return NULL;
+ }
+
+ trace_qio_channel_socket_new_fd(ioc, fd);
+
+ return ioc;
+}
+
+
+int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ Error **errp)
+{
+ int fd;
+
+ trace_qio_channel_socket_connect_sync(ioc, addr);
+ fd = socket_connect(addr, errp, NULL, NULL);
+ if (fd < 0) {
+ trace_qio_channel_socket_connect_fail(ioc);
+ return -1;
+ }
+
+ trace_qio_channel_socket_connect_complete(ioc, fd);
+ if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
+ close(fd);
+ return -1;
+ }
+
+ return 0;
+}
+
+
+static int qio_channel_socket_connect_worker(QIOTask *task,
+ Error **errp,
+ gpointer opaque)
+{
+ QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
+ SocketAddress *addr = opaque;
+ int ret;
+
+ ret = qio_channel_socket_connect_sync(ioc,
+ addr,
+ errp);
+
+ object_unref(OBJECT(ioc));
+ return ret;
+}
+
+
+void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy)
+{
+ QIOTask *task = qio_task_new(
+ OBJECT(ioc), callback, opaque, destroy);
+ SocketAddress *addrCopy;
+
+ qapi_copy_SocketAddress(&addrCopy, addr);
+
+ /* socket_connect() does a non-blocking connect(), but it
+ * still blocks in DNS lookups, so we must use a thread */
+ trace_qio_channel_socket_connect_async(ioc, addr);
+ qio_task_run_in_thread(task,
+ qio_channel_socket_connect_worker,
+ addrCopy,
+ (GDestroyNotify)qapi_free_SocketAddress);
+}
+
+
+int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ Error **errp)
+{
+ int fd;
+
+ trace_qio_channel_socket_listen_sync(ioc, addr);
+ fd = socket_listen(addr, errp);
+ if (fd < 0) {
+ trace_qio_channel_socket_listen_fail(ioc);
+ return -1;
+ }
+
+ trace_qio_channel_socket_listen_complete(ioc, fd);
+ if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
+ close(fd);
+ return -1;
+ }
+
+ return 0;
+}
+
+
+static int qio_channel_socket_listen_worker(QIOTask *task,
+ Error **errp,
+ gpointer opaque)
+{
+ QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
+ SocketAddress *addr = opaque;
+ int ret;
+
+ ret = qio_channel_socket_listen_sync(ioc,
+ addr,
+ errp);
+
+ object_unref(OBJECT(ioc));
+ return ret;
+}
+
+
+void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
+ SocketAddress *addr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy)
+{
+ QIOTask *task = qio_task_new(
+ OBJECT(ioc), callback, opaque, destroy);
+ SocketAddress *addrCopy;
+
+ qapi_copy_SocketAddress(&addrCopy, addr);
+
+ /* socket_listen() blocks in DNS lookups, so we must use a thread */
+ trace_qio_channel_socket_listen_async(ioc, addr);
+ qio_task_run_in_thread(task,
+ qio_channel_socket_listen_worker,
+ addrCopy,
+ (GDestroyNotify)qapi_free_SocketAddress);
+}
+
+
+int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
+ SocketAddress *localAddr,
+ SocketAddress *remoteAddr,
+ Error **errp)
+{
+ int fd;
+
+ trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
+ fd = socket_dgram(localAddr, remoteAddr, errp);
+ if (fd < 0) {
+ trace_qio_channel_socket_dgram_fail(ioc);
+ return -1;
+ }
+
+ trace_qio_channel_socket_dgram_complete(ioc, fd);
+ if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
+ close(fd);
+ return -1;
+ }
+
+ return 0;
+}
+
+
+struct QIOChannelSocketDGramWorkerData {
+ SocketAddress *localAddr;
+ SocketAddress *remoteAddr;
+};
+
+
+static void qio_channel_socket_dgram_worker_free(gpointer opaque)
+{
+ struct QIOChannelSocketDGramWorkerData *data = opaque;
+ qapi_free_SocketAddress(data->localAddr);
+ qapi_free_SocketAddress(data->remoteAddr);
+ g_free(data);
+}
+
+static int qio_channel_socket_dgram_worker(QIOTask *task,
+ Error **errp,
+ gpointer opaque)
+{
+ QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
+ struct QIOChannelSocketDGramWorkerData *data = opaque;
+ int ret;
+
+ /* socket_dgram() blocks in DNS lookups, so we must use a thread */
+ ret = qio_channel_socket_dgram_sync(ioc,
+ data->localAddr,
+ data->remoteAddr,
+ errp);
+
+ object_unref(OBJECT(ioc));
+ return ret;
+}
+
+
+void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
+ SocketAddress *localAddr,
+ SocketAddress *remoteAddr,
+ QIOTaskFunc callback,
+ gpointer opaque,
+ GDestroyNotify destroy)
+{
+ QIOTask *task = qio_task_new(
+ OBJECT(ioc), callback, opaque, destroy);
+ struct QIOChannelSocketDGramWorkerData *data = g_new0(
+ struct QIOChannelSocketDGramWorkerData, 1);
+
+ qapi_copy_SocketAddress(&data->localAddr, localAddr);
+ qapi_copy_SocketAddress(&data->remoteAddr, remoteAddr);
+
+ trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
+ qio_task_run_in_thread(task,
+ qio_channel_socket_dgram_worker,
+ data,
+ qio_channel_socket_dgram_worker_free);
+}
+
+
+QIOChannelSocket *
+qio_channel_socket_accept(QIOChannelSocket *ioc,
+ Error **errp)
+{
+ QIOChannelSocket *cioc;
+
+ cioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
+ cioc->fd = -1;
+ cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
+ cioc->localAddrLen = sizeof(ioc->localAddr);
+
+ retry:
+ trace_qio_channel_socket_accept(ioc);
+ cioc->fd = accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
+ &cioc->remoteAddrLen);
+ if (cioc->fd < 0) {
+ trace_qio_channel_socket_accept_fail(ioc);
+ if (socket_error() == EINTR) {
+ goto retry;
+ }
+ goto error;
+ }
+
+ if (getsockname(cioc->fd, (struct sockaddr *)&ioc->localAddr,
+ &ioc->localAddrLen) < 0) {
+ error_setg_errno(errp, socket_error(),
+ "Unable to query local socket address");
+ goto error;
+ }
+
+ trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
+ return cioc;
+
+ error:
+ object_unref(OBJECT(cioc));
+ return NULL;
+}
+
+static void qio_channel_socket_init(Object *obj)
+{
+ QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
+ ioc->fd = -1;
+}
+
+static void qio_channel_socket_finalize(Object *obj)
+{
+ QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
+ if (ioc->fd != -1) {
+ close(ioc->fd);
+ ioc->fd = -1;
+ }
+}
+
+
+#ifndef WIN32
+static void qio_channel_socket_copy_fds(struct msghdr *msg,
+ int **fds, size_t *nfds)
+{
+ struct cmsghdr *cmsg;
+
+ *nfds = 0;
+ *fds = NULL;
+
+ for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
+ int fd_size, i;
+ int gotfds;
+
+ if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
+ cmsg->cmsg_level != SOL_SOCKET ||
+ cmsg->cmsg_type != SCM_RIGHTS) {
+ continue;
+ }
+
+ fd_size = cmsg->cmsg_len - CMSG_LEN(0);
+
+ if (!fd_size) {
+ continue;
+ }
+
+ gotfds = fd_size / sizeof(int);
+ *fds = g_renew(int, *fds, *nfds + gotfds);
+ memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
+
+ for (i = 0; i < gotfds; i++) {
+ int fd = (*fds)[*nfds + i];
+ if (fd < 0) {
+ continue;
+ }
+
+ /* O_NONBLOCK is preserved across SCM_RIGHTS so reset it */
+ qemu_set_block(fd);
+
+#ifndef MSG_CMSG_CLOEXEC
+ qemu_set_cloexec(fd);
+#endif
+ }
+ *nfds += gotfds;
+ }
+}
+
+
+static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ int **fds,
+ size_t *nfds,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ ssize_t ret;
+ struct msghdr msg = { NULL, };
+ char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
+ int sflags = 0;
+
+#ifdef MSG_CMSG_CLOEXEC
+ sflags |= MSG_CMSG_CLOEXEC;
+#endif
+
+ msg.msg_iov = (struct iovec *)iov;
+ msg.msg_iovlen = niov;
+ if (fds && nfds) {
+ msg.msg_control = control;
+ msg.msg_controllen = sizeof(control);
+ }
+
+ retry:
+ ret = recvmsg(sioc->fd, &msg, sflags);
+ if (ret < 0) {
+ if (socket_error() == EAGAIN ||
+ socket_error() == EWOULDBLOCK) {
+ return QIO_CHANNEL_ERR_BLOCK;
+ }
+ if (socket_error() == EINTR) {
+ goto retry;
+ }
+
+ error_setg_errno(errp, socket_error(),
+ "Unable to read from socket");
+ return -1;
+ }
+
+ if (fds && nfds) {
+ qio_channel_socket_copy_fds(&msg, fds, nfds);
+ }
+
+ return ret;
+}
+
+static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ int *fds,
+ size_t nfds,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ ssize_t ret;
+ struct msghdr msg = { NULL, };
+
+ msg.msg_iov = (struct iovec *)iov;
+ msg.msg_iovlen = niov;
+
+ if (nfds) {
+ char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
+ size_t fdsize = sizeof(int) * nfds;
+ struct cmsghdr *cmsg;
+
+ if (nfds > SOCKET_MAX_FDS) {
+ error_setg_errno(errp, -EINVAL,
+ "Only %d FDs can be sent, got %zu",
+ SOCKET_MAX_FDS, nfds);
+ return -1;
+ }
+
+ msg.msg_control = control;
+ msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
+
+ cmsg = CMSG_FIRSTHDR(&msg);
+ cmsg->cmsg_len = CMSG_LEN(fdsize);
+ cmsg->cmsg_level = SOL_SOCKET;
+ cmsg->cmsg_type = SCM_RIGHTS;
+ memcpy(CMSG_DATA(cmsg), fds, fdsize);
+ }
+
+ retry:
+ ret = sendmsg(sioc->fd, &msg, 0);
+ if (ret <= 0) {
+ if (socket_error() == EAGAIN ||
+ socket_error() == EWOULDBLOCK) {
+ return QIO_CHANNEL_ERR_BLOCK;
+ }
+ if (socket_error() == EINTR) {
+ goto retry;
+ }
+ error_setg_errno(errp, socket_error(),
+ "Unable to write to socket");
+ return -1;
+ }
+ return ret;
+}
+#else /* WIN32 */
+static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ int **fds,
+ size_t *nfds,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ ssize_t done = 0;
+ ssize_t i;
+
+ for (i = 0; i < niov; i++) {
+ ssize_t ret;
+ retry:
+ ret = recv(sioc->fd,
+ iov[i].iov_base,
+ iov[i].iov_len,
+ 0);
+ if (ret < 0) {
+ if (socket_error() == EAGAIN) {
+ if (done) {
+ return done;
+ } else {
+ return QIO_CHANNEL_ERR_BLOCK;
+ }
+ } else if (socket_error() == EINTR) {
+ goto retry;
+ } else {
+ error_setg_errno(errp, socket_error(),
+ "Unable to write to socket");
+ return -1;
+ }
+ }
+ done += ret;
+ if (ret < iov[i].iov_len) {
+ return done;
+ }
+ }
+
+ return done;
+}
+
+static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ int *fds,
+ size_t nfds,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ ssize_t done = 0;
+ ssize_t i;
+
+ for (i = 0; i < niov; i++) {
+ ssize_t ret;
+ retry:
+ ret = send(sioc->fd,
+ iov[i].iov_base,
+ iov[i].iov_len,
+ 0);
+ if (ret < 0) {
+ if (socket_error() == EAGAIN) {
+ if (done) {
+ return done;
+ } else {
+ return QIO_CHANNEL_ERR_BLOCK;
+ }
+ } else if (socket_error() == EINTR) {
+ goto retry;
+ } else {
+ error_setg_errno(errp, socket_error(),
+ "Unable to write to socket");
+ return -1;
+ }
+ }
+ done += ret;
+ if (ret < iov[i].iov_len) {
+ return done;
+ }
+ }
+
+ return done;
+}
+#endif /* WIN32 */
+
+static int
+qio_channel_socket_set_blocking(QIOChannel *ioc,
+ bool enabled,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+
+ if (enabled) {
+ qemu_set_block(sioc->fd);
+ } else {
+ qemu_set_nonblock(sioc->fd);
+ }
+ return 0;
+}
+
+
+static void
+qio_channel_socket_set_delay(QIOChannel *ioc,
+ bool enabled)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ int v = enabled ? 0 : 1;
+
+ qemu_setsockopt(sioc->fd,
+ IPPROTO_TCP, TCP_NODELAY,
+ &v, sizeof(v));
+}
+
+
+static void
+qio_channel_socket_set_cork(QIOChannel *ioc,
+ bool enabled)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ int v = enabled ? 1 : 0;
+
+ socket_set_cork(sioc->fd, v);
+}
+
+
+static int
+qio_channel_socket_close(QIOChannel *ioc,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+
+ if (closesocket(sioc->fd) < 0) {
+ sioc->fd = -1;
+ error_setg_errno(errp, socket_error(),
+ "Unable to close socket");
+ return -1;
+ }
+ sioc->fd = -1;
+ return 0;
+}
+
+static int
+qio_channel_socket_shutdown(QIOChannel *ioc,
+ QIOChannelShutdown how,
+ Error **errp)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ int sockhow;
+
+ switch (how) {
+ case QIO_CHANNEL_SHUTDOWN_READ:
+ sockhow = SHUT_RD;
+ break;
+ case QIO_CHANNEL_SHUTDOWN_WRITE:
+ sockhow = SHUT_WR;
+ break;
+ case QIO_CHANNEL_SHUTDOWN_BOTH:
+ default:
+ sockhow = SHUT_RDWR;
+ break;
+ }
+
+ if (shutdown(sioc->fd, sockhow) < 0) {
+ error_setg_errno(errp, socket_error(),
+ "Unable to shutdown socket");
+ return -1;
+ }
+ return 0;
+}
+
+static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
+ GIOCondition condition)
+{
+ QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+ return qio_channel_create_fd_watch(ioc,
+ sioc->fd,
+ condition);
+}
+
+static void qio_channel_socket_class_init(ObjectClass *klass,
+ void *class_data G_GNUC_UNUSED)
+{
+ QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
+
+ ioc_klass->io_writev = qio_channel_socket_writev;
+ ioc_klass->io_readv = qio_channel_socket_readv;
+ ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
+ ioc_klass->io_close = qio_channel_socket_close;
+ ioc_klass->io_shutdown = qio_channel_socket_shutdown;
+ ioc_klass->io_set_cork = qio_channel_socket_set_cork;
+ ioc_klass->io_set_delay = qio_channel_socket_set_delay;
+ ioc_klass->io_create_watch = qio_channel_socket_create_watch;
+}
+
+static const TypeInfo qio_channel_socket_info = {
+ .parent = TYPE_QIO_CHANNEL,
+ .name = TYPE_QIO_CHANNEL_SOCKET,
+ .instance_size = sizeof(QIOChannelSocket),
+ .instance_init = qio_channel_socket_init,
+ .instance_finalize = qio_channel_socket_finalize,
+ .class_init = qio_channel_socket_class_init,
+};
+
+static void qio_channel_socket_register_types(void)
+{
+ type_register_static(&qio_channel_socket_info);
+}
+
+type_init(qio_channel_socket_register_types);
diff --git a/scripts/create_config b/scripts/create_config
index 546f889..9cb176f 100755
--- a/scripts/create_config
+++ b/scripts/create_config
@@ -61,6 +61,15 @@ case $line in
value=${line#*=}
echo "#define $name $value"
;;
+ HAVE_*=y) # configuration
+ name=${line%=*}
+ echo "#define $name 1"
+ ;;
+ HAVE_*=*) # configuration
+ name=${line%=*}
+ value=${line#*=}
+ echo "#define $name $value"
+ ;;
ARCH=*) # configuration
arch=${line#*=}
arch_name=`echo $arch | LC_ALL=C tr '[a-z]' '[A-Z]'`
diff --git a/tests/.gitignore b/tests/.gitignore
index eec12cc..6164cfa 100644
--- a/tests/.gitignore
+++ b/tests/.gitignore
@@ -24,6 +24,7 @@ test-cutils
test-hbitmap
test-int128
test-iov
+test-io-channel-socket
test-io-task
test-mul64
test-opts-visitor
diff --git a/tests/Makefile b/tests/Makefile
index 515a7c7..b2e987c 100644
--- a/tests/Makefile
+++ b/tests/Makefile
@@ -85,6 +85,7 @@ check-unit-$(CONFIG_GNUTLS) +=
tests/test-crypto-tlssession$(EXESUF)
check-unit-$(CONFIG_LINUX) += tests/test-qga$(EXESUF)
check-unit-y += tests/test-timed-average$(EXESUF)
check-unit-y += tests/test-io-task$(EXESUF)
+check-unit-y += tests/test-io-channel-socket$(EXESUF)
check-block-$(CONFIG_POSIX) += tests/qemu-iotests-quick.sh
@@ -472,6 +473,8 @@ tests/test-crypto-tlssession.o-cflags := $(TASN1_CFLAGS)
tests/test-crypto-tlssession$(EXESUF): tests/test-crypto-tlssession.o \
tests/crypto-tls-x509-helpers.o tests/pkix_asn1_tab.o
$(test-crypto-obj-y)
tests/test-io-task$(EXESUF): tests/test-io-task.o $(test-io-obj-y)
+tests/test-io-channel-socket$(EXESUF): tests/test-io-channel-socket.o \
+ tests/io-channel-helpers.o $(test-io-obj-y)
libqos-obj-y = tests/libqos/pci.o tests/libqos/fw_cfg.o tests/libqos/malloc.o
libqos-obj-y += tests/libqos/i2c.o tests/libqos/libqos.o
diff --git a/tests/io-channel-helpers.c b/tests/io-channel-helpers.c
new file mode 100644
index 0000000..78d36dd
--- /dev/null
+++ b/tests/io-channel-helpers.c
@@ -0,0 +1,246 @@
+/*
+ * QEMU I/O channel test helpers
+ *
+ * Copyright (c) 2015 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "io-channel-helpers.h"
+
+struct QIOChannelTest {
+ QIOChannel *src;
+ QIOChannel *dst;
+ bool blocking;
+ size_t len;
+ size_t niov;
+ char *input;
+ struct iovec *inputv;
+ char *output;
+ struct iovec *outputv;
+ Error *writeerr;
+ Error *readerr;
+};
+
+
+static void test_skip_iovec(struct iovec **iov,
+ size_t *niov,
+ size_t skip,
+ struct iovec *old)
+{
+ size_t offset = 0;
+ size_t i;
+
+ for (i = 0; i < *niov; i++) {
+ if (skip < (*iov)[i].iov_len) {
+ old->iov_len = (*iov)[i].iov_len;
+ old->iov_base = (*iov)[i].iov_base;
+
+ (*iov)[i].iov_len -= skip;
+ (*iov)[i].iov_base += skip;
+ break;
+ } else {
+ skip -= (*iov)[i].iov_len;
+
+ if (i == 0 && old->iov_base) {
+ (*iov)[i].iov_len = old->iov_len;
+ (*iov)[i].iov_base = old->iov_base;
+ old->iov_len = 0;
+ old->iov_base = NULL;
+ }
+
+ offset++;
+ }
+ }
+
+ *iov = *iov + offset;
+ *niov -= offset;
+}
+
+
+/* This thread sends all data using iovecs */
+static gpointer test_io_thread_writer(gpointer opaque)
+{
+ QIOChannelTest *data = opaque;
+ struct iovec *iov = data->inputv;
+ size_t niov = data->niov;
+ struct iovec old = { 0 };
+
+ qio_channel_set_blocking(data->src, data->blocking, NULL);
+
+ while (niov) {
+ ssize_t ret;
+ ret = qio_channel_writev(data->src,
+ iov,
+ niov,
+ &data->writeerr);
+ if (ret == QIO_CHANNEL_ERR_BLOCK) {
+ if (data->blocking) {
+ error_setg(&data->writeerr,
+ "Unexpected I/O blocking");
+ break;
+ } else {
+ qio_channel_wait(data->src,
+ G_IO_OUT);
+ continue;
+ }
+ } else if (ret < 0) {
+ break;
+ } else if (ret == 0) {
+ error_setg(&data->writeerr,
+ "Unexpected zero length write");
+ break;
+ }
+
+ test_skip_iovec(&iov, &niov, ret, &old);
+ }
+
+ return NULL;
+}
+
+
+/* This thread receives all data using iovecs */
+static gpointer test_io_thread_reader(gpointer opaque)
+{
+ QIOChannelTest *data = opaque;
+ struct iovec *iov = data->outputv;
+ size_t niov = data->niov;
+ struct iovec old = { 0 };
+
+ qio_channel_set_blocking(data->dst, data->blocking, NULL);
+
+ while (niov) {
+ ssize_t ret;
+
+ ret = qio_channel_readv(data->dst,
+ iov,
+ niov,
+ &data->readerr);
+
+ if (ret == QIO_CHANNEL_ERR_BLOCK) {
+ if (data->blocking) {
+ error_setg(&data->writeerr,
+ "Unexpected I/O blocking");
+ break;
+ } else {
+ qio_channel_wait(data->dst,
+ G_IO_IN);
+ continue;
+ }
+ } else if (ret < 0) {
+ break;
+ } else if (ret == 0) {
+ break;
+ }
+
+ test_skip_iovec(&iov, &niov, ret, &old);
+ }
+
+ return NULL;
+}
+
+
+QIOChannelTest *qio_channel_test_new(void)
+{
+ QIOChannelTest *data = g_new0(QIOChannelTest, 1);
+ size_t i;
+ size_t offset;
+
+
+ /* We'll send 1 MB of data */
+#define CHUNK_COUNT 250
+#define CHUNK_LEN 4194
+
+ data->len = CHUNK_COUNT * CHUNK_LEN;
+ data->input = g_new0(char, data->len);
+ data->output = g_new0(gchar, data->len);
+
+ /* Fill input with a pattern */
+ for (i = 0; i < data->len; i += CHUNK_LEN) {
+ memset(data->input + i, (i / CHUNK_LEN), CHUNK_LEN);
+ }
+
+ /* We'll split the data across a bunch of IO vecs */
+ data->niov = CHUNK_COUNT;
+ data->inputv = g_new0(struct iovec, data->niov);
+ data->outputv = g_new0(struct iovec, data->niov);
+
+ for (i = 0, offset = 0; i < data->niov; i++, offset += CHUNK_LEN) {
+ data->inputv[i].iov_base = data->input + offset;
+ data->outputv[i].iov_base = data->output + offset;
+ data->inputv[i].iov_len = CHUNK_LEN;
+ data->outputv[i].iov_len = CHUNK_LEN;
+ }
+
+ return data;
+}
+
+void qio_channel_test_run_threads(QIOChannelTest *test,
+ bool blocking,
+ QIOChannel *src,
+ QIOChannel *dst)
+{
+ GThread *reader, *writer;
+
+ test->src = src;
+ test->dst = dst;
+ test->blocking = blocking;
+
+ reader = g_thread_new("reader",
+ test_io_thread_reader,
+ test);
+ writer = g_thread_new("writer",
+ test_io_thread_writer,
+ test);
+
+ g_thread_join(reader);
+ g_thread_join(writer);
+
+ test->dst = test->src = NULL;
+}
+
+
+void qio_channel_test_run_writer(QIOChannelTest *test,
+ QIOChannel *src)
+{
+ test->src = src;
+ test_io_thread_writer(test);
+ test->src = NULL;
+}
+
+
+void qio_channel_test_run_reader(QIOChannelTest *test,
+ QIOChannel *dst)
+{
+ test->dst = dst;
+ test_io_thread_reader(test);
+ test->dst = NULL;
+}
+
+
+void qio_channel_test_validate(QIOChannelTest *test)
+{
+ g_assert_cmpint(memcmp(test->input,
+ test->output,
+ test->len), ==, 0);
+ g_assert(test->readerr == NULL);
+ g_assert(test->writeerr == NULL);
+
+ g_free(test->inputv);
+ g_free(test->outputv);
+ g_free(test->input);
+ g_free(test->output);
+ g_free(test);
+}
diff --git a/tests/io-channel-helpers.h b/tests/io-channel-helpers.h
new file mode 100644
index 0000000..fedc64f
--- /dev/null
+++ b/tests/io-channel-helpers.h
@@ -0,0 +1,42 @@
+/*
+ * QEMU I/O channel test helpers
+ *
+ * Copyright (c) 2015 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "io/channel.h"
+
+#ifndef TEST_IO_CHANNEL_HELPERS
+#define TEST_IO_CHANNEL_HELPERS
+
+typedef struct QIOChannelTest QIOChannelTest;
+
+QIOChannelTest *qio_channel_test_new(void);
+
+void qio_channel_test_run_threads(QIOChannelTest *test,
+ bool blocking,
+ QIOChannel *src,
+ QIOChannel *dst);
+
+void qio_channel_test_run_writer(QIOChannelTest *test,
+ QIOChannel *src);
+void qio_channel_test_run_reader(QIOChannelTest *test,
+ QIOChannel *dst);
+
+void qio_channel_test_validate(QIOChannelTest *test);
+
+#endif /* TEST_IO_CHANNEL_HELPERS */
diff --git a/tests/test-io-channel-socket.c b/tests/test-io-channel-socket.c
new file mode 100644
index 0000000..194d043
--- /dev/null
+++ b/tests/test-io-channel-socket.c
@@ -0,0 +1,399 @@
+/*
+ * QEMU I/O channel sockets test
+ *
+ * Copyright (c) 2015 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "io/channel-socket.h"
+#include "io-channel-helpers.h"
+#ifdef HAVE_IFADDRS_H
+#include <ifaddrs.h>
+#endif
+
+static int check_protocol_support(bool *has_ipv4, bool *has_ipv6)
+{
+#ifdef HAVE_IFADDRS_H
+ struct ifaddrs *ifaddr = NULL, *ifa;
+ struct addrinfo hints = { 0 };
+ struct addrinfo *ai = NULL;
+ int gaierr;
+
+ *has_ipv4 = *has_ipv6 = false;
+
+ if (getifaddrs(&ifaddr) < 0) {
+ g_printerr("Failed to lookup interface addresses: %s\n",
+ strerror(errno));
+ return -1;
+ }
+
+ for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next) {
+ if (!ifa->ifa_addr) {
+ continue;
+ }
+
+ if (ifa->ifa_addr->sa_family == AF_INET) {
+ *has_ipv4 = true;
+ }
+ if (ifa->ifa_addr->sa_family == AF_INET6) {
+ *has_ipv6 = true;
+ }
+ }
+
+ freeifaddrs(ifaddr);
+
+ hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
+ hints.ai_family = AF_INET6;
+ hints.ai_socktype = SOCK_STREAM;
+
+ gaierr = getaddrinfo("::1", NULL, &hints, &ai);
+ if (gaierr != 0) {
+ if (gaierr == EAI_ADDRFAMILY ||
+ gaierr == EAI_FAMILY ||
+ gaierr == EAI_NONAME) {
+ *has_ipv6 = false;
+ } else {
+ g_printerr("Failed to resolve ::1 address: %s\n",
+ gai_strerror(gaierr));
+ return -1;
+ }
+ }
+
+ freeaddrinfo(ai);
+
+ return 0;
+#else
+ *has_ipv4 = *has_ipv6 = false;
+
+ return -1;
+#endif
+}
+
+
+static void test_io_channel_set_socket_bufs(QIOChannel *src,
+ QIOChannel *dst)
+{
+ int buflen = 64 * 1024;
+
+ /*
+ * Make the socket buffers small so that we see
+ * the effects of partial reads/writes
+ */
+ setsockopt(((QIOChannelSocket *)src)->fd,
+ SOL_SOCKET, SO_SNDBUF,
+ (char *)&buflen,
+ sizeof(buflen));
+
+ setsockopt(((QIOChannelSocket *)dst)->fd,
+ SOL_SOCKET, SO_SNDBUF,
+ (char *)&buflen,
+ sizeof(buflen));
+}
+
+
+static void test_io_channel_setup_sync(SocketAddress *listen_addr,
+ SocketAddress *connect_addr,
+ QIOChannel **src,
+ QIOChannel **dst)
+{
+ QIOChannelSocket *lioc;
+
+ lioc = qio_channel_socket_new();
+ qio_channel_socket_listen_sync(lioc, listen_addr, &error_abort);
+
+ if (listen_addr->type == SOCKET_ADDRESS_KIND_INET) {
+ SocketAddress *laddr = qio_channel_socket_get_local_address(
+ lioc, &error_abort);
+
+ g_free(connect_addr->u.inet->port);
+ connect_addr->u.inet->port = g_strdup(laddr->u.inet->port);
+
+ qapi_free_SocketAddress(laddr);
+ }
+
+ *src = QIO_CHANNEL(qio_channel_socket_new());
+ qio_channel_socket_connect_sync(
+ QIO_CHANNEL_SOCKET(*src), connect_addr, &error_abort);
+ qio_channel_set_delay(*src, false);
+
+ *dst = QIO_CHANNEL(qio_channel_socket_accept(lioc, &error_abort));
+ g_assert(*dst);
+
+ test_io_channel_set_socket_bufs(*src, *dst);
+
+ object_unref(OBJECT(lioc));
+}
+
+
+struct TestIOChannelData {
+ bool err;
+ GMainLoop *loop;
+};
+
+
+static void test_io_channel_complete(Object *src,
+ Error *err,
+ gpointer opaque)
+{
+ struct TestIOChannelData *data = opaque;
+ data->err = err != NULL;
+ g_main_loop_quit(data->loop);
+}
+
+
+static void test_io_channel_setup_async(SocketAddress *listen_addr,
+ SocketAddress *connect_addr,
+ QIOChannel **src,
+ QIOChannel **dst)
+{
+ QIOChannelSocket *lioc;
+ struct TestIOChannelData data;
+
+ data.loop = g_main_loop_new(g_main_context_default(),
+ TRUE);
+
+ lioc = qio_channel_socket_new();
+ qio_channel_socket_listen_async(
+ lioc, listen_addr,
+ test_io_channel_complete, &data, NULL);
+
+ g_main_loop_run(data.loop);
+ g_main_context_iteration(g_main_context_default(), FALSE);
+
+ g_assert(!data.err);
+
+ if (listen_addr->type == SOCKET_ADDRESS_KIND_INET) {
+ SocketAddress *laddr = qio_channel_socket_get_local_address(
+ lioc, &error_abort);
+
+ g_free(connect_addr->u.inet->port);
+ connect_addr->u.inet->port = g_strdup(laddr->u.inet->port);
+
+ qapi_free_SocketAddress(laddr);
+ }
+
+ *src = QIO_CHANNEL(qio_channel_socket_new());
+
+ qio_channel_socket_connect_async(
+ QIO_CHANNEL_SOCKET(*src), connect_addr,
+ test_io_channel_complete, &data, NULL);
+
+ g_main_loop_run(data.loop);
+ g_main_context_iteration(g_main_context_default(), FALSE);
+
+ g_assert(!data.err);
+
+ *dst = QIO_CHANNEL(qio_channel_socket_accept(lioc, &error_abort));
+ g_assert(*dst);
+
+ qio_channel_set_delay(*src, false);
+ test_io_channel_set_socket_bufs(*src, *dst);
+
+ object_unref(OBJECT(lioc));
+
+ g_main_loop_unref(data.loop);
+}
+
+
+static void test_io_channel(bool async,
+ SocketAddress *listen_addr,
+ SocketAddress *connect_addr)
+{
+ QIOChannel *src, *dst;
+ QIOChannelTest *test;
+ if (async) {
+ test_io_channel_setup_async(listen_addr, connect_addr, &src, &dst);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_threads(test, true, src, dst);
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(src));
+ object_unref(OBJECT(dst));
+
+ test_io_channel_setup_async(listen_addr, connect_addr, &src, &dst);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_threads(test, false, src, dst);
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(src));
+ object_unref(OBJECT(dst));
+ } else {
+ test_io_channel_setup_sync(listen_addr, connect_addr, &src, &dst);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_threads(test, true, src, dst);
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(src));
+ object_unref(OBJECT(dst));
+
+ test_io_channel_setup_sync(listen_addr, connect_addr, &src, &dst);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_threads(test, false, src, dst);
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(src));
+ object_unref(OBJECT(dst));
+ }
+}
+
+
+static void test_io_channel_ipv4(bool async)
+{
+ SocketAddress *listen_addr = g_new0(SocketAddress, 1);
+ SocketAddress *connect_addr = g_new0(SocketAddress, 1);
+
+ listen_addr->type = SOCKET_ADDRESS_KIND_INET;
+ listen_addr->u.inet = g_new0(InetSocketAddress, 1);
+ listen_addr->u.inet->host = g_strdup("0.0.0.0");
+ listen_addr->u.inet->port = NULL; /* Auto-select */
+
+ connect_addr->type = SOCKET_ADDRESS_KIND_INET;
+ connect_addr->u.inet = g_new0(InetSocketAddress, 1);
+ connect_addr->u.inet->host = g_strdup("127.0.0.1");
+ connect_addr->u.inet->port = NULL; /* Filled in later */
+
+ test_io_channel(async, listen_addr, connect_addr);
+
+ qapi_free_SocketAddress(listen_addr);
+ qapi_free_SocketAddress(connect_addr);
+}
+
+
+static void test_io_channel_ipv4_sync(void)
+{
+ return test_io_channel_ipv4(false);
+}
+
+
+static void test_io_channel_ipv4_async(void)
+{
+ return test_io_channel_ipv4(true);
+}
+
+
+static void test_io_channel_ipv6(bool async)
+{
+ SocketAddress *listen_addr = g_new0(SocketAddress, 1);
+ SocketAddress *connect_addr = g_new0(SocketAddress, 1);
+
+ listen_addr->type = SOCKET_ADDRESS_KIND_INET;
+ listen_addr->u.inet = g_new0(InetSocketAddress, 1);
+ listen_addr->u.inet->host = g_strdup("::");
+ listen_addr->u.inet->port = NULL; /* Auto-select */
+
+ connect_addr->type = SOCKET_ADDRESS_KIND_INET;
+ connect_addr->u.inet = g_new0(InetSocketAddress, 1);
+ connect_addr->u.inet->host = g_strdup("::1");
+ connect_addr->u.inet->port = NULL; /* Filled in later */
+
+ test_io_channel(async, listen_addr, connect_addr);
+
+ qapi_free_SocketAddress(listen_addr);
+ qapi_free_SocketAddress(connect_addr);
+}
+
+
+static void test_io_channel_ipv6_sync(void)
+{
+ return test_io_channel_ipv6(false);
+}
+
+
+static void test_io_channel_ipv6_async(void)
+{
+ return test_io_channel_ipv6(true);
+}
+
+
+#ifndef _WIN32
+static void test_io_channel_unix(bool async)
+{
+ SocketAddress *listen_addr = g_new0(SocketAddress, 1);
+ SocketAddress *connect_addr = g_new0(SocketAddress, 1);
+
+#define TEST_SOCKET "test-io-channel-socket.sock"
+ listen_addr->type = SOCKET_ADDRESS_KIND_UNIX;
+ listen_addr->u.q_unix = g_new0(UnixSocketAddress, 1);
+ listen_addr->u.q_unix->path = g_strdup(TEST_SOCKET);
+
+ connect_addr->type = SOCKET_ADDRESS_KIND_UNIX;
+ connect_addr->u.q_unix = g_new0(UnixSocketAddress, 1);
+ connect_addr->u.q_unix->path = g_strdup(TEST_SOCKET);
+
+ test_io_channel(async, listen_addr, connect_addr);
+
+ qapi_free_SocketAddress(listen_addr);
+ qapi_free_SocketAddress(connect_addr);
+ unlink(TEST_SOCKET);
+}
+
+
+static void test_io_channel_unix_sync(void)
+{
+ return test_io_channel_unix(false);
+}
+
+
+static void test_io_channel_unix_async(void)
+{
+ return test_io_channel_unix(true);
+}
+#endif /* _WIN32 */
+
+
+int main(int argc, char **argv)
+{
+ bool has_ipv4, has_ipv6;
+
+ module_call_init(MODULE_INIT_QOM);
+
+ g_test_init(&argc, &argv, NULL);
+
+ /* We're creating actual IPv4/6 sockets, so we should
+ * check if the host running tests actually supports
+ * each protocol to avoid breaking tests on machines
+ * with either IPv4 or IPv6 disabled.
+ */
+ if (check_protocol_support(&has_ipv4, &has_ipv6) < 0) {
+ return 1;
+ }
+
+ if (has_ipv4) {
+ g_test_add_func("/io/channel/socket/ipv4-sync",
+ test_io_channel_ipv4_sync);
+ g_test_add_func("/io/channel/socket/ipv4-async",
+ test_io_channel_ipv4_async);
+ }
+ if (has_ipv6) {
+ g_test_add_func("/io/channel/socket/ipv6-sync",
+ test_io_channel_ipv6_sync);
+ g_test_add_func("/io/channel/socket/ipv6-async",
+ test_io_channel_ipv6_async);
+ }
+
+#ifndef _WIN32
+ g_test_add_func("/io/channel/socket/unix-sync",
+ test_io_channel_unix_sync);
+ g_test_add_func("/io/channel/socket/unix-async",
+ test_io_channel_unix_async);
+#endif /* _WIN32 */
+
+ return g_test_run();
+}
diff --git a/trace-events b/trace-events
index 20356b5..5eb386e 100644
--- a/trace-events
+++ b/trace-events
@@ -1803,3 +1803,22 @@ qio_task_thread_start(void *task, void *worker, void
*opaque) "Task thread start
qio_task_thread_run(void *task) "Task thread run task=%p"
qio_task_thread_exit(void *task) "Task thread exit task=%p"
qio_task_thread_result(void *task) "Task thread result task=%p"
+
+# io/channel-socket.c
+qio_channel_socket_new(void *ioc) "Socket new ioc=%p"
+qio_channel_socket_new_fd(void *ioc, int fd) "Socket new ioc=%p fd=%d"
+qio_channel_socket_connect_sync(void *ioc, void *addr) "Socket connect sync
ioc=%p addr=%p"
+qio_channel_socket_connect_async(void *ioc, void *addr) "Socket connect async
ioc=%p addr=%p"
+qio_channel_socket_connect_fail(void *ioc) "Socket connect fail ioc=%p"
+qio_channel_socket_connect_complete(void *ioc, int fd) "Socket connect
complete ioc=%p fd=%d"
+qio_channel_socket_listen_sync(void *ioc, void *addr) "Socket listen sync
ioc=%p addr=%p"
+qio_channel_socket_listen_async(void *ioc, void *addr) "Socket listen async
ioc=%p addr=%p"
+qio_channel_socket_listen_fail(void *ioc) "Socket listen fail ioc=%p"
+qio_channel_socket_listen_complete(void *ioc, int fd) "Socket listen complete
ioc=%p fd=%d"
+qio_channel_socket_dgram_sync(void *ioc, void *localAddr, void *remoteAddr)
"Socket dgram sync ioc=%p localAddr=%p remoteAddr=%p"
+qio_channel_socket_dgram_async(void *ioc, void *localAddr, void *remoteAddr)
"Socket dgram async ioc=%p localAddr=%p remoteAddr=%p"
+qio_channel_socket_dgram_fail(void *ioc) "Socket dgram fail ioc=%p"
+qio_channel_socket_dgram_complete(void *ioc, int fd) "Socket dgram complete
ioc=%p fd=%d"
+qio_channel_socket_accept(void *ioc) "Socket accept start ioc=%p"
+qio_channel_socket_accept_fail(void *ioc) "Socket accept fail ioc=%p"
+qio_channel_socket_accept_complete(void *ioc, void *cioc, int fd) "Socket
accept complete ioc=%p cioc=%p fd=%d"
diff --git a/util/qemu-sockets.c b/util/qemu-sockets.c
index 5a31d16..922efb3 100644
--- a/util/qemu-sockets.c
+++ b/util/qemu-sockets.c
@@ -1086,7 +1086,7 @@ socket_sockaddr_to_address_unix(struct sockaddr_storage
*sa,
}
#endif /* WIN32 */
-static SocketAddress *
+SocketAddress *
socket_sockaddr_to_address(struct sockaddr_storage *sa,
socklen_t salen,
Error **errp)
--
2.5.0
- [Qemu-devel] [PULL v2 0/9] Introduce I/O channels framework, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 1/9] io: add abstract QIOChannel classes, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 2/9] io: add helper module for creating watches on FDs, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 5/9] io: add QIOChannelFile class, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 3/9] io: add QIOTask class for async operations, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 6/9] io: add QIOChannelTLS class, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 8/9] io: add QIOChannelCommand class, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 9/9] io: add QIOChannelBuffer class, Daniel P. Berrange, 2015/12/17
- [Qemu-devel] [PULL v2 4/9] io: add QIOChannelSocket class,
Daniel P. Berrange <=
- [Qemu-devel] [PULL v2 7/9] io: add QIOChannelWebsock class, Daniel P. Berrange, 2015/12/17
- Re: [Qemu-devel] [PULL v2 0/9] Introduce I/O channels framework, Peter Maydell, 2015/12/17