io/channel.c

   1 /*
   2  * QEMU I/O channels
   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
  21 #include "qemu/osdep.h"
  22 #include "block/aio-wait.h"
  23 #include "io/channel.h"
  24 #include "qapi/error.h"
  25 #include "qemu/main-loop.h"
  26 #include "qemu/module.h"
  27 #include "qemu/iov.h"
  28
  29 bool qio_channel_has_feature(QIOChannel *ioc,
  30                              QIOChannelFeature feature)
  31 {
  32     return ioc->features & (1 << feature);
  33 }
  34
  35
  36 void qio_channel_set_feature(QIOChannel *ioc,
  37                              QIOChannelFeature feature)
  38 {
  39     ioc->features |= (1 << feature);
  40 }
  41
  42
  43 void qio_channel_set_name(QIOChannel *ioc,
  44                           const char *name)
  45 {
  46     g_free(ioc->name);
  47     ioc->name = g_strdup(name);
  48 }
  49
  50
  51 ssize_t qio_channel_readv_full(QIOChannel *ioc,
  52                                const struct iovec *iov,
  53                                size_t niov,
  54                                int **fds,
  55                                size_t *nfds,
  56                                int flags,
  57                                Error **errp)
  58 {
  59     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
  60
  61     if ((fds || nfds) &&
  62         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
  63         error_setg_errno(errp, EINVAL,
  64                          "Channel does not support file descriptor passing");
  65         return -1;
  66     }
  67
  68     if ((flags & QIO_CHANNEL_READ_FLAG_MSG_PEEK) &&
  69         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_READ_MSG_PEEK)) {
  70         error_setg_errno(errp, EINVAL,
  71                          "Channel does not support peek read");
  72         return -1;
  73     }
  74
  75     return klass->io_readv(ioc, iov, niov, fds, nfds, flags, errp);
  76 }
  77
  78
  79 ssize_t qio_channel_writev_full(QIOChannel *ioc,
  80                                 const struct iovec *iov,
  81                                 size_t niov,
  82                                 int *fds,
  83                                 size_t nfds,
  84                                 int flags,
  85                                 Error **errp)
  86 {
  87     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
  88
  89     if (fds || nfds) {
  90         if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
  91             error_setg_errno(errp, EINVAL,
  92                              "Channel does not support file descriptor passing");
  93             return -1;
  94         }
  95         if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
  96             error_setg_errno(errp, EINVAL,
  97                              "Zero Copy does not support file descriptor passing");
  98             return -1;
  99         }
 100     }
 101
 102     if ((flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) &&
 103         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
 104         error_setg_errno(errp, EINVAL,
 105                          "Requested Zero Copy feature is not available");
 106         return -1;
 107     }
 108
 109     return klass->io_writev(ioc, iov, niov, fds, nfds, flags, errp);
 110 }
 111
 112
 113 int coroutine_mixed_fn qio_channel_readv_all_eof(QIOChannel *ioc,
 114                                                  const struct iovec *iov,
 115                                                  size_t niov,
 116                                                  Error **errp)
 117 {
 118     return qio_channel_readv_full_all_eof(ioc, iov, niov, NULL, NULL, errp);
 119 }
 120
 121 int coroutine_mixed_fn qio_channel_readv_all(QIOChannel *ioc,
 122                                              const struct iovec *iov,
 123                                              size_t niov,
 124                                              Error **errp)
 125 {
 126     return qio_channel_readv_full_all(ioc, iov, niov, NULL, NULL, errp);
 127 }
 128
 129 int coroutine_mixed_fn qio_channel_readv_full_all_eof(QIOChannel *ioc,
 130                                                       const struct iovec *iov,
 131                                                       size_t niov,
 132                                                       int **fds, size_t *nfds,
 133                                                       Error **errp)
 134 {
 135     int ret = -1;
 136     struct iovec *local_iov = g_new(struct iovec, niov);
 137     struct iovec *local_iov_head = local_iov;
 138     unsigned int nlocal_iov = niov;
 139     int **local_fds = fds;
 140     size_t *local_nfds = nfds;
 141     bool partial = false;
 142
 143     if (nfds) {
 144         *nfds = 0;
 145     }
 146
 147     if (fds) {
 148         *fds = NULL;
 149     }
 150
 151     nlocal_iov = iov_copy(local_iov, nlocal_iov,
 152                           iov, niov,
 153                           0, iov_size(iov, niov));
 154
 155     while ((nlocal_iov > 0) || local_fds) {
 156         ssize_t len;
 157         len = qio_channel_readv_full(ioc, local_iov, nlocal_iov, local_fds,
 158                                      local_nfds, 0, errp);
 159         if (len == QIO_CHANNEL_ERR_BLOCK) {
 160             if (qemu_in_coroutine()) {
 161                 qio_channel_yield(ioc, G_IO_IN);
 162             } else {
 163                 qio_channel_wait(ioc, G_IO_IN);
 164             }
 165             continue;
 166         }
 167
 168         if (len == 0) {
 169             if (local_nfds && *local_nfds) {
 170                 /*
 171                  * Got some FDs, but no data yet. This isn't an EOF
 172                  * scenario (yet), so carry on to try to read data
 173                  * on next loop iteration
 174                  */
 175                 goto next_iter;
 176             } else if (!partial) {
 177                 /* No fds and no data - EOF before any data read */
 178                 ret = 0;
 179                 goto cleanup;
 180             } else {
 181                 len = -1;
 182                 error_setg(errp,
 183                            "Unexpected end-of-file before all data were read");
 184                 /* Fallthrough into len < 0 handling */
 185             }
 186         }
 187
 188         if (len < 0) {
 189             /* Close any FDs we previously received */
 190             if (nfds && fds) {
 191                 size_t i;
 192                 for (i = 0; i < (*nfds); i++) {
 193                     close((*fds)[i]);
 194                 }
 195                 g_free(*fds);
 196                 *fds = NULL;
 197                 *nfds = 0;
 198             }
 199             goto cleanup;
 200         }
 201
 202         if (nlocal_iov) {
 203             iov_discard_front(&local_iov, &nlocal_iov, len);
 204         }
 205
 206 next_iter:
 207         partial = true;
 208         local_fds = NULL;
 209         local_nfds = NULL;
 210     }
 211
 212     ret = 1;
 213
 214  cleanup:
 215     g_free(local_iov_head);
 216     return ret;
 217 }
 218
 219 int coroutine_mixed_fn qio_channel_readv_full_all(QIOChannel *ioc,
 220                                                   const struct iovec *iov,
 221                                                   size_t niov,
 222                                                   int **fds, size_t *nfds,
 223                                                   Error **errp)
 224 {
 225     int ret = qio_channel_readv_full_all_eof(ioc, iov, niov, fds, nfds, errp);
 226
 227     if (ret == 0) {
 228         error_setg(errp, "Unexpected end-of-file before all data were read");
 229         return -1;
 230     }
 231     if (ret == 1) {
 232         return 0;
 233     }
 234
 235     return ret;
 236 }
 237
 238 int coroutine_mixed_fn qio_channel_writev_all(QIOChannel *ioc,
 239                                               const struct iovec *iov,
 240                                               size_t niov,
 241                                               Error **errp)
 242 {
 243     return qio_channel_writev_full_all(ioc, iov, niov, NULL, 0, 0, errp);
 244 }
 245
 246 int coroutine_mixed_fn qio_channel_writev_full_all(QIOChannel *ioc,
 247                                                    const struct iovec *iov,
 248                                                    size_t niov,
 249                                                    int *fds, size_t nfds,
 250                                                    int flags, Error **errp)
 251 {
 252     int ret = -1;
 253     struct iovec *local_iov = g_new(struct iovec, niov);
 254     struct iovec *local_iov_head = local_iov;
 255     unsigned int nlocal_iov = niov;
 256
 257     nlocal_iov = iov_copy(local_iov, nlocal_iov,
 258                           iov, niov,
 259                           0, iov_size(iov, niov));
 260
 261     while (nlocal_iov > 0) {
 262         ssize_t len;
 263
 264         len = qio_channel_writev_full(ioc, local_iov, nlocal_iov, fds,
 265                                             nfds, flags, errp);
 266
 267         if (len == QIO_CHANNEL_ERR_BLOCK) {
 268             if (qemu_in_coroutine()) {
 269                 qio_channel_yield(ioc, G_IO_OUT);
 270             } else {
 271                 qio_channel_wait(ioc, G_IO_OUT);
 272             }
 273             continue;
 274         }
 275         if (len < 0) {
 276             goto cleanup;
 277         }
 278
 279         iov_discard_front(&local_iov, &nlocal_iov, len);
 280
 281         fds = NULL;
 282         nfds = 0;
 283     }
 284
 285     ret = 0;
 286  cleanup:
 287     g_free(local_iov_head);
 288     return ret;
 289 }
 290
 291 ssize_t qio_channel_readv(QIOChannel *ioc,
 292                           const struct iovec *iov,
 293                           size_t niov,
 294                           Error **errp)
 295 {
 296     return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, 0, errp);
 297 }
 298
 299
 300 ssize_t qio_channel_writev(QIOChannel *ioc,
 301                            const struct iovec *iov,
 302                            size_t niov,
 303                            Error **errp)
 304 {
 305     return qio_channel_writev_full(ioc, iov, niov, NULL, 0, 0, errp);
 306 }
 307
 308
 309 ssize_t qio_channel_read(QIOChannel *ioc,
 310                          char *buf,
 311                          size_t buflen,
 312                          Error **errp)
 313 {
 314     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
 315     return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, 0, errp);
 316 }
 317
 318
 319 ssize_t qio_channel_write(QIOChannel *ioc,
 320                           const char *buf,
 321                           size_t buflen,
 322                           Error **errp)
 323 {
 324     struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
 325     return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, 0, errp);
 326 }
 327
 328
 329 int coroutine_mixed_fn qio_channel_read_all_eof(QIOChannel *ioc,
 330                                                 char *buf,
 331                                                 size_t buflen,
 332                                                 Error **errp)
 333 {
 334     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
 335     return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
 336 }
 337
 338
 339 int coroutine_mixed_fn qio_channel_read_all(QIOChannel *ioc,
 340                                             char *buf,
 341                                             size_t buflen,
 342                                             Error **errp)
 343 {
 344     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
 345     return qio_channel_readv_all(ioc, &iov, 1, errp);
 346 }
 347
 348
 349 int coroutine_mixed_fn qio_channel_write_all(QIOChannel *ioc,
 350                                              const char *buf,
 351                                              size_t buflen,
 352                                              Error **errp)
 353 {
 354     struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
 355     return qio_channel_writev_all(ioc, &iov, 1, errp);
 356 }
 357
 358
 359 int qio_channel_set_blocking(QIOChannel *ioc,
 360                               bool enabled,
 361                               Error **errp)
 362 {
 363     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 364     return klass->io_set_blocking(ioc, enabled, errp);
 365 }
 366
 367
 368 int qio_channel_close(QIOChannel *ioc,
 369                       Error **errp)
 370 {
 371     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 372     return klass->io_close(ioc, errp);
 373 }
 374
 375
 376 GSource *qio_channel_create_watch(QIOChannel *ioc,
 377                                   GIOCondition condition)
 378 {
 379     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 380     GSource *ret = klass->io_create_watch(ioc, condition);
 381
 382     if (ioc->name) {
 383         g_source_set_name(ret, ioc->name);
 384     }
 385
 386     return ret;
 387 }
 388
 389
 390 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
 391                                     AioContext *ctx,
 392                                     IOHandler *io_read,
 393                                     IOHandler *io_write,
 394                                     void *opaque)
 395 {
 396     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 397
 398     klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
 399 }
 400
 401 guint qio_channel_add_watch_full(QIOChannel *ioc,
 402                                  GIOCondition condition,
 403                                  QIOChannelFunc func,
 404                                  gpointer user_data,
 405                                  GDestroyNotify notify,
 406                                  GMainContext *context)
 407 {
 408     GSource *source;
 409     guint id;
 410
 411     source = qio_channel_create_watch(ioc, condition);
 412
 413     g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
 414
 415     id = g_source_attach(source, context);
 416     g_source_unref(source);
 417
 418     return id;
 419 }
 420
 421 guint qio_channel_add_watch(QIOChannel *ioc,
 422                             GIOCondition condition,
 423                             QIOChannelFunc func,
 424                             gpointer user_data,
 425                             GDestroyNotify notify)
 426 {
 427     return qio_channel_add_watch_full(ioc, condition, func,
 428                                       user_data, notify, NULL);
 429 }
 430
 431 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
 432                                       GIOCondition condition,
 433                                       QIOChannelFunc func,
 434                                       gpointer user_data,
 435                                       GDestroyNotify notify,
 436                                       GMainContext *context)
 437 {
 438     GSource *source;
 439     guint id;
 440
 441     id = qio_channel_add_watch_full(ioc, condition, func,
 442                                     user_data, notify, context);
 443     source = g_main_context_find_source_by_id(context, id);
 444     g_source_ref(source);
 445     return source;
 446 }
 447
 448
 449 int qio_channel_shutdown(QIOChannel *ioc,
 450                          QIOChannelShutdown how,
 451                          Error **errp)
 452 {
 453     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 454
 455     if (!klass->io_shutdown) {
 456         error_setg(errp, "Data path shutdown not supported");
 457         return -1;
 458     }
 459
 460     return klass->io_shutdown(ioc, how, errp);
 461 }
 462
 463
 464 void qio_channel_set_delay(QIOChannel *ioc,
 465                            bool enabled)
 466 {
 467     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 468
 469     if (klass->io_set_delay) {
 470         klass->io_set_delay(ioc, enabled);
 471     }
 472 }
 473
 474
 475 void qio_channel_set_cork(QIOChannel *ioc,
 476                           bool enabled)
 477 {
 478     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 479
 480     if (klass->io_set_cork) {
 481         klass->io_set_cork(ioc, enabled);
 482     }
 483 }
 484
 485
 486 off_t qio_channel_io_seek(QIOChannel *ioc,
 487                           off_t offset,
 488                           int whence,
 489                           Error **errp)
 490 {
 491     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 492
 493     if (!klass->io_seek) {
 494         error_setg(errp, "Channel does not support random access");
 495         return -1;
 496     }
 497
 498     return klass->io_seek(ioc, offset, whence, errp);
 499 }
 500
 501 int qio_channel_flush(QIOChannel *ioc,
 502                                 Error **errp)
 503 {
 504     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 505
 506     if (!klass->io_flush ||
 507         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
 508         return 0;
 509     }
 510
 511     return klass->io_flush(ioc, errp);
 512 }
 513
 514
 515 static void qio_channel_restart_read(void *opaque)
 516 {
 517     QIOChannel *ioc = opaque;
 518     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 519
 520     if (!co) {
 521         return;
 522     }
 523
 524     /* Assert that aio_co_wake() reenters the coroutine directly */
 525     assert(qemu_get_current_aio_context() ==
 526            qemu_coroutine_get_aio_context(co));
 527     aio_co_wake(co);
 528 }
 529
 530 static void qio_channel_restart_write(void *opaque)
 531 {
 532     QIOChannel *ioc = opaque;
 533     Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
 534
 535     if (!co) {
 536         return;
 537     }
 538
 539     /* Assert that aio_co_wake() reenters the coroutine directly */
 540     assert(qemu_get_current_aio_context() ==
 541            qemu_coroutine_get_aio_context(co));
 542     aio_co_wake(co);
 543 }
 544
 545 static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
 546 {
 547     IOHandler *rd_handler = NULL, *wr_handler = NULL;
 548     AioContext *ctx;
 549
 550     if (ioc->read_coroutine) {
 551         rd_handler = qio_channel_restart_read;
 552     }
 553     if (ioc->write_coroutine) {
 554         wr_handler = qio_channel_restart_write;
 555     }
 556
 557     ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
 558     qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
 559 }
 560
 561 void qio_channel_attach_aio_context(QIOChannel *ioc,
 562                                     AioContext *ctx)
 563 {
 564     assert(!ioc->read_coroutine);
 565     assert(!ioc->write_coroutine);
 566     ioc->ctx = ctx;
 567 }
 568
 569 void qio_channel_detach_aio_context(QIOChannel *ioc)
 570 {
 571     ioc->read_coroutine = NULL;
 572     ioc->write_coroutine = NULL;
 573     qio_channel_set_aio_fd_handlers(ioc);
 574     ioc->ctx = NULL;
 575 }
 576
 577 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
 578                                     GIOCondition condition)
 579 {
 580     AioContext *ioc_ctx = ioc->ctx ?: qemu_get_aio_context();
 581
 582     assert(qemu_in_coroutine());
 583     assert(in_aio_context_home_thread(ioc_ctx));
 584
 585     if (condition == G_IO_IN) {
 586         assert(!ioc->read_coroutine);
 587         ioc->read_coroutine = qemu_coroutine_self();
 588     } else if (condition == G_IO_OUT) {
 589         assert(!ioc->write_coroutine);
 590         ioc->write_coroutine = qemu_coroutine_self();
 591     } else {
 592         abort();
 593     }
 594     qio_channel_set_aio_fd_handlers(ioc);
 595     qemu_coroutine_yield();
 596     assert(in_aio_context_home_thread(ioc_ctx));
 597
 598     /* Allow interrupting the operation by reentering the coroutine other than
 599      * through the aio_fd_handlers. */
 600     if (condition == G_IO_IN) {
 601         assert(ioc->read_coroutine == NULL);
 602         qio_channel_set_aio_fd_handlers(ioc);
 603     } else if (condition == G_IO_OUT) {
 604         assert(ioc->write_coroutine == NULL);
 605         qio_channel_set_aio_fd_handlers(ioc);
 606     }
 607 }
 608
 609 void qio_channel_wake_read(QIOChannel *ioc)
 610 {
 611     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 612     if (co) {
 613         aio_co_wake(co);
 614     }
 615 }
 616
 617 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
 618                                           GIOCondition condition,
 619                                           gpointer opaque)
 620 {
 621     GMainLoop *loop = opaque;
 622
 623     g_main_loop_quit(loop);
 624     return FALSE;
 625 }
 626
 627
 628 void qio_channel_wait(QIOChannel *ioc,
 629                       GIOCondition condition)
 630 {
 631     GMainContext *ctxt = g_main_context_new();
 632     GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
 633     GSource *source;
 634
 635     source = qio_channel_create_watch(ioc, condition);
 636
 637     g_source_set_callback(source,
 638                           (GSourceFunc)qio_channel_wait_complete,
 639                           loop,
 640                           NULL);
 641
 642     g_source_attach(source, ctxt);
 643
 644     g_main_loop_run(loop);
 645
 646     g_source_unref(source);
 647     g_main_loop_unref(loop);
 648     g_main_context_unref(ctxt);
 649 }
 650
 651
 652 static void qio_channel_finalize(Object *obj)
 653 {
 654     QIOChannel *ioc = QIO_CHANNEL(obj);
 655
 656     g_free(ioc->name);
 657
 658 #ifdef _WIN32
 659     if (ioc->event) {
 660         CloseHandle(ioc->event);
 661     }
 662 #endif
 663 }
 664
 665 static const TypeInfo qio_channel_info = {
 666     .parent = TYPE_OBJECT,
 667     .name = TYPE_QIO_CHANNEL,
 668     .instance_size = sizeof(QIOChannel),
 669     .instance_finalize = qio_channel_finalize,
 670     .abstract = true,
 671     .class_size = sizeof(QIOChannelClass),
 672 };
 673
 674
 675 static void qio_channel_register_types(void)
 676 {
 677     type_register_static(&qio_channel_info);
 678 }
 679
 680
 681 type_init(qio_channel_register_types);