colo: Consolidate the duplicate code chunk into a routine

[qemu.git] / io / channel-socket.c

diff --git a/io/channel-socket.c b/io/channel-socket.c
index 10a5b3136ee0683607501b47ed3a14e7ccd5b824..563e297357c35dc0c9c97494fb347a2b66c395b4 100644 (file)
--- a/io/channel-socket.c
+++ b/io/channel-socket.c
@@ -18,9 +18,12 @@
  *
  */
 
  *
  */
 

+#include "qemu/osdep.h"
+#include "qapi/error.h"

 #include "io/channel-socket.h"
 #include "io/channel-watch.h"
 #include "trace.h"
 #include "io/channel-socket.h"
 #include "io/channel-watch.h"
 #include "trace.h"

+#include "qapi/clone-visitor.h"

 
 #define SOCKET_MAX_FDS 16
 
 
 #define SOCKET_MAX_FDS 16
 


@@ -52,7 +55,11 @@ qio_channel_socket_new(void)
     sioc->fd = -1;
 
     ioc = QIO_CHANNEL(sioc);
     sioc->fd = -1;
 
     ioc = QIO_CHANNEL(sioc);

-    ioc->features |= (1 << QIO_CHANNEL_FEATURE_SHUTDOWN);
+    qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_SHUTDOWN);
+
+#ifdef WIN32
+    ioc->event = CreateEvent(NULL, FALSE, FALSE, NULL);
+#endif

 
     trace_qio_channel_socket_new(sioc);
 
 
     trace_qio_channel_socket_new(sioc);
 


@@ -77,11 +84,11 @@ qio_channel_socket_set_fd(QIOChannelSocket *sioc,
 
     if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
                     &sioc->remoteAddrLen) < 0) {
 
     if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
                     &sioc->remoteAddrLen) < 0) {

-        if (socket_error() == ENOTCONN) {
+        if (errno == ENOTCONN) {

             memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
             sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
         } else {
             memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
             sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
         } else {

-            error_setg_errno(errp, socket_error(),
+            error_setg_errno(errp, errno,

                              "Unable to query remote socket address");
             goto error;
         }
                              "Unable to query remote socket address");
             goto error;
         }


@@ -89,7 +96,7 @@ qio_channel_socket_set_fd(QIOChannelSocket *sioc,
 
     if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
                     &sioc->localAddrLen) < 0) {
 
     if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
                     &sioc->localAddrLen) < 0) {

-        error_setg_errno(errp, socket_error(),
+        error_setg_errno(errp, errno,

                          "Unable to query local socket address");
         goto error;
     }
                          "Unable to query local socket address");
         goto error;
     }


@@ -97,7 +104,7 @@ qio_channel_socket_set_fd(QIOChannelSocket *sioc,
 #ifndef WIN32
     if (sioc->localAddr.ss_family == AF_UNIX) {
         QIOChannel *ioc = QIO_CHANNEL(sioc);
 #ifndef WIN32
     if (sioc->localAddr.ss_family == AF_UNIX) {
         QIOChannel *ioc = QIO_CHANNEL(sioc);

-        ioc->features |= (1 << QIO_CHANNEL_FEATURE_FD_PASS);
+        qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS);

     }
 #endif /* WIN32 */
 
     }
 #endif /* WIN32 */
 


@@ -133,7 +140,7 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
     int fd;
 
     trace_qio_channel_socket_connect_sync(ioc, addr);
     int fd;
 
     trace_qio_channel_socket_connect_sync(ioc, addr);

-    fd = socket_connect(addr, errp, NULL, NULL);
+    fd = socket_connect(addr, errp);

     if (fd < 0) {
         trace_qio_channel_socket_connect_fail(ioc);
         return -1;
     if (fd < 0) {
         trace_qio_channel_socket_connect_fail(ioc);
         return -1;


@@ -149,20 +156,16 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
 }
 
 
 }
 
 

-static int qio_channel_socket_connect_worker(QIOTask *task,
-                                             Error **errp,
-                                             gpointer opaque)
+static void qio_channel_socket_connect_worker(QIOTask *task,
+                                              gpointer opaque)

 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     SocketAddress *addr = opaque;
 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     SocketAddress *addr = opaque;

-    int ret;
+    Error *err = NULL;

 
 

-    ret = qio_channel_socket_connect_sync(ioc,
-                                          addr,
-                                          errp);
+    qio_channel_socket_connect_sync(ioc, addr, &err);

 
 

-    object_unref(OBJECT(ioc));
-    return ret;
+    qio_task_set_error(task, err);

 }
 
 
 }
 
 


@@ -176,7 +179,7 @@ void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
         OBJECT(ioc), callback, opaque, destroy);
     SocketAddress *addrCopy;
 
         OBJECT(ioc), callback, opaque, destroy);
     SocketAddress *addrCopy;
 

-    qapi_copy_SocketAddress(&addrCopy, addr);
+    addrCopy = QAPI_CLONE(SocketAddress, addr);

 
     /* socket_connect() does a non-blocking connect(), but it
      * still blocks in DNS lookups, so we must use a thread */
 
     /* socket_connect() does a non-blocking connect(), but it
      * still blocks in DNS lookups, so we must use a thread */


@@ -206,25 +209,22 @@ int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
         close(fd);
         return -1;
     }
         close(fd);
         return -1;
     }

+    qio_channel_set_feature(QIO_CHANNEL(ioc), QIO_CHANNEL_FEATURE_LISTEN);

 
     return 0;
 }
 
 
 
     return 0;
 }
 
 

-static int qio_channel_socket_listen_worker(QIOTask *task,
-                                            Error **errp,
-                                            gpointer opaque)
+static void qio_channel_socket_listen_worker(QIOTask *task,
+                                             gpointer opaque)

 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     SocketAddress *addr = opaque;
 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     SocketAddress *addr = opaque;

-    int ret;
+    Error *err = NULL;

 
 

-    ret = qio_channel_socket_listen_sync(ioc,
-                                         addr,
-                                         errp);
+    qio_channel_socket_listen_sync(ioc, addr, &err);

 
 

-    object_unref(OBJECT(ioc));
-    return ret;
+    qio_task_set_error(task, err);

 }
 
 
 }
 
 


@@ -238,7 +238,7 @@ void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
         OBJECT(ioc), callback, opaque, destroy);
     SocketAddress *addrCopy;
 
         OBJECT(ioc), callback, opaque, destroy);
     SocketAddress *addrCopy;
 

-    qapi_copy_SocketAddress(&addrCopy, addr);
+    addrCopy = QAPI_CLONE(SocketAddress, addr);

 
     /* socket_listen() blocks in DNS lookups, so we must use a thread */
     trace_qio_channel_socket_listen_async(ioc, addr);
 
     /* socket_listen() blocks in DNS lookups, so we must use a thread */
     trace_qio_channel_socket_listen_async(ioc, addr);


@@ -257,7 +257,7 @@ int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
     int fd;
 
     trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
     int fd;
 
     trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);

-    fd = socket_dgram(localAddr, remoteAddr, errp);
+    fd = socket_dgram(remoteAddr, localAddr, errp);

     if (fd < 0) {
         trace_qio_channel_socket_dgram_fail(ioc);
         return -1;
     if (fd < 0) {
         trace_qio_channel_socket_dgram_fail(ioc);
         return -1;


@@ -287,22 +287,18 @@ static void qio_channel_socket_dgram_worker_free(gpointer opaque)
     g_free(data);
 }
 
     g_free(data);
 }
 

-static int qio_channel_socket_dgram_worker(QIOTask *task,
-                                           Error **errp,
-                                           gpointer opaque)
+static void qio_channel_socket_dgram_worker(QIOTask *task,
+                                            gpointer opaque)

 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     struct QIOChannelSocketDGramWorkerData *data = opaque;
 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
     struct QIOChannelSocketDGramWorkerData *data = opaque;

-    int ret;
+    Error *err = NULL;

 
     /* socket_dgram() blocks in DNS lookups, so we must use a thread */
 
     /* socket_dgram() blocks in DNS lookups, so we must use a thread */

-    ret = qio_channel_socket_dgram_sync(ioc,
-                                        data->localAddr,
-                                        data->remoteAddr,
-                                        errp);
+    qio_channel_socket_dgram_sync(ioc, data->localAddr,
+                                  data->remoteAddr, &err);

 
 

-    object_unref(OBJECT(ioc));
-    return ret;
+    qio_task_set_error(task, err);

 }
 
 
 }
 
 


@@ -318,8 +314,8 @@ void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
     struct QIOChannelSocketDGramWorkerData *data = g_new0(
         struct QIOChannelSocketDGramWorkerData, 1);
 
     struct QIOChannelSocketDGramWorkerData *data = g_new0(
         struct QIOChannelSocketDGramWorkerData, 1);
 

-    qapi_copy_SocketAddress(&data->localAddr, localAddr);
-    qapi_copy_SocketAddress(&data->remoteAddr, remoteAddr);
+    data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
+    data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);

 
     trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
     qio_task_run_in_thread(task,
 
     trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
     qio_task_run_in_thread(task,


@@ -335,33 +331,34 @@ qio_channel_socket_accept(QIOChannelSocket *ioc,
 {
     QIOChannelSocket *cioc;
 
 {
     QIOChannelSocket *cioc;
 

-    cioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
-    cioc->fd = -1;
+    cioc = qio_channel_socket_new();

     cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
     cioc->localAddrLen = sizeof(ioc->localAddr);
 
  retry:
     trace_qio_channel_socket_accept(ioc);
     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);
+    cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
+                           &cioc->remoteAddrLen);

     if (cioc->fd < 0) {
     if (cioc->fd < 0) {

-        trace_qio_channel_socket_accept_fail(ioc);
-        if (socket_error() == EINTR) {
+        if (errno == EINTR) {

             goto retry;
         }
             goto retry;
         }

+        error_setg_errno(errp, errno, "Unable to accept connection");
+        trace_qio_channel_socket_accept_fail(ioc);

         goto error;
     }
 
     if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
                     &cioc->localAddrLen) < 0) {
         goto error;
     }
 
     if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
                     &cioc->localAddrLen) < 0) {

-        error_setg_errno(errp, socket_error(),
+        error_setg_errno(errp, errno,

                          "Unable to query local socket address");
         goto error;
     }
 
 #ifndef WIN32
     if (cioc->localAddr.ss_family == AF_UNIX) {
                          "Unable to query local socket address");
         goto error;
     }
 
 #ifndef WIN32
     if (cioc->localAddr.ss_family == AF_UNIX) {

-        QIO_CHANNEL(cioc)->features |= (1 << QIO_CHANNEL_FEATURE_FD_PASS);
+        QIOChannel *ioc_local = QIO_CHANNEL(cioc);
+        qio_channel_set_feature(ioc_local, QIO_CHANNEL_FEATURE_FD_PASS);

     }
 #endif /* WIN32 */
 
     }
 #endif /* WIN32 */
 


@@ -382,8 +379,22 @@ static void qio_channel_socket_init(Object *obj)
 static void qio_channel_socket_finalize(Object *obj)
 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
 static void qio_channel_socket_finalize(Object *obj)
 {
     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);

+

     if (ioc->fd != -1) {
     if (ioc->fd != -1) {

-        close(ioc->fd);
+        QIOChannel *ioc_local = QIO_CHANNEL(ioc);
+        if (qio_channel_has_feature(ioc_local, QIO_CHANNEL_FEATURE_LISTEN)) {
+            Error *err = NULL;
+
+            socket_listen_cleanup(ioc->fd, &err);
+            if (err) {
+                error_report_err(err);
+                err = NULL;
+            }
+        }
+#ifdef WIN32
+        WSAEventSelect(ioc->fd, NULL, 0);
+#endif
+        closesocket(ioc->fd);

         ioc->fd = -1;
     }
 }
         ioc->fd = -1;
     }
 }


@@ -449,6 +460,8 @@ static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
     char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
     int sflags = 0;
 
     char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
     int sflags = 0;
 

+    memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
+

 #ifdef MSG_CMSG_CLOEXEC
     sflags |= MSG_CMSG_CLOEXEC;
 #endif
 #ifdef MSG_CMSG_CLOEXEC
     sflags |= MSG_CMSG_CLOEXEC;
 #endif


@@ -463,15 +476,14 @@ static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
  retry:
     ret = recvmsg(sioc->fd, &msg, sflags);
     if (ret < 0) {
  retry:
     ret = recvmsg(sioc->fd, &msg, sflags);
     if (ret < 0) {

-        if (socket_error() == EAGAIN ||
-            socket_error() == EWOULDBLOCK) {
+        if (errno == EAGAIN) {

             return QIO_CHANNEL_ERR_BLOCK;
         }
             return QIO_CHANNEL_ERR_BLOCK;
         }

-        if (socket_error() == EINTR) {
+        if (errno == EINTR) {

             goto retry;
         }
 
             goto retry;
         }
 

-        error_setg_errno(errp, socket_error(),
+        error_setg_errno(errp, errno,

                          "Unable to read from socket");
         return -1;
     }
                          "Unable to read from socket");
         return -1;
     }


@@ -493,16 +505,18 @@ static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
     ssize_t ret;
     struct msghdr msg = { NULL, };
     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
     ssize_t ret;
     struct msghdr msg = { NULL, };

-    char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)] = { 0 };
+    char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];

     size_t fdsize = sizeof(int) * nfds;
     struct cmsghdr *cmsg;
 
     size_t fdsize = sizeof(int) * nfds;
     struct cmsghdr *cmsg;
 

+    memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
+

     msg.msg_iov = (struct iovec *)iov;
     msg.msg_iovlen = niov;
 
     if (nfds) {
         if (nfds > SOCKET_MAX_FDS) {
     msg.msg_iov = (struct iovec *)iov;
     msg.msg_iovlen = niov;
 
     if (nfds) {
         if (nfds > SOCKET_MAX_FDS) {

-            error_setg_errno(errp, -EINVAL,
+            error_setg_errno(errp, EINVAL,

                              "Only %d FDs can be sent, got %zu",
                              SOCKET_MAX_FDS, nfds);
             return -1;
                              "Only %d FDs can be sent, got %zu",
                              SOCKET_MAX_FDS, nfds);
             return -1;


@@ -521,14 +535,13 @@ static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
  retry:
     ret = sendmsg(sioc->fd, &msg, 0);
     if (ret <= 0) {
  retry:
     ret = sendmsg(sioc->fd, &msg, 0);
     if (ret <= 0) {

-        if (socket_error() == EAGAIN ||
-            socket_error() == EWOULDBLOCK) {
+        if (errno == EAGAIN) {

             return QIO_CHANNEL_ERR_BLOCK;
         }
             return QIO_CHANNEL_ERR_BLOCK;
         }

-        if (socket_error() == EINTR) {
+        if (errno == EINTR) {

             goto retry;
         }
             goto retry;
         }

-        error_setg_errno(errp, socket_error(),
+        error_setg_errno(errp, errno,

                          "Unable to write to socket");
         return -1;
     }
                          "Unable to write to socket");
         return -1;
     }


@@ -554,17 +567,17 @@ static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
                    iov[i].iov_len,
                    0);
         if (ret < 0) {
                    iov[i].iov_len,
                    0);
         if (ret < 0) {

-            if (socket_error() == EAGAIN) {
+            if (errno == EAGAIN) {

                 if (done) {
                     return done;
                 } else {
                     return QIO_CHANNEL_ERR_BLOCK;
                 }
                 if (done) {
                     return done;
                 } else {
                     return QIO_CHANNEL_ERR_BLOCK;
                 }

-            } else if (socket_error() == EINTR) {
+            } else if (errno == EINTR) {

                 goto retry;
             } else {
                 goto retry;
             } else {

-                error_setg_errno(errp, socket_error(),
-                                 "Unable to write to socket");
+                error_setg_errno(errp, errno,
+                                 "Unable to read from socket");

                 return -1;
             }
         }
                 return -1;
             }
         }


@@ -596,16 +609,16 @@ static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
                    iov[i].iov_len,
                    0);
         if (ret < 0) {
                    iov[i].iov_len,
                    0);
         if (ret < 0) {

-            if (socket_error() == EAGAIN) {
+            if (errno == EAGAIN) {

                 if (done) {
                     return done;
                 } else {
                     return QIO_CHANNEL_ERR_BLOCK;
                 }
                 if (done) {
                     return done;
                 } else {
                     return QIO_CHANNEL_ERR_BLOCK;
                 }

-            } else if (socket_error() == EINTR) {
+            } else if (errno == EINTR) {

                 goto retry;
             } else {
                 goto retry;
             } else {

-                error_setg_errno(errp, socket_error(),
+                error_setg_errno(errp, errno,

                                  "Unable to write to socket");
                 return -1;
             }
                                  "Unable to write to socket");
                 return -1;
             }


@@ -666,13 +679,18 @@ qio_channel_socket_close(QIOChannel *ioc,
 {
     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 
 {
     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 

-    if (closesocket(sioc->fd) < 0) {
+    if (sioc->fd != -1) {
+#ifdef WIN32
+        WSAEventSelect(sioc->fd, NULL, 0);
+#endif
+        if (closesocket(sioc->fd) < 0) {
+            sioc->fd = -1;
+            error_setg_errno(errp, errno,
+                             "Unable to close socket");
+            return -1;
+        }

         sioc->fd = -1;
         sioc->fd = -1;

-        error_setg_errno(errp, socket_error(),
-                         "Unable to close socket");
-        return -1;

     }
     }

-    sioc->fd = -1;

     return 0;
 }
 
     return 0;
 }
 


@@ -698,20 +716,30 @@ qio_channel_socket_shutdown(QIOChannel *ioc,
     }
 
     if (shutdown(sioc->fd, sockhow) < 0) {
     }
 
     if (shutdown(sioc->fd, sockhow) < 0) {

-        error_setg_errno(errp, socket_error(),
+        error_setg_errno(errp, errno,

                          "Unable to shutdown socket");
         return -1;
     }
     return 0;
 }
 
                          "Unable to shutdown socket");
         return -1;
     }
     return 0;
 }
 

+static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
+                                                  AioContext *ctx,
+                                                  IOHandler *io_read,
+                                                  IOHandler *io_write,
+                                                  void *opaque)
+{
+    QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
+    aio_set_fd_handler(ctx, sioc->fd, false, io_read, io_write, NULL, opaque);
+}
+

 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
                                                 GIOCondition condition)
 {
     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 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);
+    return qio_channel_create_socket_watch(ioc,
+                                           sioc->fd,
+                                           condition);

 }
 
 static void qio_channel_socket_class_init(ObjectClass *klass,
 }
 
 static void qio_channel_socket_class_init(ObjectClass *klass,


@@ -727,6 +755,7 @@ static void qio_channel_socket_class_init(ObjectClass *klass,
     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;
     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;

+    ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;

 }
 
 static const TypeInfo qio_channel_socket_info = {
 }
 
 static const TypeInfo qio_channel_socket_info = {