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#include "qemu/osdep.h"
26#include "sysemu/block-backend.h"
27#include "block/throttle-groups.h"
28#include "qemu/queue.h"
29#include "qemu/thread.h"
30#include "sysemu/qtest.h"
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
56typedef struct ThrottleGroup {
57 char *name;
58
59 QemuMutex lock;
60 ThrottleState ts;
61 QLIST_HEAD(, BlockBackendPublic) head;
62 BlockBackend *tokens[2];
63 bool any_timer_armed[2];
64 QEMUClockType clock_type;
65
66
67 unsigned refcount;
68 QTAILQ_ENTRY(ThrottleGroup) list;
69} ThrottleGroup;
70
71static QemuMutex throttle_groups_lock;
72static QTAILQ_HEAD(, ThrottleGroup) throttle_groups =
73 QTAILQ_HEAD_INITIALIZER(throttle_groups);
74
75
76
77
78
79
80
81
82
83ThrottleState *throttle_group_incref(const char *name)
84{
85 ThrottleGroup *tg = NULL;
86 ThrottleGroup *iter;
87
88 qemu_mutex_lock(&throttle_groups_lock);
89
90
91 QTAILQ_FOREACH(iter, &throttle_groups, list) {
92 if (!strcmp(name, iter->name)) {
93 tg = iter;
94 break;
95 }
96 }
97
98
99 if (!tg) {
100 tg = g_new0(ThrottleGroup, 1);
101 tg->name = g_strdup(name);
102 tg->clock_type = QEMU_CLOCK_REALTIME;
103
104 if (qtest_enabled()) {
105
106 tg->clock_type = QEMU_CLOCK_VIRTUAL;
107 }
108 qemu_mutex_init(&tg->lock);
109 throttle_init(&tg->ts);
110 QLIST_INIT(&tg->head);
111
112 QTAILQ_INSERT_TAIL(&throttle_groups, tg, list);
113 }
114
115 tg->refcount++;
116
117 qemu_mutex_unlock(&throttle_groups_lock);
118
119 return &tg->ts;
120}
121
122
123
124
125
126
127
128
129void throttle_group_unref(ThrottleState *ts)
130{
131 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
132
133 qemu_mutex_lock(&throttle_groups_lock);
134 if (--tg->refcount == 0) {
135 QTAILQ_REMOVE(&throttle_groups, tg, list);
136 qemu_mutex_destroy(&tg->lock);
137 g_free(tg->name);
138 g_free(tg);
139 }
140 qemu_mutex_unlock(&throttle_groups_lock);
141}
142
143
144
145
146
147
148
149const char *throttle_group_get_name(BlockBackend *blk)
150{
151 BlockBackendPublic *blkp = blk_get_public(blk);
152 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
153 return tg->name;
154}
155
156
157
158
159
160
161
162
163
164static BlockBackend *throttle_group_next_blk(BlockBackend *blk)
165{
166 BlockBackendPublic *blkp = blk_get_public(blk);
167 ThrottleState *ts = blkp->throttle_state;
168 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
169 BlockBackendPublic *next = QLIST_NEXT(blkp, round_robin);
170
171 if (!next) {
172 next = QLIST_FIRST(&tg->head);
173 }
174
175 return blk_by_public(next);
176}
177
178
179
180
181
182
183
184
185
186
187static inline bool blk_has_pending_reqs(BlockBackend *blk,
188 bool is_write)
189{
190 const BlockBackendPublic *blkp = blk_get_public(blk);
191 return blkp->pending_reqs[is_write];
192}
193
194
195
196
197
198
199
200
201
202
203
204static BlockBackend *next_throttle_token(BlockBackend *blk, bool is_write)
205{
206 BlockBackendPublic *blkp = blk_get_public(blk);
207 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
208 BlockBackend *token, *start;
209
210 start = token = tg->tokens[is_write];
211
212
213 token = throttle_group_next_blk(token);
214 while (token != start && !blk_has_pending_reqs(token, is_write)) {
215 token = throttle_group_next_blk(token);
216 }
217
218
219
220
221
222 if (token == start && !blk_has_pending_reqs(token, is_write)) {
223 token = blk;
224 }
225
226
227 assert(token == blk || blk_has_pending_reqs(token, is_write));
228
229 return token;
230}
231
232
233
234
235
236
237
238
239
240
241
242static bool throttle_group_schedule_timer(BlockBackend *blk, bool is_write)
243{
244 BlockBackendPublic *blkp = blk_get_public(blk);
245 ThrottleState *ts = blkp->throttle_state;
246 ThrottleTimers *tt = &blkp->throttle_timers;
247 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
248 bool must_wait;
249
250 if (atomic_read(&blkp->io_limits_disabled)) {
251 return false;
252 }
253
254
255 if (tg->any_timer_armed[is_write]) {
256 return true;
257 }
258
259 must_wait = throttle_schedule_timer(ts, tt, is_write);
260
261
262 if (must_wait) {
263 tg->tokens[is_write] = blk;
264 tg->any_timer_armed[is_write] = true;
265 }
266
267 return must_wait;
268}
269
270
271
272
273
274
275
276static bool coroutine_fn throttle_group_co_restart_queue(BlockBackend *blk,
277 bool is_write)
278{
279 BlockBackendPublic *blkp = blk_get_public(blk);
280 bool ret;
281
282 qemu_co_mutex_lock(&blkp->throttled_reqs_lock);
283 ret = qemu_co_queue_next(&blkp->throttled_reqs[is_write]);
284 qemu_co_mutex_unlock(&blkp->throttled_reqs_lock);
285
286 return ret;
287}
288
289
290
291
292
293
294
295
296static void schedule_next_request(BlockBackend *blk, bool is_write)
297{
298 BlockBackendPublic *blkp = blk_get_public(blk);
299 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
300 bool must_wait;
301 BlockBackend *token;
302
303
304 token = next_throttle_token(blk, is_write);
305 if (!blk_has_pending_reqs(token, is_write)) {
306 return;
307 }
308
309
310 must_wait = throttle_group_schedule_timer(token, is_write);
311
312
313 if (!must_wait) {
314
315 if (qemu_in_coroutine() &&
316 throttle_group_co_restart_queue(blk, is_write)) {
317 token = blk;
318 } else {
319 ThrottleTimers *tt = &blk_get_public(token)->throttle_timers;
320 int64_t now = qemu_clock_get_ns(tg->clock_type);
321 timer_mod(tt->timers[is_write], now);
322 tg->any_timer_armed[is_write] = true;
323 }
324 tg->tokens[is_write] = token;
325 }
326}
327
328
329
330
331
332
333
334
335
336void coroutine_fn throttle_group_co_io_limits_intercept(BlockBackend *blk,
337 unsigned int bytes,
338 bool is_write)
339{
340 bool must_wait;
341 BlockBackend *token;
342
343 BlockBackendPublic *blkp = blk_get_public(blk);
344 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
345 qemu_mutex_lock(&tg->lock);
346
347
348 token = next_throttle_token(blk, is_write);
349 must_wait = throttle_group_schedule_timer(token, is_write);
350
351
352 if (must_wait || blkp->pending_reqs[is_write]) {
353 blkp->pending_reqs[is_write]++;
354 qemu_mutex_unlock(&tg->lock);
355 qemu_co_mutex_lock(&blkp->throttled_reqs_lock);
356 qemu_co_queue_wait(&blkp->throttled_reqs[is_write],
357 &blkp->throttled_reqs_lock);
358 qemu_co_mutex_unlock(&blkp->throttled_reqs_lock);
359 qemu_mutex_lock(&tg->lock);
360 blkp->pending_reqs[is_write]--;
361 }
362
363
364 throttle_account(blkp->throttle_state, is_write, bytes);
365
366
367 schedule_next_request(blk, is_write);
368
369 qemu_mutex_unlock(&tg->lock);
370}
371
372typedef struct {
373 BlockBackend *blk;
374 bool is_write;
375} RestartData;
376
377static void coroutine_fn throttle_group_restart_queue_entry(void *opaque)
378{
379 RestartData *data = opaque;
380 BlockBackend *blk = data->blk;
381 bool is_write = data->is_write;
382 BlockBackendPublic *blkp = blk_get_public(blk);
383 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
384 bool empty_queue;
385
386 empty_queue = !throttle_group_co_restart_queue(blk, is_write);
387
388
389
390 if (empty_queue) {
391 qemu_mutex_lock(&tg->lock);
392 schedule_next_request(blk, is_write);
393 qemu_mutex_unlock(&tg->lock);
394 }
395
396 g_free(data);
397}
398
399static void throttle_group_restart_queue(BlockBackend *blk, bool is_write)
400{
401 Coroutine *co;
402 RestartData *rd = g_new0(RestartData, 1);
403
404 rd->blk = blk;
405 rd->is_write = is_write;
406
407 co = qemu_coroutine_create(throttle_group_restart_queue_entry, rd);
408 aio_co_enter(blk_get_aio_context(blk), co);
409}
410
411void throttle_group_restart_blk(BlockBackend *blk)
412{
413 BlockBackendPublic *blkp = blk_get_public(blk);
414
415 if (blkp->throttle_state) {
416 throttle_group_restart_queue(blk, 0);
417 throttle_group_restart_queue(blk, 1);
418 }
419}
420
421
422
423
424
425
426
427
428void throttle_group_config(BlockBackend *blk, ThrottleConfig *cfg)
429{
430 BlockBackendPublic *blkp = blk_get_public(blk);
431 ThrottleState *ts = blkp->throttle_state;
432 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
433 qemu_mutex_lock(&tg->lock);
434 throttle_config(ts, tg->clock_type, cfg);
435 qemu_mutex_unlock(&tg->lock);
436
437 throttle_group_restart_blk(blk);
438}
439
440
441
442
443
444
445
446
447void throttle_group_get_config(BlockBackend *blk, ThrottleConfig *cfg)
448{
449 BlockBackendPublic *blkp = blk_get_public(blk);
450 ThrottleState *ts = blkp->throttle_state;
451 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
452 qemu_mutex_lock(&tg->lock);
453 throttle_get_config(ts, cfg);
454 qemu_mutex_unlock(&tg->lock);
455}
456
457
458
459
460
461
462
463static void timer_cb(BlockBackend *blk, bool is_write)
464{
465 BlockBackendPublic *blkp = blk_get_public(blk);
466 ThrottleState *ts = blkp->throttle_state;
467 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
468
469
470 qemu_mutex_lock(&tg->lock);
471 tg->any_timer_armed[is_write] = false;
472 qemu_mutex_unlock(&tg->lock);
473
474
475 throttle_group_restart_queue(blk, is_write);
476}
477
478static void read_timer_cb(void *opaque)
479{
480 timer_cb(opaque, false);
481}
482
483static void write_timer_cb(void *opaque)
484{
485 timer_cb(opaque, true);
486}
487
488
489
490
491
492
493
494
495void throttle_group_register_blk(BlockBackend *blk, const char *groupname)
496{
497 int i;
498 BlockBackendPublic *blkp = blk_get_public(blk);
499 ThrottleState *ts = throttle_group_incref(groupname);
500 ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
501 blkp->throttle_state = ts;
502
503 qemu_mutex_lock(&tg->lock);
504
505 for (i = 0; i < 2; i++) {
506 if (!tg->tokens[i]) {
507 tg->tokens[i] = blk;
508 }
509 }
510
511 QLIST_INSERT_HEAD(&tg->head, blkp, round_robin);
512
513 throttle_timers_init(&blkp->throttle_timers,
514 blk_get_aio_context(blk),
515 tg->clock_type,
516 read_timer_cb,
517 write_timer_cb,
518 blk);
519
520 qemu_mutex_unlock(&tg->lock);
521}
522
523
524
525
526
527
528
529
530
531
532
533void throttle_group_unregister_blk(BlockBackend *blk)
534{
535 BlockBackendPublic *blkp = blk_get_public(blk);
536 ThrottleGroup *tg = container_of(blkp->throttle_state, ThrottleGroup, ts);
537 int i;
538
539 assert(blkp->pending_reqs[0] == 0 && blkp->pending_reqs[1] == 0);
540 assert(qemu_co_queue_empty(&blkp->throttled_reqs[0]));
541 assert(qemu_co_queue_empty(&blkp->throttled_reqs[1]));
542
543 qemu_mutex_lock(&tg->lock);
544 for (i = 0; i < 2; i++) {
545 if (tg->tokens[i] == blk) {
546 BlockBackend *token = throttle_group_next_blk(blk);
547
548 if (token == blk) {
549 token = NULL;
550 }
551 tg->tokens[i] = token;
552 }
553 }
554
555
556 QLIST_REMOVE(blkp, round_robin);
557 throttle_timers_destroy(&blkp->throttle_timers);
558 qemu_mutex_unlock(&tg->lock);
559
560 throttle_group_unref(&tg->ts);
561 blkp->throttle_state = NULL;
562}
563
564static void throttle_groups_init(void)
565{
566 qemu_mutex_init(&throttle_groups_lock);
567}
568
569block_init(throttle_groups_init);
570