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.h"
  18 #include "qemu-queue.h"
  19 #include "qemu_socket.h"
  20
  21 struct AioHandler
  22 {
  23     GPollFD pfd;
  24     IOHandler *io_read;
  25     IOHandler *io_write;
  26     AioFlushHandler *io_flush;
  27     int deleted;
  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->io_flush = io_flush;
  87         node->opaque = opaque;
  88
  89         node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP : 0);
  90         node->pfd.events |= (io_write ? G_IO_OUT : 0);
  91     }
  92 }
  93
  94 void aio_set_event_notifier(AioContext *ctx,
  95                             EventNotifier *notifier,
  96                             EventNotifierHandler *io_read,
  97                             AioFlushEventNotifierHandler *io_flush)
  98 {
  99     aio_set_fd_handler(ctx, event_notifier_get_fd(notifier),
 100                        (IOHandler *)io_read, NULL,
 101                        (AioFlushHandler *)io_flush, notifier);
 102 }
 103
 104 bool aio_pending(AioContext *ctx)
 105 {
 106     AioHandler *node;
 107
 108     QLIST_FOREACH(node, &ctx->aio_handlers, node) {
 109         int revents;
 110
 111         /*
 112          * FIXME: right now we cannot get G_IO_HUP and G_IO_ERR because
 113          * main-loop.c is still select based (due to the slirp legacy).
 114          * If main-loop.c ever switches to poll, G_IO_ERR should be
 115          * tested too.  Dispatching G_IO_ERR to both handlers should be
 116          * okay, since handlers need to be ready for spurious wakeups.
 117          */
 118         revents = node->pfd.revents & node->pfd.events;
 119         if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read) {
 120             return true;
 121         }
 122         if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write) {
 123             return true;
 124         }
 125     }
 126
 127     return false;
 128 }
 129
 130 bool aio_poll(AioContext *ctx, bool blocking)
 131 {
 132     static struct timeval tv0;
 133     AioHandler *node;
 134     fd_set rdfds, wrfds;
 135     int max_fd = -1;
 136     int ret;
 137     bool busy, progress;
 138
 139     progress = false;
 140
 141     /*
 142      * If there are callbacks left that have been queued, we need to call then.
 143      * Do not call select in this case, because it is possible that the caller
 144      * does not need a complete flush (as is the case for qemu_aio_wait loops).
 145      */
 146     if (aio_bh_poll(ctx)) {
 147         blocking = false;
 148         progress = true;
 149     }
 150
 151     /*
 152      * Then dispatch any pending callbacks from the GSource.
 153      *
 154      * We have to walk very carefully in case qemu_aio_set_fd_handler is
 155      * called while we're walking.
 156      */
 157     node = QLIST_FIRST(&ctx->aio_handlers);
 158     while (node) {
 159         AioHandler *tmp;
 160         int revents;
 161
 162         ctx->walking_handlers++;
 163
 164         revents = node->pfd.revents & node->pfd.events;
 165         node->pfd.revents = 0;
 166
 167         /* See comment in aio_pending.  */
 168         if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read) {
 169             node->io_read(node->opaque);
 170             progress = true;
 171         }
 172         if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write) {
 173             node->io_write(node->opaque);
 174             progress = true;
 175         }
 176
 177         tmp = node;
 178         node = QLIST_NEXT(node, node);
 179
 180         ctx->walking_handlers--;
 181
 182         if (!ctx->walking_handlers && tmp->deleted) {
 183             QLIST_REMOVE(tmp, node);
 184             g_free(tmp);
 185         }
 186     }
 187
 188     if (progress && !blocking) {
 189         return true;
 190     }
 191
 192     ctx->walking_handlers++;
 193
 194     FD_ZERO(&rdfds);
 195     FD_ZERO(&wrfds);
 196
 197     /* fill fd sets */
 198     busy = false;
 199     QLIST_FOREACH(node, &ctx->aio_handlers, node) {
 200         /* If there aren't pending AIO operations, don't invoke callbacks.
 201          * Otherwise, if there are no AIO requests, qemu_aio_wait() would
 202          * wait indefinitely.
 203          */
 204         if (!node->deleted && node->io_flush) {
 205             if (node->io_flush(node->opaque) == 0) {
 206                 continue;
 207             }
 208             busy = true;
 209         }
 210         if (!node->deleted && node->io_read) {
 211             FD_SET(node->pfd.fd, &rdfds);
 212             max_fd = MAX(max_fd, node->pfd.fd + 1);
 213         }
 214         if (!node->deleted && node->io_write) {
 215             FD_SET(node->pfd.fd, &wrfds);
 216             max_fd = MAX(max_fd, node->pfd.fd + 1);
 217         }
 218     }
 219
 220     ctx->walking_handlers--;
 221
 222     /* No AIO operations?  Get us out of here */
 223     if (!busy) {
 224         return progress;
 225     }
 226
 227     /* wait until next event */
 228     ret = select(max_fd, &rdfds, &wrfds, NULL, blocking ? NULL : &tv0);
 229
 230     /* if we have any readable fds, dispatch event */
 231     if (ret > 0) {
 232         /* we have to walk very carefully in case
 233          * qemu_aio_set_fd_handler is called while we're walking */
 234         node = QLIST_FIRST(&ctx->aio_handlers);
 235         while (node) {
 236             AioHandler *tmp;
 237
 238             ctx->walking_handlers++;
 239
 240             if (!node->deleted &&
 241                 FD_ISSET(node->pfd.fd, &rdfds) &&
 242                 node->io_read) {
 243                 node->io_read(node->opaque);
 244                 progress = true;
 245             }
 246             if (!node->deleted &&
 247                 FD_ISSET(node->pfd.fd, &wrfds) &&
 248                 node->io_write) {
 249                 node->io_write(node->opaque);
 250                 progress = true;
 251             }
 252
 253             tmp = node;
 254             node = QLIST_NEXT(node, node);
 255
 256             ctx->walking_handlers--;
 257
 258             if (!ctx->walking_handlers && tmp->deleted) {
 259                 QLIST_REMOVE(tmp, node);
 260                 g_free(tmp);
 261             }
 262         }
 263     }
 264
 265     return progress;
 266 }