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

[Stefan Hajnoczi <stefanha@gmail.com>]

[-- 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 --]