[Qemu-devel] [PATCH v3 0/6] Block Throttle Group Support

[Qemu-devel] [PATCH v3 0/6] Block Throttle Group Support

[Alberto Garcia]

Hello,

here's the new version of the block throttling patches. After the
discussions from the previous thread, this one is significantly
rewritten from the original code.

The main change is in the group throttling API, which is -I believe-
much simpler now. The code in block.c and throttle.c also needs far
fewer changes.

Here's what's new:

- The following calls are no longer part of the API, and are now
  handled internally:

  throttle_group_incref()
  throttle_group_unref()
  throttle_group_set_token()
  throttle_group_token()
  throttle_group_lock()
  throttle_group_unlock()

- New throttle_group_config() and throttle_group_get_config() calls,
  that can be called concurrently by the members of the same group.

- bdrv_throttle_group_add/remove() have been merged into
  throttle_group_register/unregister_bs() and moved to
  throttle-groups.c.

  The result of this is that in the sequence throttle_group_incref() +
  bdrv_throttle_group_add() + throttle_group_register_bs() only this
  last calls remains and takes care of everything else.

- The algorithm in bdrv_io_limits_intercept() and friends has been
  simplified and moved to throttle-groups.c.

- The any_timer_armed array no longer exists. The group throttling
  code checks the timers directly to see if any of them is pending.

- The tests have been adapted to the new API.

- The code is hopefully better documented now.

There's still the question of what to do with the QMP API to get all
throttling groups, but since there is no consensus yet I haven't
included it in this series.

And I think that's all.

As usual, comments and feedback are welcome.

Berto

Alberto Garcia (5):
  throttle: Add throttle group infrastructure
  throttle: Add throttle group infrastructure tests
  throttle: Add throttle group support
  throttle: add the name of the ThrottleGroup to BlockDeviceInfo
  throttle: Update throttle infrastructure copyright

Benoît Canet (1):
  throttle: Extract timers from ThrottleState into a separate structure

 block.c                         |  82 ++++-----
 block/Makefile.objs             |   1 +
 block/qapi.c                    |   8 +-
 block/throttle-groups.c         | 371 ++++++++++++++++++++++++++++++++++++++++
 blockdev.c                      |  19 +-
 hmp.c                           |  10 +-
 include/block/block.h           |   3 +-
 include/block/block_int.h       |   9 +-
 include/block/throttle-groups.h |  45 +++++
 include/qemu/throttle.h         |  46 +++--
 qapi/block-core.json            |   8 +-
 qemu-options.hx                 |   1 +
 qmp-commands.hx                 |   3 +-
 tests/test-throttle.c           | 119 +++++++++----
 util/throttle.c                 |  81 +++++----
 15 files changed, 665 insertions(+), 141 deletions(-)
 create mode 100644 block/throttle-groups.c
 create mode 100644 include/block/throttle-groups.h

-- 
2.1.4

[Qemu-devel] [PATCH 1/6] throttle: Extract timers from ThrottleState into a separate structure

[Alberto Garcia]

From: Benoît Canet <benoit.canet@nodalink.com>

Group throttling will share ThrottleState between multiple bs.
As a consequence the ThrottleState will be accessed by multiple aio
context.

Timers are tied to their aio context so they must go out of the
ThrottleState structure.

This commit paves the way for each bs of a common ThrottleState to
have its own timer.

Signed-off-by: Benoit Canet <benoit.canet@nodalink.com>
Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 block.c                   | 35 ++++++++++++--------
 include/block/block_int.h |  1 +
 include/qemu/throttle.h   | 38 ++++++++++++++--------
 tests/test-throttle.c     | 82 ++++++++++++++++++++++++++---------------------
 util/throttle.c           | 73 ++++++++++++++++++++++++-----------------
 5 files changed, 135 insertions(+), 94 deletions(-)

diff --git a/block.c b/block.c
index 28ea19a..da0cb48 100644
--- a/block.c
+++ b/block.c
@@ -130,7 +130,7 @@ void bdrv_set_io_limits(BlockDriverState *bs,
 {
     int i;
 
-    throttle_config(&bs->throttle_state, cfg);
+    throttle_config(&bs->throttle_state, &bs->throttle_timers, cfg);
 
     for (i = 0; i < 2; i++) {
         qemu_co_enter_next(&bs->throttled_reqs[i]);
@@ -163,7 +163,7 @@ void bdrv_io_limits_disable(BlockDriverState *bs)
 
     bdrv_start_throttled_reqs(bs);
 
-    throttle_destroy(&bs->throttle_state);
+    throttle_timers_destroy(&bs->throttle_timers);
 }
 
 static void bdrv_throttle_read_timer_cb(void *opaque)
@@ -182,12 +182,13 @@ static void bdrv_throttle_write_timer_cb(void *opaque)
 void bdrv_io_limits_enable(BlockDriverState *bs)
 {
     assert(!bs->io_limits_enabled);
-    throttle_init(&bs->throttle_state,
-                  bdrv_get_aio_context(bs),
-                  QEMU_CLOCK_VIRTUAL,
-                  bdrv_throttle_read_timer_cb,
-                  bdrv_throttle_write_timer_cb,
-                  bs);
+    throttle_init(&bs->throttle_state);
+    throttle_timers_init(&bs->throttle_timers,
+                         bdrv_get_aio_context(bs),
+                         QEMU_CLOCK_VIRTUAL,
+                         bdrv_throttle_read_timer_cb,
+                         bdrv_throttle_write_timer_cb,
+                         bs);
     bs->io_limits_enabled = true;
 }
 
@@ -201,7 +202,9 @@ static void bdrv_io_limits_intercept(BlockDriverState *bs,
                                      bool is_write)
 {
     /* does this io must wait */
-    bool must_wait = throttle_schedule_timer(&bs->throttle_state, is_write);
+    bool must_wait = throttle_schedule_timer(&bs->throttle_state,
+                                             &bs->throttle_timers,
+                                             is_write);
 
     /* if must wait or any request of this type throttled queue the IO */
     if (must_wait ||
@@ -214,7 +217,8 @@ static void bdrv_io_limits_intercept(BlockDriverState *bs,
 
 
     /* if the next request must wait -> do nothing */
-    if (throttle_schedule_timer(&bs->throttle_state, is_write)) {
+    if (throttle_schedule_timer(&bs->throttle_state, &bs->throttle_timers,
+                                is_write)) {
         return;
     }
 
@@ -2049,6 +2053,9 @@ static void bdrv_move_feature_fields(BlockDriverState *bs_dest,
     memcpy(&bs_dest->throttle_state,
            &bs_src->throttle_state,
            sizeof(ThrottleState));
+    memcpy(&bs_dest->throttle_timers,
+           &bs_src->throttle_timers,
+           sizeof(ThrottleTimers));
     bs_dest->throttled_reqs[0]  = bs_src->throttled_reqs[0];
     bs_dest->throttled_reqs[1]  = bs_src->throttled_reqs[1];
     bs_dest->io_limits_enabled  = bs_src->io_limits_enabled;
@@ -2110,7 +2117,7 @@ void bdrv_swap(BlockDriverState *bs_new, BlockDriverState *bs_old)
     assert(QLIST_EMPTY(&bs_new->dirty_bitmaps));
     assert(bs_new->job == NULL);
     assert(bs_new->io_limits_enabled == false);
-    assert(!throttle_have_timer(&bs_new->throttle_state));
+    assert(!throttle_timers_are_initialized(&bs_new->throttle_timers));
 
     tmp = *bs_new;
     *bs_new = *bs_old;
@@ -2127,7 +2134,7 @@ void bdrv_swap(BlockDriverState *bs_new, BlockDriverState *bs_old)
     /* Check a few fields that should remain attached to the device */
     assert(bs_new->job == NULL);
     assert(bs_new->io_limits_enabled == false);
-    assert(!throttle_have_timer(&bs_new->throttle_state));
+    assert(!throttle_timers_are_initialized(&bs_new->throttle_timers));
 
     /* insert the nodes back into the graph node list if needed */
     if (bs_new->node_name[0] != '\0') {
@@ -5793,7 +5800,7 @@ void bdrv_detach_aio_context(BlockDriverState *bs)
     }
 
     if (bs->io_limits_enabled) {
-        throttle_detach_aio_context(&bs->throttle_state);
+        throttle_timers_detach_aio_context(&bs->throttle_timers);
     }
     if (bs->drv->bdrv_detach_aio_context) {
         bs->drv->bdrv_detach_aio_context(bs);
@@ -5829,7 +5836,7 @@ void bdrv_attach_aio_context(BlockDriverState *bs,
         bs->drv->bdrv_attach_aio_context(bs, new_context);
     }
     if (bs->io_limits_enabled) {
-        throttle_attach_aio_context(&bs->throttle_state, new_context);
+        throttle_timers_attach_aio_context(&bs->throttle_timers, new_context);
     }
 
     QLIST_FOREACH(ban, &bs->aio_notifiers, list) {
diff --git a/include/block/block_int.h b/include/block/block_int.h
index b340e7e..58b3a5f 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -360,6 +360,7 @@ struct BlockDriverState {
 
     /* I/O throttling */
     ThrottleState throttle_state;
+    ThrottleTimers throttle_timers;
     CoQueue      throttled_reqs[2];
     bool         io_limits_enabled;
 
diff --git a/include/qemu/throttle.h b/include/qemu/throttle.h
index b890613..2c560db 100644
--- a/include/qemu/throttle.h
+++ b/include/qemu/throttle.h
@@ -65,14 +65,17 @@ typedef struct ThrottleConfig {
 typedef struct ThrottleState {
     ThrottleConfig cfg;       /* configuration */
     int64_t previous_leak;    /* timestamp of the last leak done */
-    QEMUTimer * timers[2];    /* timers used to do the throttling */
+} ThrottleState;
+
+typedef struct ThrottleTimers {
+    QEMUTimer *timers[2];     /* timers used to do the throttling */
     QEMUClockType clock_type; /* the clock used */
 
     /* Callbacks */
     QEMUTimerCB *read_timer_cb;
     QEMUTimerCB *write_timer_cb;
     void *timer_opaque;
-} ThrottleState;
+} ThrottleTimers;
 
 /* operations on single leaky buckets */
 void throttle_leak_bucket(LeakyBucket *bkt, int64_t delta);
@@ -86,20 +89,23 @@ bool throttle_compute_timer(ThrottleState *ts,
                             int64_t *next_timestamp);
 
 /* init/destroy cycle */
-void throttle_init(ThrottleState *ts,
-                   AioContext *aio_context,
-                   QEMUClockType clock_type,
-                   void (read_timer)(void *),
-                   void (write_timer)(void *),
-                   void *timer_opaque);
+void throttle_init(ThrottleState *ts);
+
+void throttle_timers_init(ThrottleTimers *tt,
+                          AioContext *aio_context,
+                          QEMUClockType clock_type,
+                          QEMUTimerCB *read_timer_cb,
+                          QEMUTimerCB *write_timer_cb,
+                          void *timer_opaque);
 
-void throttle_destroy(ThrottleState *ts);
+void throttle_timers_destroy(ThrottleTimers *tt);
 
-void throttle_detach_aio_context(ThrottleState *ts);
+void throttle_timers_detach_aio_context(ThrottleTimers *tt);
 
-void throttle_attach_aio_context(ThrottleState *ts, AioContext *new_context);
+void throttle_timers_attach_aio_context(ThrottleTimers *tt,
+                                        AioContext *new_context);
 
-bool throttle_have_timer(ThrottleState *ts);
+bool throttle_timers_are_initialized(ThrottleTimers *tt);
 
 /* configuration */
 bool throttle_enabled(ThrottleConfig *cfg);
@@ -108,12 +114,16 @@ bool throttle_conflicting(ThrottleConfig *cfg);
 
 bool throttle_is_valid(ThrottleConfig *cfg);
 
-void throttle_config(ThrottleState *ts, ThrottleConfig *cfg);
+void throttle_config(ThrottleState *ts,
+                     ThrottleTimers *tt,
+                     ThrottleConfig *cfg);
 
 void throttle_get_config(ThrottleState *ts, ThrottleConfig *cfg);
 
 /* usage */
-bool throttle_schedule_timer(ThrottleState *ts, bool is_write);
+bool throttle_schedule_timer(ThrottleState *ts,
+                             ThrottleTimers *tt,
+                             bool is_write);
 
 void throttle_account(ThrottleState *ts, bool is_write, uint64_t size);
 
diff --git a/tests/test-throttle.c b/tests/test-throttle.c
index d8ba415..458f577 100644
--- a/tests/test-throttle.c
+++ b/tests/test-throttle.c
@@ -20,6 +20,7 @@ static AioContext     *ctx;
 static LeakyBucket    bkt;
 static ThrottleConfig cfg;
 static ThrottleState  ts;
+static ThrottleTimers tt;
 
 /* useful function */
 static bool double_cmp(double x, double y)
@@ -103,17 +104,19 @@ static void test_init(void)
 {
     int i;
 
-    /* fill the structure with crap */
+    /* fill the structures with crap */
     memset(&ts, 1, sizeof(ts));
+    memset(&tt, 1, sizeof(tt));
 
-    /* init the structure */
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
+    /* init structures */
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
 
     /* check initialized fields */
-    g_assert(ts.clock_type == QEMU_CLOCK_VIRTUAL);
-    g_assert(ts.timers[0]);
-    g_assert(ts.timers[1]);
+    g_assert(tt.clock_type == QEMU_CLOCK_VIRTUAL);
+    g_assert(tt.timers[0]);
+    g_assert(tt.timers[1]);
 
     /* check other fields where cleared */
     g_assert(!ts.previous_leak);
@@ -124,17 +127,18 @@ static void test_init(void)
         g_assert(!ts.cfg.buckets[i].level);
     }
 
-    throttle_destroy(&ts);
+    throttle_timers_destroy(&tt);
 }
 
 static void test_destroy(void)
 {
     int i;
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
-    throttle_destroy(&ts);
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
+    throttle_timers_destroy(&tt);
     for (i = 0; i < 2; i++) {
-        g_assert(!ts.timers[i]);
+        g_assert(!tt.timers[i]);
     }
 }
 
@@ -170,11 +174,12 @@ static void test_config_functions(void)
 
     orig_cfg.op_size = 1;
 
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
     /* structure reset by throttle_init previous_leak should be null */
     g_assert(!ts.previous_leak);
-    throttle_config(&ts, &orig_cfg);
+    throttle_config(&ts, &tt, &orig_cfg);
 
     /* has previous leak been initialized by throttle_config ? */
     g_assert(ts.previous_leak);
@@ -182,7 +187,7 @@ static void test_config_functions(void)
     /* get back the fixed configuration */
     throttle_get_config(&ts, &final_cfg);
 
-    throttle_destroy(&ts);
+    throttle_timers_destroy(&tt);
 
     g_assert(final_cfg.buckets[THROTTLE_BPS_TOTAL].avg == 153);
     g_assert(final_cfg.buckets[THROTTLE_BPS_READ].avg  == 56);
@@ -323,43 +328,47 @@ static void test_is_valid(void)
 
 static void test_have_timer(void)
 {
-    /* zero the structure */
+    /* zero structures */
     memset(&ts, 0, sizeof(ts));
+    memset(&tt, 0, sizeof(tt));
 
     /* no timer set should return false */
-    g_assert(!throttle_have_timer(&ts));
+    g_assert(!throttle_timers_are_initialized(&tt));
 
-    /* init the structure */
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
+    /* init structures */
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
 
     /* timer set by init should return true */
-    g_assert(throttle_have_timer(&ts));
+    g_assert(throttle_timers_are_initialized(&tt));
 
-    throttle_destroy(&ts);
+    throttle_timers_destroy(&tt);
 }
 
 static void test_detach_attach(void)
 {
-    /* zero the structure */
+    /* zero structures */
     memset(&ts, 0, sizeof(ts));
+    memset(&tt, 0, sizeof(tt));
 
     /* init the structure */
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
 
     /* timer set by init should return true */
-    g_assert(throttle_have_timer(&ts));
+    g_assert(throttle_timers_are_initialized(&tt));
 
     /* timer should no longer exist after detaching */
-    throttle_detach_aio_context(&ts);
-    g_assert(!throttle_have_timer(&ts));
+    throttle_timers_detach_aio_context(&tt);
+    g_assert(!throttle_timers_are_initialized(&tt));
 
     /* timer should exist again after attaching */
-    throttle_attach_aio_context(&ts, ctx);
-    g_assert(throttle_have_timer(&ts));
+    throttle_timers_attach_aio_context(&tt, ctx);
+    g_assert(throttle_timers_are_initialized(&tt));
 
-    throttle_destroy(&ts);
+    throttle_timers_destroy(&tt);
 }
 
 static bool do_test_accounting(bool is_ops, /* are we testing bps or ops */
@@ -387,9 +396,10 @@ static bool do_test_accounting(bool is_ops, /* are we testing bps or ops */
 
     cfg.op_size = op_size;
 
-    throttle_init(&ts, ctx, QEMU_CLOCK_VIRTUAL,
-                  read_timer_cb, write_timer_cb, &ts);
-    throttle_config(&ts, &cfg);
+    throttle_init(&ts);
+    throttle_timers_init(&tt, ctx, QEMU_CLOCK_VIRTUAL,
+                         read_timer_cb, write_timer_cb, &ts);
+    throttle_config(&ts, &tt, &cfg);
 
     /* account a read */
     throttle_account(&ts, false, size);
@@ -414,7 +424,7 @@ static bool do_test_accounting(bool is_ops, /* are we testing bps or ops */
         return false;
     }
 
-    throttle_destroy(&ts);
+    throttle_timers_destroy(&tt);
 
     return true;
 }
diff --git a/util/throttle.c b/util/throttle.c
index f976ac7..d76a48e 100644
--- a/util/throttle.c
+++ b/util/throttle.c
@@ -159,29 +159,36 @@ bool throttle_compute_timer(ThrottleState *ts,
 }
 
 /* Add timers to event loop */
-void throttle_attach_aio_context(ThrottleState *ts, AioContext *new_context)
+void throttle_timers_attach_aio_context(ThrottleTimers *tt,
+                                        AioContext *new_context)
 {
-    ts->timers[0] = aio_timer_new(new_context, ts->clock_type, SCALE_NS,
-                                  ts->read_timer_cb, ts->timer_opaque);
-    ts->timers[1] = aio_timer_new(new_context, ts->clock_type, SCALE_NS,
-                                  ts->write_timer_cb, ts->timer_opaque);
+    tt->timers[0] = aio_timer_new(new_context, tt->clock_type, SCALE_NS,
+                                  tt->read_timer_cb, tt->timer_opaque);
+    tt->timers[1] = aio_timer_new(new_context, tt->clock_type, SCALE_NS,
+                                  tt->write_timer_cb, tt->timer_opaque);
 }
 
 /* To be called first on the ThrottleState */
-void throttle_init(ThrottleState *ts,
-                   AioContext *aio_context,
-                   QEMUClockType clock_type,
-                   QEMUTimerCB *read_timer_cb,
-                   QEMUTimerCB *write_timer_cb,
-                   void *timer_opaque)
+void throttle_init(ThrottleState *ts)
 {
     memset(ts, 0, sizeof(ThrottleState));
+}
+
+/* To be called first on the ThrottleTimers */
+void throttle_timers_init(ThrottleTimers *tt,
+                          AioContext *aio_context,
+                          QEMUClockType clock_type,
+                          QEMUTimerCB *read_timer_cb,
+                          QEMUTimerCB *write_timer_cb,
+                          void *timer_opaque)
+{
+    memset(tt, 0, sizeof(ThrottleTimers));
 
-    ts->clock_type = clock_type;
-    ts->read_timer_cb = read_timer_cb;
-    ts->write_timer_cb = write_timer_cb;
-    ts->timer_opaque = timer_opaque;
-    throttle_attach_aio_context(ts, aio_context);
+    tt->clock_type = clock_type;
+    tt->read_timer_cb = read_timer_cb;
+    tt->write_timer_cb = write_timer_cb;
+    tt->timer_opaque = timer_opaque;
+    throttle_timers_attach_aio_context(tt, aio_context);
 }
 
 /* destroy a timer */
@@ -195,25 +202,25 @@ static void throttle_timer_destroy(QEMUTimer **timer)
 }
 
 /* Remove timers from event loop */
-void throttle_detach_aio_context(ThrottleState *ts)
+void throttle_timers_detach_aio_context(ThrottleTimers *tt)
 {
     int i;
 
     for (i = 0; i < 2; i++) {
-        throttle_timer_destroy(&ts->timers[i]);
+        throttle_timer_destroy(&tt->timers[i]);
     }
 }
 
-/* To be called last on the ThrottleState */
-void throttle_destroy(ThrottleState *ts)
+/* To be called last on the ThrottleTimers */
+void throttle_timers_destroy(ThrottleTimers *tt)
 {
-    throttle_detach_aio_context(ts);
+    throttle_timers_detach_aio_context(tt);
 }
 
 /* is any throttling timer configured */
-bool throttle_have_timer(ThrottleState *ts)
+bool throttle_timers_are_initialized(ThrottleTimers *tt)
 {
-    if (ts->timers[0]) {
+    if (tt->timers[0]) {
         return true;
     }
 
@@ -324,9 +331,12 @@ static void throttle_cancel_timer(QEMUTimer *timer)
 /* Used to configure the throttle
  *
  * @ts: the throttle state we are working on
+ * @tt: the throttle timers we use in this aio context
  * @cfg: the config to set
  */
-void throttle_config(ThrottleState *ts, ThrottleConfig *cfg)
+void throttle_config(ThrottleState *ts,
+                     ThrottleTimers *tt,
+                     ThrottleConfig *cfg)
 {
     int i;
 
@@ -336,10 +346,10 @@ void throttle_config(ThrottleState *ts, ThrottleConfig *cfg)
         throttle_fix_bucket(&ts->cfg.buckets[i]);
     }
 
-    ts->previous_leak = qemu_clock_get_ns(ts->clock_type);
+    ts->previous_leak = qemu_clock_get_ns(tt->clock_type);
 
     for (i = 0; i < 2; i++) {
-        throttle_cancel_timer(ts->timers[i]);
+        throttle_cancel_timer(tt->timers[i]);
     }
 }
 
@@ -358,12 +368,15 @@ void throttle_get_config(ThrottleState *ts, ThrottleConfig *cfg)
  *
  * NOTE: this function is not unit tested due to it's usage of timer_mod
  *
+ * @tt:       the timers structure
  * @is_write: the type of operation (read/write)
  * @ret:      true if the timer has been scheduled else false
  */
-bool throttle_schedule_timer(ThrottleState *ts, bool is_write)
+bool throttle_schedule_timer(ThrottleState *ts,
+                             ThrottleTimers *tt,
+                             bool is_write)
 {
-    int64_t now = qemu_clock_get_ns(ts->clock_type);
+    int64_t now = qemu_clock_get_ns(tt->clock_type);
     int64_t next_timestamp;
     bool must_wait;
 
@@ -378,12 +391,12 @@ bool throttle_schedule_timer(ThrottleState *ts, bool is_write)
     }
 
     /* request throttled and timer pending -> do nothing */
-    if (timer_pending(ts->timers[is_write])) {
+    if (timer_pending(tt->timers[is_write])) {
         return true;
     }
 
     /* request throttled and timer not pending -> arm timer */
-    timer_mod(ts->timers[is_write], next_timestamp);
+    timer_mod(tt->timers[is_write], next_timestamp);
     return true;
 }
 
-- 
2.1.4

[Qemu-devel] [PATCH 2/6] throttle: Add throttle group infrastructure

[Alberto Garcia]

Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 block/Makefile.objs             |   1 +
 block/throttle-groups.c         | 244 ++++++++++++++++++++++++++++++++++++++++
 include/block/block_int.h       |   1 +
 include/block/throttle-groups.h |  41 +++++++
 4 files changed, 287 insertions(+)
 create mode 100644 block/throttle-groups.c
 create mode 100644 include/block/throttle-groups.h

diff --git a/block/Makefile.objs b/block/Makefile.objs
index db2933e..1e61ce0 100644
--- a/block/Makefile.objs
+++ b/block/Makefile.objs
@@ -10,6 +10,7 @@ block-obj-$(CONFIG_WIN32) += raw-win32.o win32-aio.o
 block-obj-$(CONFIG_POSIX) += raw-posix.o
 block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
 block-obj-y += null.o mirror.o
+block-obj-y += throttle-groups.o
 
 block-obj-y += nbd.o nbd-client.o sheepdog.o
 block-obj-$(CONFIG_LIBISCSI) += iscsi.o
diff --git a/block/throttle-groups.c b/block/throttle-groups.c
new file mode 100644
index 0000000..e355fed
--- /dev/null
+++ b/block/throttle-groups.c
@@ -0,0 +1,244 @@
+/*
+ * QEMU block throttling group infrastructure
+ *
+ * Copyright (C) Nodalink, EURL. 2014
+ * Copyright (C) Igalia, S.L. 2015
+ *
+ * Authors:
+ *   Benoît Canet <benoit.canet@nodalink.com>
+ *   Alberto Garcia <berto@igalia.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License as
+ * published by the Free Software Foundation; either version 2 or
+ * (at your option) version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "block/throttle-groups.h"
+
+/* The ThrottleGroup structure (with its ThrottleState) is shared
+ * among different BlockDriverState and it's independent from
+ * AioContext, so in order to use it from different threads it needs
+ * its own locking.
+ *
+ * This locking is however handled internally in this file, so it's
+ * transparent to outside users.
+ *
+ * The whole ThrottleGroup structure is private and invisible to
+ * outside users, that only use it through its ThrottleState.
+ */
+typedef struct ThrottleGroup {
+    char *name; /* This is constant during the lifetime of the group */
+
+    QemuMutex lock;
+    ThrottleState ts;
+    QLIST_HEAD(, BlockDriverState) head;
+    BlockDriverState *tokens[2];
+
+    /* These two are protected by throttle_groups_lock */
+    unsigned refcount;
+    QTAILQ_ENTRY(ThrottleGroup) list;
+} ThrottleGroup;
+
+static QemuMutex throttle_groups_lock;
+static QTAILQ_HEAD(, ThrottleGroup) throttle_groups =
+    QTAILQ_HEAD_INITIALIZER(throttle_groups);
+
+/* Increments the reference count of a ThrottleGroup given its name.
+ *
+ * If no ThrottleGroup is found with the given name a new one is
+ * created.
+ *
+ * @name: the name of the ThrottleGroup
+ * @ret:  the ThrottleGroup
+ */
+static ThrottleGroup *throttle_group_incref(const char *name)
+{
+    ThrottleGroup *tg = NULL;
+    ThrottleGroup *iter;
+
+    qemu_mutex_lock(&throttle_groups_lock);
+
+    /* Look for an existing group with that name */
+    QTAILQ_FOREACH(iter, &throttle_groups, list) {
+        if (!strcmp(name, iter->name)) {
+            tg = iter;
+            break;
+        }
+    }
+
+    /* Create a new one if not found */
+    if (!tg) {
+        tg = g_new0(ThrottleGroup, 1);
+        tg->name = g_strdup(name);
+        qemu_mutex_init(&tg->lock);
+        throttle_init(&tg->ts);
+        QLIST_INIT(&tg->head);
+
+        QTAILQ_INSERT_TAIL(&throttle_groups, tg, list);
+    }
+
+    tg->refcount++;
+
+    qemu_mutex_unlock(&throttle_groups_lock);
+
+    return tg;
+}
+
+/* Decrease the reference count of a ThrottleGroup.
+ *
+ * When the reference count reaches zero the ThrottleGroup is
+ * destroyed.
+ *
+ * @tg:  The ThrottleGroup to unref
+ */
+static void throttle_group_unref(ThrottleGroup *tg)
+{
+    qemu_mutex_lock(&throttle_groups_lock);
+    if (--tg->refcount == 0) {
+        QTAILQ_REMOVE(&throttle_groups, tg, list);
+        qemu_mutex_destroy(&tg->lock);
+        g_free(tg->name);
+        g_free(tg);
+    }
+    qemu_mutex_unlock(&throttle_groups_lock);
+}
+
+/* Get the name from a ThrottleState's ThrottleGroup. The name (and
+ * the pointer) is guaranteed to remain constant during the lifetime
+ * of the group.
+ *
+ * @ts:   the throttle state whose group we are inspecting
+ * @ret:  the name of the group.
+ */
+const char *throttle_group_get_name(ThrottleState *ts)
+{
+    ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
+    return tg->name;
+}
+
+/* Return the next BlockDriverState in the round-robin sequence
+ *
+ * This takes care of simulating a circular list
+ *
+ * @bs:  the current BlockDriverState
+ * @ret: the next BlockDriverState in the sequence
+ */
+static BlockDriverState *throttle_group_next_bs(BlockDriverState *bs)
+{
+    ThrottleState *ts = bs->throttle_state;
+    ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
+    BlockDriverState *next = QLIST_NEXT(bs, round_robin);
+
+    if (!next) {
+        return QLIST_FIRST(&tg->head);
+    }
+
+    return next;
+}
+
+/* Update the throttle configuration for a particular group. Similar
+ * to throttle_config(), but guarantees atomicity within the
+ * throttling group.
+ *
+ * @ts:  the ThrottleState of the group to update.
+ * @tt:  the throttle timers we use in this aio context
+ * @cfg: the configuration to set
+ */
+void throttle_group_config(ThrottleState *ts,
+                           ThrottleTimers *tt,
+                           ThrottleConfig *cfg)
+{
+    ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
+    qemu_mutex_lock(&tg->lock);
+    throttle_config(ts, tt, cfg);
+    qemu_mutex_unlock(&tg->lock);
+}
+
+/* Get the throttle configuration from a particular group. Similar to
+ * throttle_get_config(), but guarantees atomicity within the
+ * throttling group.
+ *
+ * @ts:  the ThrottleState of the group.
+ * @cfg: the configuration will be written here
+ */
+void throttle_group_get_config(ThrottleState *ts, ThrottleConfig *cfg)
+{
+    ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
+    qemu_mutex_lock(&tg->lock);
+    throttle_get_config(ts, cfg);
+    qemu_mutex_unlock(&tg->lock);
+}
+
+/* Register a BlockDriverState in the throttling group, also updating
+ * its throttle_state pointer to point to it. If a throttling group
+ * with that name does not exist yet, it will be created.
+ *
+ * @bs:        the BlockDriverState to insert
+ * @groupname: the name of the group
+ */
+void throttle_group_register_bs(BlockDriverState *bs, const char *groupname)
+{
+    int i;
+    ThrottleGroup *tg = throttle_group_incref(groupname);
+
+    bs->throttle_state = &tg->ts;
+
+    qemu_mutex_lock(&tg->lock);
+    /* If the ThrottleGroup is new set this BlockDriverState as the token */
+    for (i = 0; i < 2; i++) {
+        if (!tg->tokens[i]) {
+            tg->tokens[i] = bs;
+        }
+    }
+
+    QLIST_INSERT_HEAD(&tg->head, bs, round_robin);
+    qemu_mutex_unlock(&tg->lock);
+}
+
+/* Unregister a BlockDriverState from its group, removing it from the
+ * list and setting the throttle_state pointer to NULL.
+ *
+ * The group will be destroyed if it's empty after this operation.
+ *
+ * @bs: the BlockDriverState to remove
+ */
+void throttle_group_unregister_bs(BlockDriverState *bs)
+{
+    ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
+    int i;
+
+    qemu_mutex_lock(&tg->lock);
+    for (i = 0; i < 2; i++) {
+        if (tg->tokens[i] == bs) {
+            BlockDriverState *token = throttle_group_next_bs(bs);
+            /* Take care of the case where this is the last bs in the group */
+            if (token == bs) {
+                token = NULL;
+            }
+            tg->tokens[i] = token;
+        }
+    }
+
+    /* remove the current bs from the list */
+    QLIST_REMOVE(bs, round_robin);
+    qemu_mutex_unlock(&tg->lock);
+
+    throttle_group_unref(tg);
+    bs->throttle_state = NULL;
+}
+
+static void throttle_groups_init(void)
+{
+    qemu_mutex_init(&throttle_groups_lock);
+}
+
+block_init(throttle_groups_init);
diff --git a/include/block/block_int.h b/include/block/block_int.h
index 58b3a5f..1314fd3 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -363,6 +363,7 @@ struct BlockDriverState {
     ThrottleTimers throttle_timers;
     CoQueue      throttled_reqs[2];
     bool         io_limits_enabled;
+    QLIST_ENTRY(BlockDriverState) round_robin;
 
     /* I/O stats (display with "info blockstats"). */
     BlockAcctStats stats;
diff --git a/include/block/throttle-groups.h b/include/block/throttle-groups.h
new file mode 100644
index 0000000..23a172f
--- /dev/null
+++ b/include/block/throttle-groups.h
@@ -0,0 +1,41 @@
+/*
+ * QEMU block throttling group infrastructure
+ *
+ * Copyright (C) Nodalink, EURL. 2014
+ * Copyright (C) Igalia, S.L. 2015
+ *
+ * Authors:
+ *   Benoît Canet <benoit.canet@nodalink.com>
+ *   Alberto Garcia <berto@igalia.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License as
+ * published by the Free Software Foundation; either version 2 or
+ * (at your option) version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef THROTTLE_GROUPS_H
+#define THROTTLE_GROUPS_H
+
+#include "qemu/throttle.h"
+#include "block/block_int.h"
+
+const char *throttle_group_get_name(ThrottleState *ts);
+
+void throttle_group_config(ThrottleState *ts,
+                           ThrottleTimers *tt,
+                           ThrottleConfig *cfg);
+void throttle_group_get_config(ThrottleState *ts, ThrottleConfig *cfg);
+
+void throttle_group_register_bs(BlockDriverState *bs, const char *groupname);
+void throttle_group_unregister_bs(BlockDriverState *bs);
+
+#endif
-- 
2.1.4

[Qemu-devel] [PATCH 3/6] throttle: Add throttle group infrastructure tests

[Alberto Garcia]

Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 tests/test-throttle.c | 37 +++++++++++++++++++++++++++++++++++--
 1 file changed, 35 insertions(+), 2 deletions(-)

diff --git a/tests/test-throttle.c b/tests/test-throttle.c
index 458f577..995d41d 100644
--- a/tests/test-throttle.c
+++ b/tests/test-throttle.c
@@ -1,10 +1,12 @@
 /*
  * Throttle infrastructure tests
  *
- * Copyright Nodalink, SARL. 2013
+ * Copyright Nodalink, EURL. 2013-2014
+ * Copyright Igalia, S.L. 2015
  *
  * Authors:
- *  Benoît Canet     <benoit.canet@irqsave.net>
+ *  Benoît Canet     <benoit.canet@nodalink.com>
+ *  Alberto Garcia   <berto@igalia.com>
  *
  * This work is licensed under the terms of the GNU LGPL, version 2 or later.
  * See the COPYING.LIB file in the top-level directory.
@@ -15,6 +17,7 @@
 #include "block/aio.h"
 #include "qemu/throttle.h"
 #include "qemu/error-report.h"
+#include "block/throttle-groups.h"
 
 static AioContext     *ctx;
 static LeakyBucket    bkt;
@@ -500,6 +503,35 @@ static void test_accounting(void)
                                 (64.0 / 13)));
 }
 
+static void test_groups(void)
+{
+    BlockDriverState *bdrv1, *bdrv2, *bdrv3;
+
+    bdrv1 = bdrv_new();
+    bdrv2 = bdrv_new();
+    bdrv3 = bdrv_new();
+
+    g_assert(bdrv1->throttle_state == NULL);
+    g_assert(bdrv2->throttle_state == NULL);
+    g_assert(bdrv3->throttle_state == NULL);
+
+    throttle_group_register_bs(bdrv1, "bar");
+    throttle_group_register_bs(bdrv2, "foo");
+    throttle_group_register_bs(bdrv3, "bar");
+
+    g_assert(!strcmp(throttle_group_get_name(bdrv1->throttle_state), "bar"));
+    g_assert(!strcmp(throttle_group_get_name(bdrv2->throttle_state), "foo"));
+    g_assert(bdrv1->throttle_state == bdrv3->throttle_state);
+
+    throttle_group_unregister_bs(bdrv1);
+    throttle_group_unregister_bs(bdrv2);
+    throttle_group_unregister_bs(bdrv3);
+
+    g_assert(bdrv1->throttle_state == NULL);
+    g_assert(bdrv2->throttle_state == NULL);
+    g_assert(bdrv3->throttle_state == NULL);
+}
+
 int main(int argc, char **argv)
 {
     GSource *src;
@@ -533,6 +565,7 @@ int main(int argc, char **argv)
     g_test_add_func("/throttle/config/is_valid",    test_is_valid);
     g_test_add_func("/throttle/config_functions",   test_config_functions);
     g_test_add_func("/throttle/accounting",         test_accounting);
+    g_test_add_func("/throttle/groups",             test_groups);
     return g_test_run();
 }
 
-- 
2.1.4

[Qemu-devel] [PATCH 4/6] throttle: Add throttle group support

[Alberto Garcia]

The throttle group support use a cooperative round robin scheduling
algorithm.

The principles of the algorithm are simple:
- Each BDS of the group is used as a token in a circular way.
- The active BDS computes if a wait must be done and arms the right
  timer.
- If a wait must be done the token timer will be armed so the token
  will become the next active BDS.

Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 block.c                         |  65 ++++++++++----------
 block/qapi.c                    |   5 +-
 block/throttle-groups.c         | 127 ++++++++++++++++++++++++++++++++++++++++
 blockdev.c                      |  19 +++++-
 hmp.c                           |   4 +-
 include/block/block.h           |   3 +-
 include/block/block_int.h       |   9 +--
 include/block/throttle-groups.h |   4 ++
 qapi/block-core.json            |   4 +-
 qemu-options.hx                 |   1 +
 qmp-commands.hx                 |   3 +-
 11 files changed, 198 insertions(+), 46 deletions(-)

diff --git a/block.c b/block.c
index da0cb48..cd8582f 100644
--- a/block.c
+++ b/block.c
@@ -36,6 +36,7 @@
 #include "qmp-commands.h"
 #include "qemu/timer.h"
 #include "qapi-event.h"
+#include "block/throttle-groups.h"
 
 #ifdef CONFIG_BSD
 #include <sys/types.h>
@@ -130,7 +131,7 @@ void bdrv_set_io_limits(BlockDriverState *bs,
 {
     int i;
 
-    throttle_config(&bs->throttle_state, &bs->throttle_timers, cfg);
+    throttle_group_config(bs->throttle_state, &bs->throttle_timers, cfg);
 
     for (i = 0; i < 2; i++) {
         qemu_co_enter_next(&bs->throttled_reqs[i]);
@@ -160,9 +161,8 @@ static bool bdrv_start_throttled_reqs(BlockDriverState *bs)
 void bdrv_io_limits_disable(BlockDriverState *bs)
 {
     bs->io_limits_enabled = false;
-
     bdrv_start_throttled_reqs(bs);
-
+    throttle_group_unregister_bs(bs);
     throttle_timers_destroy(&bs->throttle_timers);
 }
 
@@ -179,10 +179,11 @@ static void bdrv_throttle_write_timer_cb(void *opaque)
 }
 
 /* should be called before bdrv_set_io_limits if a limit is set */
-void bdrv_io_limits_enable(BlockDriverState *bs)
+void bdrv_io_limits_enable(BlockDriverState *bs, const char *group)
 {
     assert(!bs->io_limits_enabled);
-    throttle_init(&bs->throttle_state);
+
+    throttle_group_register_bs(bs, group ? group : bdrv_get_device_name(bs));
     throttle_timers_init(&bs->throttle_timers,
                          bdrv_get_aio_context(bs),
                          QEMU_CLOCK_VIRTUAL,
@@ -192,6 +193,23 @@ void bdrv_io_limits_enable(BlockDriverState *bs)
     bs->io_limits_enabled = true;
 }
 
+void bdrv_io_limits_update_group(BlockDriverState *bs, const char *group)
+{
+    /* this bs is not part of any group */
+    if (!bs->throttle_state) {
+        return;
+    }
+
+    /* this bs is a part of the same group than the one we want */
+    if (!g_strcmp0(throttle_group_get_name(bs->throttle_state), group)) {
+        return;
+    }
+
+    /* need to change the group this bs belong to */
+    bdrv_io_limits_disable(bs);
+    bdrv_io_limits_enable(bs, group);
+}
+
 /* This function makes an IO wait if needed
  *
  * @nb_sectors: the number of sectors of the IO
@@ -201,29 +219,7 @@ static void bdrv_io_limits_intercept(BlockDriverState *bs,
                                      unsigned int bytes,
                                      bool is_write)
 {
-    /* does this io must wait */
-    bool must_wait = throttle_schedule_timer(&bs->throttle_state,
-                                             &bs->throttle_timers,
-                                             is_write);
-
-    /* if must wait or any request of this type throttled queue the IO */
-    if (must_wait ||
-        !qemu_co_queue_empty(&bs->throttled_reqs[is_write])) {
-        qemu_co_queue_wait(&bs->throttled_reqs[is_write]);
-    }
-
-    /* the IO will be executed, do the accounting */
-    throttle_account(&bs->throttle_state, is_write, bytes);
-
-
-    /* if the next request must wait -> do nothing */
-    if (throttle_schedule_timer(&bs->throttle_state, &bs->throttle_timers,
-                                is_write)) {
-        return;
-    }
-
-    /* else queue next request for execution */
-    qemu_co_queue_next(&bs->throttled_reqs[is_write]);
+    throttle_group_io_limits_intercept(bs, bytes, is_write);
 }
 
 size_t bdrv_opt_mem_align(BlockDriverState *bs)
@@ -2050,15 +2046,16 @@ static void bdrv_move_feature_fields(BlockDriverState *bs_dest,
     bs_dest->enable_write_cache = bs_src->enable_write_cache;
 
     /* i/o throttled req */
-    memcpy(&bs_dest->throttle_state,
-           &bs_src->throttle_state,
-           sizeof(ThrottleState));
+    bs_dest->throttle_state     = bs_src->throttle_state,
+    bs_dest->io_limits_enabled  = bs_src->io_limits_enabled;
+    bs_dest->throttled_reqs[0]  = bs_src->throttled_reqs[0];
+    bs_dest->throttled_reqs[1]  = bs_src->throttled_reqs[1];
+    memcpy(&bs_dest->round_robin,
+           &bs_src->round_robin,
+           sizeof(bs_dest->round_robin));
     memcpy(&bs_dest->throttle_timers,
            &bs_src->throttle_timers,
            sizeof(ThrottleTimers));
-    bs_dest->throttled_reqs[0]  = bs_src->throttled_reqs[0];
-    bs_dest->throttled_reqs[1]  = bs_src->throttled_reqs[1];
-    bs_dest->io_limits_enabled  = bs_src->io_limits_enabled;
 
     /* r/w error */
     bs_dest->on_read_error      = bs_src->on_read_error;
diff --git a/block/qapi.c b/block/qapi.c
index 1808e67..abeeb38 100644
--- a/block/qapi.c
+++ b/block/qapi.c
@@ -24,6 +24,7 @@
 
 #include "block/qapi.h"
 #include "block/block_int.h"
+#include "block/throttle-groups.h"
 #include "block/write-threshold.h"
 #include "qmp-commands.h"
 #include "qapi-visit.h"
@@ -63,7 +64,9 @@ BlockDeviceInfo *bdrv_block_device_info(BlockDriverState *bs)
 
     if (bs->io_limits_enabled) {
         ThrottleConfig cfg;
-        throttle_get_config(&bs->throttle_state, &cfg);
+
+        throttle_group_get_config(bs->throttle_state, &cfg);
+
         info->bps     = cfg.buckets[THROTTLE_BPS_TOTAL].avg;
         info->bps_rd  = cfg.buckets[THROTTLE_BPS_READ].avg;
         info->bps_wr  = cfg.buckets[THROTTLE_BPS_WRITE].avg;
diff --git a/block/throttle-groups.c b/block/throttle-groups.c
index e355fed..9075c46 100644
--- a/block/throttle-groups.c
+++ b/block/throttle-groups.c
@@ -23,6 +23,8 @@
  */
 
 #include "block/throttle-groups.h"
+#include "qemu/queue.h"
+#include "qemu/thread.h"
 
 /* The ThrottleGroup structure (with its ThrottleState) is shared
  * among different BlockDriverState and it's independent from
@@ -145,6 +147,131 @@ static BlockDriverState *throttle_group_next_bs(BlockDriverState *bs)
     return next;
 }
 
+/* Return the next BlockDriverState in the round-robin sequence with
+ * pending I/O requests.
+ *
+ * @bs:        the current BlockDriverState
+ * @is_write:  the type of operation (read/write)
+ * @ret:       the next BlockDriverState with pending requests, or bs
+ *             if there is none.
+ */
+static BlockDriverState *next_throttle_token(BlockDriverState *bs,
+                                             bool is_write)
+{
+    ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
+    BlockDriverState *token, *start;
+
+    start = token = tg->tokens[is_write];
+
+    /* get next bs round in round robin style */
+    token = throttle_group_next_bs(token);
+    while (token != start  &&
+           qemu_co_queue_empty(&token->throttled_reqs[is_write])) {
+        token = throttle_group_next_bs(token);
+    }
+
+    /* If no IO are queued for scheduling on the next round robin token
+     * then decide the token is the current bs because chances are
+     * the current bs get the current request queued.
+     */
+    if (token == start &&
+        qemu_co_queue_empty(&token->throttled_reqs[is_write])) {
+        token = bs;
+    }
+
+    return token;
+}
+
+/* Check if the next I/O request for a BlockDriverState needs to be
+ * throttled or not. If there's no timer set in this group, set one
+ * and update the token accordingly.
+ *
+ * @bs:         the current BlockDriverState
+ * @is_write:   the type of operation (read/write)
+ * @ret:        whether the I/O request needs to be throttled or not
+ */
+static bool throttle_group_schedule_timer(BlockDriverState *bs,
+                                          bool is_write)
+{
+    ThrottleState *ts = bs->throttle_state;
+    ThrottleTimers *tt = &bs->throttle_timers;
+    ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
+    bool must_wait;
+    BlockDriverState *iter;
+
+    /* Check if any of the timers in this group is already armed */
+    QLIST_FOREACH(iter, &tg->head, round_robin) {
+        if (timer_pending(iter->throttle_timers.timers[is_write])) {
+            return true;
+        }
+    }
+
+    must_wait = throttle_schedule_timer(ts, tt, is_write);
+
+    /* If a timer just got armed, set bs as the current token */
+    if (must_wait) {
+        tg->tokens[is_write] = bs;
+    }
+
+    return must_wait;
+}
+
+/* Check if an I/O request needs to be throttled, wait and set a timer
+ * if necessary, and schedule the next request using a round robin
+ * algorithm.
+ *
+ * @bs:        the current BlockDriverState
+ * @bytes:     the number of bytes for this I/O
+ * @is_write:  the type of operation (read/write)
+ */
+void throttle_group_io_limits_intercept(BlockDriverState *bs,
+                                        unsigned int bytes,
+                                        bool is_write)
+{
+    bool must_wait;
+    BlockDriverState *token;
+
+    ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
+    qemu_mutex_lock(&tg->lock);
+
+    /* First we check if this I/O has to be throttled. */
+    token = next_throttle_token(bs, is_write);
+    must_wait = throttle_group_schedule_timer(token, is_write);
+
+    /* Wait if there's a timer set or queued requests of this type */
+    if (must_wait || !qemu_co_queue_empty(&bs->throttled_reqs[is_write])) {
+        qemu_mutex_unlock(&tg->lock);
+        qemu_co_queue_wait(&bs->throttled_reqs[is_write]);
+        qemu_mutex_lock(&tg->lock);
+    }
+
+    /* The I/O will be executed, so do the accounting */
+    throttle_account(bs->throttle_state, is_write, bytes);
+
+    /* Now we check if there's any pending request to schedule next */
+    token = next_throttle_token(bs, is_write);
+    if (!qemu_co_queue_empty(&token->throttled_reqs[is_write])) {
+
+        /* If it doesn't have to wait, queue it for immediate execution */
+        must_wait = throttle_group_schedule_timer(token, is_write);
+
+        if (!must_wait) {
+            /* Give preference to requests from the current bs */
+            if (!qemu_co_queue_empty(&bs->throttled_reqs[is_write])) {
+                token = bs;
+                qemu_co_queue_next(&bs->throttled_reqs[is_write]);
+            } else {
+                ThrottleTimers *tt = &token->throttle_timers;
+                int64_t now = qemu_clock_get_ns(tt->clock_type);
+                timer_mod(tt->timers[is_write], now + 1);
+            }
+            tg->tokens[is_write] = token;
+        }
+    }
+
+    qemu_mutex_unlock(&tg->lock);
+}
+
 /* Update the throttle configuration for a particular group. Similar
  * to throttle_config(), but guarantees atomicity within the
  * throttling group.
diff --git a/blockdev.c b/blockdev.c
index b9c1c0c..89fc7a8 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -357,6 +357,7 @@ static BlockBackend *blockdev_init(const char *file, QDict *bs_opts,
     const char *id;
     bool has_driver_specific_opts;
     BlockdevDetectZeroesOptions detect_zeroes;
+    const char *throttling_group;
 
     /* Check common options by copying from bs_opts to opts, all other options
      * stay in bs_opts for processing by bdrv_open(). */
@@ -459,6 +460,8 @@ static BlockBackend *blockdev_init(const char *file, QDict *bs_opts,
 
     cfg.op_size = qemu_opt_get_number(opts, "throttling.iops-size", 0);
 
+    throttling_group = qemu_opt_get(opts, "throttling.group");
+
     if (!check_throttle_config(&cfg, &error)) {
         error_propagate(errp, error);
         goto early_err;
@@ -547,7 +550,7 @@ static BlockBackend *blockdev_init(const char *file, QDict *bs_opts,
 
     /* disk I/O throttling */
     if (throttle_enabled(&cfg)) {
-        bdrv_io_limits_enable(bs);
+        bdrv_io_limits_enable(bs, throttling_group);
         bdrv_set_io_limits(bs, &cfg);
     }
 
@@ -711,6 +714,8 @@ DriveInfo *drive_new(QemuOpts *all_opts, BlockInterfaceType block_default_type)
 
         { "iops_size",      "throttling.iops-size" },
 
+        { "group",          "throttling.group" },
+
         { "readonly",       "read-only" },
     };
 
@@ -1877,7 +1882,9 @@ void qmp_block_set_io_throttle(const char *device, int64_t bps, int64_t bps_rd,
                                bool has_iops_wr_max,
                                int64_t iops_wr_max,
                                bool has_iops_size,
-                               int64_t iops_size, Error **errp)
+                               int64_t iops_size,
+                               bool has_group,
+                               const char *group, Error **errp)
 {
     ThrottleConfig cfg;
     BlockDriverState *bs;
@@ -1929,9 +1936,11 @@ void qmp_block_set_io_throttle(const char *device, int64_t bps, int64_t bps_rd,
     aio_context_acquire(aio_context);
 
     if (!bs->io_limits_enabled && throttle_enabled(&cfg)) {
-        bdrv_io_limits_enable(bs);
+        bdrv_io_limits_enable(bs, has_group ? group : NULL);
     } else if (bs->io_limits_enabled && !throttle_enabled(&cfg)) {
         bdrv_io_limits_disable(bs);
+    } else if (bs->io_limits_enabled && throttle_enabled(&cfg)) {
+        bdrv_io_limits_update_group(bs, has_group ? group : NULL);
     }
 
     if (bs->io_limits_enabled) {
@@ -2991,6 +3000,10 @@ QemuOptsList qemu_common_drive_opts = {
             .type = QEMU_OPT_NUMBER,
             .help = "when limiting by iops max size of an I/O in bytes",
         },{
+            .name = "throttling.group",
+            .type = QEMU_OPT_STRING,
+            .help = "name of the block throttling group",
+        },{
             .name = "copy-on-read",
             .type = QEMU_OPT_BOOL,
             .help = "copy read data from backing file into image file",
diff --git a/hmp.c b/hmp.c
index 71c28bc..b81e4ba 100644
--- a/hmp.c
+++ b/hmp.c
@@ -1261,7 +1261,9 @@ void hmp_block_set_io_throttle(Monitor *mon, const QDict *qdict)
                               false,
                               0,
                               false, /* No default I/O size */
-                              0, &err);
+                              0,
+                              false,
+                              NULL, &err);
     hmp_handle_error(mon, &err);
 }
 
diff --git a/include/block/block.h b/include/block/block.h
index 649c269..2bb6222 100644
--- a/include/block/block.h
+++ b/include/block/block.h
@@ -162,8 +162,9 @@ void bdrv_stats_print(Monitor *mon, const QObject *data);
 void bdrv_info_stats(Monitor *mon, QObject **ret_data);
 
 /* disk I/O throttling */
-void bdrv_io_limits_enable(BlockDriverState *bs);
+void bdrv_io_limits_enable(BlockDriverState *bs, const char *group);
 void bdrv_io_limits_disable(BlockDriverState *bs);
+void bdrv_io_limits_update_group(BlockDriverState *bs, const char *group);
 
 void bdrv_init(void);
 void bdrv_init_with_whitelist(void);
diff --git a/include/block/block_int.h b/include/block/block_int.h
index 1314fd3..c02e963 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -358,12 +358,13 @@ struct BlockDriverState {
     /* number of in-flight serialising requests */
     unsigned int serialising_in_flight;
 
-    /* I/O throttling */
-    ThrottleState throttle_state;
-    ThrottleTimers throttle_timers;
-    CoQueue      throttled_reqs[2];
+    /* I/O throttling - following elements protected by ThrottleGroup lock */
+    ThrottleState *throttle_state;
     bool         io_limits_enabled;
+    CoQueue      throttled_reqs[2];
     QLIST_ENTRY(BlockDriverState) round_robin;
+    /* timers have their own locking */
+    ThrottleTimers throttle_timers;
 
     /* I/O stats (display with "info blockstats"). */
     BlockAcctStats stats;
diff --git a/include/block/throttle-groups.h b/include/block/throttle-groups.h
index 23a172f..ddb962b 100644
--- a/include/block/throttle-groups.h
+++ b/include/block/throttle-groups.h
@@ -38,4 +38,8 @@ void throttle_group_get_config(ThrottleState *ts, ThrottleConfig *cfg);
 void throttle_group_register_bs(BlockDriverState *bs, const char *groupname);
 void throttle_group_unregister_bs(BlockDriverState *bs);
 
+void throttle_group_io_limits_intercept(BlockDriverState *bs,
+                                        unsigned int bytes,
+                                        bool is_write);
+
 #endif
diff --git a/qapi/block-core.json b/qapi/block-core.json
index a3fdaf0..487b147 100644
--- a/qapi/block-core.json
+++ b/qapi/block-core.json
@@ -989,6 +989,8 @@
 #
 # @iops_size: #optional an I/O size in bytes (Since 1.7)
 #
+# @group: #optional throttle group name (Since 2.3)
+#
 # Returns: Nothing on success
 #          If @device is not a valid block device, DeviceNotFound
 #
@@ -1000,7 +1002,7 @@
             '*bps_max': 'int', '*bps_rd_max': 'int',
             '*bps_wr_max': 'int', '*iops_max': 'int',
             '*iops_rd_max': 'int', '*iops_wr_max': 'int',
-            '*iops_size': 'int' } }
+            '*iops_size': 'int', '*group': 'str' } }
 
 ##
 # @block-stream:
diff --git a/qemu-options.hx b/qemu-options.hx
index 837624d..6db0613 100644
--- a/qemu-options.hx
+++ b/qemu-options.hx
@@ -454,6 +454,7 @@ DEF("drive", HAS_ARG, QEMU_OPTION_drive,
     "       [[,bps_max=bm]|[[,bps_rd_max=rm][,bps_wr_max=wm]]]\n"
     "       [[,iops_max=im]|[[,iops_rd_max=irm][,iops_wr_max=iwm]]]\n"
     "       [[,iops_size=is]]\n"
+    "       [[,group=g]]\n"
     "                use 'file' as a drive image\n", QEMU_ARCH_ALL)
 STEXI
 @item -drive @var{option}[,@var{option}[,@var{option}[,...]]]
diff --git a/qmp-commands.hx b/qmp-commands.hx
index c12334a..d866d57 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -1704,7 +1704,7 @@ EQMP
 
     {
         .name       = "block_set_io_throttle",
-        .args_type  = "device:B,bps:l,bps_rd:l,bps_wr:l,iops:l,iops_rd:l,iops_wr:l,bps_max:l?,bps_rd_max:l?,bps_wr_max:l?,iops_max:l?,iops_rd_max:l?,iops_wr_max:l?,iops_size:l?",
+        .args_type  = "device:B,bps:l,bps_rd:l,bps_wr:l,iops:l,iops_rd:l,iops_wr:l,bps_max:l?,bps_rd_max:l?,bps_wr_max:l?,iops_max:l?,iops_rd_max:l?,iops_wr_max:l?,iops_size:l?,group:s?",
         .mhandler.cmd_new = qmp_marshal_input_block_set_io_throttle,
     },
 
@@ -1730,6 +1730,7 @@ Arguments:
 - "iops_rd_max":  read I/O operations max (json-int)
 - "iops_wr_max":  write I/O operations max (json-int)
 - "iops_size":  I/O size in bytes when limiting (json-int)
+- "group": throttle group name (json-string)
 
 Example:
 
-- 
2.1.4

[Qemu-devel] [PATCH 5/6] throttle: add the name of the ThrottleGroup to BlockDeviceInfo

[Alberto Garcia]

Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 block/qapi.c         | 3 +++
 hmp.c                | 6 ++++--
 qapi/block-core.json | 4 +++-
 3 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/block/qapi.c b/block/qapi.c
index abeeb38..a0c7e6e 100644
--- a/block/qapi.c
+++ b/block/qapi.c
@@ -91,6 +91,9 @@ BlockDeviceInfo *bdrv_block_device_info(BlockDriverState *bs)
 
         info->has_iops_size = cfg.op_size;
         info->iops_size = cfg.op_size;
+
+        info->has_group = true;
+        info->group = g_strdup(throttle_group_get_name(bs->throttle_state));
     }
 
     info->write_threshold = bdrv_write_threshold_get(bs);
diff --git a/hmp.c b/hmp.c
index b81e4ba..716bf82 100644
--- a/hmp.c
+++ b/hmp.c
@@ -374,7 +374,8 @@ static void print_block_info(Monitor *mon, BlockInfo *info,
                         " iops_max=%" PRId64
                         " iops_rd_max=%" PRId64
                         " iops_wr_max=%" PRId64
-                        " iops_size=%" PRId64 "\n",
+                        " iops_size=%" PRId64
+                        " group=%s\n",
                         inserted->bps,
                         inserted->bps_rd,
                         inserted->bps_wr,
@@ -387,7 +388,8 @@ static void print_block_info(Monitor *mon, BlockInfo *info,
                         inserted->iops_max,
                         inserted->iops_rd_max,
                         inserted->iops_wr_max,
-                        inserted->iops_size);
+                        inserted->iops_size,
+                        inserted->group);
     }
 
     if (verbose) {
diff --git a/qapi/block-core.json b/qapi/block-core.json
index 487b147..0ae5e2e 100644
--- a/qapi/block-core.json
+++ b/qapi/block-core.json
@@ -255,6 +255,8 @@
 #
 # @iops_size: #optional an I/O size in bytes (Since 1.7)
 #
+# @group: #optional throttle group name (Since 2.3)
+#
 # @cache: the cache mode used for the block device (since: 2.3)
 #
 # @write_threshold: configured write threshold for the device.
@@ -274,7 +276,7 @@
             '*bps_max': 'int', '*bps_rd_max': 'int',
             '*bps_wr_max': 'int', '*iops_max': 'int',
             '*iops_rd_max': 'int', '*iops_wr_max': 'int',
-            '*iops_size': 'int', 'cache': 'BlockdevCacheInfo',
+            '*iops_size': 'int', '*group': 'str', 'cache': 'BlockdevCacheInfo',
             'write_threshold': 'int' } }
 
 ##
-- 
2.1.4

[Qemu-devel] [PATCH 6/6] throttle: Update throttle infrastructure copyright

[Alberto Garcia]

Signed-off-by: Alberto Garcia <berto@igalia.com>
---
 include/qemu/throttle.h | 8 +++++---
 util/throttle.c         | 8 +++++---
 2 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/include/qemu/throttle.h b/include/qemu/throttle.h
index 2c560db..5af76f0 100644
--- a/include/qemu/throttle.h
+++ b/include/qemu/throttle.h
@@ -1,10 +1,12 @@
 /*
  * QEMU throttling infrastructure
  *
- * Copyright (C) Nodalink, SARL. 2013
+ * Copyright (C) Nodalink, EURL. 2013-2014
+ * Copyright (C) Igalia, S.L. 2015
  *
- * Author:
- *   Benoît Canet <benoit.canet@irqsave.net>
+ * Authors:
+ *   Benoît Canet <benoit.canet@nodalink.com>
+ *   Alberto Garcia <berto@igalia.com>
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License as
diff --git a/util/throttle.c b/util/throttle.c
index d76a48e..706c131 100644
--- a/util/throttle.c
+++ b/util/throttle.c
@@ -1,10 +1,12 @@
 /*
  * QEMU throttling infrastructure
  *
- * Copyright (C) Nodalink, SARL. 2013
+ * Copyright (C) Nodalink, EURL. 2013-2014
+ * Copyright (C) Igalia, S.L. 2015
  *
- * Author:
- *   Benoît Canet <benoit.canet@irqsave.net>
+ * Authors:
+ *   Benoît Canet <benoit.canet@nodalink.com>
+ *   Alberto Garcia <berto@igalia.com>
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License as
-- 
2.1.4

Re: [Qemu-devel] [PATCH 2/6] throttle: Add throttle group infrastructure

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 624 bytes --]

On Tue, Mar 10, 2015 at 05:30:46PM +0200, Alberto Garcia wrote:
> +typedef struct ThrottleGroup {
> +    char *name; /* This is constant during the lifetime of the group */

Is this also protected by throttle_groups_lock?

I guess throttle_groups_lock must be held in order to read this field -
otherwise there is a risk that the object is freed unless you have
already incremented the refcount.

> +
> +    QemuMutex lock;

What does this lock protect?  I guess 'ts', 'head', and 'tokens' fields.
Please document it.

> +    ThrottleState ts;
> +    QLIST_HEAD(, BlockDriverState) head;
> +    BlockDriverState *tokens[2];

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 1/6] throttle: Extract timers from ThrottleState into a separate structure

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1007 bytes --]

On Tue, Mar 10, 2015 at 05:30:45PM +0200, Alberto Garcia wrote:
> From: Benoît Canet <benoit.canet@nodalink.com>
> 
> Group throttling will share ThrottleState between multiple bs.
> As a consequence the ThrottleState will be accessed by multiple aio
> context.
> 
> Timers are tied to their aio context so they must go out of the
> ThrottleState structure.
> 
> This commit paves the way for each bs of a common ThrottleState to
> have its own timer.
> 
> Signed-off-by: Benoit Canet <benoit.canet@nodalink.com>
> Signed-off-by: Alberto Garcia <berto@igalia.com>
> ---
>  block.c                   | 35 ++++++++++++--------
>  include/block/block_int.h |  1 +
>  include/qemu/throttle.h   | 38 ++++++++++++++--------
>  tests/test-throttle.c     | 82 ++++++++++++++++++++++++++---------------------
>  util/throttle.c           | 73 ++++++++++++++++++++++++-----------------
>  5 files changed, 135 insertions(+), 94 deletions(-)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 3/6] throttle: Add throttle group infrastructure tests

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 293 bytes --]

On Tue, Mar 10, 2015 at 05:30:47PM +0200, Alberto Garcia wrote:
> Signed-off-by: Alberto Garcia <berto@igalia.com>
> ---
>  tests/test-throttle.c | 37 +++++++++++++++++++++++++++++++++++--
>  1 file changed, 35 insertions(+), 2 deletions(-)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 2/6] throttle: Add throttle group infrastructure

[Alberto Garcia]

On Tue, Mar 24, 2015 at 03:03:07PM +0000, Stefan Hajnoczi wrote:

> > +typedef struct ThrottleGroup {
> > +    char *name; /* This is constant during the lifetime of the group */
> 
> Is this also protected by throttle_groups_lock?
> 
> I guess throttle_groups_lock must be held in order to read this
> field - otherwise there is a risk that the object is freed unless
> you have already incremented the refcount.

The creation and destruction of ThrottleGroup objects are protected by
throttle_groups_lock. That includes handling the memory for that field
and its contents.

Once the ThrottleGroup is created the 'name' field doesn't need any
additional locking since it remains constant during the lifetime of
the group.

The only way to read it from the outside is throttle_group_get_name()
and that's safe (until you release the reference to the group, that
is).

> > +    QemuMutex lock;
> 
> What does this lock protect?  I guess 'ts', 'head', and 'tokens'
> fields.  Please document it.

I thought it was obvious in the code but since you ask I guess it's
not :) I can add an additional comment.

Berto

Re: [Qemu-devel] [PATCH 4/6] throttle: Add throttle group support

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 3804 bytes --]

On Tue, Mar 10, 2015 at 05:30:48PM +0200, Alberto Garcia wrote:
> @@ -179,10 +179,11 @@ static void bdrv_throttle_write_timer_cb(void *opaque)
>  }
>  
>  /* should be called before bdrv_set_io_limits if a limit is set */
> -void bdrv_io_limits_enable(BlockDriverState *bs)
> +void bdrv_io_limits_enable(BlockDriverState *bs, const char *group)
>  {
>      assert(!bs->io_limits_enabled);
> -    throttle_init(&bs->throttle_state);
> +
> +    throttle_group_register_bs(bs, group ? group : bdrv_get_device_name(bs));

Please don't allow a NULL group argument.  blockdev_init() should pick
the group name instead of requiring bdrv_io_limits_enable() to generate
a unique name.

This way we avoid silently adding all BDS without a BlockBackend to a ""
throttling group.  I think that's a bug waiting to happen :).

> @@ -201,29 +219,7 @@ static void bdrv_io_limits_intercept(BlockDriverState *bs,
>                                       unsigned int bytes,
>                                       bool is_write)
>  {
> -    /* does this io must wait */
> -    bool must_wait = throttle_schedule_timer(&bs->throttle_state,
> -                                             &bs->throttle_timers,
> -                                             is_write);
> -
> -    /* if must wait or any request of this type throttled queue the IO */
> -    if (must_wait ||
> -        !qemu_co_queue_empty(&bs->throttled_reqs[is_write])) {
> -        qemu_co_queue_wait(&bs->throttled_reqs[is_write]);
> -    }
> -
> -    /* the IO will be executed, do the accounting */
> -    throttle_account(&bs->throttle_state, is_write, bytes);
> -
> -
> -    /* if the next request must wait -> do nothing */
> -    if (throttle_schedule_timer(&bs->throttle_state, &bs->throttle_timers,
> -                                is_write)) {
> -        return;
> -    }
> -
> -    /* else queue next request for execution */
> -    qemu_co_queue_next(&bs->throttled_reqs[is_write]);
> +    throttle_group_io_limits_intercept(bs, bytes, is_write);
>  }

bdrv_io_limits_intercept() is no longer useful.  Can you replace
bdrv_io_limits_intercept() calls with
throttle_group_io_limits_intercept() to eliminate this indirection?

>  
>  size_t bdrv_opt_mem_align(BlockDriverState *bs)
> @@ -2050,15 +2046,16 @@ static void bdrv_move_feature_fields(BlockDriverState *bs_dest,
>      bs_dest->enable_write_cache = bs_src->enable_write_cache;
>  
>      /* i/o throttled req */
> -    memcpy(&bs_dest->throttle_state,
> -           &bs_src->throttle_state,
> -           sizeof(ThrottleState));
> +    bs_dest->throttle_state     = bs_src->throttle_state,
> +    bs_dest->io_limits_enabled  = bs_src->io_limits_enabled;
> +    bs_dest->throttled_reqs[0]  = bs_src->throttled_reqs[0];
> +    bs_dest->throttled_reqs[1]  = bs_src->throttled_reqs[1];
> +    memcpy(&bs_dest->round_robin,
> +           &bs_src->round_robin,
> +           sizeof(bs_dest->round_robin));

The bdrv_swap() code is tricky to think about...I think this is okay
because bs_dest and bs_src linked list pointers will be unchanged
(although they are now pointing to a different block driver instance).

Just highlighting this in case anyone else spots a problem with the
pointer swapping.

> @@ -145,6 +147,131 @@ static BlockDriverState *throttle_group_next_bs(BlockDriverState *bs)
>      return next;
>  }
>  
> +/* Return the next BlockDriverState in the round-robin sequence with
> + * pending I/O requests.
> + *
> + * @bs:        the current BlockDriverState
> + * @is_write:  the type of operation (read/write)
> + * @ret:       the next BlockDriverState with pending requests, or bs
> + *             if there is none.

Assumes tg->lock is held.  It's helpful to document this.

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 4/6] throttle: Add throttle group support

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 2585 bytes --]

On Tue, Mar 10, 2015 at 05:30:48PM +0200, Alberto Garcia wrote:
> +/* Return the next BlockDriverState in the round-robin sequence with
> + * pending I/O requests.
> + *
> + * @bs:        the current BlockDriverState
> + * @is_write:  the type of operation (read/write)
> + * @ret:       the next BlockDriverState with pending requests, or bs
> + *             if there is none.
> + */
> +static BlockDriverState *next_throttle_token(BlockDriverState *bs,
> +                                             bool is_write)
> +{
> +    ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
> +    BlockDriverState *token, *start;
> +
> +    start = token = tg->tokens[is_write];
> +
> +    /* get next bs round in round robin style */
> +    token = throttle_group_next_bs(token);
> +    while (token != start  &&
> +           qemu_co_queue_empty(&token->throttled_reqs[is_write])) {

It's not safe to access bs->throttled_reqs[].  There are many of other
places that access bs->throttled_reqs[] without holding tg->lock (e.g.
block.c).

throttled_reqs needs to be protected by tg->lock in order for this to
work.

> +/* Check if an I/O request needs to be throttled, wait and set a timer
> + * if necessary, and schedule the next request using a round robin
> + * algorithm.
> + *
> + * @bs:        the current BlockDriverState
> + * @bytes:     the number of bytes for this I/O
> + * @is_write:  the type of operation (read/write)
> + */
> +void throttle_group_io_limits_intercept(BlockDriverState *bs,
> +                                        unsigned int bytes,
> +                                        bool is_write)

This function must be called from coroutine context because it uses
qemu_co_queue_wait() and may yield.  Please mark the function with
coroutine_fn (see include/block/coroutine.h).  This serves as
documentation.

> +{
> +    bool must_wait;
> +    BlockDriverState *token;
> +
> +    ThrottleGroup *tg = container_of(bs->throttle_state, ThrottleGroup, ts);
> +    qemu_mutex_lock(&tg->lock);
> +
> +    /* First we check if this I/O has to be throttled. */
> +    token = next_throttle_token(bs, is_write);
> +    must_wait = throttle_group_schedule_timer(token, is_write);
> +
> +    /* Wait if there's a timer set or queued requests of this type */
> +    if (must_wait || !qemu_co_queue_empty(&bs->throttled_reqs[is_write])) {
> +        qemu_mutex_unlock(&tg->lock);
> +        qemu_co_queue_wait(&bs->throttled_reqs[is_write]);

Here bs->throttled_reqs[] is used without holding tg->lock.  It can race
with token->throttled_reqs[] users.

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH v3 0/6] Block Throttle Group Support

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 250 bytes --]

On Tue, Mar 10, 2015 at 05:30:44PM +0200, Alberto Garcia wrote:
> Here's what's new:

Thanks for the cleanup!  The locking isn't safe yet but the series is in
pretty good shape overall.  Please see comments on individual patches
for details.

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 4/6] throttle: Add throttle group support

[Alberto Garcia]

On Tue, Mar 24, 2015 at 04:31:45PM +0000, Stefan Hajnoczi wrote:

> > +    /* get next bs round in round robin style */
> > +    token = throttle_group_next_bs(token);
> > +    while (token != start  &&
> > +           qemu_co_queue_empty(&token->throttled_reqs[is_write])) {
> 
> It's not safe to access bs->throttled_reqs[].  There are many of
> other places that access bs->throttled_reqs[] without holding
> tg->lock (e.g. block.c).
>
> throttled_reqs needs to be protected by tg->lock in order for this
> to work.

Good catch, but I don't think that's the solution. throttled_reqs
cannot be protected by that lock because it must release it before
calling qemu_co_queue_wait(&bs->throttled_reqs[is_write]). Otherwise
it will cause a deadlock.

My assumption was that throttled_reqs would be protected by the BDS's
AioContext lock, but I overlooked the fact that this part of the
algorithm needs to access other BDSs' queues so we indeed have a
problem.

I will give it a thought to see what I can come up with. Since we only
need to check whether other BDSs have pending requests maybe I can
implement this with a flag.

Thanks!

Berto

Re: [Qemu-devel] [PATCH 2/6] throttle: Add throttle group infrastructure

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1461 bytes --]

On Tue, Mar 24, 2015 at 04:33:48PM +0100, Alberto Garcia wrote:
> On Tue, Mar 24, 2015 at 03:03:07PM +0000, Stefan Hajnoczi wrote:
> 
> > > +typedef struct ThrottleGroup {
> > > +    char *name; /* This is constant during the lifetime of the group */
> > 
> > Is this also protected by throttle_groups_lock?
> > 
> > I guess throttle_groups_lock must be held in order to read this
> > field - otherwise there is a risk that the object is freed unless
> > you have already incremented the refcount.
> 
> The creation and destruction of ThrottleGroup objects are protected by
> throttle_groups_lock. That includes handling the memory for that field
> and its contents.
> 
> Once the ThrottleGroup is created the 'name' field doesn't need any
> additional locking since it remains constant during the lifetime of
> the group.
> 
> The only way to read it from the outside is throttle_group_get_name()
> and that's safe (until you release the reference to the group, that
> is).

Right, the race condition is when the group is released.

Looking at this again, the assumption isn't that throttle_groups_lock is
held.  The AioContext lock is held by throttle_group_get_name() users
and that's why there is no race when releasing the reference.

If someone adds a throttle_group_get_name() caller in the future without
holding AioContext, then we'd be in trouble.  That is why documenting
the locking constraints is useful.

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 4/6] throttle: Add throttle group support

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1463 bytes --]

On Tue, Mar 24, 2015 at 06:48:37PM +0100, Alberto Garcia wrote:
> On Tue, Mar 24, 2015 at 04:31:45PM +0000, Stefan Hajnoczi wrote:
> 
> > > +    /* get next bs round in round robin style */
> > > +    token = throttle_group_next_bs(token);
> > > +    while (token != start  &&
> > > +           qemu_co_queue_empty(&token->throttled_reqs[is_write])) {
> > 
> > It's not safe to access bs->throttled_reqs[].  There are many of
> > other places that access bs->throttled_reqs[] without holding
> > tg->lock (e.g. block.c).
> >
> > throttled_reqs needs to be protected by tg->lock in order for this
> > to work.
> 
> Good catch, but I don't think that's the solution. throttled_reqs
> cannot be protected by that lock because it must release it before
> calling qemu_co_queue_wait(&bs->throttled_reqs[is_write]). Otherwise
> it will cause a deadlock.
> 
> My assumption was that throttled_reqs would be protected by the BDS's
> AioContext lock, but I overlooked the fact that this part of the
> algorithm needs to access other BDSs' queues so we indeed have a
> problem.
> 
> I will give it a thought to see what I can come up with. Since we only
> need to check whether other BDSs have pending requests maybe I can
> implement this with a flag.

A flag is a good solution since it is not possible to acquire other BDS'
AioContext lock from arbitrary block layer code (it's only allowed when
the QEMU global mutex is held).

Stefan

[-- Attachment #2: Type: application/pgp-signature, Size: 473 bytes --]

Re: [Qemu-devel] [PATCH 2/6] throttle: Add throttle group infrastructure

[Alberto Garcia]

> > > > +typedef struct ThrottleGroup {
> > > > +    char *name; /* This is constant during the lifetime of the group */
> > > 
> > > Is this also protected by throttle_groups_lock?
> > > 
> > > I guess throttle_groups_lock must be held in order to read this
> > > field - otherwise there is a risk that the object is freed
> > > unless you have already incremented the refcount.
> > 
> > The creation and destruction of ThrottleGroup objects are
> > protected by throttle_groups_lock. That includes handling the
> > memory for that field and its contents.
> > 
> > Once the ThrottleGroup is created the 'name' field doesn't need
> > any additional locking since it remains constant during the
> > lifetime of the group.
> > 
> > The only way to read it from the outside is
> > throttle_group_get_name() and that's safe (until you release the
> > reference to the group, that is).
> 
> Right, the race condition is when the group is released.
> 
> Looking at this again, the assumption isn't that
> throttle_groups_lock is held.  The AioContext lock is held by
> throttle_group_get_name() users and that's why there is no race when
> releasing the reference.
> 
> If someone adds a throttle_group_get_name() caller in the future
> without holding AioContext, then we'd be in trouble.  That is why
> documenting the locking constraints is useful.

But that would only happen if you access bs->throttle_state without
holding bs's AioContext. I understand that it's implicit that you
should hold the context there.

Maybe I can update the throttle_group_* API to use BlockDriverState
in all cases instead of ThrottleState, it's probably more clear like
that.

Berto