From: Jan Kara <jack@suse.cz>

Subject: [PATCH] sbitmap: Avoid lockups when waker gets preempted

References: bsc#1209118

Patch-mainline: Never, sbitmap code got rewritten

When the process performing wakeup on the wait queue gets preempted and this

wait queue has the only waiter, we can loose many wakeups and in the end we

may end up with waiter in the wait queue without anybody able to wake him up.

The scenario can by like: We have waitqueue 0 active.

  CPU0                              CPU1

  __sbq_wake_up()

    decrements wait_cnt to 0

    wake_up_nr() -> wakes waiters

                                    queues IOs until all tags are used

                                    blk_mq_get_tag() finds no tag ->

                                      adds waiter to waitqueue 0

                                    all IOs complete - wakeups discarded because 

                                      waitqueue 0 still has wait_cnt == 0.

   set wait_cnt to 8

And waiter in waitqueue 0 sleeps indefinitely now because there's no IO to

complete to wake him up.

Fix the problem by waking up some waiters if we don't find any valid wait queue

to perform wake up on. This makes sure we don't loose all the wake ups.

Signed-off-by: Jan Kara <jack@suse.cz>

---

 lib/sbitmap.c |   22 ++++++++++++++++++----

 1 file changed, 18 insertions(+), 4 deletions(-)

--- a/lib/sbitmap.c

+++ b/lib/sbitmap.c

@@ -532,6 +532,7 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_min_shal

 static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)

 {

 	int i, wake_index;

+	struct sbq_wait_state *active_ws = NULL;

 	if (!atomic_read(&sbq->ws_active))

 		return NULL;

@@ -540,15 +541,28 @@ static struct sbq_wait_state *sbq_wake_p

 	for (i = 0; i < SBQ_WAIT_QUEUES; i++) {

 		struct sbq_wait_state *ws = &sbq->ws[wake_index];

-		if (waitqueue_active(&ws->wait) && atomic_read(&ws->wait_cnt) > 0) {

-			if (wake_index != atomic_read(&sbq->wake_index))

-				atomic_set(&sbq->wake_index, wake_index);

-			return ws;

+		if (waitqueue_active(&ws->wait)) {

+			if (atomic_read(&ws->wait_cnt) > 0) {

+				if (wake_index != atomic_read(&sbq->wake_index))

+					atomic_set(&sbq->wake_index, wake_index);

+				return ws;

+			}

+			active_ws = ws;

 		}

 		wake_index = sbq_index_inc(wake_index);

 	}

+	/*

+	 * There are active waitqueues but all are in the process of being

+	 * woken. Perform wakeup on some waitqueue to avoid loosing the wakeup.

+	 * This is actually important in case task performing wakeup gets

+	 * preempted and lots of other wakeup events happen before it gets

+	 * scheduled again.

+	 */

+	if (active_ws)

+		wake_up_nr(&active_ws->wait, READ_ONCE(sbq->wake_batch));

+

 	return NULL;

 }