#include "sysemu/block-backend.h"
#include "block/throttle-groups.h"
#include "qemu/throttle-options.h"
+#include "qemu/main-loop.h"
#include "qemu/queue.h"
#include "qemu/thread.h"
#include "sysemu/qtest.h"
static void throttle_group_obj_init(Object *obj);
static void throttle_group_obj_complete(UserCreatable *obj, Error **errp);
+static void timer_cb(ThrottleGroupMember *tgm, ThrottleDirection direction);
/* The ThrottleGroup structure (with its ThrottleState) is shared
* among different ThrottleGroupMembers and it's independent from
* access some other ThrottleGroupMember's timers only after verifying that
* that ThrottleGroupMember has throttled requests in the queue.
*/
-typedef struct ThrottleGroup {
+struct ThrottleGroup {
Object parent_obj;
/* refuse individual property change if initialization is complete */
QemuMutex lock; /* This lock protects the following four fields */
ThrottleState ts;
QLIST_HEAD(, ThrottleGroupMember) head;
- ThrottleGroupMember *tokens[2];
- bool any_timer_armed[2];
+ ThrottleGroupMember *tokens[THROTTLE_MAX];
+ bool any_timer_armed[THROTTLE_MAX];
QEMUClockType clock_type;
/* This field is protected by the global QEMU mutex */
QTAILQ_ENTRY(ThrottleGroup) list;
-} ThrottleGroup;
+};
/* This is protected by the global QEMU mutex */
static QTAILQ_HEAD(, ThrottleGroup) throttle_groups =
* This assumes that tg->lock is held.
*
* @tgm: the ThrottleGroupMember
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
* @ret: whether the ThrottleGroupMember has pending requests.
*/
static inline bool tgm_has_pending_reqs(ThrottleGroupMember *tgm,
- bool is_write)
+ ThrottleDirection direction)
{
- return tgm->pending_reqs[is_write];
+ return tgm->pending_reqs[direction];
}
/* Return the next ThrottleGroupMember in the round-robin sequence with pending
* This assumes that tg->lock is held.
*
* @tgm: the current ThrottleGroupMember
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
* @ret: the next ThrottleGroupMember with pending requests, or tgm if
* there is none.
*/
static ThrottleGroupMember *next_throttle_token(ThrottleGroupMember *tgm,
- bool is_write)
+ ThrottleDirection direction)
{
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
* it's being drained. Skip the round-robin search and return tgm
* immediately if it has pending requests. Otherwise we could be
* forcing it to wait for other member's throttled requests. */
- if (tgm_has_pending_reqs(tgm, is_write) &&
- atomic_read(&tgm->io_limits_disabled)) {
+ if (tgm_has_pending_reqs(tgm, direction) &&
+ qatomic_read(&tgm->io_limits_disabled)) {
return tgm;
}
- start = token = tg->tokens[is_write];
+ start = token = tg->tokens[direction];
/* get next bs round in round robin style */
token = throttle_group_next_tgm(token);
- while (token != start && !tgm_has_pending_reqs(token, is_write)) {
+ while (token != start && !tgm_has_pending_reqs(token, direction)) {
token = throttle_group_next_tgm(token);
}
* then decide the token is the current tgm because chances are
* the current tgm got the current request queued.
*/
- if (token == start && !tgm_has_pending_reqs(token, is_write)) {
+ if (token == start && !tgm_has_pending_reqs(token, direction)) {
token = tgm;
}
/* Either we return the original TGM, or one with pending requests */
- assert(token == tgm || tgm_has_pending_reqs(token, is_write));
+ assert(token == tgm || tgm_has_pending_reqs(token, direction));
return token;
}
* This assumes that tg->lock is held.
*
* @tgm: the current ThrottleGroupMember
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
* @ret: whether the I/O request needs to be throttled or not
*/
static bool throttle_group_schedule_timer(ThrottleGroupMember *tgm,
- bool is_write)
+ ThrottleDirection direction)
{
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
ThrottleTimers *tt = &tgm->throttle_timers;
bool must_wait;
- if (atomic_read(&tgm->io_limits_disabled)) {
+ if (qatomic_read(&tgm->io_limits_disabled)) {
return false;
}
/* Check if any of the timers in this group is already armed */
- if (tg->any_timer_armed[is_write]) {
+ if (tg->any_timer_armed[direction]) {
return true;
}
- must_wait = throttle_schedule_timer(ts, tt, is_write);
+ must_wait = throttle_schedule_timer(ts, tt, direction);
/* If a timer just got armed, set tgm as the current token */
if (must_wait) {
- tg->tokens[is_write] = tgm;
- tg->any_timer_armed[is_write] = true;
+ tg->tokens[direction] = tgm;
+ tg->any_timer_armed[direction] = true;
}
return must_wait;
* any request was actually pending.
*
* @tgm: the current ThrottleGroupMember
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
*/
static bool coroutine_fn throttle_group_co_restart_queue(ThrottleGroupMember *tgm,
- bool is_write)
+ ThrottleDirection direction)
{
bool ret;
qemu_co_mutex_lock(&tgm->throttled_reqs_lock);
- ret = qemu_co_queue_next(&tgm->throttled_reqs[is_write]);
+ ret = qemu_co_queue_next(&tgm->throttled_reqs[direction]);
qemu_co_mutex_unlock(&tgm->throttled_reqs_lock);
return ret;
* This assumes that tg->lock is held.
*
* @tgm: the current ThrottleGroupMember
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
*/
-static void schedule_next_request(ThrottleGroupMember *tgm, bool is_write)
+static void coroutine_mixed_fn schedule_next_request(ThrottleGroupMember *tgm,
+ ThrottleDirection direction)
{
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
ThrottleGroupMember *token;
/* Check if there's any pending request to schedule next */
- token = next_throttle_token(tgm, is_write);
- if (!tgm_has_pending_reqs(token, is_write)) {
+ token = next_throttle_token(tgm, direction);
+ if (!tgm_has_pending_reqs(token, direction)) {
return;
}
/* Set a timer for the request if it needs to be throttled */
- must_wait = throttle_group_schedule_timer(token, is_write);
+ must_wait = throttle_group_schedule_timer(token, direction);
/* If it doesn't have to wait, queue it for immediate execution */
if (!must_wait) {
/* Give preference to requests from the current tgm */
if (qemu_in_coroutine() &&
- throttle_group_co_restart_queue(tgm, is_write)) {
+ throttle_group_co_restart_queue(tgm, direction)) {
token = tgm;
} else {
ThrottleTimers *tt = &token->throttle_timers;
int64_t now = qemu_clock_get_ns(tg->clock_type);
- timer_mod(tt->timers[is_write], now);
- tg->any_timer_armed[is_write] = true;
+ timer_mod(tt->timers[direction], now);
+ tg->any_timer_armed[direction] = true;
}
- tg->tokens[is_write] = token;
+ tg->tokens[direction] = token;
}
}
*
* @tgm: the current ThrottleGroupMember
* @bytes: the number of bytes for this I/O
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
*/
void coroutine_fn throttle_group_co_io_limits_intercept(ThrottleGroupMember *tgm,
- unsigned int bytes,
- bool is_write)
+ int64_t bytes,
+ ThrottleDirection direction)
{
bool must_wait;
ThrottleGroupMember *token;
ThrottleGroup *tg = container_of(tgm->throttle_state, ThrottleGroup, ts);
+
+ assert(bytes >= 0);
+ assert(direction < THROTTLE_MAX);
+
qemu_mutex_lock(&tg->lock);
/* First we check if this I/O has to be throttled. */
- token = next_throttle_token(tgm, is_write);
- must_wait = throttle_group_schedule_timer(token, is_write);
+ token = next_throttle_token(tgm, direction);
+ must_wait = throttle_group_schedule_timer(token, direction);
/* Wait if there's a timer set or queued requests of this type */
- if (must_wait || tgm->pending_reqs[is_write]) {
- tgm->pending_reqs[is_write]++;
+ if (must_wait || tgm->pending_reqs[direction]) {
+ tgm->pending_reqs[direction]++;
qemu_mutex_unlock(&tg->lock);
qemu_co_mutex_lock(&tgm->throttled_reqs_lock);
- qemu_co_queue_wait(&tgm->throttled_reqs[is_write],
+ qemu_co_queue_wait(&tgm->throttled_reqs[direction],
&tgm->throttled_reqs_lock);
qemu_co_mutex_unlock(&tgm->throttled_reqs_lock);
qemu_mutex_lock(&tg->lock);
- tgm->pending_reqs[is_write]--;
+ tgm->pending_reqs[direction]--;
}
/* The I/O will be executed, so do the accounting */
- throttle_account(tgm->throttle_state, is_write, bytes);
+ throttle_account(tgm->throttle_state, direction, bytes);
/* Schedule the next request */
- schedule_next_request(tgm, is_write);
+ schedule_next_request(tgm, direction);
qemu_mutex_unlock(&tg->lock);
}
typedef struct {
ThrottleGroupMember *tgm;
- bool is_write;
+ ThrottleDirection direction;
} RestartData;
static void coroutine_fn throttle_group_restart_queue_entry(void *opaque)
ThrottleGroupMember *tgm = data->tgm;
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
- bool is_write = data->is_write;
+ ThrottleDirection direction = data->direction;
bool empty_queue;
- empty_queue = !throttle_group_co_restart_queue(tgm, is_write);
+ empty_queue = !throttle_group_co_restart_queue(tgm, direction);
/* If the request queue was empty then we have to take care of
* scheduling the next one */
if (empty_queue) {
qemu_mutex_lock(&tg->lock);
- schedule_next_request(tgm, is_write);
+ schedule_next_request(tgm, direction);
qemu_mutex_unlock(&tg->lock);
}
g_free(data);
+
+ qatomic_dec(&tgm->restart_pending);
+ aio_wait_kick();
}
-static void throttle_group_restart_queue(ThrottleGroupMember *tgm, bool is_write)
+static void throttle_group_restart_queue(ThrottleGroupMember *tgm,
+ ThrottleDirection direction)
{
Coroutine *co;
RestartData *rd = g_new0(RestartData, 1);
rd->tgm = tgm;
- rd->is_write = is_write;
+ rd->direction = direction;
+
+ /* This function is called when a timer is fired or when
+ * throttle_group_restart_tgm() is called. Either way, there can
+ * be no timer pending on this tgm at this point */
+ assert(!timer_pending(tgm->throttle_timers.timers[direction]));
+
+ qatomic_inc(&tgm->restart_pending);
co = qemu_coroutine_create(throttle_group_restart_queue_entry, rd);
aio_co_enter(tgm->aio_context, co);
void throttle_group_restart_tgm(ThrottleGroupMember *tgm)
{
+ ThrottleDirection dir;
+
if (tgm->throttle_state) {
- throttle_group_restart_queue(tgm, 0);
- throttle_group_restart_queue(tgm, 1);
+ for (dir = THROTTLE_READ; dir < THROTTLE_MAX; dir++) {
+ QEMUTimer *t = tgm->throttle_timers.timers[dir];
+ if (timer_pending(t)) {
+ /* If there's a pending timer on this tgm, fire it now */
+ timer_del(t);
+ timer_cb(tgm, dir);
+ } else {
+ /* Else run the next request from the queue manually */
+ throttle_group_restart_queue(tgm, dir);
+ }
+ }
}
}
* because it had been throttled.
*
* @tgm: the ThrottleGroupMember whose request had been throttled
- * @is_write: the type of operation (read/write)
+ * @direction: the ThrottleDirection
*/
-static void timer_cb(ThrottleGroupMember *tgm, bool is_write)
+static void timer_cb(ThrottleGroupMember *tgm, ThrottleDirection direction)
{
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
/* The timer has just been fired, so we can update the flag */
qemu_mutex_lock(&tg->lock);
- tg->any_timer_armed[is_write] = false;
+ tg->any_timer_armed[direction] = false;
qemu_mutex_unlock(&tg->lock);
/* Run the request that was waiting for this timer */
- throttle_group_restart_queue(tgm, is_write);
+ throttle_group_restart_queue(tgm, direction);
}
static void read_timer_cb(void *opaque)
{
- timer_cb(opaque, false);
+ timer_cb(opaque, THROTTLE_READ);
}
static void write_timer_cb(void *opaque)
{
- timer_cb(opaque, true);
+ timer_cb(opaque, THROTTLE_WRITE);
}
/* Register a ThrottleGroupMember from the throttling group, also initializing
const char *groupname,
AioContext *ctx)
{
- int i;
+ ThrottleDirection dir;
ThrottleState *ts = throttle_group_incref(groupname);
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
tgm->throttle_state = ts;
tgm->aio_context = ctx;
+ qatomic_set(&tgm->restart_pending, 0);
- qemu_mutex_lock(&tg->lock);
+ QEMU_LOCK_GUARD(&tg->lock);
/* If the ThrottleGroup is new set this ThrottleGroupMember as the token */
- for (i = 0; i < 2; i++) {
- if (!tg->tokens[i]) {
- tg->tokens[i] = tgm;
+ for (dir = THROTTLE_READ; dir < THROTTLE_MAX; dir++) {
+ if (!tg->tokens[dir]) {
+ tg->tokens[dir] = tgm;
}
+ qemu_co_queue_init(&tgm->throttled_reqs[dir]);
}
QLIST_INSERT_HEAD(&tg->head, tgm, round_robin);
write_timer_cb,
tgm);
qemu_co_mutex_init(&tgm->throttled_reqs_lock);
- qemu_co_queue_init(&tgm->throttled_reqs[0]);
- qemu_co_queue_init(&tgm->throttled_reqs[1]);
-
- qemu_mutex_unlock(&tg->lock);
}
/* Unregister a ThrottleGroupMember from its group, removing it from the list,
ThrottleState *ts = tgm->throttle_state;
ThrottleGroup *tg = container_of(ts, ThrottleGroup, ts);
ThrottleGroupMember *token;
- int i;
+ ThrottleDirection dir;
if (!ts) {
/* Discard already unregistered tgm */
return;
}
- assert(tgm->pending_reqs[0] == 0 && tgm->pending_reqs[1] == 0);
- assert(qemu_co_queue_empty(&tgm->throttled_reqs[0]));
- assert(qemu_co_queue_empty(&tgm->throttled_reqs[1]));
-
- qemu_mutex_lock(&tg->lock);
- for (i = 0; i < 2; i++) {
- if (timer_pending(tgm->throttle_timers.timers[i])) {
- tg->any_timer_armed[i] = false;
- schedule_next_request(tgm, i);
- }
- if (tg->tokens[i] == tgm) {
- token = throttle_group_next_tgm(tgm);
- /* Take care of the case where this is the last tgm in the group */
- if (token == tgm) {
- token = NULL;
+ /* Wait for throttle_group_restart_queue_entry() coroutines to finish */
+ AIO_WAIT_WHILE(tgm->aio_context, qatomic_read(&tgm->restart_pending) > 0);
+
+ WITH_QEMU_LOCK_GUARD(&tg->lock) {
+ for (dir = THROTTLE_READ; dir < THROTTLE_MAX; dir++) {
+ assert(tgm->pending_reqs[dir] == 0);
+ assert(qemu_co_queue_empty(&tgm->throttled_reqs[dir]));
+ assert(!timer_pending(tgm->throttle_timers.timers[dir]));
+ if (tg->tokens[dir] == tgm) {
+ token = throttle_group_next_tgm(tgm);
+ /* Take care of the case where this is the last tgm in the group */
+ if (token == tgm) {
+ token = NULL;
+ }
+ tg->tokens[dir] = token;
}
- tg->tokens[i] = token;
}
- }
- /* remove the current tgm from the list */
- QLIST_REMOVE(tgm, round_robin);
- throttle_timers_destroy(&tgm->throttle_timers);
- qemu_mutex_unlock(&tg->lock);
+ /* remove the current tgm from the list */
+ QLIST_REMOVE(tgm, round_robin);
+ throttle_timers_destroy(&tgm->throttle_timers);
+ }
throttle_group_unref(&tg->ts);
tgm->throttle_state = NULL;
{
ThrottleGroup *tg = container_of(tgm->throttle_state, ThrottleGroup, ts);
ThrottleTimers *tt = &tgm->throttle_timers;
- int i;
+ ThrottleDirection dir;
/* Requests must have been drained */
- assert(tgm->pending_reqs[0] == 0 && tgm->pending_reqs[1] == 0);
- assert(qemu_co_queue_empty(&tgm->throttled_reqs[0]));
- assert(qemu_co_queue_empty(&tgm->throttled_reqs[1]));
+ for (dir = THROTTLE_READ; dir < THROTTLE_MAX; dir++) {
+ assert(tgm->pending_reqs[dir] == 0);
+ assert(qemu_co_queue_empty(&tgm->throttled_reqs[dir]));
+ }
/* Kick off next ThrottleGroupMember, if necessary */
- qemu_mutex_lock(&tg->lock);
- for (i = 0; i < 2; i++) {
- if (timer_pending(tt->timers[i])) {
- tg->any_timer_armed[i] = false;
- schedule_next_request(tgm, i);
+ WITH_QEMU_LOCK_GUARD(&tg->lock) {
+ for (dir = THROTTLE_READ; dir < THROTTLE_MAX; dir++) {
+ if (timer_pending(tt->timers[dir])) {
+ tg->any_timer_armed[dir] = false;
+ schedule_next_request(tgm, dir);
+ }
}
}
- qemu_mutex_unlock(&tg->lock);
throttle_timers_detach_aio_context(tt);
tgm->aio_context = NULL;
/* set group name to object id if it exists */
if (!tg->name && tg->parent_obj.parent) {
- tg->name = object_get_canonical_path_component(OBJECT(obj));
+ tg->name = g_strdup(object_get_canonical_path_component(OBJECT(obj)));
}
/* We must have a group name at this point */
assert(tg->name);
ThrottleGroup *tg = THROTTLE_GROUP(obj);
ThrottleConfig *cfg;
ThrottleParamInfo *info = opaque;
- Error *local_err = NULL;
int64_t value;
/* If we have finished initialization, don't accept individual property
* transaction, as certain combinations are invalid.
*/
if (tg->is_initialized) {
- error_setg(&local_err, "Property cannot be set after initialization");
- goto ret;
+ error_setg(errp, "Property cannot be set after initialization");
+ return;
}
- visit_type_int64(v, name, &value, &local_err);
- if (local_err) {
- goto ret;
+ if (!visit_type_int64(v, name, &value, errp)) {
+ return;
}
if (value < 0) {
- error_setg(&local_err, "Property values cannot be negative");
- goto ret;
+ error_setg(errp, "Property values cannot be negative");
+ return;
}
cfg = &tg->ts.cfg;
break;
case BURST_LENGTH:
if (value > UINT_MAX) {
- error_setg(&local_err, "%s value must be in the"
- "range [0, %u]", info->name, UINT_MAX);
- goto ret;
+ error_setg(errp, "%s value must be in the" "range [0, %u]",
+ info->name, UINT_MAX);
+ return;
}
cfg->buckets[info->type].burst_length = value;
break;
cfg->op_size = value;
break;
}
-
-ret:
- error_propagate(errp, local_err);
- return;
-
}
static void throttle_group_get(Object *obj, Visitor *v, const char *name,
{
ThrottleGroup *tg = THROTTLE_GROUP(obj);
ThrottleConfig cfg;
- ThrottleLimits arg = { 0 };
- ThrottleLimits *argp = &arg;
+ ThrottleLimits *argp;
Error *local_err = NULL;
- visit_type_ThrottleLimits(v, name, &argp, &local_err);
- if (local_err) {
- goto ret;
+ if (!visit_type_ThrottleLimits(v, name, &argp, errp)) {
+ return;
}
qemu_mutex_lock(&tg->lock);
throttle_get_config(&tg->ts, &cfg);
unlock:
qemu_mutex_unlock(&tg->lock);
-ret:
+ qapi_free_ThrottleLimits(argp);
error_propagate(errp, local_err);
return;
}
"int",
throttle_group_get,
throttle_group_set,
- NULL, &properties[i],
- &error_abort);
+ NULL, &properties[i]);
}
/* ThrottleLimits */
"limits", "ThrottleLimits",
throttle_group_get_limits,
throttle_group_set_limits,
- NULL, NULL,
- &error_abort);
+ NULL, NULL);
}
static const TypeInfo throttle_group_info = {
projects / mirror_qemu.git / blobdiff