4 * Copyright (c) 2015 Red Hat, Inc.
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.
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.
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/>.
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"
29 bool qio_channel_has_feature(QIOChannel *ioc,
30 QIOChannelFeature feature)
32 return ioc->features & (1 << feature);
36 void qio_channel_set_feature(QIOChannel *ioc,
37 QIOChannelFeature feature)
39 ioc->features |= (1 << feature);
43 void qio_channel_set_name(QIOChannel *ioc,
47 ioc->name = g_strdup(name);
51 ssize_t qio_channel_readv_full(QIOChannel *ioc,
52 const struct iovec *iov,
59 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
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");
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");
75 return klass->io_readv(ioc, iov, niov, fds, nfds, flags, errp);
79 ssize_t qio_channel_writev_full(QIOChannel *ioc,
80 const struct iovec *iov,
87 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
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");
95 if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
96 error_setg_errno(errp, EINVAL,
97 "Zero Copy does not support file descriptor passing");
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");
109 return klass->io_writev(ioc, iov, niov, fds, nfds, flags, errp);
113 int coroutine_mixed_fn qio_channel_readv_all_eof(QIOChannel *ioc,
114 const struct iovec *iov,
118 return qio_channel_readv_full_all_eof(ioc, iov, niov, NULL, NULL, errp);
121 int coroutine_mixed_fn qio_channel_readv_all(QIOChannel *ioc,
122 const struct iovec *iov,
126 return qio_channel_readv_full_all(ioc, iov, niov, NULL, NULL, errp);
129 int coroutine_mixed_fn qio_channel_readv_full_all_eof(QIOChannel *ioc,
130 const struct iovec *iov,
132 int **fds, size_t *nfds,
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;
151 nlocal_iov = iov_copy(local_iov, nlocal_iov,
153 0, iov_size(iov, niov));
155 while ((nlocal_iov > 0) || local_fds) {
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);
163 qio_channel_wait(ioc, G_IO_IN);
169 if (local_nfds && *local_nfds) {
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
176 } else if (!partial) {
177 /* No fds and no data - EOF before any data read */
183 "Unexpected end-of-file before all data were read");
184 /* Fallthrough into len < 0 handling */
189 /* Close any FDs we previously received */
192 for (i = 0; i < (*nfds); i++) {
203 iov_discard_front(&local_iov, &nlocal_iov, len);
215 g_free(local_iov_head);
219 int coroutine_mixed_fn qio_channel_readv_full_all(QIOChannel *ioc,
220 const struct iovec *iov,
222 int **fds, size_t *nfds,
225 int ret = qio_channel_readv_full_all_eof(ioc, iov, niov, fds, nfds, errp);
228 error_setg(errp, "Unexpected end-of-file before all data were read");
238 int coroutine_mixed_fn qio_channel_writev_all(QIOChannel *ioc,
239 const struct iovec *iov,
243 return qio_channel_writev_full_all(ioc, iov, niov, NULL, 0, 0, errp);
246 int coroutine_mixed_fn qio_channel_writev_full_all(QIOChannel *ioc,
247 const struct iovec *iov,
249 int *fds, size_t nfds,
250 int flags, Error **errp)
253 struct iovec *local_iov = g_new(struct iovec, niov);
254 struct iovec *local_iov_head = local_iov;
255 unsigned int nlocal_iov = niov;
257 nlocal_iov = iov_copy(local_iov, nlocal_iov,
259 0, iov_size(iov, niov));
261 while (nlocal_iov > 0) {
264 len = qio_channel_writev_full(ioc, local_iov, nlocal_iov, fds,
267 if (len == QIO_CHANNEL_ERR_BLOCK) {
268 if (qemu_in_coroutine()) {
269 qio_channel_yield(ioc, G_IO_OUT);
271 qio_channel_wait(ioc, G_IO_OUT);
279 iov_discard_front(&local_iov, &nlocal_iov, len);
287 g_free(local_iov_head);
291 ssize_t qio_channel_readv(QIOChannel *ioc,
292 const struct iovec *iov,
296 return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, 0, errp);
300 ssize_t qio_channel_writev(QIOChannel *ioc,
301 const struct iovec *iov,
305 return qio_channel_writev_full(ioc, iov, niov, NULL, 0, 0, errp);
309 ssize_t qio_channel_read(QIOChannel *ioc,
314 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
315 return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, 0, errp);
319 ssize_t qio_channel_write(QIOChannel *ioc,
324 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
325 return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, 0, errp);
329 int coroutine_mixed_fn qio_channel_read_all_eof(QIOChannel *ioc,
334 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
335 return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
339 int coroutine_mixed_fn qio_channel_read_all(QIOChannel *ioc,
344 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
345 return qio_channel_readv_all(ioc, &iov, 1, errp);
349 int coroutine_mixed_fn qio_channel_write_all(QIOChannel *ioc,
354 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
355 return qio_channel_writev_all(ioc, &iov, 1, errp);
359 int qio_channel_set_blocking(QIOChannel *ioc,
363 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
364 return klass->io_set_blocking(ioc, enabled, errp);
368 void qio_channel_set_follow_coroutine_ctx(QIOChannel *ioc, bool enabled)
370 ioc->follow_coroutine_ctx = enabled;
374 int qio_channel_close(QIOChannel *ioc,
377 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
378 return klass->io_close(ioc, errp);
382 GSource *qio_channel_create_watch(QIOChannel *ioc,
383 GIOCondition condition)
385 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
386 GSource *ret = klass->io_create_watch(ioc, condition);
389 g_source_set_name(ret, ioc->name);
396 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
397 AioContext *read_ctx,
399 AioContext *write_ctx,
403 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
405 klass->io_set_aio_fd_handler(ioc, read_ctx, io_read, write_ctx, io_write,
409 guint qio_channel_add_watch_full(QIOChannel *ioc,
410 GIOCondition condition,
413 GDestroyNotify notify,
414 GMainContext *context)
419 source = qio_channel_create_watch(ioc, condition);
421 g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
423 id = g_source_attach(source, context);
424 g_source_unref(source);
429 guint qio_channel_add_watch(QIOChannel *ioc,
430 GIOCondition condition,
433 GDestroyNotify notify)
435 return qio_channel_add_watch_full(ioc, condition, func,
436 user_data, notify, NULL);
439 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
440 GIOCondition condition,
443 GDestroyNotify notify,
444 GMainContext *context)
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);
457 int qio_channel_shutdown(QIOChannel *ioc,
458 QIOChannelShutdown how,
461 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
463 if (!klass->io_shutdown) {
464 error_setg(errp, "Data path shutdown not supported");
468 return klass->io_shutdown(ioc, how, errp);
472 void qio_channel_set_delay(QIOChannel *ioc,
475 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
477 if (klass->io_set_delay) {
478 klass->io_set_delay(ioc, enabled);
483 void qio_channel_set_cork(QIOChannel *ioc,
486 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
488 if (klass->io_set_cork) {
489 klass->io_set_cork(ioc, enabled);
494 off_t qio_channel_io_seek(QIOChannel *ioc,
499 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
501 if (!klass->io_seek) {
502 error_setg(errp, "Channel does not support random access");
506 return klass->io_seek(ioc, offset, whence, errp);
509 int qio_channel_flush(QIOChannel *ioc,
512 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
514 if (!klass->io_flush ||
515 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
519 return klass->io_flush(ioc, errp);
523 static void qio_channel_restart_read(void *opaque)
525 QIOChannel *ioc = opaque;
526 Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
532 /* Assert that aio_co_wake() reenters the coroutine directly */
533 assert(qemu_get_current_aio_context() ==
534 qemu_coroutine_get_aio_context(co));
538 static void qio_channel_restart_write(void *opaque)
540 QIOChannel *ioc = opaque;
541 Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
547 /* Assert that aio_co_wake() reenters the coroutine directly */
548 assert(qemu_get_current_aio_context() ==
549 qemu_coroutine_get_aio_context(co));
553 static void coroutine_fn
554 qio_channel_set_fd_handlers(QIOChannel *ioc, GIOCondition condition)
556 AioContext *ctx = ioc->follow_coroutine_ctx ?
557 qemu_coroutine_get_aio_context(qemu_coroutine_self()) :
558 iohandler_get_aio_context();
559 AioContext *read_ctx = NULL;
560 IOHandler *io_read = NULL;
561 AioContext *write_ctx = NULL;
562 IOHandler *io_write = NULL;
564 if (condition == G_IO_IN) {
565 ioc->read_coroutine = qemu_coroutine_self();
568 io_read = qio_channel_restart_read;
571 * Thread safety: if the other coroutine is set and its AioContext
572 * matches ours, then there is mutual exclusion between read and write
573 * because they share a single thread and it's safe to set both read
574 * and write fd handlers here. If the AioContext does not match ours,
575 * then both threads may run in parallel but there is no shared state
578 if (ioc->write_coroutine && ioc->write_ctx == ctx) {
580 io_write = qio_channel_restart_write;
582 } else if (condition == G_IO_OUT) {
583 ioc->write_coroutine = qemu_coroutine_self();
584 ioc->write_ctx = ctx;
586 io_write = qio_channel_restart_write;
587 if (ioc->read_coroutine && ioc->read_ctx == ctx) {
589 io_read = qio_channel_restart_read;
595 qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
596 write_ctx, io_write, ioc);
599 static void coroutine_fn
600 qio_channel_clear_fd_handlers(QIOChannel *ioc, GIOCondition condition)
602 AioContext *read_ctx = NULL;
603 IOHandler *io_read = NULL;
604 AioContext *write_ctx = NULL;
605 IOHandler *io_write = NULL;
608 if (condition == G_IO_IN) {
612 if (ioc->write_coroutine && ioc->write_ctx == ctx) {
614 io_write = qio_channel_restart_write;
616 } else if (condition == G_IO_OUT) {
617 ctx = ioc->write_ctx;
620 if (ioc->read_coroutine && ioc->read_ctx == ctx) {
622 io_read = qio_channel_restart_read;
628 qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
629 write_ctx, io_write, ioc);
632 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
633 GIOCondition condition)
637 assert(qemu_in_coroutine());
638 ioc_ctx = qemu_coroutine_get_aio_context(qemu_coroutine_self());
640 if (condition == G_IO_IN) {
641 assert(!ioc->read_coroutine);
642 } else if (condition == G_IO_OUT) {
643 assert(!ioc->write_coroutine);
647 qio_channel_set_fd_handlers(ioc, condition);
648 qemu_coroutine_yield();
649 assert(in_aio_context_home_thread(ioc_ctx));
651 /* Allow interrupting the operation by reentering the coroutine other than
652 * through the aio_fd_handlers. */
653 if (condition == G_IO_IN) {
654 assert(ioc->read_coroutine == NULL);
655 } else if (condition == G_IO_OUT) {
656 assert(ioc->write_coroutine == NULL);
658 qio_channel_clear_fd_handlers(ioc, condition);
661 void qio_channel_wake_read(QIOChannel *ioc)
663 Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
669 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
670 GIOCondition condition,
673 GMainLoop *loop = opaque;
675 g_main_loop_quit(loop);
680 void qio_channel_wait(QIOChannel *ioc,
681 GIOCondition condition)
683 GMainContext *ctxt = g_main_context_new();
684 GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
687 source = qio_channel_create_watch(ioc, condition);
689 g_source_set_callback(source,
690 (GSourceFunc)qio_channel_wait_complete,
694 g_source_attach(source, ctxt);
696 g_main_loop_run(loop);
698 g_source_unref(source);
699 g_main_loop_unref(loop);
700 g_main_context_unref(ctxt);
704 static void qio_channel_finalize(Object *obj)
706 QIOChannel *ioc = QIO_CHANNEL(obj);
708 /* Must not have coroutines in qio_channel_yield() */
709 assert(!ioc->read_coroutine);
710 assert(!ioc->write_coroutine);
716 CloseHandle(ioc->event);
721 static const TypeInfo qio_channel_info = {
722 .parent = TYPE_OBJECT,
723 .name = TYPE_QIO_CHANNEL,
724 .instance_size = sizeof(QIOChannel),
725 .instance_finalize = qio_channel_finalize,
727 .class_size = sizeof(QIOChannelClass),
731 static void qio_channel_register_types(void)
733 type_register_static(&qio_channel_info);
737 type_init(qio_channel_register_types);