async.c

   1 /*
   2  * QEMU System Emulator
   3  *
   4  * Copyright (c) 2003-2008 Fabrice Bellard
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy
   7  * of this software and associated documentation files (the "Software"), to deal
   8  * in the Software without restriction, including without limitation the rights
   9  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10  * copies of the Software, and to permit persons to whom the Software is
  11  * furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  19  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  22  * THE SOFTWARE.
  23  */
  24
  25 #include "qemu-common.h"
  26 #include "block/aio.h"
  27 #include "block/thread-pool.h"
  28 #include "qemu/main-loop.h"
  29
  30 /***********************************************************/
  31 /* bottom halves (can be seen as timers which expire ASAP) */
  32
  33 struct QEMUBH {
  34     AioContext *ctx;
  35     QEMUBHFunc *cb;
  36     void *opaque;
  37     QEMUBH *next;
  38     bool scheduled;
  39     bool idle;
  40     bool deleted;
  41 };
  42
  43 QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
  44 {
  45     QEMUBH *bh;
  46     bh = g_malloc0(sizeof(QEMUBH));
  47     bh->ctx = ctx;
  48     bh->cb = cb;
  49     bh->opaque = opaque;
  50     qemu_mutex_lock(&ctx->bh_lock);
  51     bh->next = ctx->first_bh;
  52     /* Make sure that the members are ready before putting bh into list */
  53     smp_wmb();
  54     ctx->first_bh = bh;
  55     qemu_mutex_unlock(&ctx->bh_lock);
  56     return bh;
  57 }
  58
  59 /* Multiple occurrences of aio_bh_poll cannot be called concurrently */
  60 int aio_bh_poll(AioContext *ctx)
  61 {
  62     QEMUBH *bh, **bhp, *next;
  63     int ret;
  64
  65     ctx->walking_bh++;
  66
  67     ret = 0;
  68     for (bh = ctx->first_bh; bh; bh = next) {
  69         /* Make sure that fetching bh happens before accessing its members */
  70         smp_read_barrier_depends();
  71         next = bh->next;
  72         if (!bh->deleted && bh->scheduled) {
  73             bh->scheduled = 0;
  74             /* Paired with write barrier in bh schedule to ensure reading for
  75              * idle & callbacks coming after bh's scheduling.
  76              */
  77             smp_rmb();
  78             if (!bh->idle)
  79                 ret = 1;
  80             bh->idle = 0;
  81             bh->cb(bh->opaque);
  82         }
  83     }
  84
  85     ctx->walking_bh--;
  86
  87     /* remove deleted bhs */
  88     if (!ctx->walking_bh) {
  89         qemu_mutex_lock(&ctx->bh_lock);
  90         bhp = &ctx->first_bh;
  91         while (*bhp) {
  92             bh = *bhp;
  93             if (bh->deleted) {
  94                 *bhp = bh->next;
  95                 g_free(bh);
  96             } else {
  97                 bhp = &bh->next;
  98             }
  99         }
 100         qemu_mutex_unlock(&ctx->bh_lock);
 101     }
 102
 103     return ret;
 104 }
 105
 106 void qemu_bh_schedule_idle(QEMUBH *bh)
 107 {
 108     if (bh->scheduled)
 109         return;
 110     bh->idle = 1;
 111     /* Make sure that idle & any writes needed by the callback are done
 112      * before the locations are read in the aio_bh_poll.
 113      */
 114     smp_wmb();
 115     bh->scheduled = 1;
 116 }
 117
 118 void qemu_bh_schedule(QEMUBH *bh)
 119 {
 120     AioContext *ctx;
 121
 122     if (bh->scheduled)
 123         return;
 124     ctx = bh->ctx;
 125     bh->idle = 0;
 126     /* Make sure that:
 127      * 1. idle & any writes needed by the callback are done before the
 128      *    locations are read in the aio_bh_poll.
 129      * 2. ctx is loaded before scheduled is set and the callback has a chance
 130      *    to execute.
 131      */
 132     smp_mb();
 133     bh->scheduled = 1;
 134     aio_notify(ctx);
 135 }
 136
 137
 138 /* This func is async.
 139  */
 140 void qemu_bh_cancel(QEMUBH *bh)
 141 {
 142     bh->scheduled = 0;
 143 }
 144
 145 /* This func is async.The bottom half will do the delete action at the finial
 146  * end.
 147  */
 148 void qemu_bh_delete(QEMUBH *bh)
 149 {
 150     bh->scheduled = 0;
 151     bh->deleted = 1;
 152 }
 153
 154 static gboolean
 155 aio_ctx_prepare(GSource *source, gint    *timeout)
 156 {
 157     AioContext *ctx = (AioContext *) source;
 158     QEMUBH *bh;
 159     int deadline;
 160
 161     /* We assume there is no timeout already supplied */
 162     *timeout = -1;
 163     for (bh = ctx->first_bh; bh; bh = bh->next) {
 164         if (!bh->deleted && bh->scheduled) {
 165             if (bh->idle) {
 166                 /* idle bottom halves will be polled at least
 167                  * every 10ms */
 168                 *timeout = 10;
 169             } else {
 170                 /* non-idle bottom halves will be executed
 171                  * immediately */
 172                 *timeout = 0;
 173                 return true;
 174             }
 175         }
 176     }
 177
 178     deadline = qemu_timeout_ns_to_ms(timerlistgroup_deadline_ns(&ctx->tlg));
 179     if (deadline == 0) {
 180         *timeout = 0;
 181         return true;
 182     } else {
 183         *timeout = qemu_soonest_timeout(*timeout, deadline);
 184     }
 185
 186     return false;
 187 }
 188
 189 static gboolean
 190 aio_ctx_check(GSource *source)
 191 {
 192     AioContext *ctx = (AioContext *) source;
 193     QEMUBH *bh;
 194
 195     for (bh = ctx->first_bh; bh; bh = bh->next) {
 196         if (!bh->deleted && bh->scheduled) {
 197             return true;
 198         }
 199     }
 200     return aio_pending(ctx) || (timerlistgroup_deadline_ns(&ctx->tlg) == 0);
 201 }
 202
 203 static gboolean
 204 aio_ctx_dispatch(GSource     *source,
 205                  GSourceFunc  callback,
 206                  gpointer     user_data)
 207 {
 208     AioContext *ctx = (AioContext *) source;
 209
 210     assert(callback == NULL);
 211     aio_poll(ctx, false);
 212     return true;
 213 }
 214
 215 static void
 216 aio_ctx_finalize(GSource     *source)
 217 {
 218     AioContext *ctx = (AioContext *) source;
 219
 220     thread_pool_free(ctx->thread_pool);
 221     aio_set_event_notifier(ctx, &ctx->notifier, NULL);
 222     event_notifier_cleanup(&ctx->notifier);
 223     rfifolock_destroy(&ctx->lock);
 224     qemu_mutex_destroy(&ctx->bh_lock);
 225     g_array_free(ctx->pollfds, TRUE);
 226     timerlistgroup_deinit(&ctx->tlg);
 227 }
 228
 229 static GSourceFuncs aio_source_funcs = {
 230     aio_ctx_prepare,
 231     aio_ctx_check,
 232     aio_ctx_dispatch,
 233     aio_ctx_finalize
 234 };
 235
 236 GSource *aio_get_g_source(AioContext *ctx)
 237 {
 238     g_source_ref(&ctx->source);
 239     return &ctx->source;
 240 }
 241
 242 ThreadPool *aio_get_thread_pool(AioContext *ctx)
 243 {
 244     if (!ctx->thread_pool) {
 245         ctx->thread_pool = thread_pool_new(ctx);
 246     }
 247     return ctx->thread_pool;
 248 }
 249
 250 void aio_notify(AioContext *ctx)
 251 {
 252     event_notifier_set(&ctx->notifier);
 253 }
 254
 255 static void aio_timerlist_notify(void *opaque)
 256 {
 257     aio_notify(opaque);
 258 }
 259
 260 static void aio_rfifolock_cb(void *opaque)
 261 {
 262     /* Kick owner thread in case they are blocked in aio_poll() */
 263     aio_notify(opaque);
 264 }
 265
 266 AioContext *aio_context_new(void)
 267 {
 268     AioContext *ctx;
 269     ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));
 270     ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
 271     ctx->thread_pool = NULL;
 272     qemu_mutex_init(&ctx->bh_lock);
 273     rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx);
 274     event_notifier_init(&ctx->notifier, false);
 275     aio_set_event_notifier(ctx, &ctx->notifier,
 276                            (EventNotifierHandler *)
 277                            event_notifier_test_and_clear);
 278     timerlistgroup_init(&ctx->tlg, aio_timerlist_notify, ctx);
 279
 280     return ctx;
 281 }
 282
 283 void aio_context_ref(AioContext *ctx)
 284 {
 285     g_source_ref(&ctx->source);
 286 }
 287
 288 void aio_context_unref(AioContext *ctx)
 289 {
 290     g_source_unref(&ctx->source);
 291 }
 292
 293 void aio_context_acquire(AioContext *ctx)
 294 {
 295     rfifolock_lock(&ctx->lock);
 296 }
 297
 298 void aio_context_release(AioContext *ctx)
 299 {
 300     rfifolock_unlock(&ctx->lock);
 301 }