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 void qio_channel_set_follow_coroutine_ctx(QIOChannel *ioc, bool enabled)
 369 {
 370     ioc->follow_coroutine_ctx = enabled;
 371 }
 372
 373
 374 int qio_channel_close(QIOChannel *ioc,
 375                       Error **errp)
 376 {
 377     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 378     return klass->io_close(ioc, errp);
 379 }
 380
 381
 382 GSource *qio_channel_create_watch(QIOChannel *ioc,
 383                                   GIOCondition condition)
 384 {
 385     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 386     GSource *ret = klass->io_create_watch(ioc, condition);
 387
 388     if (ioc->name) {
 389         g_source_set_name(ret, ioc->name);
 390     }
 391
 392     return ret;
 393 }
 394
 395
 396 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
 397                                     AioContext *read_ctx,
 398                                     IOHandler *io_read,
 399                                     AioContext *write_ctx,
 400                                     IOHandler *io_write,
 401                                     void *opaque)
 402 {
 403     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 404
 405     klass->io_set_aio_fd_handler(ioc, read_ctx, io_read, write_ctx, io_write,
 406             opaque);
 407 }
 408
 409 guint qio_channel_add_watch_full(QIOChannel *ioc,
 410                                  GIOCondition condition,
 411                                  QIOChannelFunc func,
 412                                  gpointer user_data,
 413                                  GDestroyNotify notify,
 414                                  GMainContext *context)
 415 {
 416     GSource *source;
 417     guint id;
 418
 419     source = qio_channel_create_watch(ioc, condition);
 420
 421     g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
 422
 423     id = g_source_attach(source, context);
 424     g_source_unref(source);
 425
 426     return id;
 427 }
 428
 429 guint qio_channel_add_watch(QIOChannel *ioc,
 430                             GIOCondition condition,
 431                             QIOChannelFunc func,
 432                             gpointer user_data,
 433                             GDestroyNotify notify)
 434 {
 435     return qio_channel_add_watch_full(ioc, condition, func,
 436                                       user_data, notify, NULL);
 437 }
 438
 439 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
 440                                       GIOCondition condition,
 441                                       QIOChannelFunc func,
 442                                       gpointer user_data,
 443                                       GDestroyNotify notify,
 444                                       GMainContext *context)
 445 {
 446     GSource *source;
 447     guint id;
 448
 449     id = qio_channel_add_watch_full(ioc, condition, func,
 450                                     user_data, notify, context);
 451     source = g_main_context_find_source_by_id(context, id);
 452     g_source_ref(source);
 453     return source;
 454 }
 455
 456
 457 ssize_t qio_channel_pwritev(QIOChannel *ioc, const struct iovec *iov,
 458                             size_t niov, off_t offset, Error **errp)
 459 {
 460     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 461
 462     if (!klass->io_pwritev) {
 463         error_setg(errp, "Channel does not support pwritev");
 464         return -1;
 465     }
 466
 467     if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_SEEKABLE)) {
 468         error_setg_errno(errp, EINVAL, "Requested channel is not seekable");
 469         return -1;
 470     }
 471
 472     return klass->io_pwritev(ioc, iov, niov, offset, errp);
 473 }
 474
 475 ssize_t qio_channel_pwrite(QIOChannel *ioc, char *buf, size_t buflen,
 476                            off_t offset, Error **errp)
 477 {
 478     struct iovec iov = {
 479         .iov_base = buf,
 480         .iov_len = buflen
 481     };
 482
 483     return qio_channel_pwritev(ioc, &iov, 1, offset, errp);
 484 }
 485
 486 ssize_t qio_channel_preadv(QIOChannel *ioc, const struct iovec *iov,
 487                            size_t niov, off_t offset, Error **errp)
 488 {
 489     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 490
 491     if (!klass->io_preadv) {
 492         error_setg(errp, "Channel does not support preadv");
 493         return -1;
 494     }
 495
 496     if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_SEEKABLE)) {
 497         error_setg_errno(errp, EINVAL, "Requested channel is not seekable");
 498         return -1;
 499     }
 500
 501     return klass->io_preadv(ioc, iov, niov, offset, errp);
 502 }
 503
 504 ssize_t qio_channel_pread(QIOChannel *ioc, char *buf, size_t buflen,
 505                           off_t offset, Error **errp)
 506 {
 507     struct iovec iov = {
 508         .iov_base = buf,
 509         .iov_len = buflen
 510     };
 511
 512     return qio_channel_preadv(ioc, &iov, 1, offset, errp);
 513 }
 514
 515 int qio_channel_shutdown(QIOChannel *ioc,
 516                          QIOChannelShutdown how,
 517                          Error **errp)
 518 {
 519     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 520
 521     if (!klass->io_shutdown) {
 522         error_setg(errp, "Data path shutdown not supported");
 523         return -1;
 524     }
 525
 526     return klass->io_shutdown(ioc, how, errp);
 527 }
 528
 529
 530 void qio_channel_set_delay(QIOChannel *ioc,
 531                            bool enabled)
 532 {
 533     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 534
 535     if (klass->io_set_delay) {
 536         klass->io_set_delay(ioc, enabled);
 537     }
 538 }
 539
 540
 541 void qio_channel_set_cork(QIOChannel *ioc,
 542                           bool enabled)
 543 {
 544     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 545
 546     if (klass->io_set_cork) {
 547         klass->io_set_cork(ioc, enabled);
 548     }
 549 }
 550
 551 int qio_channel_get_peerpid(QIOChannel *ioc,
 552                              unsigned int *pid,
 553                              Error **errp)
 554 {
 555     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 556
 557     if (!klass->io_peerpid) {
 558         error_setg(errp, "Channel does not support peer pid");
 559         return -1;
 560     }
 561     klass->io_peerpid(ioc, pid, errp);
 562     return 0;
 563 }
 564
 565 off_t qio_channel_io_seek(QIOChannel *ioc,
 566                           off_t offset,
 567                           int whence,
 568                           Error **errp)
 569 {
 570     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 571
 572     if (!klass->io_seek) {
 573         error_setg(errp, "Channel does not support random access");
 574         return -1;
 575     }
 576
 577     return klass->io_seek(ioc, offset, whence, errp);
 578 }
 579
 580 int qio_channel_flush(QIOChannel *ioc,
 581                                 Error **errp)
 582 {
 583     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 584
 585     if (!klass->io_flush ||
 586         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
 587         return 0;
 588     }
 589
 590     return klass->io_flush(ioc, errp);
 591 }
 592
 593
 594 static void qio_channel_restart_read(void *opaque)
 595 {
 596     QIOChannel *ioc = opaque;
 597     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 598
 599     if (!co) {
 600         return;
 601     }
 602
 603     /* Assert that aio_co_wake() reenters the coroutine directly */
 604     assert(qemu_get_current_aio_context() ==
 605            qemu_coroutine_get_aio_context(co));
 606     aio_co_wake(co);
 607 }
 608
 609 static void qio_channel_restart_write(void *opaque)
 610 {
 611     QIOChannel *ioc = opaque;
 612     Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
 613
 614     if (!co) {
 615         return;
 616     }
 617
 618     /* Assert that aio_co_wake() reenters the coroutine directly */
 619     assert(qemu_get_current_aio_context() ==
 620            qemu_coroutine_get_aio_context(co));
 621     aio_co_wake(co);
 622 }
 623
 624 static void coroutine_fn
 625 qio_channel_set_fd_handlers(QIOChannel *ioc, GIOCondition condition)
 626 {
 627     AioContext *ctx = ioc->follow_coroutine_ctx ?
 628         qemu_coroutine_get_aio_context(qemu_coroutine_self()) :
 629         iohandler_get_aio_context();
 630     AioContext *read_ctx = NULL;
 631     IOHandler *io_read = NULL;
 632     AioContext *write_ctx = NULL;
 633     IOHandler *io_write = NULL;
 634
 635     if (condition == G_IO_IN) {
 636         ioc->read_coroutine = qemu_coroutine_self();
 637         ioc->read_ctx = ctx;
 638         read_ctx = ctx;
 639         io_read = qio_channel_restart_read;
 640
 641         /*
 642          * Thread safety: if the other coroutine is set and its AioContext
 643          * matches ours, then there is mutual exclusion between read and write
 644          * because they share a single thread and it's safe to set both read
 645          * and write fd handlers here. If the AioContext does not match ours,
 646          * then both threads may run in parallel but there is no shared state
 647          * to worry about.
 648          */
 649         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
 650             write_ctx = ctx;
 651             io_write = qio_channel_restart_write;
 652         }
 653     } else if (condition == G_IO_OUT) {
 654         ioc->write_coroutine = qemu_coroutine_self();
 655         ioc->write_ctx = ctx;
 656         write_ctx = ctx;
 657         io_write = qio_channel_restart_write;
 658         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
 659             read_ctx = ctx;
 660             io_read = qio_channel_restart_read;
 661         }
 662     } else {
 663         abort();
 664     }
 665
 666     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
 667             write_ctx, io_write, ioc);
 668 }
 669
 670 static void coroutine_fn
 671 qio_channel_clear_fd_handlers(QIOChannel *ioc, GIOCondition condition)
 672 {
 673     AioContext *read_ctx = NULL;
 674     IOHandler *io_read = NULL;
 675     AioContext *write_ctx = NULL;
 676     IOHandler *io_write = NULL;
 677     AioContext *ctx;
 678
 679     if (condition == G_IO_IN) {
 680         ctx = ioc->read_ctx;
 681         read_ctx = ctx;
 682         io_read = NULL;
 683         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
 684             write_ctx = ctx;
 685             io_write = qio_channel_restart_write;
 686         }
 687     } else if (condition == G_IO_OUT) {
 688         ctx = ioc->write_ctx;
 689         write_ctx = ctx;
 690         io_write = NULL;
 691         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
 692             read_ctx = ctx;
 693             io_read = qio_channel_restart_read;
 694         }
 695     } else {
 696         abort();
 697     }
 698
 699     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
 700             write_ctx, io_write, ioc);
 701 }
 702
 703 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
 704                                     GIOCondition condition)
 705 {
 706     AioContext *ioc_ctx;
 707
 708     assert(qemu_in_coroutine());
 709     ioc_ctx = qemu_coroutine_get_aio_context(qemu_coroutine_self());
 710
 711     if (condition == G_IO_IN) {
 712         assert(!ioc->read_coroutine);
 713     } else if (condition == G_IO_OUT) {
 714         assert(!ioc->write_coroutine);
 715     } else {
 716         abort();
 717     }
 718     qio_channel_set_fd_handlers(ioc, condition);
 719     qemu_coroutine_yield();
 720     assert(in_aio_context_home_thread(ioc_ctx));
 721
 722     /* Allow interrupting the operation by reentering the coroutine other than
 723      * through the aio_fd_handlers. */
 724     if (condition == G_IO_IN) {
 725         assert(ioc->read_coroutine == NULL);
 726     } else if (condition == G_IO_OUT) {
 727         assert(ioc->write_coroutine == NULL);
 728     }
 729     qio_channel_clear_fd_handlers(ioc, condition);
 730 }
 731
 732 void qio_channel_wake_read(QIOChannel *ioc)
 733 {
 734     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 735     if (co) {
 736         aio_co_wake(co);
 737     }
 738 }
 739
 740 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
 741                                           GIOCondition condition,
 742                                           gpointer opaque)
 743 {
 744     GMainLoop *loop = opaque;
 745
 746     g_main_loop_quit(loop);
 747     return FALSE;
 748 }
 749
 750
 751 void qio_channel_wait(QIOChannel *ioc,
 752                       GIOCondition condition)
 753 {
 754     GMainContext *ctxt = g_main_context_new();
 755     GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
 756     GSource *source;
 757
 758     source = qio_channel_create_watch(ioc, condition);
 759
 760     g_source_set_callback(source,
 761                           (GSourceFunc)qio_channel_wait_complete,
 762                           loop,
 763                           NULL);
 764
 765     g_source_attach(source, ctxt);
 766
 767     g_main_loop_run(loop);
 768
 769     g_source_unref(source);
 770     g_main_loop_unref(loop);
 771     g_main_context_unref(ctxt);
 772 }
 773
 774
 775 static void qio_channel_finalize(Object *obj)
 776 {
 777     QIOChannel *ioc = QIO_CHANNEL(obj);
 778
 779     /* Must not have coroutines in qio_channel_yield() */
 780     assert(!ioc->read_coroutine);
 781     assert(!ioc->write_coroutine);
 782
 783     g_free(ioc->name);
 784
 785 #ifdef _WIN32
 786     if (ioc->event) {
 787         CloseHandle(ioc->event);
 788     }
 789 #endif
 790 }
 791
 792 static const TypeInfo qio_channel_info = {
 793     .parent = TYPE_OBJECT,
 794     .name = TYPE_QIO_CHANNEL,
 795     .instance_size = sizeof(QIOChannel),
 796     .instance_finalize = qio_channel_finalize,
 797     .abstract = true,
 798     .class_size = sizeof(QIOChannelClass),
 799 };
 800
 801
 802 static void qio_channel_register_types(void)
 803 {
 804     type_register_static(&qio_channel_info);
 805 }
 806
 807
 808 type_init(qio_channel_register_types);