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