io/channel-socket.c

   1 /*
   2  * QEMU I/O channels sockets driver
   3  *
   4  * Copyright (c) 2015 Red Hat, Inc.
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2.1 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General Public
  17  * License along with this library; if not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 #include "qemu/osdep.h"
  21 #include "qapi/error.h"
  22 #include "qapi/qapi-visit-sockets.h"
  23 #include "qemu/module.h"
  24 #include "io/channel-socket.h"
  25 #include "io/channel-watch.h"
  26 #include "trace.h"
  27 #include "qapi/clone-visitor.h"
  28 #ifdef CONFIG_LINUX
  29 #include <linux/errqueue.h>
  30 #include <sys/socket.h>
  31
  32 #if (defined(MSG_ZEROCOPY) && defined(SO_ZEROCOPY))
  33 #define QEMU_MSG_ZEROCOPY
  34 #endif
  35 #endif
  36
  37 #define SOCKET_MAX_FDS 16
  38
  39 SocketAddress *
  40 qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
  41                                      Error **errp)
  42 {
  43     return socket_sockaddr_to_address(&ioc->localAddr,
  44                                       ioc->localAddrLen,
  45                                       errp);
  46 }
  47
  48 SocketAddress *
  49 qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
  50                                       Error **errp)
  51 {
  52     return socket_sockaddr_to_address(&ioc->remoteAddr,
  53                                       ioc->remoteAddrLen,
  54                                       errp);
  55 }
  56
  57 QIOChannelSocket *
  58 qio_channel_socket_new(void)
  59 {
  60     QIOChannelSocket *sioc;
  61     QIOChannel *ioc;
  62
  63     sioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
  64     sioc->fd = -1;
  65     sioc->zero_copy_queued = 0;
  66     sioc->zero_copy_sent = 0;
  67
  68     ioc = QIO_CHANNEL(sioc);
  69     qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_SHUTDOWN);
  70
  71 #ifdef WIN32
  72     ioc->event = CreateEvent(NULL, FALSE, FALSE, NULL);
  73 #endif
  74
  75     trace_qio_channel_socket_new(sioc);
  76
  77     return sioc;
  78 }
  79
  80
  81 static int
  82 qio_channel_socket_set_fd(QIOChannelSocket *sioc,
  83                           int fd,
  84                           Error **errp)
  85 {
  86     if (sioc->fd != -1) {
  87         error_setg(errp, "Socket is already open");
  88         return -1;
  89     }
  90
  91     sioc->fd = fd;
  92     sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
  93     sioc->localAddrLen = sizeof(sioc->localAddr);
  94
  95
  96     if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
  97                     &sioc->remoteAddrLen) < 0) {
  98         if (errno == ENOTCONN) {
  99             memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
 100             sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
 101         } else {
 102             error_setg_errno(errp, errno,
 103                              "Unable to query remote socket address");
 104             goto error;
 105         }
 106     }
 107
 108     if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
 109                     &sioc->localAddrLen) < 0) {
 110         error_setg_errno(errp, errno,
 111                          "Unable to query local socket address");
 112         goto error;
 113     }
 114
 115 #ifndef WIN32
 116     if (sioc->localAddr.ss_family == AF_UNIX) {
 117         QIOChannel *ioc = QIO_CHANNEL(sioc);
 118         qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS);
 119     }
 120 #endif /* WIN32 */
 121
 122     return 0;
 123
 124  error:
 125     sioc->fd = -1; /* Let the caller close FD on failure */
 126     return -1;
 127 }
 128
 129 QIOChannelSocket *
 130 qio_channel_socket_new_fd(int fd,
 131                           Error **errp)
 132 {
 133     QIOChannelSocket *ioc;
 134
 135     ioc = qio_channel_socket_new();
 136     if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
 137         object_unref(OBJECT(ioc));
 138         return NULL;
 139     }
 140
 141     trace_qio_channel_socket_new_fd(ioc, fd);
 142
 143     return ioc;
 144 }
 145
 146
 147 int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
 148                                     SocketAddress *addr,
 149                                     Error **errp)
 150 {
 151     int fd;
 152
 153     trace_qio_channel_socket_connect_sync(ioc, addr);
 154     fd = socket_connect(addr, errp);
 155     if (fd < 0) {
 156         trace_qio_channel_socket_connect_fail(ioc);
 157         return -1;
 158     }
 159
 160     trace_qio_channel_socket_connect_complete(ioc, fd);
 161     if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
 162         close(fd);
 163         return -1;
 164     }
 165
 166 #ifdef QEMU_MSG_ZEROCOPY
 167     int ret, v = 1;
 168     ret = setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, &v, sizeof(v));
 169     if (ret == 0) {
 170         /* Zero copy available on host */
 171         qio_channel_set_feature(QIO_CHANNEL(ioc),
 172                                 QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY);
 173     }
 174 #endif
 175
 176     return 0;
 177 }
 178
 179
 180 static void qio_channel_socket_connect_worker(QIOTask *task,
 181                                               gpointer opaque)
 182 {
 183     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
 184     SocketAddress *addr = opaque;
 185     Error *err = NULL;
 186
 187     qio_channel_socket_connect_sync(ioc, addr, &err);
 188
 189     qio_task_set_error(task, err);
 190 }
 191
 192
 193 void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
 194                                       SocketAddress *addr,
 195                                       QIOTaskFunc callback,
 196                                       gpointer opaque,
 197                                       GDestroyNotify destroy,
 198                                       GMainContext *context)
 199 {
 200     QIOTask *task = qio_task_new(
 201         OBJECT(ioc), callback, opaque, destroy);
 202     SocketAddress *addrCopy;
 203
 204     addrCopy = QAPI_CLONE(SocketAddress, addr);
 205
 206     /* socket_connect() does a non-blocking connect(), but it
 207      * still blocks in DNS lookups, so we must use a thread */
 208     trace_qio_channel_socket_connect_async(ioc, addr);
 209     qio_task_run_in_thread(task,
 210                            qio_channel_socket_connect_worker,
 211                            addrCopy,
 212                            (GDestroyNotify)qapi_free_SocketAddress,
 213                            context);
 214 }
 215
 216
 217 int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
 218                                    SocketAddress *addr,
 219                                    int num,
 220                                    Error **errp)
 221 {
 222     int fd;
 223
 224     trace_qio_channel_socket_listen_sync(ioc, addr, num);
 225     fd = socket_listen(addr, num, errp);
 226     if (fd < 0) {
 227         trace_qio_channel_socket_listen_fail(ioc);
 228         return -1;
 229     }
 230
 231     trace_qio_channel_socket_listen_complete(ioc, fd);
 232     if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
 233         close(fd);
 234         return -1;
 235     }
 236     qio_channel_set_feature(QIO_CHANNEL(ioc), QIO_CHANNEL_FEATURE_LISTEN);
 237
 238     return 0;
 239 }
 240
 241
 242 struct QIOChannelListenWorkerData {
 243     SocketAddress *addr;
 244     int num; /* amount of expected connections */
 245 };
 246
 247 static void qio_channel_listen_worker_free(gpointer opaque)
 248 {
 249     struct QIOChannelListenWorkerData *data = opaque;
 250
 251     qapi_free_SocketAddress(data->addr);
 252     g_free(data);
 253 }
 254
 255 static void qio_channel_socket_listen_worker(QIOTask *task,
 256                                              gpointer opaque)
 257 {
 258     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
 259     struct QIOChannelListenWorkerData *data = opaque;
 260     Error *err = NULL;
 261
 262     qio_channel_socket_listen_sync(ioc, data->addr, data->num, &err);
 263
 264     qio_task_set_error(task, err);
 265 }
 266
 267
 268 void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
 269                                      SocketAddress *addr,
 270                                      int num,
 271                                      QIOTaskFunc callback,
 272                                      gpointer opaque,
 273                                      GDestroyNotify destroy,
 274                                      GMainContext *context)
 275 {
 276     QIOTask *task = qio_task_new(
 277         OBJECT(ioc), callback, opaque, destroy);
 278     struct QIOChannelListenWorkerData *data;
 279
 280     data = g_new0(struct QIOChannelListenWorkerData, 1);
 281     data->addr = QAPI_CLONE(SocketAddress, addr);
 282     data->num = num;
 283
 284     /* socket_listen() blocks in DNS lookups, so we must use a thread */
 285     trace_qio_channel_socket_listen_async(ioc, addr, num);
 286     qio_task_run_in_thread(task,
 287                            qio_channel_socket_listen_worker,
 288                            data,
 289                            qio_channel_listen_worker_free,
 290                            context);
 291 }
 292
 293
 294 int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
 295                                   SocketAddress *localAddr,
 296                                   SocketAddress *remoteAddr,
 297                                   Error **errp)
 298 {
 299     int fd;
 300
 301     trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
 302     fd = socket_dgram(remoteAddr, localAddr, errp);
 303     if (fd < 0) {
 304         trace_qio_channel_socket_dgram_fail(ioc);
 305         return -1;
 306     }
 307
 308     trace_qio_channel_socket_dgram_complete(ioc, fd);
 309     if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
 310         close(fd);
 311         return -1;
 312     }
 313
 314     return 0;
 315 }
 316
 317
 318 struct QIOChannelSocketDGramWorkerData {
 319     SocketAddress *localAddr;
 320     SocketAddress *remoteAddr;
 321 };
 322
 323
 324 static void qio_channel_socket_dgram_worker_free(gpointer opaque)
 325 {
 326     struct QIOChannelSocketDGramWorkerData *data = opaque;
 327     qapi_free_SocketAddress(data->localAddr);
 328     qapi_free_SocketAddress(data->remoteAddr);
 329     g_free(data);
 330 }
 331
 332 static void qio_channel_socket_dgram_worker(QIOTask *task,
 333                                             gpointer opaque)
 334 {
 335     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
 336     struct QIOChannelSocketDGramWorkerData *data = opaque;
 337     Error *err = NULL;
 338
 339     /* socket_dgram() blocks in DNS lookups, so we must use a thread */
 340     qio_channel_socket_dgram_sync(ioc, data->localAddr,
 341                                   data->remoteAddr, &err);
 342
 343     qio_task_set_error(task, err);
 344 }
 345
 346
 347 void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
 348                                     SocketAddress *localAddr,
 349                                     SocketAddress *remoteAddr,
 350                                     QIOTaskFunc callback,
 351                                     gpointer opaque,
 352                                     GDestroyNotify destroy,
 353                                     GMainContext *context)
 354 {
 355     QIOTask *task = qio_task_new(
 356         OBJECT(ioc), callback, opaque, destroy);
 357     struct QIOChannelSocketDGramWorkerData *data = g_new0(
 358         struct QIOChannelSocketDGramWorkerData, 1);
 359
 360     data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
 361     data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);
 362
 363     trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
 364     qio_task_run_in_thread(task,
 365                            qio_channel_socket_dgram_worker,
 366                            data,
 367                            qio_channel_socket_dgram_worker_free,
 368                            context);
 369 }
 370
 371
 372 QIOChannelSocket *
 373 qio_channel_socket_accept(QIOChannelSocket *ioc,
 374                           Error **errp)
 375 {
 376     QIOChannelSocket *cioc;
 377
 378     cioc = qio_channel_socket_new();
 379     cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
 380     cioc->localAddrLen = sizeof(ioc->localAddr);
 381
 382  retry:
 383     trace_qio_channel_socket_accept(ioc);
 384     cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
 385                            &cioc->remoteAddrLen);
 386     if (cioc->fd < 0) {
 387         if (errno == EINTR) {
 388             goto retry;
 389         }
 390         error_setg_errno(errp, errno, "Unable to accept connection");
 391         trace_qio_channel_socket_accept_fail(ioc);
 392         goto error;
 393     }
 394
 395     if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
 396                     &cioc->localAddrLen) < 0) {
 397         error_setg_errno(errp, errno,
 398                          "Unable to query local socket address");
 399         goto error;
 400     }
 401
 402 #ifndef WIN32
 403     if (cioc->localAddr.ss_family == AF_UNIX) {
 404         QIOChannel *ioc_local = QIO_CHANNEL(cioc);
 405         qio_channel_set_feature(ioc_local, QIO_CHANNEL_FEATURE_FD_PASS);
 406     }
 407 #endif /* WIN32 */
 408
 409     trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
 410     return cioc;
 411
 412  error:
 413     object_unref(OBJECT(cioc));
 414     return NULL;
 415 }
 416
 417 static void qio_channel_socket_init(Object *obj)
 418 {
 419     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
 420     ioc->fd = -1;
 421 }
 422
 423 static void qio_channel_socket_finalize(Object *obj)
 424 {
 425     QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
 426
 427     if (ioc->fd != -1) {
 428         QIOChannel *ioc_local = QIO_CHANNEL(ioc);
 429         if (qio_channel_has_feature(ioc_local, QIO_CHANNEL_FEATURE_LISTEN)) {
 430             Error *err = NULL;
 431
 432             socket_listen_cleanup(ioc->fd, &err);
 433             if (err) {
 434                 error_report_err(err);
 435                 err = NULL;
 436             }
 437         }
 438 #ifdef WIN32
 439         WSAEventSelect(ioc->fd, NULL, 0);
 440 #endif
 441         closesocket(ioc->fd);
 442         ioc->fd = -1;
 443     }
 444 }
 445
 446
 447 #ifndef WIN32
 448 static void qio_channel_socket_copy_fds(struct msghdr *msg,
 449                                         int **fds, size_t *nfds)
 450 {
 451     struct cmsghdr *cmsg;
 452
 453     *nfds = 0;
 454     *fds = NULL;
 455
 456     for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
 457         int fd_size, i;
 458         int gotfds;
 459
 460         if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
 461             cmsg->cmsg_level != SOL_SOCKET ||
 462             cmsg->cmsg_type != SCM_RIGHTS) {
 463             continue;
 464         }
 465
 466         fd_size = cmsg->cmsg_len - CMSG_LEN(0);
 467
 468         if (!fd_size) {
 469             continue;
 470         }
 471
 472         gotfds = fd_size / sizeof(int);
 473         *fds = g_renew(int, *fds, *nfds + gotfds);
 474         memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
 475
 476         for (i = 0; i < gotfds; i++) {
 477             int fd = (*fds)[*nfds + i];
 478             if (fd < 0) {
 479                 continue;
 480             }
 481
 482             /* O_NONBLOCK is preserved across SCM_RIGHTS so reset it */
 483             qemu_socket_set_block(fd);
 484
 485 #ifndef MSG_CMSG_CLOEXEC
 486             qemu_set_cloexec(fd);
 487 #endif
 488         }
 489         *nfds += gotfds;
 490     }
 491 }
 492
 493
 494 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
 495                                         const struct iovec *iov,
 496                                         size_t niov,
 497                                         int **fds,
 498                                         size_t *nfds,
 499                                         Error **errp)
 500 {
 501     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 502     ssize_t ret;
 503     struct msghdr msg = { NULL, };
 504     char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
 505     int sflags = 0;
 506
 507     memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
 508
 509     msg.msg_iov = (struct iovec *)iov;
 510     msg.msg_iovlen = niov;
 511     if (fds && nfds) {
 512         msg.msg_control = control;
 513         msg.msg_controllen = sizeof(control);
 514 #ifdef MSG_CMSG_CLOEXEC
 515         sflags |= MSG_CMSG_CLOEXEC;
 516 #endif
 517
 518     }
 519
 520  retry:
 521     ret = recvmsg(sioc->fd, &msg, sflags);
 522     if (ret < 0) {
 523         if (errno == EAGAIN) {
 524             return QIO_CHANNEL_ERR_BLOCK;
 525         }
 526         if (errno == EINTR) {
 527             goto retry;
 528         }
 529
 530         error_setg_errno(errp, errno,
 531                          "Unable to read from socket");
 532         return -1;
 533     }
 534
 535     if (fds && nfds) {
 536         qio_channel_socket_copy_fds(&msg, fds, nfds);
 537     }
 538
 539     return ret;
 540 }
 541
 542 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
 543                                          const struct iovec *iov,
 544                                          size_t niov,
 545                                          int *fds,
 546                                          size_t nfds,
 547                                          int flags,
 548                                          Error **errp)
 549 {
 550     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 551     ssize_t ret;
 552     struct msghdr msg = { NULL, };
 553     char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
 554     size_t fdsize = sizeof(int) * nfds;
 555     struct cmsghdr *cmsg;
 556     int sflags = 0;
 557
 558     memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
 559
 560     msg.msg_iov = (struct iovec *)iov;
 561     msg.msg_iovlen = niov;
 562
 563     if (nfds) {
 564         if (nfds > SOCKET_MAX_FDS) {
 565             error_setg_errno(errp, EINVAL,
 566                              "Only %d FDs can be sent, got %zu",
 567                              SOCKET_MAX_FDS, nfds);
 568             return -1;
 569         }
 570
 571         msg.msg_control = control;
 572         msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
 573
 574         cmsg = CMSG_FIRSTHDR(&msg);
 575         cmsg->cmsg_len = CMSG_LEN(fdsize);
 576         cmsg->cmsg_level = SOL_SOCKET;
 577         cmsg->cmsg_type = SCM_RIGHTS;
 578         memcpy(CMSG_DATA(cmsg), fds, fdsize);
 579     }
 580
 581     if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
 582 #ifdef QEMU_MSG_ZEROCOPY
 583         sflags = MSG_ZEROCOPY;
 584 #else
 585         /*
 586          * We expect QIOChannel class entry point to have
 587          * blocked this code path already
 588          */
 589         g_assert_not_reached();
 590 #endif
 591     }
 592
 593  retry:
 594     ret = sendmsg(sioc->fd, &msg, sflags);
 595     if (ret <= 0) {
 596         switch (errno) {
 597         case EAGAIN:
 598             return QIO_CHANNEL_ERR_BLOCK;
 599         case EINTR:
 600             goto retry;
 601         case ENOBUFS:
 602             if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
 603                 error_setg_errno(errp, errno,
 604                                  "Process can't lock enough memory for using MSG_ZEROCOPY");
 605                 return -1;
 606             }
 607             break;
 608         }
 609
 610         error_setg_errno(errp, errno,
 611                          "Unable to write to socket");
 612         return -1;
 613     }
 614
 615     if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
 616         sioc->zero_copy_queued++;
 617     }
 618
 619     return ret;
 620 }
 621 #else /* WIN32 */
 622 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
 623                                         const struct iovec *iov,
 624                                         size_t niov,
 625                                         int **fds,
 626                                         size_t *nfds,
 627                                         Error **errp)
 628 {
 629     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 630     ssize_t done = 0;
 631     ssize_t i;
 632
 633     for (i = 0; i < niov; i++) {
 634         ssize_t ret;
 635     retry:
 636         ret = recv(sioc->fd,
 637                    iov[i].iov_base,
 638                    iov[i].iov_len,
 639                    0);
 640         if (ret < 0) {
 641             if (errno == EAGAIN) {
 642                 if (done) {
 643                     return done;
 644                 } else {
 645                     return QIO_CHANNEL_ERR_BLOCK;
 646                 }
 647             } else if (errno == EINTR) {
 648                 goto retry;
 649             } else {
 650                 error_setg_errno(errp, errno,
 651                                  "Unable to read from socket");
 652                 return -1;
 653             }
 654         }
 655         done += ret;
 656         if (ret < iov[i].iov_len) {
 657             return done;
 658         }
 659     }
 660
 661     return done;
 662 }
 663
 664 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
 665                                          const struct iovec *iov,
 666                                          size_t niov,
 667                                          int *fds,
 668                                          size_t nfds,
 669                                          int flags,
 670                                          Error **errp)
 671 {
 672     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 673     ssize_t done = 0;
 674     ssize_t i;
 675
 676     for (i = 0; i < niov; i++) {
 677         ssize_t ret;
 678     retry:
 679         ret = send(sioc->fd,
 680                    iov[i].iov_base,
 681                    iov[i].iov_len,
 682                    0);
 683         if (ret < 0) {
 684             if (errno == EAGAIN) {
 685                 if (done) {
 686                     return done;
 687                 } else {
 688                     return QIO_CHANNEL_ERR_BLOCK;
 689                 }
 690             } else if (errno == EINTR) {
 691                 goto retry;
 692             } else {
 693                 error_setg_errno(errp, errno,
 694                                  "Unable to write to socket");
 695                 return -1;
 696             }
 697         }
 698         done += ret;
 699         if (ret < iov[i].iov_len) {
 700             return done;
 701         }
 702     }
 703
 704     return done;
 705 }
 706 #endif /* WIN32 */
 707
 708
 709 #ifdef QEMU_MSG_ZEROCOPY
 710 static int qio_channel_socket_flush(QIOChannel *ioc,
 711                                     Error **errp)
 712 {
 713     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 714     struct msghdr msg = {};
 715     struct sock_extended_err *serr;
 716     struct cmsghdr *cm;
 717     char control[CMSG_SPACE(sizeof(*serr))];
 718     int received;
 719     int ret;
 720
 721     if (sioc->zero_copy_queued == sioc->zero_copy_sent) {
 722         return 0;
 723     }
 724
 725     msg.msg_control = control;
 726     msg.msg_controllen = sizeof(control);
 727     memset(control, 0, sizeof(control));
 728
 729     ret = 1;
 730
 731     while (sioc->zero_copy_sent < sioc->zero_copy_queued) {
 732         received = recvmsg(sioc->fd, &msg, MSG_ERRQUEUE);
 733         if (received < 0) {
 734             switch (errno) {
 735             case EAGAIN:
 736                 /* Nothing on errqueue, wait until something is available */
 737                 qio_channel_wait(ioc, G_IO_ERR);
 738                 continue;
 739             case EINTR:
 740                 continue;
 741             default:
 742                 error_setg_errno(errp, errno,
 743                                  "Unable to read errqueue");
 744                 return -1;
 745             }
 746         }
 747
 748         cm = CMSG_FIRSTHDR(&msg);
 749         if (cm->cmsg_level != SOL_IP   && cm->cmsg_type != IP_RECVERR &&
 750             cm->cmsg_level != SOL_IPV6 && cm->cmsg_type != IPV6_RECVERR) {
 751             error_setg_errno(errp, EPROTOTYPE,
 752                              "Wrong cmsg in errqueue");
 753             return -1;
 754         }
 755
 756         serr = (void *) CMSG_DATA(cm);
 757         if (serr->ee_errno != SO_EE_ORIGIN_NONE) {
 758             error_setg_errno(errp, serr->ee_errno,
 759                              "Error on socket");
 760             return -1;
 761         }
 762         if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY) {
 763             error_setg_errno(errp, serr->ee_origin,
 764                              "Error not from zero copy");
 765             return -1;
 766         }
 767
 768         /* No errors, count successfully finished sendmsg()*/
 769         sioc->zero_copy_sent += serr->ee_data - serr->ee_info + 1;
 770
 771         /* If any sendmsg() succeeded using zero copy, return 0 at the end */
 772         if (serr->ee_code != SO_EE_CODE_ZEROCOPY_COPIED) {
 773             ret = 0;
 774         }
 775     }
 776
 777     return ret;
 778 }
 779
 780 #endif /* QEMU_MSG_ZEROCOPY */
 781
 782 static int
 783 qio_channel_socket_set_blocking(QIOChannel *ioc,
 784                                 bool enabled,
 785                                 Error **errp)
 786 {
 787     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 788
 789     if (enabled) {
 790         qemu_socket_set_block(sioc->fd);
 791     } else {
 792         qemu_socket_set_nonblock(sioc->fd);
 793     }
 794     return 0;
 795 }
 796
 797
 798 static void
 799 qio_channel_socket_set_delay(QIOChannel *ioc,
 800                              bool enabled)
 801 {
 802     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 803     int v = enabled ? 0 : 1;
 804
 805     setsockopt(sioc->fd,
 806                IPPROTO_TCP, TCP_NODELAY,
 807                &v, sizeof(v));
 808 }
 809
 810
 811 static void
 812 qio_channel_socket_set_cork(QIOChannel *ioc,
 813                             bool enabled)
 814 {
 815     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 816     int v = enabled ? 1 : 0;
 817
 818     socket_set_cork(sioc->fd, v);
 819 }
 820
 821
 822 static int
 823 qio_channel_socket_close(QIOChannel *ioc,
 824                          Error **errp)
 825 {
 826     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 827     int rc = 0;
 828     Error *err = NULL;
 829
 830     if (sioc->fd != -1) {
 831 #ifdef WIN32
 832         WSAEventSelect(sioc->fd, NULL, 0);
 833 #endif
 834         if (qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_LISTEN)) {
 835             socket_listen_cleanup(sioc->fd, errp);
 836         }
 837
 838         if (closesocket(sioc->fd) < 0) {
 839             sioc->fd = -1;
 840             error_setg_errno(&err, errno, "Unable to close socket");
 841             error_propagate(errp, err);
 842             return -1;
 843         }
 844         sioc->fd = -1;
 845     }
 846     return rc;
 847 }
 848
 849 static int
 850 qio_channel_socket_shutdown(QIOChannel *ioc,
 851                             QIOChannelShutdown how,
 852                             Error **errp)
 853 {
 854     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 855     int sockhow;
 856
 857     switch (how) {
 858     case QIO_CHANNEL_SHUTDOWN_READ:
 859         sockhow = SHUT_RD;
 860         break;
 861     case QIO_CHANNEL_SHUTDOWN_WRITE:
 862         sockhow = SHUT_WR;
 863         break;
 864     case QIO_CHANNEL_SHUTDOWN_BOTH:
 865     default:
 866         sockhow = SHUT_RDWR;
 867         break;
 868     }
 869
 870     if (shutdown(sioc->fd, sockhow) < 0) {
 871         error_setg_errno(errp, errno,
 872                          "Unable to shutdown socket");
 873         return -1;
 874     }
 875     return 0;
 876 }
 877
 878 static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
 879                                                   AioContext *ctx,
 880                                                   IOHandler *io_read,
 881                                                   IOHandler *io_write,
 882                                                   void *opaque)
 883 {
 884     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 885     aio_set_fd_handler(ctx, sioc->fd, false,
 886                        io_read, io_write, NULL, NULL, opaque);
 887 }
 888
 889 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
 890                                                 GIOCondition condition)
 891 {
 892     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 893     return qio_channel_create_socket_watch(ioc,
 894                                            sioc->fd,
 895                                            condition);
 896 }
 897
 898 static void qio_channel_socket_class_init(ObjectClass *klass,
 899                                           void *class_data G_GNUC_UNUSED)
 900 {
 901     QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
 902
 903     ioc_klass->io_writev = qio_channel_socket_writev;
 904     ioc_klass->io_readv = qio_channel_socket_readv;
 905     ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
 906     ioc_klass->io_close = qio_channel_socket_close;
 907     ioc_klass->io_shutdown = qio_channel_socket_shutdown;
 908     ioc_klass->io_set_cork = qio_channel_socket_set_cork;
 909     ioc_klass->io_set_delay = qio_channel_socket_set_delay;
 910     ioc_klass->io_create_watch = qio_channel_socket_create_watch;
 911     ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;
 912 #ifdef QEMU_MSG_ZEROCOPY
 913     ioc_klass->io_flush = qio_channel_socket_flush;
 914 #endif
 915 }
 916
 917 static const TypeInfo qio_channel_socket_info = {
 918     .parent = TYPE_QIO_CHANNEL,
 919     .name = TYPE_QIO_CHANNEL_SOCKET,
 920     .instance_size = sizeof(QIOChannelSocket),
 921     .instance_init = qio_channel_socket_init,
 922     .instance_finalize = qio_channel_socket_finalize,
 923     .class_init = qio_channel_socket_class_init,
 924 };
 925
 926 static void qio_channel_socket_register_types(void)
 927 {
 928     type_register_static(&qio_channel_socket_info);
 929 }
 930
 931 type_init(qio_channel_socket_register_types);