static inline void mirror_wait_for_io(MirrorBlockJob *s)
{
    assert(!s->waiting_for_io);
    s->waiting_for_io = true;
    qemu_coroutine_yield();
    s->waiting_for_io = false;
}

static void test_nesting(void)
{
    Coroutine *root;
    NestData nd = {
        .n_enter  = 0,
        .n_return = 0,
        .max      = 128,
    };

    root = qemu_coroutine_create(nest);
    qemu_coroutine_enter(root, &nd);

    /* Must enter and return from max nesting level */
    g_assert_cmpint(nd.n_enter, ==, nd.max);
    g_assert_cmpint(nd.n_return, ==, nd.max);
}

/*
 * Check that yield/enter transfer control correctly
 */

static void coroutine_fn yield_5_times(void *opaque)
{
    bool *done = opaque;
    int i;

    for (i = 0; i < 5; i++) {
        qemu_coroutine_yield();
    }
    *done = true;
}

ssize_t coroutine_fn
qemu_co_sendv_recvv(int sockfd, struct iovec *iov, unsigned iov_cnt,
                    size_t offset, size_t bytes, bool do_send)
{
    size_t done = 0;
    ssize_t ret;
    while (done < bytes) {
        ret = iov_send_recv(sockfd, iov, iov_cnt,
                            offset + done, bytes - done, do_send);
        if (ret > 0) {
            done += ret;
        } else if (ret < 0) {
            if (errno == EAGAIN) {
                qemu_coroutine_yield();
            } else if (done == 0) {
                return -1;
            } else {
                break;
            }
        } else if (ret == 0 && !do_send) {
            /* write (send) should never return 0.
             * read (recv) returns 0 for end-of-file (-data).
             * In both cases there's little point retrying,
             * but we do for write anyway, just in case */
            break;
        }
    }
    return done;
}

static coroutine_fn int qemu_gluster_co_rw(BlockDriverState *bs,
        int64_t sector_num, int nb_sectors, QEMUIOVector *qiov, int write)
{
    int ret;
    GlusterAIOCB *acb = g_slice_new(GlusterAIOCB);
    BDRVGlusterState *s = bs->opaque;
    size_t size = nb_sectors * BDRV_SECTOR_SIZE;
    off_t offset = sector_num * BDRV_SECTOR_SIZE;

    acb->size = size;
    acb->ret = 0;
    acb->coroutine = qemu_coroutine_self();

    if (write) {
        ret = glfs_pwritev_async(s->fd, qiov->iov, qiov->niov, offset, 0,
            &gluster_finish_aiocb, acb);
    } else {
        ret = glfs_preadv_async(s->fd, qiov->iov, qiov->niov, offset, 0,
            &gluster_finish_aiocb, acb);
    }

    if (ret < 0) {
        ret = -errno;
        goto out;
    }

    qemu_coroutine_yield();
    ret = acb->ret;

out:
    g_slice_free(GlusterAIOCB, acb);
    return ret;
}

/* A non-blocking call returned EAGAIN, so yield, ensuring the
 * handlers are set up so that we'll be rescheduled when there is an
 * interesting event on the socket.
 */
static coroutine_fn void co_yield(BDRVSSHState *s, BlockDriverState *bs)
{
    int r;
    IOHandler *rd_handler = NULL, *wr_handler = NULL;
    Coroutine *co = qemu_coroutine_self();

    r = libssh2_session_block_directions(s->session);

    if (r & LIBSSH2_SESSION_BLOCK_INBOUND) {
        rd_handler = restart_coroutine;
    }
    if (r & LIBSSH2_SESSION_BLOCK_OUTBOUND) {
        wr_handler = restart_coroutine;
    }

    DPRINTF("s->sock=%d rd_handler=%p wr_handler=%p", s->sock,
            rd_handler, wr_handler);

    aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock,
                       false, rd_handler, wr_handler, NULL, co);
    qemu_coroutine_yield();
    DPRINTF("s->sock=%d - back", s->sock);
    aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock, false,
                       NULL, NULL, NULL, NULL);
}

static coroutine_fn int qemu_gluster_co_discard(BlockDriverState *bs,
        int64_t sector_num, int nb_sectors)
{
    int ret;
    GlusterAIOCB *acb = g_slice_new(GlusterAIOCB);
    BDRVGlusterState *s = bs->opaque;
    size_t size = nb_sectors * BDRV_SECTOR_SIZE;
    off_t offset = sector_num * BDRV_SECTOR_SIZE;

    acb->size = 0;
    acb->ret = 0;
    acb->coroutine = qemu_coroutine_self();

    ret = glfs_discard_async(s->fd, offset, size, &gluster_finish_aiocb, acb);
    if (ret < 0) {
        ret = -errno;
        goto out;
    }

    qemu_coroutine_yield();
    ret = acb->ret;

out:
    g_slice_free(GlusterAIOCB, acb);
    return ret;
}

void coroutine_fn qemu_co_queue_wait(CoQueue *queue)
{
    Coroutine *self = qemu_coroutine_self();
    QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
    qemu_coroutine_yield();
    assert(qemu_in_coroutine());
}

static void nbd_co_receive_reply(NBDClientSession *s,
                                 NBDRequest *request,
                                 NBDReply *reply,
                                 QEMUIOVector *qiov)
{
    int ret;

    /* Wait until we're woken up by nbd_read_reply_entry.  */
    qemu_coroutine_yield();
    *reply = s->reply;
    if (reply->handle != request->handle ||
        !s->ioc) {
        reply->error = EIO;
    } else {
        if (qiov && reply->error == 0) {
            ret = nbd_rwv(s->ioc, qiov->iov, qiov->niov, request->len, true,
                          NULL);
            if (ret != request->len) {
                reply->error = EIO;
            }
        }

        /* Tell the read handler to read another header.  */
        s->reply.handle = 0;
    }
}

static void nbd_co_receive_reply(NbdClientSession *s,
    struct nbd_request *request, struct nbd_reply *reply,
    QEMUIOVector *qiov, int offset)
{
    int ret;

    /* Wait until we're woken up by the read handler.  TODO: perhaps
     * peek at the next reply and avoid yielding if it's ours?  */
    qemu_coroutine_yield();
    *reply = s->reply;
    if (reply->handle != request->handle) {
        reply->error = EIO;
    } else {
        if (qiov && reply->error == 0) {
            ret = qemu_co_recvv(s->sock, qiov->iov, qiov->niov,
                                offset, request->len);
            if (ret != request->len) {
                reply->error = EIO;
            }
        }

        /* Tell the read handler to read another header.  */
        s->reply.handle = 0;
    }
}

static void mirror_drain(MirrorBlockJob *s)
{
    while (s->in_flight > 0) {
        s->waiting_for_io = true;
        qemu_coroutine_yield();
        s->waiting_for_io = false;
    }
}

ssize_t nbd_wr_syncv(QIOChannel *ioc,
                     struct iovec *iov,
                     size_t niov,
                     size_t length,
                     bool do_read)
{
    ssize_t done = 0;
    Error *local_err = NULL;
    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, length);

    while (nlocal_iov > 0) {
        ssize_t len;
        if (do_read) {
            len = qio_channel_readv(ioc, local_iov, nlocal_iov, &local_err);
        } else {
            len = qio_channel_writev(ioc, local_iov, nlocal_iov, &local_err);
        }
        if (len == QIO_CHANNEL_ERR_BLOCK) {
            if (qemu_in_coroutine()) {
                /* XXX figure out if we can create a variant on
                 * qio_channel_yield() that works with AIO contexts
                 * and consider using that in this branch */
                qemu_coroutine_yield();
            } else if (done) {
                /* XXX this is needed by nbd_reply_ready.  */
                qio_channel_wait(ioc,
                                 do_read ? G_IO_IN : G_IO_OUT);
            } else {
                return -EAGAIN;
            }
            continue;
        }
        if (len < 0) {
            TRACE("I/O error: %s", error_get_pretty(local_err));
            error_free(local_err);
            /* XXX handle Error objects */
            done = -EIO;
            goto cleanup;
        }

        if (do_read && len == 0) {
            break;
        }

        iov_discard_front(&local_iov, &nlocal_iov, len);
        done += len;
    }

 cleanup:
    g_free(local_iov_head);
    return done;
}

void coroutine_fn yield_until_fd_readable(int fd)
{
    FDYieldUntilData data;

    assert(qemu_in_coroutine());
    data.co = qemu_coroutine_self();
    data.fd = fd;
    qemu_set_fd_handler(fd, fd_coroutine_enter, NULL, &data);
    qemu_coroutine_yield();
}

static void coroutine_fn mutex_fn(void *opaque)
{
    CoMutex *m = opaque;
    qemu_co_mutex_lock(m);
    assert(!locked);
    locked = true;
    qemu_coroutine_yield();
    locked = false;
    qemu_co_mutex_unlock(m);
    done++;
}

static void coroutine_fn lockable_fn(void *opaque)
{
    QemuLockable *x = opaque;
    qemu_lockable_lock(x);
    assert(!locked);
    locked = true;
    qemu_coroutine_yield();
    locked = false;
    qemu_lockable_unlock(x);
    done++;
}

void async_wait_fd(int fd, int flags)
{
    action_t *action = pthread_getspecific(my_thread_state_key);

    action->fd = fd;
    action->ready = 0;
    action->flags = flags;

    while (!action->ready) {
        qemu_coroutine_yield();
    }
}

static void coroutine_fn verify_entered_step_2(void *opaque)
{
    Coroutine *caller = (Coroutine *)opaque;

    g_assert(qemu_coroutine_entered(caller));
    g_assert(qemu_coroutine_entered(qemu_coroutine_self()));
    qemu_coroutine_yield();

    /* Once more to check it still works after yielding */
    g_assert(qemu_coroutine_entered(caller));
    g_assert(qemu_coroutine_entered(qemu_coroutine_self()));
}

void block_job_yield(BlockJob *job)
{
    assert(job->busy);

    /* Check cancellation *before* setting busy = false, too!  */
    if (block_job_is_cancelled(job)) {
        return;
    }

    job->busy = false;
    qemu_coroutine_yield();
    job->busy = true;
}

void coroutine_fn qio_channel_yield(QIOChannel *ioc,
                                    GIOCondition condition)
{
    QIOChannelYieldData data;

    assert(qemu_in_coroutine());
    data.ioc = ioc;
    data.co = qemu_coroutine_self();
    qio_channel_add_watch(ioc,
                          condition,
                          qio_channel_yield_enter,
                          &data,
                          NULL);
    qemu_coroutine_yield();
}

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

static coroutine_fn void nbd_read_reply_entry(void *opaque)
{
    NBDClientSession *s = opaque;
    uint64_t i;
    int ret = 0;
    Error *local_err = NULL;

    while (!s->quit) {
        assert(s->reply.handle == 0);
        ret = nbd_receive_reply(s->ioc, &s->reply, &local_err);
        if (ret < 0) {
            error_report_err(local_err);
        }
        if (ret <= 0) {
            break;
        }

        /* There's no need for a mutex on the receive side, because the
         * handler acts as a synchronization point and ensures that only
         * one coroutine is called until the reply finishes.
         */
        i = HANDLE_TO_INDEX(s, s->reply.handle);
        if (i >= MAX_NBD_REQUESTS ||
            !s->requests[i].coroutine ||
            !s->requests[i].receiving) {
            break;
        }

        /* We're woken up again by the request itself.  Note that there
         * is no race between yielding and reentering read_reply_co.  This
         * is because:
         *
         * - if the request runs on the same AioContext, it is only
         *   entered after we yield
         *
         * - if the request runs on a different AioContext, reentering
         *   read_reply_co happens through a bottom half, which can only
         *   run after we yield.
         */
        aio_co_wake(s->requests[i].coroutine);
        qemu_coroutine_yield();
    }

    s->quit = true;
    nbd_recv_coroutines_wake_all(s);
    s->read_reply_co = NULL;
}