Commit 6c0ca7ae authored by Omar Sandoval's avatar Omar Sandoval Committed by Jens Axboe

sbitmap: fix wakeup hang after sbq resize

When we resize a struct sbitmap_queue, we update the wakeup batch size,
but we don't update the wait count in the struct sbq_wait_states. If we
resized down from a size which could use a bigger batch size, these
counts could be too large and cause us to miss necessary wakeups. To fix
this, update the wait counts when we resize (ensuring some careful
memory ordering so that it's safe w.r.t. concurrent clears).

This also fixes a theoretical issue where two threads could end up
bumping the wait count up by the batch size, which could also
potentially lead to hangs.
Reported-by: default avatarMartin Raiber <martin@urbackup.org>
Fixes: e3a2b3f9 ("blk-mq: allow changing of queue depth through sysfs")
Fixes: 2971c35f ("blk-mq: bitmap tag: fix race on blk_mq_bitmap_tags::wake_cnt")
Signed-off-by: default avatarOmar Sandoval <osandov@fb.com>
Signed-off-by: default avatarJens Axboe <axboe@fb.com>
parent f66227de
...@@ -239,7 +239,19 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_init_node); ...@@ -239,7 +239,19 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_init_node);
void sbitmap_queue_resize(struct sbitmap_queue *sbq, unsigned int depth) void sbitmap_queue_resize(struct sbitmap_queue *sbq, unsigned int depth)
{ {
sbq->wake_batch = sbq_calc_wake_batch(depth); unsigned int wake_batch = sbq_calc_wake_batch(depth);
int i;
if (sbq->wake_batch != wake_batch) {
WRITE_ONCE(sbq->wake_batch, wake_batch);
/*
* Pairs with the memory barrier in sbq_wake_up() to ensure that
* the batch size is updated before the wait counts.
*/
smp_mb__before_atomic();
for (i = 0; i < SBQ_WAIT_QUEUES; i++)
atomic_set(&sbq->ws[i].wait_cnt, 1);
}
sbitmap_resize(&sbq->sb, depth); sbitmap_resize(&sbq->sb, depth);
} }
EXPORT_SYMBOL_GPL(sbitmap_queue_resize); EXPORT_SYMBOL_GPL(sbitmap_queue_resize);
...@@ -297,6 +309,7 @@ static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq) ...@@ -297,6 +309,7 @@ static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
static void sbq_wake_up(struct sbitmap_queue *sbq) static void sbq_wake_up(struct sbitmap_queue *sbq)
{ {
struct sbq_wait_state *ws; struct sbq_wait_state *ws;
unsigned int wake_batch;
int wait_cnt; int wait_cnt;
/* /*
...@@ -313,10 +326,22 @@ static void sbq_wake_up(struct sbitmap_queue *sbq) ...@@ -313,10 +326,22 @@ static void sbq_wake_up(struct sbitmap_queue *sbq)
return; return;
wait_cnt = atomic_dec_return(&ws->wait_cnt); wait_cnt = atomic_dec_return(&ws->wait_cnt);
if (unlikely(wait_cnt < 0)) if (wait_cnt <= 0) {
wait_cnt = atomic_inc_return(&ws->wait_cnt); wake_batch = READ_ONCE(sbq->wake_batch);
if (wait_cnt == 0) { /*
atomic_add(sbq->wake_batch, &ws->wait_cnt); * Pairs with the memory barrier in sbitmap_queue_resize() to
* ensure that we see the batch size update before the wait
* count is reset.
*/
smp_mb__before_atomic();
/*
* If there are concurrent callers to sbq_wake_up(), the last
* one to decrement the wait count below zero will bump it back
* up. If there is a concurrent resize, the count reset will
* either cause the cmpxchg to fail or overwrite after the
* cmpxchg.
*/
atomic_cmpxchg(&ws->wait_cnt, wait_cnt, wait_cnt + wake_batch);
sbq_index_atomic_inc(&sbq->wake_index); sbq_index_atomic_inc(&sbq->wake_index);
wake_up(&ws->wait); wake_up(&ws->wait);
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment