Re: [Qemu-devel] [PATCH V5] fsdev: add IO throttle support to fsdev devices

[Alberto Garcia]

On Thu 20 Oct 2016 04:14:23 PM CEST, Pradeep Jagadeesh wrote:
> +static void fsdev_throttle_schedule_next_request(FsThrottle *fst, bool 
> is_write)
> +{
> +    bool must_wait;
> +
> +    if (qemu_co_queue_empty(&fst->throttled_reqs[is_write])) {
> +        return;
> +    }
> +    must_wait = fsdev_throttle_check_for_wait(fst, is_write);
> +    if (!must_wait) {
> +        if (qemu_in_coroutine() &&
> +            qemu_co_queue_next(&fst->throttled_reqs[is_write])) {
> +            ;
> +       } else {
> +           int64_t now = qemu_clock_get_ns(fst->tt.clock_type);
> +           timer_mod(fst->tt.timers[is_write], now + 1);
> +       }
> +   }
> +}

I think you don't need to schedule the timer here. This was added for
group throttling for disks because once a disk completes an I/O request,
the next request might be in a different disk (due to the round-robin
algorithm). So we put a timer in that disk to handle that.

In your case that cannot happen, see fsdev_throttle_request() below[1]
for a suggestion of how it can look like.

> +static void fsdev_throttle_timer_cb(FsThrottle *fst, bool is_write)
> +{
> +    bool empty_queue;
> +    empty_queue = !qemu_co_enter_next(&fst->throttled_reqs[is_write]);
> +    if (empty_queue) {
> +        fsdev_throttle_schedule_next_request(fst, is_write);
> +    }
> +}

This doesn't make any sense, if the throttled_reqs queue is empty then
you're calling fsdev_throttle_schedule_next_request(), but how can a
next request be scheduled if the queue is empty? :-)

You don't need fsdev_throttle_timer_cb() at all, but you can simply do
this instead:

> +static void fsdev_throttle_read_timer_cb(void *opaque)
> +{
       FsThrottle *fst = opaque;
       qemu_co_enter_next(&fst->throttled_reqs[false]);
> +}
> +
> +static void fsdev_throttle_write_timer_cb(void *opaque)
> +{
       FsThrottle *fst = opaque;
       qemu_co_enter_next(&fst->throttled_reqs[true]);
> +}

> +static int get_num_bytes(struct iovec *iov, int iovcnt)
> +{
> +    int index = 0;
> +    int bytes = 0;
> +
> +    while (index < iovcnt) {
> +        bytes += iov[index].iov_len;
> +        index++;
> +    }
> +    return bytes;

You're looping over the iov vector, use a for loop instead, no?:

for (i = 0; i < iovcnt; i++) {
   bytes += iov[i].iov_len;
}

> +void fsdev_throttle_request(FsThrottle *fst, bool is_write,
> +                            struct iovec *iov, int iovcnt)

First, mark the function as running in a coroutine context:

void coroutine_fn fsdev_throttle_request(FsThrottle *fst, bool is_write,
                                         struct iovec *iov, int iovcnt)

> +{
> +    int bytes;
> +
> +    if (throttle_enabled(&fst->cfg)) {

I think it's probably better to check that in the caller, but I don't
have a strong opinion. If you keep it like this, put the definition of
'bytes' inside the if block instead.

> +        bytes = get_num_bytes(iov, iovcnt);
> +        bool must_wait = fsdev_throttle_check_for_wait(fst, is_write);
> +        if (must_wait || 
> !qemu_co_queue_empty(&fst->throttled_reqs[is_write])) {
> +            qemu_co_queue_wait(&fst->throttled_reqs[is_write]);
> +       }
> +       throttle_account(&fst->ts, is_write, bytes);
> +
> +       fsdev_throttle_schedule_next_request(fst, is_write);

So this is the [1] that I mentioned earlier. In your case instead of
fsdev_throttle_schedule_next_request() I think you simply something
like:

    if (!qemu_co_queue_empty(&fst->throttled_reqs[is_write])) {
       /* If there's a timer set, don't do anything else, the timer
          callback will schedule the next request */
       if (fsdev_throttle_check_for_wait(fst, is_write)) {
           return;
       }

       /* If a timer is not set, then schedule the next request now */
       qemu_co_queue_next(&fst->throttled_reqs[is_write]);
    }

Berto