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 "io/channel.h"
23 #include "qapi/error.h"
24 #include "qemu/main-loop.h"
25 #include "qemu/module.h"
28 bool qio_channel_has_feature(QIOChannel
*ioc
,
29 QIOChannelFeature feature
)
31 return ioc
->features
& (1 << feature
);
35 void qio_channel_set_feature(QIOChannel
*ioc
,
36 QIOChannelFeature feature
)
38 ioc
->features
|= (1 << feature
);
42 void qio_channel_set_name(QIOChannel
*ioc
,
46 ioc
->name
= g_strdup(name
);
50 ssize_t
qio_channel_readv_full(QIOChannel
*ioc
,
51 const struct iovec
*iov
,
57 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
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");
66 return klass
->io_readv(ioc
, iov
, niov
, fds
, nfds
, errp
);
70 ssize_t
qio_channel_writev_full(QIOChannel
*ioc
,
71 const struct iovec
*iov
,
77 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
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");
86 return klass
->io_writev(ioc
, iov
, niov
, fds
, nfds
, errp
);
90 int qio_channel_readv_all_eof(QIOChannel
*ioc
,
91 const struct iovec
*iov
,
95 return qio_channel_readv_full_all_eof(ioc
, iov
, niov
, NULL
, NULL
, errp
);
98 int qio_channel_readv_all(QIOChannel
*ioc
,
99 const struct iovec
*iov
,
103 return qio_channel_readv_full_all(ioc
, iov
, niov
, NULL
, NULL
, errp
);
106 int qio_channel_readv_full_all_eof(QIOChannel
*ioc
,
107 const struct iovec
*iov
,
109 int **fds
, size_t *nfds
,
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;
128 nlocal_iov
= iov_copy(local_iov
, nlocal_iov
,
130 0, iov_size(iov
, niov
));
132 while ((nlocal_iov
> 0) || local_fds
) {
134 len
= qio_channel_readv_full(ioc
, local_iov
, nlocal_iov
, local_fds
,
136 if (len
== QIO_CHANNEL_ERR_BLOCK
) {
137 if (qemu_in_coroutine()) {
138 qio_channel_yield(ioc
, G_IO_IN
);
140 qio_channel_wait(ioc
, G_IO_IN
);
146 if (local_nfds
&& *local_nfds
) {
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
153 } else if (!partial
) {
154 /* No fds and no data - EOF before any data read */
160 "Unexpected end-of-file before all data were read");
161 /* Fallthrough into len < 0 handling */
166 /* Close any FDs we previously received */
169 for (i
= 0; i
< (*nfds
); i
++) {
180 iov_discard_front(&local_iov
, &nlocal_iov
, len
);
192 g_free(local_iov_head
);
196 int qio_channel_readv_full_all(QIOChannel
*ioc
,
197 const struct iovec
*iov
,
199 int **fds
, size_t *nfds
,
202 int ret
= qio_channel_readv_full_all_eof(ioc
, iov
, niov
, fds
, nfds
, errp
);
205 error_setg(errp
, "Unexpected end-of-file before all data were read");
215 int qio_channel_writev_all(QIOChannel
*ioc
,
216 const struct iovec
*iov
,
220 return qio_channel_writev_full_all(ioc
, iov
, niov
, NULL
, 0, errp
);
223 int qio_channel_writev_full_all(QIOChannel
*ioc
,
224 const struct iovec
*iov
,
226 int *fds
, size_t nfds
,
230 struct iovec
*local_iov
= g_new(struct iovec
, niov
);
231 struct iovec
*local_iov_head
= local_iov
;
232 unsigned int nlocal_iov
= niov
;
234 nlocal_iov
= iov_copy(local_iov
, nlocal_iov
,
236 0, iov_size(iov
, niov
));
238 while (nlocal_iov
> 0) {
240 len
= qio_channel_writev_full(ioc
, local_iov
, nlocal_iov
, fds
, nfds
,
242 if (len
== QIO_CHANNEL_ERR_BLOCK
) {
243 if (qemu_in_coroutine()) {
244 qio_channel_yield(ioc
, G_IO_OUT
);
246 qio_channel_wait(ioc
, G_IO_OUT
);
254 iov_discard_front(&local_iov
, &nlocal_iov
, len
);
262 g_free(local_iov_head
);
266 ssize_t
qio_channel_readv(QIOChannel
*ioc
,
267 const struct iovec
*iov
,
271 return qio_channel_readv_full(ioc
, iov
, niov
, NULL
, NULL
, errp
);
275 ssize_t
qio_channel_writev(QIOChannel
*ioc
,
276 const struct iovec
*iov
,
280 return qio_channel_writev_full(ioc
, iov
, niov
, NULL
, 0, errp
);
284 ssize_t
qio_channel_read(QIOChannel
*ioc
,
289 struct iovec iov
= { .iov_base
= buf
, .iov_len
= buflen
};
290 return qio_channel_readv_full(ioc
, &iov
, 1, NULL
, NULL
, errp
);
294 ssize_t
qio_channel_write(QIOChannel
*ioc
,
299 struct iovec iov
= { .iov_base
= (char *)buf
, .iov_len
= buflen
};
300 return qio_channel_writev_full(ioc
, &iov
, 1, NULL
, 0, errp
);
304 int qio_channel_read_all_eof(QIOChannel
*ioc
,
309 struct iovec iov
= { .iov_base
= buf
, .iov_len
= buflen
};
310 return qio_channel_readv_all_eof(ioc
, &iov
, 1, errp
);
314 int qio_channel_read_all(QIOChannel
*ioc
,
319 struct iovec iov
= { .iov_base
= buf
, .iov_len
= buflen
};
320 return qio_channel_readv_all(ioc
, &iov
, 1, errp
);
324 int qio_channel_write_all(QIOChannel
*ioc
,
329 struct iovec iov
= { .iov_base
= (char *)buf
, .iov_len
= buflen
};
330 return qio_channel_writev_all(ioc
, &iov
, 1, errp
);
334 int qio_channel_set_blocking(QIOChannel
*ioc
,
338 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
339 return klass
->io_set_blocking(ioc
, enabled
, errp
);
343 int qio_channel_close(QIOChannel
*ioc
,
346 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
347 return klass
->io_close(ioc
, errp
);
351 GSource
*qio_channel_create_watch(QIOChannel
*ioc
,
352 GIOCondition condition
)
354 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
355 GSource
*ret
= klass
->io_create_watch(ioc
, condition
);
358 g_source_set_name(ret
, ioc
->name
);
365 void qio_channel_set_aio_fd_handler(QIOChannel
*ioc
,
371 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
373 klass
->io_set_aio_fd_handler(ioc
, ctx
, io_read
, io_write
, opaque
);
376 guint
qio_channel_add_watch_full(QIOChannel
*ioc
,
377 GIOCondition condition
,
380 GDestroyNotify notify
,
381 GMainContext
*context
)
386 source
= qio_channel_create_watch(ioc
, condition
);
388 g_source_set_callback(source
, (GSourceFunc
)func
, user_data
, notify
);
390 id
= g_source_attach(source
, context
);
391 g_source_unref(source
);
396 guint
qio_channel_add_watch(QIOChannel
*ioc
,
397 GIOCondition condition
,
400 GDestroyNotify notify
)
402 return qio_channel_add_watch_full(ioc
, condition
, func
,
403 user_data
, notify
, NULL
);
406 GSource
*qio_channel_add_watch_source(QIOChannel
*ioc
,
407 GIOCondition condition
,
410 GDestroyNotify notify
,
411 GMainContext
*context
)
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
);
424 int qio_channel_shutdown(QIOChannel
*ioc
,
425 QIOChannelShutdown how
,
428 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
430 if (!klass
->io_shutdown
) {
431 error_setg(errp
, "Data path shutdown not supported");
435 return klass
->io_shutdown(ioc
, how
, errp
);
439 void qio_channel_set_delay(QIOChannel
*ioc
,
442 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
444 if (klass
->io_set_delay
) {
445 klass
->io_set_delay(ioc
, enabled
);
450 void qio_channel_set_cork(QIOChannel
*ioc
,
453 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
455 if (klass
->io_set_cork
) {
456 klass
->io_set_cork(ioc
, enabled
);
461 off_t
qio_channel_io_seek(QIOChannel
*ioc
,
466 QIOChannelClass
*klass
= QIO_CHANNEL_GET_CLASS(ioc
);
468 if (!klass
->io_seek
) {
469 error_setg(errp
, "Channel does not support random access");
473 return klass
->io_seek(ioc
, offset
, whence
, errp
);
477 static void qio_channel_restart_read(void *opaque
)
479 QIOChannel
*ioc
= opaque
;
480 Coroutine
*co
= ioc
->read_coroutine
;
482 /* Assert that aio_co_wake() reenters the coroutine directly */
483 assert(qemu_get_current_aio_context() ==
484 qemu_coroutine_get_aio_context(co
));
488 static void qio_channel_restart_write(void *opaque
)
490 QIOChannel
*ioc
= opaque
;
491 Coroutine
*co
= ioc
->write_coroutine
;
493 /* Assert that aio_co_wake() reenters the coroutine directly */
494 assert(qemu_get_current_aio_context() ==
495 qemu_coroutine_get_aio_context(co
));
499 static void qio_channel_set_aio_fd_handlers(QIOChannel
*ioc
)
501 IOHandler
*rd_handler
= NULL
, *wr_handler
= NULL
;
504 if (ioc
->read_coroutine
) {
505 rd_handler
= qio_channel_restart_read
;
507 if (ioc
->write_coroutine
) {
508 wr_handler
= qio_channel_restart_write
;
511 ctx
= ioc
->ctx
? ioc
->ctx
: iohandler_get_aio_context();
512 qio_channel_set_aio_fd_handler(ioc
, ctx
, rd_handler
, wr_handler
, ioc
);
515 void qio_channel_attach_aio_context(QIOChannel
*ioc
,
518 assert(!ioc
->read_coroutine
);
519 assert(!ioc
->write_coroutine
);
523 void qio_channel_detach_aio_context(QIOChannel
*ioc
)
525 ioc
->read_coroutine
= NULL
;
526 ioc
->write_coroutine
= NULL
;
527 qio_channel_set_aio_fd_handlers(ioc
);
531 void coroutine_fn
qio_channel_yield(QIOChannel
*ioc
,
532 GIOCondition condition
)
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();
544 qio_channel_set_aio_fd_handlers(ioc
);
545 qemu_coroutine_yield();
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
);
559 static gboolean
qio_channel_wait_complete(QIOChannel
*ioc
,
560 GIOCondition condition
,
563 GMainLoop
*loop
= opaque
;
565 g_main_loop_quit(loop
);
570 void qio_channel_wait(QIOChannel
*ioc
,
571 GIOCondition condition
)
573 GMainContext
*ctxt
= g_main_context_new();
574 GMainLoop
*loop
= g_main_loop_new(ctxt
, TRUE
);
577 source
= qio_channel_create_watch(ioc
, condition
);
579 g_source_set_callback(source
,
580 (GSourceFunc
)qio_channel_wait_complete
,
584 g_source_attach(source
, ctxt
);
586 g_main_loop_run(loop
);
588 g_source_unref(source
);
589 g_main_loop_unref(loop
);
590 g_main_context_unref(ctxt
);
594 static void qio_channel_finalize(Object
*obj
)
596 QIOChannel
*ioc
= QIO_CHANNEL(obj
);
602 CloseHandle(ioc
->event
);
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
,
613 .class_size
= sizeof(QIOChannelClass
),
617 static void qio_channel_register_types(void)
619 type_register_static(&qio_channel_info
);
623 type_init(qio_channel_register_types
);