NoaSoftware
/
TencentOS-kernel
forked from TencentOS/TencentOS-kernel
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

blk-wbt: improve waking of tasks 

commit <ffa358dcaae1f2f00926484e712e06daa8953cb4>
commit <b78820937b4762b7d30b807d7156bec1d89e4dd3>
commit <c45e6a037a536530bd25781ac7c989e52deb2a63>
commit <38cfb5a45ee013bfab5d1ae4c4738815e744b440>

We have two potential issues:

1) After commit 2887e41b910b, we only wake one process at the time when
   we finish an IO. We really want to wake up as many tasks as can
   queue IO. Before this commit, we woke up everyone, which could cause
   a thundering herd issue.

2) A task can potentially consume two wakeups, causing us to (in
   practice) miss a wakeup.

Fix both by providing our own wakeup function, which stops
__wake_up_common() from waking up more tasks if we fail to get a
queueing token. With the strict ordering we have on the wait list, this
wakes the right tasks and the right amount of tasks.

Signed-off-by: Chunguang Xu <brookxu@tencent.com>
This commit is contained in:
Chunguang Xu 2020-07-30 16:17:38 +08:00 committed by Chen Lei
parent f0c2827b13
commit 43c1da6894
1 changed files with 90 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
block/blk-wbt.c
View File

@ -113,7 +113,7 @@ static void rwb_wake_all(struct rq_wb *rwb)
for (i = 0; i < WBT_NUM_RWQ; i++) {
struct rq_wait *rqw = &rwb->rq_wait[i];
if (waitqueue_active(&rqw->wait))
if (wq_has_sleeper(&rqw->wait))
wake_up_all(&rqw->wait);
}
}
@ -153,11 +153,11 @@ void __wbt_done(struct rq_wb *rwb, enum wbt_flags wb_acct)
if (inflight && inflight >= limit)
return;
if (waitqueue_active(&rqw->wait)) {
if (wq_has_sleeper(&rqw->wait)) {
int diff = limit - inflight;
if (!inflight || diff >= rwb->wb_background / 2)
wake_up(&rqw->wait);
wake_up_all(&rqw->wait);
}
}
@ -480,6 +480,13 @@ static inline unsigned int get_limit(struct rq_wb *rwb, unsigned long rw)
{
unsigned int limit;
/*
* If we got disabled, just return UINT_MAX. This ensures that
* we'll properly inc a new IO, and dec+wakeup at the end.
*/
if (!rwb_enabled(rwb))
return UINT_MAX;
/*
* At this point we know it's a buffered write. If this is
* kswapd trying to free memory, or REQ_SYNC is set, set, then
@ -502,53 +509,92 @@ static inline unsigned int get_limit(struct rq_wb *rwb, unsigned long rw)
return limit;
}
struct wbt_wait_data {
struct wait_queue_entry wq;
struct task_struct *task;
struct rq_wb *rwb;
struct rq_wait *rqw;
unsigned long rw;
bool got_token;
};
static int wbt_wake_function(struct wait_queue_entry *curr, unsigned int mode,
int wake_flags, void *key)
{
struct wbt_wait_data *data = container_of(curr, struct wbt_wait_data,
wq);
/*
* If we fail to get a budget, return -1 to interrupt the wake up
* loop in __wake_up_common.
*/
if (!atomic_inc_below(&data->rqw->inflight, get_limit(data->rwb, data->rw)))
return -1;
data->got_token = true;
list_del_init(&curr->entry);
wake_up_process(data->task);
return 1;
}
/*
* Block if we will exceed our limit, or if we are currently waiting for
* the timer to kick off queuing again.
*/
static void __wbt_wait(struct rq_wb *rwb, unsigned long rw, spinlock_t *lock)
static void __wbt_wait(struct rq_wb *rwb, enum wbt_flags wb_acct,
unsigned long rw, spinlock_t *lock)
__releases(lock)
__acquires(lock)
{
struct rq_wait *rqw = get_rq_wait(rwb, current_is_kswapd());
DECLARE_WAITQUEUE(wait, current);
struct wbt_wait_data data = {
.wq = {
.func = wbt_wake_function,
.entry = LIST_HEAD_INIT(data.wq.entry),
},
.task = current,
.rwb = rwb,
.rqw = rqw,
.rw = rw,
.got_token = false
};
bool has_sleeper;
/*
* inc it here even if disabled, since we'll dec it at completion.
* this only happens if the task was sleeping in __wbt_wait(),
* and someone turned it off at the same time.
*/
if (!rwb_enabled(rwb)) {
atomic_inc(&rqw->inflight);
return;
}
if (!waitqueue_active(&rqw->wait) &&
has_sleeper = wq_has_sleeper(&rqw->wait);
if (!has_sleeper &&
atomic_inc_below(&rqw->inflight, get_limit(rwb, rw)))
return;
add_wait_queue_exclusive(&rqw->wait, &wait);
prepare_to_wait_exclusive(&rqw->wait, &data.wq, TASK_UNINTERRUPTIBLE);
do {
set_current_state(TASK_UNINTERRUPTIBLE);
if (data.got_token)
break;
if (!rwb_enabled(rwb)) {
atomic_inc(&rqw->inflight);
if (!has_sleeper &&
atomic_inc_below(&rqw->inflight, get_limit(rwb, rw))) {
finish_wait(&rqw->wait, &data.wq);
/*
* We raced with wbt_wake_function() getting a token,
* which means we now have two. Put our local token
* and wake anyone else potentially waiting for one.
*/
if (data.got_token)
__wbt_done(rwb, wb_acct);
break;
}
if (atomic_inc_below(&rqw->inflight, get_limit(rwb, rw)))
break;
if (lock) {
spin_unlock_irq(lock);
io_schedule();
spin_lock_irq(lock);
} else
io_schedule();
has_sleeper = false;
} while (1);
__set_current_state(TASK_RUNNING);
remove_wait_queue(&rqw->wait, &wait);
finish_wait(&rqw->wait, &data.wq);
}
static inline bool wbt_should_throttle(struct rq_wb *rwb, struct bio *bio)
@ -570,6 +616,22 @@ static inline bool wbt_should_throttle(struct rq_wb *rwb, struct bio *bio)
return true;
}
static enum wbt_flags bio_to_wbt_flags(struct rq_wb *rwb, struct bio *bio)
{
enum wbt_flags flags = 0;
if (!rwb_enabled(rwb))
return 0;
if (bio_op(bio) == REQ_OP_READ) {
flags = WBT_READ;
} else if (wbt_should_throttle(rwb, bio)) {
if (current_is_kswapd())
flags |= WBT_KSWAPD;
flags |= WBT_TRACKED;
}
return flags;
}
/*
* Returns true if the IO request should be accounted, false if not.
* May sleep, if we have exceeded the writeback limits. Caller can pass
@ -578,6 +640,7 @@ static inline bool wbt_should_throttle(struct rq_wb *rwb, struct bio *bio)
*/
enum wbt_flags wbt_wait(struct rq_wb *rwb, struct bio *bio, spinlock_t *lock)
{
enum wbt_flags flags;
unsigned int ret = 0;
if (!rwb_enabled(rwb))
@ -592,7 +655,8 @@ enum wbt_flags wbt_wait(struct rq_wb *rwb, struct bio *bio, spinlock_t *lock)
return ret;
}
__wbt_wait(rwb, bio->bi_opf, lock);
flags = bio_to_wbt_flags(rwb, bio);
__wbt_wait(rwb, flags, bio->bi_opf, lock);
if (!blk_stat_is_active(rwb->cb))
rwb_arm_timer(rwb);