block/throttle-groups.c

   1 /*
   2  * QEMU block throttling group infrastructure
   3  *
   4  * Copyright (C) Nodalink, EURL. 2014
   5  * Copyright (C) Igalia, S.L. 2015
   6  *
   7  * Authors:
   8  *   Benoît Canet <benoit.canet@nodalink.com>
   9  *   Alberto Garcia <berto@igalia.com>
  10  *
  11  * This program is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU General Public License as
  13  * published by the Free Software Foundation; either version 2 or
  14  * (at your option) version 3 of the License.
  15  *
  16  * This program is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  * GNU General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU General Public License
  22  * along with this program; if not, see <http://www.gnu.org/licenses/>.
  23  */
  24
  25 #include "block/throttle-groups.h"
  26 #include "qemu/queue.h"
  27 #include "qemu/thread.h"
  28 #include "sysemu/qtest.h"
  29
  30 /* The ThrottleGroup structure (with its ThrottleState) is shared
  31  * among different BlockDriverState and it's independent from
  32  * AioContext, so in order to use it from different threads it needs
  33  * its own locking.
  34  *
  35  * This locking is however handled internally in this file, so it's
  36  * transparent to outside users.
  37  *
  38  * The whole ThrottleGroup structure is private and invisible to
  39  * outside users, that only use it through its ThrottleState.
  40  *
  41  * In addition to the ThrottleGroup structure, BlockDriverState has
  42  * fields that need to be accessed by other members of the group and
  43  * therefore also need to be protected by this lock. Once a BDS is
  44  * registered in a group those fields can be accessed by other threads
  45  * any time.
  46  *
  47  * Again, all this is handled internally and is mostly transparent to
  48  * the outside. The 'throttle_timers' field however has an additional
  49  * constraint because it may be temporarily invalid (see for example
  50  * bdrv_set_aio_context()). Therefore in this file a thread will
  51  * access some other BDS's timers only after verifying that that BDS
  52  * has throttled requests in the queue.
  53  */
  54 typedef struct ThrottleGroup {
  55     char *name; /* This is constant during the lifetime of the group */
  56
  57     QemuMutex lock; /* This lock protects the following four fields */
  58     ThrottleState ts;
  59     QLIST_HEAD(, BlockDriverState) head;
  60     BlockDriverState *tokens[2];
  61     bool any_timer_armed[2];
  62
  63     /* These two are protected by the global throttle_groups_lock */
  64     unsigned refcount;
  65     QTAILQ_ENTRY(ThrottleGroup) list;
  66 } ThrottleGroup;
  67
  68 static QemuMutex throttle_groups_lock;
  69 static QTAILQ_HEAD(, ThrottleGroup) throttle_groups =
  70     QTAILQ_HEAD_INITIALIZER(throttle_groups);
  71
  72 /* Increments the reference count of a ThrottleGroup given its name.
  73  *
  74  * If no ThrottleGroup is found with the given name a new one is
  75  * created.
  76  *
  77  * @name: the name of the ThrottleGroup
  78  * @ret:  the ThrottleGroup
  79  */
  80 static ThrottleGroup *throttle_group_incref(const char *name)
  81 {
  82     ThrottleGroup *tg = NULL;
  83     ThrottleGroup *iter;
  84
  85     qemu_mutex_lock(&throttle_groups_lock);
  86
  87     /* Look for an existing group with that name */
  88     QTAILQ_FOREACH(iter, &throttle_groups, list) {
  89         if (!strcmp(name, iter->name)) {
  90             tg = iter;
  91             break;
  92         }
  93     }
  94
  95     /* Create a new one if not found */
  96     if (!tg) {
  97         tg = g_new0(ThrottleGroup, 1);
  98         tg->name = g_strdup(name);
  99         qemu_mutex_init(&tg->lock);
 100         throttle_init(&tg->ts);
 101         QLIST_INIT(&tg->head);
 102
 103         QTAILQ_INSERT_TAIL(&throttle_groups, tg, list);
 104     }
 105
 106     tg->refcount++;
 107
 108     qemu_mutex_unlock(&throttle_groups_lock);
 109
 110     return tg;
 111 }
 112
 113 /* Decrease the reference count of a ThrottleGroup.
 114  *
 115  * When the reference count reaches zero the ThrottleGroup is
 116  * destroyed.
 117  *
 118  * @tg:  The ThrottleGroup to unref
 119  */
 120 static void throttle_group_unref(ThrottleGroup *tg)
 121 {
 122     qemu_mutex_lock(&throttle_groups_lock);
 123     if (--tg->refcount == 0) {
 124         QTAILQ_REMOVE(&throttle_groups, tg, list);
 125         qemu_mutex_destroy(&tg->lock);
 126         g_free(tg->name);
 127         g_free(tg);
 128     }
 129     qemu_mutex_unlock(&throttle_groups_lock);
 130 }
 131
 132 /* Get the name from a BlockDriverState's ThrottleGroup. The name (and
 133  * the pointer) is guaranteed to remain constant during the lifetime
 134  * of the group.
 135  *
 136  * @bs:   a BlockDriverState that is member of a throttling group
 137  * @ret:  the name of the group.
 138  */
 139 const char *throttle_group_get_name(BlockDriverState *bs)
 140 {
 141     ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
 142     return tg->name;
 143 }
 144
 145 /* Return the next BlockDriverState in the round-robin sequence,
 146  * simulating a circular list.
 147  *
 148  * This assumes that tg->lock is held.
 149  *
 150  * @bs:  the current BlockDriverState
 151  * @ret: the next BlockDriverState in the sequence
 152  */
 153 static BlockDriverState *throttle_group_next_bs(BlockDriverState *bs)
 154 {
 155     ThrottleState *ts = bs->throttle_state;
 156     ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
 157     BlockDriverState *next = QLIST_NEXT(bs, round_robin);
 158
 159     if (!next) {
 160         return QLIST_FIRST(&tg->head);
 161     }
 162
 163     return next;
 164 }
 165
 166 /* Return the next BlockDriverState in the round-robin sequence with
 167  * pending I/O requests.
 168  *
 169  * This assumes that tg->lock is held.
 170  *
 171  * @bs:        the current BlockDriverState
 172  * @is_write:  the type of operation (read/write)
 173  * @ret:       the next BlockDriverState with pending requests, or bs
 174  *             if there is none.
 175  */
 176 static BlockDriverState *next_throttle_token(BlockDriverState *bs,
 177                                              bool is_write)
 178 {
 179     ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
 180     BlockDriverState *token, *start;
 181
 182     start = token = tg->tokens[is_write];
 183
 184     /* get next bs round in round robin style */
 185     token = throttle_group_next_bs(token);
 186     while (token != start && !token->pending_reqs[is_write]) {
 187         token = throttle_group_next_bs(token);
 188     }
 189
 190     /* If no IO are queued for scheduling on the next round robin token
 191      * then decide the token is the current bs because chances are
 192      * the current bs get the current request queued.
 193      */
 194     if (token == start && !token->pending_reqs[is_write]) {
 195         token = bs;
 196     }
 197
 198     return token;
 199 }
 200
 201 /* Check if the next I/O request for a BlockDriverState needs to be
 202  * throttled or not. If there's no timer set in this group, set one
 203  * and update the token accordingly.
 204  *
 205  * This assumes that tg->lock is held.
 206  *
 207  * @bs:         the current BlockDriverState
 208  * @is_write:   the type of operation (read/write)
 209  * @ret:        whether the I/O request needs to be throttled or not
 210  */
 211 static bool throttle_group_schedule_timer(BlockDriverState *bs,
 212                                           bool is_write)
 213 {
 214     ThrottleState *ts = bs->throttle_state;
 215     ThrottleTimers *tt = &bs->throttle_timers;
 216     ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
 217     bool must_wait;
 218
 219     /* Check if any of the timers in this group is already armed */
 220     if (tg->any_timer_armed[is_write]) {
 221         return true;
 222     }
 223
 224     must_wait = throttle_schedule_timer(ts, tt, is_write);
 225
 226     /* If a timer just got armed, set bs as the current token */
 227     if (must_wait) {
 228         tg->tokens[is_write] = bs;
 229         tg->any_timer_armed[is_write] = true;
 230     }
 231
 232     return must_wait;
 233 }
 234
 235 /* Look for the next pending I/O request and schedule it.
 236  *
 237  * This assumes that tg->lock is held.
 238  *
 239  * @bs:        the current BlockDriverState
 240  * @is_write:  the type of operation (read/write)
 241  */
 242 static void schedule_next_request(BlockDriverState *bs, bool is_write)
 243 {
 244     ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
 245     bool must_wait;
 246     BlockDriverState *token;
 247
 248     /* Check if there's any pending request to schedule next */
 249     token = next_throttle_token(bs, is_write);
 250     if (!token->pending_reqs[is_write]) {
 251         return;
 252     }
 253
 254     /* Set a timer for the request if it needs to be throttled */
 255     must_wait = throttle_group_schedule_timer(token, is_write);
 256
 257     /* If it doesn't have to wait, queue it for immediate execution */
 258     if (!must_wait) {
 259         /* Give preference to requests from the current bs */
 260         if (qemu_in_coroutine() &&
 261             qemu_co_queue_next(&bs->throttled_reqs[is_write])) {
 262             token = bs;
 263         } else {
 264             ThrottleTimers *tt = &token->throttle_timers;
 265             int64_t now = qemu_clock_get_ns(tt->clock_type);
 266             timer_mod(tt->timers[is_write], now + 1);
 267             tg->any_timer_armed[is_write] = true;
 268         }
 269         tg->tokens[is_write] = token;
 270     }
 271 }
 272
 273 /* Check if an I/O request needs to be throttled, wait and set a timer
 274  * if necessary, and schedule the next request using a round robin
 275  * algorithm.
 276  *
 277  * @bs:        the current BlockDriverState
 278  * @bytes:     the number of bytes for this I/O
 279  * @is_write:  the type of operation (read/write)
 280  */
 281 void coroutine_fn throttle_group_co_io_limits_intercept(BlockDriverState *bs,
 282                                                         unsigned int bytes,
 283                                                         bool is_write)
 284 {
 285     bool must_wait;
 286     BlockDriverState *token;
 287
 288     ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
 289     qemu_mutex_lock(&tg->lock);
 290
 291     /* First we check if this I/O has to be throttled. */
 292     token = next_throttle_token(bs, is_write);
 293     must_wait = throttle_group_schedule_timer(token, is_write);
 294
 295     /* Wait if there's a timer set or queued requests of this type */
 296     if (must_wait || bs->pending_reqs[is_write]) {
 297         bs->pending_reqs[is_write]++;
 298         qemu_mutex_unlock(&tg->lock);
 299         qemu_co_queue_wait(&bs->throttled_reqs[is_write]);
 300         qemu_mutex_lock(&tg->lock);
 301         bs->pending_reqs[is_write]--;
 302     }
 303
 304     /* The I/O will be executed, so do the accounting */
 305     throttle_account(bs->throttle_state, is_write, bytes);
 306
 307     /* Schedule the next request */
 308     schedule_next_request(bs, is_write);
 309
 310     qemu_mutex_unlock(&tg->lock);
 311 }
 312
 313 /* Update the throttle configuration for a particular group. Similar
 314  * to throttle_config(), but guarantees atomicity within the
 315  * throttling group.
 316  *
 317  * @bs:  a BlockDriverState that is member of the group
 318  * @cfg: the configuration to set
 319  */
 320 void throttle_group_config(BlockDriverState *bs, ThrottleConfig *cfg)
 321 {
 322     ThrottleTimers *tt = &bs->throttle_timers;
 323     ThrottleState *ts = bs->throttle_state;
 324     ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
 325     qemu_mutex_lock(&tg->lock);
 326     throttle_config(ts, tt, cfg);
 327     /* throttle_config() cancels the timers */
 328     tg->any_timer_armed[0] = tg->any_timer_armed[1] = false;
 329     qemu_mutex_unlock(&tg->lock);
 330 }
 331
 332 /* Get the throttle configuration from a particular group. Similar to
 333  * throttle_get_config(), but guarantees atomicity within the
 334  * throttling group.
 335  *
 336  * @bs:  a BlockDriverState that is member of the group
 337  * @cfg: the configuration will be written here
 338  */
 339 void throttle_group_get_config(BlockDriverState *bs, ThrottleConfig *cfg)
 340 {
 341     ThrottleState *ts = bs->throttle_state;
 342     ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
 343     qemu_mutex_lock(&tg->lock);
 344     throttle_get_config(ts, cfg);
 345     qemu_mutex_unlock(&tg->lock);
 346 }
 347
 348 /* ThrottleTimers callback. This wakes up a request that was waiting
 349  * because it had been throttled.
 350  *
 351  * @bs:        the BlockDriverState whose request had been throttled
 352  * @is_write:  the type of operation (read/write)
 353  */
 354 static void timer_cb(BlockDriverState *bs, bool is_write)
 355 {
 356     ThrottleState *ts = bs->throttle_state;
 357     ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
 358     bool empty_queue;
 359
 360     /* The timer has just been fired, so we can update the flag */
 361     qemu_mutex_lock(&tg->lock);
 362     tg->any_timer_armed[is_write] = false;
 363     qemu_mutex_unlock(&tg->lock);
 364
 365     /* Run the request that was waiting for this timer */
 366     empty_queue = !qemu_co_enter_next(&bs->throttled_reqs[is_write]);
 367
 368     /* If the request queue was empty then we have to take care of
 369      * scheduling the next one */
 370     if (empty_queue) {
 371         qemu_mutex_lock(&tg->lock);
 372         schedule_next_request(bs, is_write);
 373         qemu_mutex_unlock(&tg->lock);
 374     }
 375 }
 376
 377 static void read_timer_cb(void *opaque)
 378 {
 379     timer_cb(opaque, false);
 380 }
 381
 382 static void write_timer_cb(void *opaque)
 383 {
 384     timer_cb(opaque, true);
 385 }
 386
 387 /* Register a BlockDriverState in the throttling group, also
 388  * initializing its timers and updating its throttle_state pointer to
 389  * point to it. If a throttling group with that name does not exist
 390  * yet, it will be created.
 391  *
 392  * @bs:        the BlockDriverState to insert
 393  * @groupname: the name of the group
 394  */
 395 void throttle_group_register_bs(BlockDriverState *bs, const char *groupname)
 396 {
 397     int i;
 398     ThrottleGroup *tg = throttle_group_incref(groupname);
 399     int clock_type = QEMU_CLOCK_REALTIME;
 400
 401     if (qtest_enabled()) {
 402         /* For testing block IO throttling only */
 403         clock_type = QEMU_CLOCK_VIRTUAL;
 404     }
 405
 406     bs->throttle_state = &tg->ts;
 407
 408     qemu_mutex_lock(&tg->lock);
 409     /* If the ThrottleGroup is new set this BlockDriverState as the token */
 410     for (i = 0; i < 2; i++) {
 411         if (!tg->tokens[i]) {
 412             tg->tokens[i] = bs;
 413         }
 414     }
 415
 416     QLIST_INSERT_HEAD(&tg->head, bs, round_robin);
 417
 418     throttle_timers_init(&bs->throttle_timers,
 419                          bdrv_get_aio_context(bs),
 420                          clock_type,
 421                          read_timer_cb,
 422                          write_timer_cb,
 423                          bs);
 424
 425     qemu_mutex_unlock(&tg->lock);
 426 }
 427
 428 /* Unregister a BlockDriverState from its group, removing it from the
 429  * list, destroying the timers and setting the throttle_state pointer
 430  * to NULL.
 431  *
 432  * The group will be destroyed if it's empty after this operation.
 433  *
 434  * @bs: the BlockDriverState to remove
 435  */
 436 void throttle_group_unregister_bs(BlockDriverState *bs)
 437 {
 438     ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
 439     int i;
 440
 441     qemu_mutex_lock(&tg->lock);
 442     for (i = 0; i < 2; i++) {
 443         if (tg->tokens[i] == bs) {
 444             BlockDriverState *token = throttle_group_next_bs(bs);
 445             /* Take care of the case where this is the last bs in the group */
 446             if (token == bs) {
 447                 token = NULL;
 448             }
 449             tg->tokens[i] = token;
 450         }
 451     }
 452
 453     /* remove the current bs from the list */
 454     QLIST_REMOVE(bs, round_robin);
 455     throttle_timers_destroy(&bs->throttle_timers);
 456     qemu_mutex_unlock(&tg->lock);
 457
 458     throttle_group_unref(tg);
 459     bs->throttle_state = NULL;
 460 }
 461
 462 static void throttle_groups_init(void)
 463 {
 464     qemu_mutex_init(&throttle_groups_lock);
 465 }
 466
 467 block_init(throttle_groups_init);