[qemu.git] / io / channel.c

/*
 * QEMU I/O channels
 *
 * Copyright (c) 2015 Red Hat, Inc.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, see <http://www.gnu.org/licenses/>.
 *
 */

#include "qemu/osdep.h"
#include "io/channel.h"
#include "qapi/error.h"
#include "qemu/main-loop.h"
#include "qemu/module.h"
#include "qemu/iov.h"

bool qio_channel_has_feature(QIOChannel *ioc,
                             QIOChannelFeature feature)
{
    return ioc->features & (1 << feature);
}


void qio_channel_set_feature(QIOChannel *ioc,
                             QIOChannelFeature feature)
{
    ioc->features |= (1 << feature);
}


void qio_channel_set_name(QIOChannel *ioc,
                          const char *name)
{
    g_free(ioc->name);
    ioc->name = g_strdup(name);
}


ssize_t qio_channel_readv_full(QIOChannel *ioc,
                               const struct iovec *iov,
                               size_t niov,
                               int **fds,
                               size_t *nfds,
                               Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if ((fds || nfds) &&
        !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
        error_setg_errno(errp, EINVAL,
                         "Channel does not support file descriptor passing");
        return -1;
    }

    return klass->io_readv(ioc, iov, niov, fds, nfds, errp);
}


ssize_t qio_channel_writev_full(QIOChannel *ioc,
                                const struct iovec *iov,
                                size_t niov,
                                int *fds,
                                size_t nfds,
                                Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if ((fds || nfds) &&
        !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
        error_setg_errno(errp, EINVAL,
                         "Channel does not support file descriptor passing");
        return -1;
    }

    return klass->io_writev(ioc, iov, niov, fds, nfds, errp);
}


int qio_channel_readv_all_eof(QIOChannel *ioc,
                              const struct iovec *iov,
                              size_t niov,
                              Error **errp)
{
    int ret = -1;
    struct iovec *local_iov = g_new(struct iovec, niov);
    struct iovec *local_iov_head = local_iov;
    unsigned int nlocal_iov = niov;
    bool partial = false;

    nlocal_iov = iov_copy(local_iov, nlocal_iov,
                          iov, niov,
                          0, iov_size(iov, niov));

    while (nlocal_iov > 0) {
        ssize_t len;
        len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
        if (len == QIO_CHANNEL_ERR_BLOCK) {
            if (qemu_in_coroutine()) {
                qio_channel_yield(ioc, G_IO_IN);
            } else {
                qio_channel_wait(ioc, G_IO_IN);
            }
            continue;
        } else if (len < 0) {
            goto cleanup;
        } else if (len == 0) {
            if (partial) {
                error_setg(errp,
                           "Unexpected end-of-file before all bytes were read");
            } else {
                ret = 0;
            }
            goto cleanup;
        }

        partial = true;
        iov_discard_front(&local_iov, &nlocal_iov, len);
    }

    ret = 1;

 cleanup:
    g_free(local_iov_head);
    return ret;
}

int qio_channel_readv_all(QIOChannel *ioc,
                          const struct iovec *iov,
                          size_t niov,
                          Error **errp)
{
    int ret = qio_channel_readv_all_eof(ioc, iov, niov, errp);

    if (ret == 0) {
        ret = -1;
        error_setg(errp,
                   "Unexpected end-of-file before all bytes were read");
    } else if (ret == 1) {
        ret = 0;
    }
    return ret;
}

int qio_channel_writev_all(QIOChannel *ioc,
                           const struct iovec *iov,
                           size_t niov,
                           Error **errp)
{
    int ret = -1;
    struct iovec *local_iov = g_new(struct iovec, niov);
    struct iovec *local_iov_head = local_iov;
    unsigned int nlocal_iov = niov;

    nlocal_iov = iov_copy(local_iov, nlocal_iov,
                          iov, niov,
                          0, iov_size(iov, niov));

    while (nlocal_iov > 0) {
        ssize_t len;
        len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);
        if (len == QIO_CHANNEL_ERR_BLOCK) {
            if (qemu_in_coroutine()) {
                qio_channel_yield(ioc, G_IO_OUT);
            } else {
                qio_channel_wait(ioc, G_IO_OUT);
            }
            continue;
        }
        if (len < 0) {
            goto cleanup;
        }

        iov_discard_front(&local_iov, &nlocal_iov, len);
    }

    ret = 0;
 cleanup:
    g_free(local_iov_head);
    return ret;
}

ssize_t qio_channel_readv(QIOChannel *ioc,
                          const struct iovec *iov,
                          size_t niov,
                          Error **errp)
{
    return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, errp);
}


ssize_t qio_channel_writev(QIOChannel *ioc,
                           const struct iovec *iov,
                           size_t niov,
                           Error **errp)
{
    return qio_channel_writev_full(ioc, iov, niov, NULL, 0, errp);
}


ssize_t qio_channel_read(QIOChannel *ioc,
                         char *buf,
                         size_t buflen,
                         Error **errp)
{
    struct iovec iov = { .iov_base = buf, .iov_len = buflen };
    return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, errp);
}


ssize_t qio_channel_write(QIOChannel *ioc,
                          const char *buf,
                          size_t buflen,
                          Error **errp)
{
    struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
    return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, errp);
}


int qio_channel_read_all_eof(QIOChannel *ioc,
                             char *buf,
                             size_t buflen,
                             Error **errp)
{
    struct iovec iov = { .iov_base = buf, .iov_len = buflen };
    return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
}


int qio_channel_read_all(QIOChannel *ioc,
                         char *buf,
                         size_t buflen,
                         Error **errp)
{
    struct iovec iov = { .iov_base = buf, .iov_len = buflen };
    return qio_channel_readv_all(ioc, &iov, 1, errp);
}


int qio_channel_write_all(QIOChannel *ioc,
                          const char *buf,
                          size_t buflen,
                          Error **errp)
{
    struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
    return qio_channel_writev_all(ioc, &iov, 1, errp);
}


int qio_channel_set_blocking(QIOChannel *ioc,
                              bool enabled,
                              Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
    return klass->io_set_blocking(ioc, enabled, errp);
}


int qio_channel_close(QIOChannel *ioc,
                      Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
    return klass->io_close(ioc, errp);
}


GSource *qio_channel_create_watch(QIOChannel *ioc,
                                  GIOCondition condition)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
    GSource *ret = klass->io_create_watch(ioc, condition);

    if (ioc->name) {
        g_source_set_name(ret, ioc->name);
    }

    return ret;
}


void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
                                    AioContext *ctx,
                                    IOHandler *io_read,
                                    IOHandler *io_write,
                                    void *opaque)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
}

guint qio_channel_add_watch_full(QIOChannel *ioc,
                                 GIOCondition condition,
                                 QIOChannelFunc func,
                                 gpointer user_data,
                                 GDestroyNotify notify,
                                 GMainContext *context)
{
    GSource *source;
    guint id;

    source = qio_channel_create_watch(ioc, condition);

    g_source_set_callback(source, (GSourceFunc)func, user_data, notify);

    id = g_source_attach(source, context);
    g_source_unref(source);

    return id;
}

guint qio_channel_add_watch(QIOChannel *ioc,
                            GIOCondition condition,
                            QIOChannelFunc func,
                            gpointer user_data,
                            GDestroyNotify notify)
{
    return qio_channel_add_watch_full(ioc, condition, func,
                                      user_data, notify, NULL);
}

GSource *qio_channel_add_watch_source(QIOChannel *ioc,
                                      GIOCondition condition,
                                      QIOChannelFunc func,
                                      gpointer user_data,
                                      GDestroyNotify notify,
                                      GMainContext *context)
{
    GSource *source;
    guint id;

    id = qio_channel_add_watch_full(ioc, condition, func,
                                    user_data, notify, context);
    source = g_main_context_find_source_by_id(context, id);
    g_source_ref(source);
    return source;
}


int qio_channel_shutdown(QIOChannel *ioc,
                         QIOChannelShutdown how,
                         Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if (!klass->io_shutdown) {
        error_setg(errp, "Data path shutdown not supported");
        return -1;
    }

    return klass->io_shutdown(ioc, how, errp);
}


void qio_channel_set_delay(QIOChannel *ioc,
                           bool enabled)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if (klass->io_set_delay) {
        klass->io_set_delay(ioc, enabled);
    }
}


void qio_channel_set_cork(QIOChannel *ioc,
                          bool enabled)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if (klass->io_set_cork) {
        klass->io_set_cork(ioc, enabled);
    }
}


off_t qio_channel_io_seek(QIOChannel *ioc,
                          off_t offset,
                          int whence,
                          Error **errp)
{
    QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);

    if (!klass->io_seek) {
        error_setg(errp, "Channel does not support random access");
        return -1;
    }

    return klass->io_seek(ioc, offset, whence, errp);
}


static void qio_channel_restart_read(void *opaque)
{
    QIOChannel *ioc = opaque;
    Coroutine *co = ioc->read_coroutine;

    /* Assert that aio_co_wake() reenters the coroutine directly */
    assert(qemu_get_current_aio_context() ==
           qemu_coroutine_get_aio_context(co));
    aio_co_wake(co);
}

static void qio_channel_restart_write(void *opaque)
{
    QIOChannel *ioc = opaque;
    Coroutine *co = ioc->write_coroutine;

    /* Assert that aio_co_wake() reenters the coroutine directly */
    assert(qemu_get_current_aio_context() ==
           qemu_coroutine_get_aio_context(co));
    aio_co_wake(co);
}

static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
{
    IOHandler *rd_handler = NULL, *wr_handler = NULL;
    AioContext *ctx;

    if (ioc->read_coroutine) {
        rd_handler = qio_channel_restart_read;
    }
    if (ioc->write_coroutine) {
        wr_handler = qio_channel_restart_write;
    }

    ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
    qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
}

void qio_channel_attach_aio_context(QIOChannel *ioc,
                                    AioContext *ctx)
{
    assert(!ioc->read_coroutine);
    assert(!ioc->write_coroutine);
    ioc->ctx = ctx;
}

void qio_channel_detach_aio_context(QIOChannel *ioc)
{
    ioc->read_coroutine = NULL;
    ioc->write_coroutine = NULL;
    qio_channel_set_aio_fd_handlers(ioc);
    ioc->ctx = NULL;
}

void coroutine_fn qio_channel_yield(QIOChannel *ioc,
                                    GIOCondition condition)
{
    assert(qemu_in_coroutine());
    if (condition == G_IO_IN) {
        assert(!ioc->read_coroutine);
        ioc->read_coroutine = qemu_coroutine_self();
    } else if (condition == G_IO_OUT) {
        assert(!ioc->write_coroutine);
        ioc->write_coroutine = qemu_coroutine_self();
    } else {
        abort();
    }
    qio_channel_set_aio_fd_handlers(ioc);
    qemu_coroutine_yield();

    /* Allow interrupting the operation by reentering the coroutine other than
     * through the aio_fd_handlers. */
    if (condition == G_IO_IN && ioc->read_coroutine) {
        ioc->read_coroutine = NULL;
        qio_channel_set_aio_fd_handlers(ioc);
    } else if (condition == G_IO_OUT && ioc->write_coroutine) {
        ioc->write_coroutine = NULL;
        qio_channel_set_aio_fd_handlers(ioc);
    }
}


static gboolean qio_channel_wait_complete(QIOChannel *ioc,
                                          GIOCondition condition,
                                          gpointer opaque)
{
    GMainLoop *loop = opaque;

    g_main_loop_quit(loop);
    return FALSE;
}


void qio_channel_wait(QIOChannel *ioc,
                      GIOCondition condition)
{
    GMainContext *ctxt = g_main_context_new();
    GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
    GSource *source;

    source = qio_channel_create_watch(ioc, condition);

    g_source_set_callback(source,
                          (GSourceFunc)qio_channel_wait_complete,
                          loop,
                          NULL);

    g_source_attach(source, ctxt);

    g_main_loop_run(loop);

    g_source_unref(source);
    g_main_loop_unref(loop);
    g_main_context_unref(ctxt);
}


static void qio_channel_finalize(Object *obj)
{
    QIOChannel *ioc = QIO_CHANNEL(obj);

    g_free(ioc->name);

#ifdef _WIN32
    if (ioc->event) {
        CloseHandle(ioc->event);
    }
#endif
}

static const TypeInfo qio_channel_info = {
    .parent = TYPE_OBJECT,
    .name = TYPE_QIO_CHANNEL,
    .instance_size = sizeof(QIOChannel),
    .instance_finalize = qio_channel_finalize,
    .abstract = true,
    .class_size = sizeof(QIOChannelClass),
};


static void qio_channel_register_types(void)
{
    type_register_static(&qio_channel_info);
}


type_init(qio_channel_register_types);
Commit	Line	Data
666a3af9 DB	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 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
cae9fc56	21	#include "qemu/osdep.h"
666a3af9	22	#include "io/channel.h"
da34e65c	23	#include "qapi/error.h"
c4c497d2	24	#include "qemu/main-loop.h"
0b8fa32f	25	#include "qemu/module.h"
d4622e55	26	#include "qemu/iov.h"
666a3af9 DB	27
	28	bool qio_channel_has_feature(QIOChannel *ioc,
	29	QIOChannelFeature feature)
	30	{
	31	return ioc->features & (1 << feature);
	32	}
	33
	34
d8d3c7cc FF	35	void qio_channel_set_feature(QIOChannel *ioc,
	36	QIOChannelFeature feature)
	37	{
	38	ioc->features \|= (1 << feature);
	39	}
	40
	41
20f4aa26 DB	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
666a3af9 DB	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) &&
e413ae0c	60	!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
666a3af9 DB	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) &&
e413ae0c	80	!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
666a3af9 DB	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
e8ffaa31 EB	90	int qio_channel_readv_all_eof(QIOChannel *ioc,
	91	const struct iovec *iov,
	92	size_t niov,
	93	Error **errp)
d4622e55 DB	94	{
	95	int ret = -1;
	96	struct iovec *local_iov = g_new(struct iovec, niov);
	97	struct iovec *local_iov_head = local_iov;
	98	unsigned int nlocal_iov = niov;
e8ffaa31	99	bool partial = false;
d4622e55 DB	100
	101	nlocal_iov = iov_copy(local_iov, nlocal_iov,
	102	iov, niov,
	103	0, iov_size(iov, niov));
	104
	105	while (nlocal_iov > 0) {
	106	ssize_t len;
	107	len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
	108	if (len == QIO_CHANNEL_ERR_BLOCK) {
9ffb8270 EB	109	if (qemu_in_coroutine()) {
	110	qio_channel_yield(ioc, G_IO_IN);
	111	} else {
	112	qio_channel_wait(ioc, G_IO_IN);
	113	}
d4622e55 DB	114	continue;
	115	} else if (len < 0) {
	116	goto cleanup;
	117	} else if (len == 0) {
e8ffaa31 EB	118	if (partial) {
	119	error_setg(errp,
	120	"Unexpected end-of-file before all bytes were read");
	121	} else {
	122	ret = 0;
	123	}
d4622e55 DB	124	goto cleanup;
	125	}
	126
e8ffaa31	127	partial = true;
d4622e55 DB	128	iov_discard_front(&local_iov, &nlocal_iov, len);
	129	}
	130
e8ffaa31	131	ret = 1;
d4622e55 DB	132
	133	cleanup:
	134	g_free(local_iov_head);
	135	return ret;
	136	}
	137
e8ffaa31 EB	138	int qio_channel_readv_all(QIOChannel *ioc,
	139	const struct iovec *iov,
	140	size_t niov,
	141	Error **errp)
	142	{
	143	int ret = qio_channel_readv_all_eof(ioc, iov, niov, errp);
	144
	145	if (ret == 0) {
	146	ret = -1;
	147	error_setg(errp,
	148	"Unexpected end-of-file before all bytes were read");
	149	} else if (ret == 1) {
	150	ret = 0;
	151	}
	152	return ret;
	153	}
	154
d4622e55 DB	155	int qio_channel_writev_all(QIOChannel *ioc,
	156	const struct iovec *iov,
	157	size_t niov,
	158	Error **errp)
	159	{
	160	int ret = -1;
	161	struct iovec *local_iov = g_new(struct iovec, niov);
	162	struct iovec *local_iov_head = local_iov;
	163	unsigned int nlocal_iov = niov;
	164
	165	nlocal_iov = iov_copy(local_iov, nlocal_iov,
	166	iov, niov,
	167	0, iov_size(iov, niov));
	168
	169	while (nlocal_iov > 0) {
	170	ssize_t len;
	171	len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);
	172	if (len == QIO_CHANNEL_ERR_BLOCK) {
9ffb8270 EB	173	if (qemu_in_coroutine()) {
	174	qio_channel_yield(ioc, G_IO_OUT);
	175	} else {
	176	qio_channel_wait(ioc, G_IO_OUT);
	177	}
d4622e55 DB	178	continue;
	179	}
	180	if (len < 0) {
	181	goto cleanup;
	182	}
	183
	184	iov_discard_front(&local_iov, &nlocal_iov, len);
	185	}
	186
	187	ret = 0;
	188	cleanup:
	189	g_free(local_iov_head);
	190	return ret;
	191	}
	192
666a3af9 DB	193	ssize_t qio_channel_readv(QIOChannel *ioc,
	194	const struct iovec *iov,
	195	size_t niov,
	196	Error **errp)
	197	{
	198	return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, errp);
	199	}
	200
	201
	202	ssize_t qio_channel_writev(QIOChannel *ioc,
	203	const struct iovec *iov,
	204	size_t niov,
	205	Error **errp)
	206	{
	207	return qio_channel_writev_full(ioc, iov, niov, NULL, 0, errp);
	208	}
	209
	210
	211	ssize_t qio_channel_read(QIOChannel *ioc,
	212	char *buf,
	213	size_t buflen,
	214	Error **errp)
	215	{
	216	struct iovec iov = { .iov_base = buf, .iov_len = buflen };
	217	return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, errp);
	218	}
	219
	220
	221	ssize_t qio_channel_write(QIOChannel *ioc,
	222	const char *buf,
	223	size_t buflen,
	224	Error **errp)
	225	{
	226	struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
	227	return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, errp);
	228	}
	229
	230
e8ffaa31 EB	231	int qio_channel_read_all_eof(QIOChannel *ioc,
	232	char *buf,
	233	size_t buflen,
	234	Error **errp)
	235	{
	236	struct iovec iov = { .iov_base = buf, .iov_len = buflen };
	237	return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
	238	}
	239
	240
d4622e55 DB	241	int qio_channel_read_all(QIOChannel *ioc,
	242	char *buf,
	243	size_t buflen,
	244	Error **errp)
	245	{
	246	struct iovec iov = { .iov_base = buf, .iov_len = buflen };
	247	return qio_channel_readv_all(ioc, &iov, 1, errp);
	248	}
	249
	250
	251	int qio_channel_write_all(QIOChannel *ioc,
	252	const char *buf,
	253	size_t buflen,
	254	Error **errp)
	255	{
	256	struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
	257	return qio_channel_writev_all(ioc, &iov, 1, errp);
	258	}
	259
	260
666a3af9 DB	261	int qio_channel_set_blocking(QIOChannel *ioc,
	262	bool enabled,
	263	Error **errp)
	264	{
	265	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	266	return klass->io_set_blocking(ioc, enabled, errp);
	267	}
	268
	269
	270	int qio_channel_close(QIOChannel *ioc,
	271	Error **errp)
	272	{
	273	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	274	return klass->io_close(ioc, errp);
	275	}
	276
	277
	278	GSource qio_channel_create_watch(QIOChannel ioc,
	279	GIOCondition condition)
	280	{
	281	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
20f4aa26 DB	282	GSource *ret = klass->io_create_watch(ioc, condition);
	283
	284	if (ioc->name) {
	285	g_source_set_name(ret, ioc->name);
	286	}
	287
	288	return ret;
666a3af9 DB	289	}
	290
	291
bf88c124 PB	292	void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
	293	AioContext *ctx,
	294	IOHandler *io_read,
	295	IOHandler *io_write,
	296	void *opaque)
	297	{
	298	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	299
	300	klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
	301	}
	302
315409c7 PX	303	guint qio_channel_add_watch_full(QIOChannel *ioc,
	304	GIOCondition condition,
	305	QIOChannelFunc func,
	306	gpointer user_data,
	307	GDestroyNotify notify,
	308	GMainContext *context)
666a3af9 DB	309	{
	310	GSource *source;
	311	guint id;
	312
	313	source = qio_channel_create_watch(ioc, condition);
	314
	315	g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
	316
315409c7	317	id = g_source_attach(source, context);
666a3af9 DB	318	g_source_unref(source);
	319
	320	return id;
	321	}
	322
315409c7 PX	323	guint qio_channel_add_watch(QIOChannel *ioc,
	324	GIOCondition condition,
	325	QIOChannelFunc func,
	326	gpointer user_data,
	327	GDestroyNotify notify)
	328	{
	329	return qio_channel_add_watch_full(ioc, condition, func,
	330	user_data, notify, NULL);
	331	}
	332
	333	GSource qio_channel_add_watch_source(QIOChannel ioc,
	334	GIOCondition condition,
	335	QIOChannelFunc func,
	336	gpointer user_data,
	337	GDestroyNotify notify,
	338	GMainContext *context)
	339	{
	340	GSource *source;
	341	guint id;
	342
	343	id = qio_channel_add_watch_full(ioc, condition, func,
	344	user_data, notify, context);
	345	source = g_main_context_find_source_by_id(context, id);
	346	g_source_ref(source);
	347	return source;
	348	}
	349
666a3af9 DB	350
	351	int qio_channel_shutdown(QIOChannel *ioc,
	352	QIOChannelShutdown how,
	353	Error **errp)
	354	{
	355	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	356
	357	if (!klass->io_shutdown) {
	358	error_setg(errp, "Data path shutdown not supported");
	359	return -1;
	360	}
	361
	362	return klass->io_shutdown(ioc, how, errp);
	363	}
	364
	365
	366	void qio_channel_set_delay(QIOChannel *ioc,
	367	bool enabled)
	368	{
	369	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	370
	371	if (klass->io_set_delay) {
	372	klass->io_set_delay(ioc, enabled);
	373	}
	374	}
	375
	376
	377	void qio_channel_set_cork(QIOChannel *ioc,
	378	bool enabled)
	379	{
	380	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	381
	382	if (klass->io_set_cork) {
	383	klass->io_set_cork(ioc, enabled);
	384	}
	385	}
	386
	387
	388	off_t qio_channel_io_seek(QIOChannel *ioc,
	389	off_t offset,
	390	int whence,
	391	Error **errp)
	392	{
	393	QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
	394
	395	if (!klass->io_seek) {
	396	error_setg(errp, "Channel does not support random access");
	397	return -1;
	398	}
	399
	400	return klass->io_seek(ioc, offset, whence, errp);
	401	}
	402
	403
c4c497d2 PB	404	static void qio_channel_restart_read(void *opaque)
	405	{
	406	QIOChannel *ioc = opaque;
	407	Coroutine *co = ioc->read_coroutine;
	408
2a239e6e KW	409	/* Assert that aio_co_wake() reenters the coroutine directly */
	410	assert(qemu_get_current_aio_context() ==
	411	qemu_coroutine_get_aio_context(co));
c4c497d2 PB	412	aio_co_wake(co);
c4c497d2 PB	413	}
666a3af9	414
c4c497d2	415	static void qio_channel_restart_write(void *opaque)
666a3af9	416	{
c4c497d2 PB	417	QIOChannel *ioc = opaque;
	418	Coroutine *co = ioc->write_coroutine;
	419
2a239e6e KW	420	/* Assert that aio_co_wake() reenters the coroutine directly */
	421	assert(qemu_get_current_aio_context() ==
	422	qemu_coroutine_get_aio_context(co));
c4c497d2	423	aio_co_wake(co);
666a3af9 DB	424	}
666a3af9 DB	425
c4c497d2 PB	426	static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
	427	{
	428	IOHandler rd_handler = NULL, wr_handler = NULL;
	429	AioContext *ctx;
	430
	431	if (ioc->read_coroutine) {
	432	rd_handler = qio_channel_restart_read;
	433	}
	434	if (ioc->write_coroutine) {
	435	wr_handler = qio_channel_restart_write;
	436	}
	437
	438	ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
	439	qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
	440	}
	441
	442	void qio_channel_attach_aio_context(QIOChannel *ioc,
	443	AioContext *ctx)
	444	{
8f7168b3 PB	445	assert(!ioc->read_coroutine);
8f7168b3 PB	446	assert(!ioc->write_coroutine);
c4c497d2	447	ioc->ctx = ctx;
c4c497d2 PB	448	}
	449
	450	void qio_channel_detach_aio_context(QIOChannel *ioc)
	451	{
	452	ioc->read_coroutine = NULL;
	453	ioc->write_coroutine = NULL;
	454	qio_channel_set_aio_fd_handlers(ioc);
	455	ioc->ctx = NULL;
	456	}
666a3af9 DB	457
	458	void coroutine_fn qio_channel_yield(QIOChannel *ioc,
	459	GIOCondition condition)
	460	{
666a3af9	461	assert(qemu_in_coroutine());
c4c497d2 PB	462	if (condition == G_IO_IN) {
	463	assert(!ioc->read_coroutine);
	464	ioc->read_coroutine = qemu_coroutine_self();
	465	} else if (condition == G_IO_OUT) {
	466	assert(!ioc->write_coroutine);
	467	ioc->write_coroutine = qemu_coroutine_self();
	468	} else {
	469	abort();
	470	}
	471	qio_channel_set_aio_fd_handlers(ioc);
666a3af9	472	qemu_coroutine_yield();
6886ceaf KW	473
	474	/* Allow interrupting the operation by reentering the coroutine other than
	475	* through the aio_fd_handlers. */
	476	if (condition == G_IO_IN && ioc->read_coroutine) {
	477	ioc->read_coroutine = NULL;
	478	qio_channel_set_aio_fd_handlers(ioc);
	479	} else if (condition == G_IO_OUT && ioc->write_coroutine) {
	480	ioc->write_coroutine = NULL;
	481	qio_channel_set_aio_fd_handlers(ioc);
	482	}
666a3af9 DB	483	}
	484
	485
	486	static gboolean qio_channel_wait_complete(QIOChannel *ioc,
	487	GIOCondition condition,
	488	gpointer opaque)
	489	{
	490	GMainLoop *loop = opaque;
	491
	492	g_main_loop_quit(loop);
	493	return FALSE;
	494	}
	495
	496
	497	void qio_channel_wait(QIOChannel *ioc,
	498	GIOCondition condition)
	499	{
	500	GMainContext *ctxt = g_main_context_new();
	501	GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
	502	GSource *source;
	503
	504	source = qio_channel_create_watch(ioc, condition);
	505
	506	g_source_set_callback(source,
	507	(GSourceFunc)qio_channel_wait_complete,
	508	loop,
	509	NULL);
	510
	511	g_source_attach(source, ctxt);
	512
	513	g_main_loop_run(loop);
	514
	515	g_source_unref(source);
	516	g_main_loop_unref(loop);
	517	g_main_context_unref(ctxt);
	518	}
	519
	520
a5897205 PB	521	static void qio_channel_finalize(Object *obj)
	522	{
	523	QIOChannel *ioc = QIO_CHANNEL(obj);
	524
20f4aa26 DB	525	g_free(ioc->name);
	526
	527	#ifdef _WIN32
a5897205 PB	528	if (ioc->event) {
	529	CloseHandle(ioc->event);
	530	}
a5897205	531	#endif
20f4aa26	532	}
a5897205	533
666a3af9 DB	534	static const TypeInfo qio_channel_info = {
	535	.parent = TYPE_OBJECT,
	536	.name = TYPE_QIO_CHANNEL,
	537	.instance_size = sizeof(QIOChannel),
a5897205	538	.instance_finalize = qio_channel_finalize,
666a3af9 DB	539	.abstract = true,
	540	.class_size = sizeof(QIOChannelClass),
	541	};
	542
	543
	544	static void qio_channel_register_types(void)
	545	{
	546	type_register_static(&qio_channel_info);
	547	}
	548
	549
	550	type_init(qio_channel_register_types);