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 #ifdef QEMU_MSG_ZEROCOPY
 582     if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
 583         sflags = MSG_ZEROCOPY;
 584     }
 585 #endif
 586
 587  retry:
 588     ret = sendmsg(sioc->fd, &msg, sflags);
 589     if (ret <= 0) {
 590         switch (errno) {
 591         case EAGAIN:
 592             return QIO_CHANNEL_ERR_BLOCK;
 593         case EINTR:
 594             goto retry;
 595 #ifdef QEMU_MSG_ZEROCOPY
 596         case ENOBUFS:
 597             if (sflags & MSG_ZEROCOPY) {
 598                 error_setg_errno(errp, errno,
 599                                  "Process can't lock enough memory for using MSG_ZEROCOPY");
 600                 return -1;
 601             }
 602             break;
 603 #endif
 604         }
 605
 606         error_setg_errno(errp, errno,
 607                          "Unable to write to socket");
 608         return -1;
 609     }
 610     return ret;
 611 }
 612 #else /* WIN32 */
 613 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
 614                                         const struct iovec *iov,
 615                                         size_t niov,
 616                                         int **fds,
 617                                         size_t *nfds,
 618                                         Error **errp)
 619 {
 620     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 621     ssize_t done = 0;
 622     ssize_t i;
 623
 624     for (i = 0; i < niov; i++) {
 625         ssize_t ret;
 626     retry:
 627         ret = recv(sioc->fd,
 628                    iov[i].iov_base,
 629                    iov[i].iov_len,
 630                    0);
 631         if (ret < 0) {
 632             if (errno == EAGAIN) {
 633                 if (done) {
 634                     return done;
 635                 } else {
 636                     return QIO_CHANNEL_ERR_BLOCK;
 637                 }
 638             } else if (errno == EINTR) {
 639                 goto retry;
 640             } else {
 641                 error_setg_errno(errp, errno,
 642                                  "Unable to read from socket");
 643                 return -1;
 644             }
 645         }
 646         done += ret;
 647         if (ret < iov[i].iov_len) {
 648             return done;
 649         }
 650     }
 651
 652     return done;
 653 }
 654
 655 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
 656                                          const struct iovec *iov,
 657                                          size_t niov,
 658                                          int *fds,
 659                                          size_t nfds,
 660                                          int flags,
 661                                          Error **errp)
 662 {
 663     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 664     ssize_t done = 0;
 665     ssize_t i;
 666
 667     for (i = 0; i < niov; i++) {
 668         ssize_t ret;
 669     retry:
 670         ret = send(sioc->fd,
 671                    iov[i].iov_base,
 672                    iov[i].iov_len,
 673                    0);
 674         if (ret < 0) {
 675             if (errno == EAGAIN) {
 676                 if (done) {
 677                     return done;
 678                 } else {
 679                     return QIO_CHANNEL_ERR_BLOCK;
 680                 }
 681             } else if (errno == EINTR) {
 682                 goto retry;
 683             } else {
 684                 error_setg_errno(errp, errno,
 685                                  "Unable to write to socket");
 686                 return -1;
 687             }
 688         }
 689         done += ret;
 690         if (ret < iov[i].iov_len) {
 691             return done;
 692         }
 693     }
 694
 695     return done;
 696 }
 697 #endif /* WIN32 */
 698
 699
 700 #ifdef QEMU_MSG_ZEROCOPY
 701 static int qio_channel_socket_flush(QIOChannel *ioc,
 702                                     Error **errp)
 703 {
 704     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 705     struct msghdr msg = {};
 706     struct sock_extended_err *serr;
 707     struct cmsghdr *cm;
 708     char control[CMSG_SPACE(sizeof(*serr))];
 709     int received;
 710     int ret = 1;
 711
 712     msg.msg_control = control;
 713     msg.msg_controllen = sizeof(control);
 714     memset(control, 0, sizeof(control));
 715
 716     while (sioc->zero_copy_sent < sioc->zero_copy_queued) {
 717         received = recvmsg(sioc->fd, &msg, MSG_ERRQUEUE);
 718         if (received < 0) {
 719             switch (errno) {
 720             case EAGAIN:
 721                 /* Nothing on errqueue, wait until something is available */
 722                 qio_channel_wait(ioc, G_IO_ERR);
 723                 continue;
 724             case EINTR:
 725                 continue;
 726             default:
 727                 error_setg_errno(errp, errno,
 728                                  "Unable to read errqueue");
 729                 return -1;
 730             }
 731         }
 732
 733         cm = CMSG_FIRSTHDR(&msg);
 734         if (cm->cmsg_level != SOL_IP &&
 735             cm->cmsg_type != IP_RECVERR) {
 736             error_setg_errno(errp, EPROTOTYPE,
 737                              "Wrong cmsg in errqueue");
 738             return -1;
 739         }
 740
 741         serr = (void *) CMSG_DATA(cm);
 742         if (serr->ee_errno != SO_EE_ORIGIN_NONE) {
 743             error_setg_errno(errp, serr->ee_errno,
 744                              "Error on socket");
 745             return -1;
 746         }
 747         if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY) {
 748             error_setg_errno(errp, serr->ee_origin,
 749                              "Error not from zero copy");
 750             return -1;
 751         }
 752
 753         /* No errors, count successfully finished sendmsg()*/
 754         sioc->zero_copy_sent += serr->ee_data - serr->ee_info + 1;
 755
 756         /* If any sendmsg() succeeded using zero copy, return 0 at the end */
 757         if (serr->ee_code != SO_EE_CODE_ZEROCOPY_COPIED) {
 758             ret = 0;
 759         }
 760     }
 761
 762     return ret;
 763 }
 764
 765 #endif /* QEMU_MSG_ZEROCOPY */
 766
 767 static int
 768 qio_channel_socket_set_blocking(QIOChannel *ioc,
 769                                 bool enabled,
 770                                 Error **errp)
 771 {
 772     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 773
 774     if (enabled) {
 775         qemu_socket_set_block(sioc->fd);
 776     } else {
 777         qemu_socket_set_nonblock(sioc->fd);
 778     }
 779     return 0;
 780 }
 781
 782
 783 static void
 784 qio_channel_socket_set_delay(QIOChannel *ioc,
 785                              bool enabled)
 786 {
 787     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 788     int v = enabled ? 0 : 1;
 789
 790     setsockopt(sioc->fd,
 791                IPPROTO_TCP, TCP_NODELAY,
 792                &v, sizeof(v));
 793 }
 794
 795
 796 static void
 797 qio_channel_socket_set_cork(QIOChannel *ioc,
 798                             bool enabled)
 799 {
 800     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 801     int v = enabled ? 1 : 0;
 802
 803     socket_set_cork(sioc->fd, v);
 804 }
 805
 806
 807 static int
 808 qio_channel_socket_close(QIOChannel *ioc,
 809                          Error **errp)
 810 {
 811     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 812     int rc = 0;
 813     Error *err = NULL;
 814
 815     if (sioc->fd != -1) {
 816 #ifdef WIN32
 817         WSAEventSelect(sioc->fd, NULL, 0);
 818 #endif
 819         if (qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_LISTEN)) {
 820             socket_listen_cleanup(sioc->fd, errp);
 821         }
 822
 823         if (closesocket(sioc->fd) < 0) {
 824             sioc->fd = -1;
 825             error_setg_errno(&err, errno, "Unable to close socket");
 826             error_propagate(errp, err);
 827             return -1;
 828         }
 829         sioc->fd = -1;
 830     }
 831     return rc;
 832 }
 833
 834 static int
 835 qio_channel_socket_shutdown(QIOChannel *ioc,
 836                             QIOChannelShutdown how,
 837                             Error **errp)
 838 {
 839     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 840     int sockhow;
 841
 842     switch (how) {
 843     case QIO_CHANNEL_SHUTDOWN_READ:
 844         sockhow = SHUT_RD;
 845         break;
 846     case QIO_CHANNEL_SHUTDOWN_WRITE:
 847         sockhow = SHUT_WR;
 848         break;
 849     case QIO_CHANNEL_SHUTDOWN_BOTH:
 850     default:
 851         sockhow = SHUT_RDWR;
 852         break;
 853     }
 854
 855     if (shutdown(sioc->fd, sockhow) < 0) {
 856         error_setg_errno(errp, errno,
 857                          "Unable to shutdown socket");
 858         return -1;
 859     }
 860     return 0;
 861 }
 862
 863 static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
 864                                                   AioContext *ctx,
 865                                                   IOHandler *io_read,
 866                                                   IOHandler *io_write,
 867                                                   void *opaque)
 868 {
 869     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 870     aio_set_fd_handler(ctx, sioc->fd, false,
 871                        io_read, io_write, NULL, NULL, opaque);
 872 }
 873
 874 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
 875                                                 GIOCondition condition)
 876 {
 877     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
 878     return qio_channel_create_socket_watch(ioc,
 879                                            sioc->fd,
 880                                            condition);
 881 }
 882
 883 static void qio_channel_socket_class_init(ObjectClass *klass,
 884                                           void *class_data G_GNUC_UNUSED)
 885 {
 886     QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
 887
 888     ioc_klass->io_writev = qio_channel_socket_writev;
 889     ioc_klass->io_readv = qio_channel_socket_readv;
 890     ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
 891     ioc_klass->io_close = qio_channel_socket_close;
 892     ioc_klass->io_shutdown = qio_channel_socket_shutdown;
 893     ioc_klass->io_set_cork = qio_channel_socket_set_cork;
 894     ioc_klass->io_set_delay = qio_channel_socket_set_delay;
 895     ioc_klass->io_create_watch = qio_channel_socket_create_watch;
 896     ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;
 897 #ifdef QEMU_MSG_ZEROCOPY
 898     ioc_klass->io_flush = qio_channel_socket_flush;
 899 #endif
 900 }
 901
 902 static const TypeInfo qio_channel_socket_info = {
 903     .parent = TYPE_QIO_CHANNEL,
 904     .name = TYPE_QIO_CHANNEL_SOCKET,
 905     .instance_size = sizeof(QIOChannelSocket),
 906     .instance_init = qio_channel_socket_init,
 907     .instance_finalize = qio_channel_socket_finalize,
 908     .class_init = qio_channel_socket_class_init,
 909 };
 910
 911 static void qio_channel_socket_register_types(void)
 912 {
 913     type_register_static(&qio_channel_socket_info);
 914 }
 915
 916 type_init(qio_channel_socket_register_types);