Commit e98ab097 authored by Paolo Bonzini's avatar Paolo Bonzini Committed by Kevin Wolf
Browse files

aio-posix: move pollfds to thread-local storage 



By using thread-local storage, aio_poll can stop using global data during
g_poll_ns.  This will make it possible to drop callbacks from rfifolock.

[Moved npfd = 0 assignment to end of walking_handlers region as
suggested by Paolo.  This resolves the assert(npfd == 0) assertion
failure in pollfds_cleanup().
--Stefan]

Signed-off-by: default avatarPaolo Bonzini <pbonzini@redhat.com>
Reviewed-by: default avatarStefan Hajnoczi <stefanha@redhat.com>
Message-id: 1424449612-18215-2-git-send-email-pbonzini@redhat.com
Signed-off-by: default avatarStefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: default avatarKevin Wolf <kwolf@redhat.com>
parent de50a20a

aio-posix.c

+57 −21
Original line number Diff line number Diff line
@@ -24,7 +24,6 @@ struct AioHandler 
    IOHandler *io_read;

    IOHandler *io_write;

    int deleted;

    int pollfds_idx;

    void *opaque;

    QLIST_ENTRY(AioHandler) node;

};
@@ -83,7 +82,6 @@ void aio_set_fd_handler(AioContext *ctx, 
        node->io_read = io_read;

        node->io_write = io_write;

        node->opaque = opaque;

        node->pollfds_idx = -1;



        node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0);

        node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
@@ -186,12 +184,59 @@ bool aio_dispatch(AioContext *ctx) 
    return progress;

}



/* These thread-local variables are used only in a small part of aio_poll

 * around the call to the poll() system call.  In particular they are not

 * used while aio_poll is performing callbacks, which makes it much easier

 * to think about reentrancy!

 *

 * Stack-allocated arrays would be perfect but they have size limitations;

 * heap allocation is expensive enough that we want to reuse arrays across

 * calls to aio_poll().  And because poll() has to be called without holding

 * any lock, the arrays cannot be stored in AioContext.  Thread-local data

 * has none of the disadvantages of these three options.

 */

static __thread GPollFD *pollfds;

static __thread AioHandler **nodes;

static __thread unsigned npfd, nalloc;

static __thread Notifier pollfds_cleanup_notifier;



static void pollfds_cleanup(Notifier *n, void *unused)

{

    g_assert(npfd == 0);

    g_free(pollfds);

    g_free(nodes);

    nalloc = 0;

}



static void add_pollfd(AioHandler *node)

{

    if (npfd == nalloc) {

        if (nalloc == 0) {

            pollfds_cleanup_notifier.notify = pollfds_cleanup;

            qemu_thread_atexit_add(&pollfds_cleanup_notifier);

            nalloc = 8;

        } else {

            g_assert(nalloc <= INT_MAX);

            nalloc *= 2;

        }

        pollfds = g_renew(GPollFD, pollfds, nalloc);

        nodes = g_renew(AioHandler *, nodes, nalloc);

    }

    nodes[npfd] = node;

    pollfds[npfd] = (GPollFD) {

        .fd = node->pfd.fd,

        .events = node->pfd.events,

    };

    npfd++;

}



bool aio_poll(AioContext *ctx, bool blocking)

{

    AioHandler *node;

    bool was_dispatching;

    int ret;

    int i, ret;

    bool progress;

    int64_t timeout;



    was_dispatching = ctx->dispatching;

    progress = false;
@@ -210,39 +255,30 @@ bool aio_poll(AioContext *ctx, bool blocking) 


    ctx->walking_handlers++;



    g_array_set_size(ctx->pollfds, 0);

    assert(npfd == 0);



    /* fill pollfds */

    QLIST_FOREACH(node, &ctx->aio_handlers, node) {

        node->pollfds_idx = -1;

        if (!node->deleted && node->pfd.events) {

            GPollFD pfd = {

                .fd = node->pfd.fd,

                .events = node->pfd.events,

            };

            node->pollfds_idx = ctx->pollfds->len;

            g_array_append_val(ctx->pollfds, pfd);

            add_pollfd(node);

        }

    }



    ctx->walking_handlers--;

    timeout = blocking ? aio_compute_timeout(ctx) : 0;



    /* wait until next event */

    ret = qemu_poll_ns((GPollFD *)ctx->pollfds->data,

                         ctx->pollfds->len,

                         blocking ? aio_compute_timeout(ctx) : 0);

    ret = qemu_poll_ns((GPollFD *)pollfds, npfd, timeout);



    /* if we have any readable fds, dispatch event */

    if (ret > 0) {

        QLIST_FOREACH(node, &ctx->aio_handlers, node) {

            if (node->pollfds_idx != -1) {

                GPollFD *pfd = &g_array_index(ctx->pollfds, GPollFD,

                                              node->pollfds_idx);

                node->pfd.revents = pfd->revents;

            }

        for (i = 0; i < npfd; i++) {

            nodes[i]->pfd.revents = pollfds[i].revents;

        }

    }



    npfd = 0;

    ctx->walking_handlers--;



    /* Run dispatch even if there were no readable fds to run timers */

    aio_set_dispatching(ctx, true);

    if (aio_dispatch(ctx)) {

async.c

+0 −2
Original line number Diff line number Diff line
@@ -230,7 +230,6 @@ aio_ctx_finalize(GSource *source) 
    event_notifier_cleanup(&ctx->notifier);

    rfifolock_destroy(&ctx->lock);

    qemu_mutex_destroy(&ctx->bh_lock);

    g_array_free(ctx->pollfds, TRUE);

    timerlistgroup_deinit(&ctx->tlg);

}
@@ -302,7 +301,6 @@ AioContext *aio_context_new(Error **errp) 
    aio_set_event_notifier(ctx, &ctx->notifier,

                           (EventNotifierHandler *)

                           event_notifier_test_and_clear);

    ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));

    ctx->thread_pool = NULL;

    qemu_mutex_init(&ctx->bh_lock);

    rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx);

include/block/aio.h

+0 −3
Original line number Diff line number Diff line
@@ -82,9 +82,6 @@ struct AioContext { 
    /* Used for aio_notify.  */

    EventNotifier notifier;



    /* GPollFDs for aio_poll() */

    GArray *pollfds;



    /* Thread pool for performing work and receiving completion callbacks */

    struct ThreadPool *thread_pool;