aboutsummaryrefslogtreecommitdiff
path: root/util/throttle.c
blob: f976ac7de585da8683811b1275001cf73c71c636 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
/*
 * QEMU throttling infrastructure
 *
 * Copyright (C) Nodalink, SARL. 2013
 *
 * Author:
 *   Benoît Canet <benoit.canet@irqsave.net>
 *
 * 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 "qemu/throttle.h"
#include "qemu/timer.h"
#include "block/aio.h"

/* This function make a bucket leak
 *
 * @bkt:   the bucket to make leak
 * @delta_ns: the time delta
 */
void throttle_leak_bucket(LeakyBucket *bkt, int64_t delta_ns)
{
    double leak;

    /* compute how much to leak */
    leak = (bkt->avg * (double) delta_ns) / NANOSECONDS_PER_SECOND;

    /* make the bucket leak */
    bkt->level = MAX(bkt->level - leak, 0);
}

/* Calculate the time delta since last leak and make proportionals leaks
 *
 * @now:      the current timestamp in ns
 */
static void throttle_do_leak(ThrottleState *ts, int64_t now)
{
    /* compute the time elapsed since the last leak */
    int64_t delta_ns = now - ts->previous_leak;
    int i;

    ts->previous_leak = now;

    if (delta_ns <= 0) {
        return;
    }

    /* make each bucket leak */
    for (i = 0; i < BUCKETS_COUNT; i++) {
        throttle_leak_bucket(&ts->cfg.buckets[i], delta_ns);
    }
}

/* do the real job of computing the time to wait
 *
 * @limit: the throttling limit
 * @extra: the number of operation to delay
 * @ret:   the time to wait in ns
 */
static int64_t throttle_do_compute_wait(double limit, double extra)
{
    double wait = extra * NANOSECONDS_PER_SECOND;
    wait /= limit;
    return wait;
}

/* This function compute the wait time in ns that a leaky bucket should trigger
 *
 * @bkt: the leaky bucket we operate on
 * @ret: the resulting wait time in ns or 0 if the operation can go through
 */
int64_t throttle_compute_wait(LeakyBucket *bkt)
{
    double extra; /* the number of extra units blocking the io */

    if (!bkt->avg) {
        return 0;
    }

    extra = bkt->level - bkt->max;

    if (extra <= 0) {
        return 0;
    }

    return throttle_do_compute_wait(bkt->avg, extra);
}

/* This function compute the time that must be waited while this IO
 *
 * @is_write:   true if the current IO is a write, false if it's a read
 * @ret:        time to wait
 */
static int64_t throttle_compute_wait_for(ThrottleState *ts,
                                         bool is_write)
{
    BucketType to_check[2][4] = { {THROTTLE_BPS_TOTAL,
                                   THROTTLE_OPS_TOTAL,
                                   THROTTLE_BPS_READ,
                                   THROTTLE_OPS_READ},
                                  {THROTTLE_BPS_TOTAL,
                                   THROTTLE_OPS_TOTAL,
                                   THROTTLE_BPS_WRITE,
                                   THROTTLE_OPS_WRITE}, };
    int64_t wait, max_wait = 0;
    int i;

    for (i = 0; i < 4; i++) {
        BucketType index = to_check[is_write][i];
        wait = throttle_compute_wait(&ts->cfg.buckets[index]);
        if (wait > max_wait) {
            max_wait = wait;
        }
    }

    return max_wait;
}

/* compute the timer for this type of operation
 *
 * @is_write:   the type of operation
 * @now:        the current clock timestamp
 * @next_timestamp: the resulting timer
 * @ret:        true if a timer must be set
 */
bool throttle_compute_timer(ThrottleState *ts,
                            bool is_write,
                            int64_t now,
                            int64_t *next_timestamp)
{
    int64_t wait;

    /* leak proportionally to the time elapsed */
    throttle_do_leak(ts, now);

    /* compute the wait time if any */
    wait = throttle_compute_wait_for(ts, is_write);

    /* if the code must wait compute when the next timer should fire */
    if (wait) {
        *next_timestamp = now + wait;
        return true;
    }

    /* else no need to wait at all */
    *next_timestamp = now;
    return false;
}

/* Add timers to event loop */
void throttle_attach_aio_context(ThrottleState *ts, 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);
}

/* 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)
{
    memset(ts, 0, sizeof(ThrottleState));

    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);
}

/* destroy a timer */
static void throttle_timer_destroy(QEMUTimer **timer)
{
    assert(*timer != NULL);

    timer_del(*timer);
    timer_free(*timer);
    *timer = NULL;
}

/* Remove timers from event loop */
void throttle_detach_aio_context(ThrottleState *ts)
{
    int i;

    for (i = 0; i < 2; i++) {
        throttle_timer_destroy(&ts->timers[i]);
    }
}

/* To be called last on the ThrottleState */
void throttle_destroy(ThrottleState *ts)
{
    throttle_detach_aio_context(ts);
}

/* is any throttling timer configured */
bool throttle_have_timer(ThrottleState *ts)
{
    if (ts->timers[0]) {
        return true;
    }

    return false;
}

/* Does any throttling must be done
 *
 * @cfg: the throttling configuration to inspect
 * @ret: true if throttling must be done else false
 */
bool throttle_enabled(ThrottleConfig *cfg)
{
    int i;

    for (i = 0; i < BUCKETS_COUNT; i++) {
        if (cfg->buckets[i].avg > 0) {
            return true;
        }
    }

    return false;
}

/* return true if any two throttling parameters conflicts
 *
 * @cfg: the throttling configuration to inspect
 * @ret: true if any conflict detected else false
 */
bool throttle_conflicting(ThrottleConfig *cfg)
{
    bool bps_flag, ops_flag;
    bool bps_max_flag, ops_max_flag;

    bps_flag = cfg->buckets[THROTTLE_BPS_TOTAL].avg &&
               (cfg->buckets[THROTTLE_BPS_READ].avg ||
                cfg->buckets[THROTTLE_BPS_WRITE].avg);

    ops_flag = cfg->buckets[THROTTLE_OPS_TOTAL].avg &&
               (cfg->buckets[THROTTLE_OPS_READ].avg ||
                cfg->buckets[THROTTLE_OPS_WRITE].avg);

    bps_max_flag = cfg->buckets[THROTTLE_BPS_TOTAL].max &&
                  (cfg->buckets[THROTTLE_BPS_READ].max  ||
                   cfg->buckets[THROTTLE_BPS_WRITE].max);

    ops_max_flag = cfg->buckets[THROTTLE_OPS_TOTAL].max &&
                   (cfg->buckets[THROTTLE_OPS_READ].max ||
                   cfg->buckets[THROTTLE_OPS_WRITE].max);

    return bps_flag || ops_flag || bps_max_flag || ops_max_flag;
}

/* check if a throttling configuration is valid
 * @cfg: the throttling configuration to inspect
 * @ret: true if valid else false
 */
bool throttle_is_valid(ThrottleConfig *cfg)
{
    bool invalid = false;
    int i;

    for (i = 0; i < BUCKETS_COUNT; i++) {
        if (cfg->buckets[i].avg < 0) {
            invalid = true;
        }
    }

    for (i = 0; i < BUCKETS_COUNT; i++) {
        if (cfg->buckets[i].max < 0) {
            invalid = true;
        }
    }

    return !invalid;
}

/* fix bucket parameters */
static void throttle_fix_bucket(LeakyBucket *bkt)
{
    double min;

    /* zero bucket level */
    bkt->level = 0;

    /* The following is done to cope with the Linux CFQ block scheduler
     * which regroup reads and writes by block of 100ms in the guest.
     * When they are two process one making reads and one making writes cfq
     * make a pattern looking like the following:
     * WWWWWWWWWWWRRRRRRRRRRRRRRWWWWWWWWWWWWWwRRRRRRRRRRRRRRRRR
     * Having a max burst value of 100ms of the average will help smooth the
     * throttling
     */
    min = bkt->avg / 10;
    if (bkt->avg && !bkt->max) {
        bkt->max = min;
    }
}

/* take care of canceling a timer */
static void throttle_cancel_timer(QEMUTimer *timer)
{
    assert(timer != NULL);

    timer_del(timer);
}

/* Used to configure the throttle
 *
 * @ts: the throttle state we are working on
 * @cfg: the config to set
 */
void throttle_config(ThrottleState *ts, ThrottleConfig *cfg)
{
    int i;

    ts->cfg = *cfg;

    for (i = 0; i < BUCKETS_COUNT; i++) {
        throttle_fix_bucket(&ts->cfg.buckets[i]);
    }

    ts->previous_leak = qemu_clock_get_ns(ts->clock_type);

    for (i = 0; i < 2; i++) {
        throttle_cancel_timer(ts->timers[i]);
    }
}

/* used to get config
 *
 * @ts:  the throttle state we are working on
 * @cfg: the config to write
 */
void throttle_get_config(ThrottleState *ts, ThrottleConfig *cfg)
{
    *cfg = ts->cfg;
}


/* Schedule the read or write timer if needed
 *
 * NOTE: this function is not unit tested due to it's usage of timer_mod
 *
 * @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)
{
    int64_t now = qemu_clock_get_ns(ts->clock_type);
    int64_t next_timestamp;
    bool must_wait;

    must_wait = throttle_compute_timer(ts,
                                       is_write,
                                       now,
                                       &next_timestamp);

    /* request not throttled */
    if (!must_wait) {
        return false;
    }

    /* request throttled and timer pending -> do nothing */
    if (timer_pending(ts->timers[is_write])) {
        return true;
    }

    /* request throttled and timer not pending -> arm timer */
    timer_mod(ts->timers[is_write], next_timestamp);
    return true;
}

/* do the accounting for this operation
 *
 * @is_write: the type of operation (read/write)
 * @size:     the size of the operation
 */
void throttle_account(ThrottleState *ts, bool is_write, uint64_t size)
{
    double units = 1.0;

    /* if cfg.op_size is defined and smaller than size we compute unit count */
    if (ts->cfg.op_size && size > ts->cfg.op_size) {
        units = (double) size / ts->cfg.op_size;
    }

    ts->cfg.buckets[THROTTLE_BPS_TOTAL].level += size;
    ts->cfg.buckets[THROTTLE_OPS_TOTAL].level += units;

    if (is_write) {
        ts->cfg.buckets[THROTTLE_BPS_WRITE].level += size;
        ts->cfg.buckets[THROTTLE_OPS_WRITE].level += units;
    } else {
        ts->cfg.buckets[THROTTLE_BPS_READ].level += size;
        ts->cfg.buckets[THROTTLE_OPS_READ].level += units;
    }
}