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
 552 off_t qio_channel_io_seek(QIOChannel *ioc,
 553                           off_t offset,
 554                           int whence,
 555                           Error **errp)
 556 {
 557     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 558
 559     if (!klass->io_seek) {
 560         error_setg(errp, "Channel does not support random access");
 561         return -1;
 562     }
 563
 564     return klass->io_seek(ioc, offset, whence, errp);
 565 }
 566
 567 int qio_channel_flush(QIOChannel *ioc,
 568                                 Error **errp)
 569 {
 570     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
 571
 572     if (!klass->io_flush ||
 573         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
 574         return 0;
 575     }
 576
 577     return klass->io_flush(ioc, errp);
 578 }
 579
 580
 581 static void qio_channel_restart_read(void *opaque)
 582 {
 583     QIOChannel *ioc = opaque;
 584     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 585
 586     if (!co) {
 587         return;
 588     }
 589
 590     /* Assert that aio_co_wake() reenters the coroutine directly */
 591     assert(qemu_get_current_aio_context() ==
 592            qemu_coroutine_get_aio_context(co));
 593     aio_co_wake(co);
 594 }
 595
 596 static void qio_channel_restart_write(void *opaque)
 597 {
 598     QIOChannel *ioc = opaque;
 599     Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
 600
 601     if (!co) {
 602         return;
 603     }
 604
 605     /* Assert that aio_co_wake() reenters the coroutine directly */
 606     assert(qemu_get_current_aio_context() ==
 607            qemu_coroutine_get_aio_context(co));
 608     aio_co_wake(co);
 609 }
 610
 611 static void coroutine_fn
 612 qio_channel_set_fd_handlers(QIOChannel *ioc, GIOCondition condition)
 613 {
 614     AioContext *ctx = ioc->follow_coroutine_ctx ?
 615         qemu_coroutine_get_aio_context(qemu_coroutine_self()) :
 616         iohandler_get_aio_context();
 617     AioContext *read_ctx = NULL;
 618     IOHandler *io_read = NULL;
 619     AioContext *write_ctx = NULL;
 620     IOHandler *io_write = NULL;
 621
 622     if (condition == G_IO_IN) {
 623         ioc->read_coroutine = qemu_coroutine_self();
 624         ioc->read_ctx = ctx;
 625         read_ctx = ctx;
 626         io_read = qio_channel_restart_read;
 627
 628         /*
 629          * Thread safety: if the other coroutine is set and its AioContext
 630          * matches ours, then there is mutual exclusion between read and write
 631          * because they share a single thread and it's safe to set both read
 632          * and write fd handlers here. If the AioContext does not match ours,
 633          * then both threads may run in parallel but there is no shared state
 634          * to worry about.
 635          */
 636         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
 637             write_ctx = ctx;
 638             io_write = qio_channel_restart_write;
 639         }
 640     } else if (condition == G_IO_OUT) {
 641         ioc->write_coroutine = qemu_coroutine_self();
 642         ioc->write_ctx = ctx;
 643         write_ctx = ctx;
 644         io_write = qio_channel_restart_write;
 645         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
 646             read_ctx = ctx;
 647             io_read = qio_channel_restart_read;
 648         }
 649     } else {
 650         abort();
 651     }
 652
 653     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
 654             write_ctx, io_write, ioc);
 655 }
 656
 657 static void coroutine_fn
 658 qio_channel_clear_fd_handlers(QIOChannel *ioc, GIOCondition condition)
 659 {
 660     AioContext *read_ctx = NULL;
 661     IOHandler *io_read = NULL;
 662     AioContext *write_ctx = NULL;
 663     IOHandler *io_write = NULL;
 664     AioContext *ctx;
 665
 666     if (condition == G_IO_IN) {
 667         ctx = ioc->read_ctx;
 668         read_ctx = ctx;
 669         io_read = NULL;
 670         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
 671             write_ctx = ctx;
 672             io_write = qio_channel_restart_write;
 673         }
 674     } else if (condition == G_IO_OUT) {
 675         ctx = ioc->write_ctx;
 676         write_ctx = ctx;
 677         io_write = NULL;
 678         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
 679             read_ctx = ctx;
 680             io_read = qio_channel_restart_read;
 681         }
 682     } else {
 683         abort();
 684     }
 685
 686     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
 687             write_ctx, io_write, ioc);
 688 }
 689
 690 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
 691                                     GIOCondition condition)
 692 {
 693     AioContext *ioc_ctx;
 694
 695     assert(qemu_in_coroutine());
 696     ioc_ctx = qemu_coroutine_get_aio_context(qemu_coroutine_self());
 697
 698     if (condition == G_IO_IN) {
 699         assert(!ioc->read_coroutine);
 700     } else if (condition == G_IO_OUT) {
 701         assert(!ioc->write_coroutine);
 702     } else {
 703         abort();
 704     }
 705     qio_channel_set_fd_handlers(ioc, condition);
 706     qemu_coroutine_yield();
 707     assert(in_aio_context_home_thread(ioc_ctx));
 708
 709     /* Allow interrupting the operation by reentering the coroutine other than
 710      * through the aio_fd_handlers. */
 711     if (condition == G_IO_IN) {
 712         assert(ioc->read_coroutine == NULL);
 713     } else if (condition == G_IO_OUT) {
 714         assert(ioc->write_coroutine == NULL);
 715     }
 716     qio_channel_clear_fd_handlers(ioc, condition);
 717 }
 718
 719 void qio_channel_wake_read(QIOChannel *ioc)
 720 {
 721     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
 722     if (co) {
 723         aio_co_wake(co);
 724     }
 725 }
 726
 727 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
 728                                           GIOCondition condition,
 729                                           gpointer opaque)
 730 {
 731     GMainLoop *loop = opaque;
 732
 733     g_main_loop_quit(loop);
 734     return FALSE;
 735 }
 736
 737
 738 void qio_channel_wait(QIOChannel *ioc,
 739                       GIOCondition condition)
 740 {
 741     GMainContext *ctxt = g_main_context_new();
 742     GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
 743     GSource *source;
 744
 745     source = qio_channel_create_watch(ioc, condition);
 746
 747     g_source_set_callback(source,
 748                           (GSourceFunc)qio_channel_wait_complete,
 749                           loop,
 750                           NULL);
 751
 752     g_source_attach(source, ctxt);
 753
 754     g_main_loop_run(loop);
 755
 756     g_source_unref(source);
 757     g_main_loop_unref(loop);
 758     g_main_context_unref(ctxt);
 759 }
 760
 761
 762 static void qio_channel_finalize(Object *obj)
 763 {
 764     QIOChannel *ioc = QIO_CHANNEL(obj);
 765
 766     /* Must not have coroutines in qio_channel_yield() */
 767     assert(!ioc->read_coroutine);
 768     assert(!ioc->write_coroutine);
 769
 770     g_free(ioc->name);
 771
 772 #ifdef _WIN32
 773     if (ioc->event) {
 774         CloseHandle(ioc->event);
 775     }
 776 #endif
 777 }
 778
 779 static const TypeInfo qio_channel_info = {
 780     .parent = TYPE_OBJECT,
 781     .name = TYPE_QIO_CHANNEL,
 782     .instance_size = sizeof(QIOChannel),
 783     .instance_finalize = qio_channel_finalize,
 784     .abstract = true,
 785     .class_size = sizeof(QIOChannelClass),
 786 };
 787
 788
 789 static void qio_channel_register_types(void)
 790 {
 791     type_register_static(&qio_channel_info);
 792 }
 793
 794
 795 type_init(qio_channel_register_types);