From e30529bcef7709d038ddafc7293c902a9ba1b4d0 Mon Sep 17 00:00:00 2001 From: Jeffrey Cody Date: Wed, 21 Mar 2012 21:55:07 +0100 Subject: [PATCH 40/55] block: rate-limit streaming operations RH-Author: Jeffrey Cody Message-id: <9eac69b4a55875d01a675e171e0b7f9d03982f25.1332362400.git.jcody@redhat.com> Patchwork-id: 38890 O-Subject: [RHEL6.3 qemu-kvm PATCH v8 40/54] block: rate-limit streaming operations Bugzilla: 582475 RH-Acked-by: Paolo Bonzini RH-Acked-by: Marcelo Tosatti RH-Acked-by: Kevin Wolf From: Stefan Hajnoczi This patch implements rate-limiting for image streaming. If we've exceeded the bandwidth quota for a 100 ms time slice we sleep the coroutine until the next slice begins. Signed-off-by: Stefan Hajnoczi Signed-off-by: Kevin Wolf (cherry picked from commit 5094a6c016f6e7a4fc800816d716e10ce2331396) Signed-off-by: Stefan Hajnoczi Signed-off-by: Anthony Liguori Signed-off-by: Jeff Cody --- block/stream.c | 65 ++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 59 insertions(+), 6 deletions(-) Signed-off-by: Michal Novotny --- block/stream.c | 65 ++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 59 insertions(+), 6 deletions(-) diff --git a/block/stream.c b/block/stream.c index 886f704..200fdce 100644 --- a/block/stream.c +++ b/block/stream.c @@ -23,8 +23,39 @@ enum { STREAM_BUFFER_SIZE = 512 * 1024, /* in bytes */ }; +#define SLICE_TIME 100ULL /* ms */ + +typedef struct { + int64_t next_slice_time; + uint64_t slice_quota; + uint64_t dispatched; +} RateLimit; + +static int64_t ratelimit_calculate_delay(RateLimit *limit, uint64_t n) +{ + int64_t delay_ns = 0; + int64_t now = qemu_get_clock(rt_clock); + + if (limit->next_slice_time < now) { + limit->next_slice_time = now + SLICE_TIME; + limit->dispatched = 0; + } + if (limit->dispatched + n > limit->slice_quota) { + delay_ns = limit->next_slice_time - now; + } else { + limit->dispatched += n; + } + return delay_ns; +} + +static void ratelimit_set_speed(RateLimit *limit, uint64_t speed) +{ + limit->slice_quota = speed / (1000ULL / SLICE_TIME); +} + typedef struct StreamBlockJob { BlockJob common; + RateLimit limit; BlockDriverState *base; } StreamBlockJob; @@ -72,20 +103,24 @@ static void coroutine_fn stream_run(void *opaque) } for (sector_num = 0; sector_num < end; sector_num += n) { +retry: if (block_job_is_cancelled(&s->common)) { break; } - /* TODO rate-limit */ - /* Note that even when no rate limit is applied we need to yield with - * no pending I/O here so that qemu_aio_flush() is able to return. - */ - co_sleep(rt_clock, 0); - ret = bdrv_co_is_allocated(bs, sector_num, STREAM_BUFFER_SIZE / BDRV_SECTOR_SIZE, &n); trace_stream_one_iteration(s, sector_num, n, ret); if (ret == 0) { + if (s->common.speed) { + uint64_t delay_ms = ratelimit_calculate_delay(&s->limit, n); + if (delay_ms > 0) { + co_sleep(rt_clock, delay_ms); + + /* Recheck cancellation and that sectors are unallocated */ + goto retry; + } + } ret = stream_populate(bs, sector_num, n, buf); } if (ret < 0) { @@ -94,6 +129,11 @@ static void coroutine_fn stream_run(void *opaque) /* Publish progress */ s->common.offset += n * BDRV_SECTOR_SIZE; + + /* Note that even when no rate limit is applied we need to yield + * with no pending I/O here so that qemu_aio_flush() returns. + */ + co_sleep(rt_clock, 0); } if (!s->base) { @@ -108,9 +148,22 @@ static void coroutine_fn stream_run(void *opaque) block_job_complete(&s->common, ret); } +static int stream_set_speed(BlockJob *job, int64_t value) +{ + StreamBlockJob *s = container_of(job, StreamBlockJob, common); + + if (value < 0) { + return -EINVAL; + } + job->speed = value; + ratelimit_set_speed(&s->limit, value / BDRV_SECTOR_SIZE); + return 0; +} + static BlockJobType stream_job_type = { .instance_size = sizeof(StreamBlockJob), .job_type = "stream", + .set_speed = stream_set_speed, }; int stream_start(BlockDriverState *bs, BlockDriverState *base, -- 1.7.7.6