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
);