aio-posix.c

   1 /*
   2  * QEMU aio implementation
   3  *
   4  * Copyright IBM, Corp. 2008
   5  *
   6  * Authors:
   7  *  Anthony Liguori   <aliguori@us.ibm.com>
   8  *
   9  * This work is licensed under the terms of the GNU GPL, version 2.  See
  10  * the COPYING file in the top-level directory.
  11  *
  12  * Contributions after 2012-01-13 are licensed under the terms of the
  13  * GNU GPL, version 2 or (at your option) any later version.
  14  */
  15
  16 #include "qemu-common.h"
  17 #include "block/block.h"
  18 #include "qemu/queue.h"
  19 #include "qemu/sockets.h"
  20
  21 struct AioHandler
  22 {
  23     GPollFD pfd;
  24     IOHandler *io_read;
  25     IOHandler *io_write;
  26     int deleted;
  27     int pollfds_idx;
  28     void *opaque;
  29     QLIST_ENTRY(AioHandler) node;
  30 };
  31
  32 static AioHandler *find_aio_handler(AioContext *ctx, int fd)
  33 {
  34     AioHandler *node;
  35
  36     QLIST_FOREACH(node, &ctx->aio_handlers, node) {
  37         if (node->pfd.fd == fd)
  38             if (!node->deleted)
  39                 return node;
  40     }
  41
  42     return NULL;
  43 }
  44
  45 void aio_set_fd_handler(AioContext *ctx,
  46                         int fd,
  47                         IOHandler *io_read,
  48                         IOHandler *io_write,
  49                         AioFlushHandler *io_flush,
  50                         void *opaque)
  51 {
  52     AioHandler *node;
  53
  54     node = find_aio_handler(ctx, fd);
  55
  56     /* Are we deleting the fd handler? */
  57     if (!io_read && !io_write) {
  58         if (node) {
  59             g_source_remove_poll(&ctx->source, &node->pfd);
  60
  61             /* If the lock is held, just mark the node as deleted */
  62             if (ctx->walking_handlers) {
  63                 node->deleted = 1;
  64                 node->pfd.revents = 0;
  65             } else {
  66                 /* Otherwise, delete it for real.  We can't just mark it as
  67                  * deleted because deleted nodes are only cleaned up after
  68                  * releasing the walking_handlers lock.
  69                  */
  70                 QLIST_REMOVE(node, node);
  71                 g_free(node);
  72             }
  73         }
  74     } else {
  75         if (node == NULL) {
  76             /* Alloc and insert if it's not already there */
  77             node = g_malloc0(sizeof(AioHandler));
  78             node->pfd.fd = fd;
  79             QLIST_INSERT_HEAD(&ctx->aio_handlers, node, node);
  80
  81             g_source_add_poll(&ctx->source, &node->pfd);
  82         }
  83         /* Update handler with latest information */
  84         node->io_read = io_read;
  85         node->io_write = io_write;
  86         node->opaque = opaque;
  87         node->pollfds_idx = -1;
  88
  89         node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0);
  90         node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
  91     }
  92
  93     aio_notify(ctx);
  94 }
  95
  96 void aio_set_event_notifier(AioContext *ctx,
  97                             EventNotifier *notifier,
  98                             EventNotifierHandler *io_read,
  99                             AioFlushEventNotifierHandler *io_flush)
 100 {
 101     aio_set_fd_handler(ctx, event_notifier_get_fd(notifier),
 102                        (IOHandler *)io_read, NULL,
 103                        (AioFlushHandler *)io_flush, notifier);
 104 }
 105
 106 bool aio_pending(AioContext *ctx)
 107 {
 108     AioHandler *node;
 109
 110     QLIST_FOREACH(node, &ctx->aio_handlers, node) {
 111         int revents;
 112
 113         revents = node->pfd.revents & node->pfd.events;
 114         if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read) {
 115             return true;
 116         }
 117         if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write) {
 118             return true;
 119         }
 120     }
 121
 122     return false;
 123 }
 124
 125 static bool aio_dispatch(AioContext *ctx)
 126 {
 127     AioHandler *node;
 128     bool progress = false;
 129
 130     /*
 131      * We have to walk very carefully in case qemu_aio_set_fd_handler is
 132      * called while we're walking.
 133      */
 134     node = QLIST_FIRST(&ctx->aio_handlers);
 135     while (node) {
 136         AioHandler *tmp;
 137         int revents;
 138
 139         ctx->walking_handlers++;
 140
 141         revents = node->pfd.revents & node->pfd.events;
 142         node->pfd.revents = 0;
 143
 144         if (!node->deleted &&
 145             (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) &&
 146             node->io_read) {
 147             node->io_read(node->opaque);
 148
 149             /* aio_notify() does not count as progress */
 150             if (node->opaque != &ctx->notifier) {
 151                 progress = true;
 152             }
 153         }
 154         if (!node->deleted &&
 155             (revents & (G_IO_OUT | G_IO_ERR)) &&
 156             node->io_write) {
 157             node->io_write(node->opaque);
 158             progress = true;
 159         }
 160
 161         tmp = node;
 162         node = QLIST_NEXT(node, node);
 163
 164         ctx->walking_handlers--;
 165
 166         if (!ctx->walking_handlers && tmp->deleted) {
 167             QLIST_REMOVE(tmp, node);
 168             g_free(tmp);
 169         }
 170     }
 171     return progress;
 172 }
 173
 174 bool aio_poll(AioContext *ctx, bool blocking)
 175 {
 176     AioHandler *node;
 177     int ret;
 178     bool progress;
 179
 180     progress = false;
 181
 182     /*
 183      * If there are callbacks left that have been queued, we need to call them.
 184      * Do not call select in this case, because it is possible that the caller
 185      * does not need a complete flush (as is the case for qemu_aio_wait loops).
 186      */
 187     if (aio_bh_poll(ctx)) {
 188         blocking = false;
 189         progress = true;
 190     }
 191
 192     if (aio_dispatch(ctx)) {
 193         progress = true;
 194     }
 195
 196     if (progress && !blocking) {
 197         return true;
 198     }
 199
 200     ctx->walking_handlers++;
 201
 202     g_array_set_size(ctx->pollfds, 0);
 203
 204     /* fill pollfds */
 205     QLIST_FOREACH(node, &ctx->aio_handlers, node) {
 206         node->pollfds_idx = -1;
 207         if (!node->deleted && node->pfd.events) {
 208             GPollFD pfd = {
 209                 .fd = node->pfd.fd,
 210                 .events = node->pfd.events,
 211             };
 212             node->pollfds_idx = ctx->pollfds->len;
 213             g_array_append_val(ctx->pollfds, pfd);
 214         }
 215     }
 216
 217     ctx->walking_handlers--;
 218
 219     /* early return if we only have the aio_notify() fd */
 220     if (ctx->pollfds->len == 1) {
 221         return progress;
 222     }
 223
 224     /* wait until next event */
 225     ret = g_poll((GPollFD *)ctx->pollfds->data,
 226                  ctx->pollfds->len,
 227                  blocking ? -1 : 0);
 228
 229     /* if we have any readable fds, dispatch event */
 230     if (ret > 0) {
 231         QLIST_FOREACH(node, &ctx->aio_handlers, node) {
 232             if (node->pollfds_idx != -1) {
 233                 GPollFD *pfd = &g_array_index(ctx->pollfds, GPollFD,
 234                                               node->pollfds_idx);
 235                 node->pfd.revents = pfd->revents;
 236             }
 237         }
 238         if (aio_dispatch(ctx)) {
 239             progress = true;
 240         }
 241     }
 242
 243     return progress;
 244 }