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