[PATCH RFC v3 0/3] bugfix for sbitmap

[PATCH RFC v3 0/3] bugfix for sbitmap

[Yu Kuai]

Changes in v3:
 - rename patch 2, and add some comments.
 - add patch 3, which fixes a new issue pointed out by Jan Kara.
Changes in v2:
 - split to spearate patches for different problem.
 - add fix tag

This patchset fixes problems in the wakeup path on concurrent scenarios,
the problems are found during code review, and it's not reporduced in
real test yet in my environment.

Keith reported a possible related io hung problem in nvme:

https://bugzilla.kernel.org/show_bug.cgi?id=215679

Previous versions:
v1: https://lore.kernel.org/all/20220617141125.3024491-1-yukuai3@huawei.com/
v2: https://lore.kernel.org/all/20220619080309.1630027-1-yukuai3@huawei.com/

Yu Kuai (3):
  sbitmap: fix that same waitqueue can be woken up continuously
  sbitmap: fix invalid wakeup on the wrong waitqueue
  sbitmap: fix that 'wait_cnt' can be decreased while waitqueue is empty

 lib/sbitmap.c | 70 +++++++++++++++++++++++++++++++--------------------
 1 file changed, 43 insertions(+), 27 deletions(-)

-- 
2.31.1

[PATCH RFC v3 1/3] sbitmap: fix that same waitqueue can be woken up continuously

[Yu Kuai]

From: Yu Kuai <yukuai3@huawei.com>

__sbq_wake_up		__sbq_wake_up
 sbq_wake_ptr -> assume	0
			 sbq_wake_ptr -> 0
 atomic_dec_return
			atomic_dec_return
 atomic_cmpxchg -> succeed
			 atomic_cmpxchg -> failed
			  return true

			__sbq_wake_up
			 sbq_wake_ptr
			  atomic_read(&sbq->wake_index) -> still 0
 sbq_index_atomic_inc -> inc to 1
			  if (waitqueue_active(&ws->wait))
			   if (wake_index != atomic_read(&sbq->wake_index))
			    atomic_set -> reset from 1 to 0
 wake_up_nr -> wake up first waitqueue
			    // continue to wake up in first waitqueue

Fix the problem by using atomic_cmpxchg() instead of atomic_set()
to update 'wake_index'.

Fixes: 417232880c8a ("sbitmap: Replace cmpxchg with xchg")
Signed-off-by: Yu Kuai <yukuai3@huawei.com>
---
 lib/sbitmap.c | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/lib/sbitmap.c b/lib/sbitmap.c
index 29eb0484215a..b46fce1beb3a 100644
--- a/lib/sbitmap.c
+++ b/lib/sbitmap.c
@@ -579,19 +579,24 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_min_shallow_depth);
 
 static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
 {
-	int i, wake_index;
+	int i, wake_index, old_wake_index;
 
+again:
 	if (!atomic_read(&sbq->ws_active))
 		return NULL;
 
-	wake_index = atomic_read(&sbq->wake_index);
+	old_wake_index = wake_index = atomic_read(&sbq->wake_index);
 	for (i = 0; i < SBQ_WAIT_QUEUES; i++) {
 		struct sbq_wait_state *ws = &sbq->ws[wake_index];
 
 		if (waitqueue_active(&ws->wait)) {
-			if (wake_index != atomic_read(&sbq->wake_index))
-				atomic_set(&sbq->wake_index, wake_index);
-			return ws;
+			if (wake_index == old_wake_index)
+				return ws;
+
+			if (atomic_cmpxchg(&sbq->wake_index, old_wake_index,
+					   wake_index) == old_wake_index)
+				return ws;
+			goto again;
 		}
 
 		wake_index = sbq_index_inc(wake_index);
-- 
2.31.1

Re: [PATCH RFC v3 1/3] sbitmap: fix that same waitqueue can be woken up continuously

[Jan Kara]

On Sun 10-07-22 12:21:58, Yu Kuai wrote:
> From: Yu Kuai <yukuai3@huawei.com>
> 
> __sbq_wake_up		__sbq_wake_up
>  sbq_wake_ptr -> assume	0
> 			 sbq_wake_ptr -> 0
>  atomic_dec_return
> 			atomic_dec_return
>  atomic_cmpxchg -> succeed
> 			 atomic_cmpxchg -> failed
> 			  return true
> 
> 			__sbq_wake_up
> 			 sbq_wake_ptr
> 			  atomic_read(&sbq->wake_index) -> still 0
>  sbq_index_atomic_inc -> inc to 1
> 			  if (waitqueue_active(&ws->wait))
> 			   if (wake_index != atomic_read(&sbq->wake_index))
> 			    atomic_set -> reset from 1 to 0
>  wake_up_nr -> wake up first waitqueue
> 			    // continue to wake up in first waitqueue
> 
> Fix the problem by using atomic_cmpxchg() instead of atomic_set()
> to update 'wake_index'.
> 
> Fixes: 417232880c8a ("sbitmap: Replace cmpxchg with xchg")
> Signed-off-by: Yu Kuai <yukuai3@huawei.com>

I don't think this patch is really needed after the following patches.  As
I see it, wake_index is just a performance optimization (plus a fairness
improvement) but in principle the code in sbq_wake_ptr() is always prone to
races as the waitqueue it returns needn't have any waiters by the time we
return. So for correctness the check-and-retry loop needs to happen at
higher level than inside sbq_wake_ptr() and occasional wrong setting of
wake_index will result only in a bit of unfairness and more scanning
looking for suitable waitqueue but I don't think that really justifies the
cost of atomic operations in cmpxchg loop...

								Honza
> ---
>  lib/sbitmap.c | 15 ++++++++++-----
>  1 file changed, 10 insertions(+), 5 deletions(-)
> 
> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
> index 29eb0484215a..b46fce1beb3a 100644
> --- a/lib/sbitmap.c
> +++ b/lib/sbitmap.c
> @@ -579,19 +579,24 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_min_shallow_depth);
>  
>  static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
>  {
> -	int i, wake_index;
> +	int i, wake_index, old_wake_index;
>  
> +again:
>  	if (!atomic_read(&sbq->ws_active))
>  		return NULL;
>  
> -	wake_index = atomic_read(&sbq->wake_index);
> +	old_wake_index = wake_index = atomic_read(&sbq->wake_index);
>  	for (i = 0; i < SBQ_WAIT_QUEUES; i++) {
>  		struct sbq_wait_state *ws = &sbq->ws[wake_index];
>  
>  		if (waitqueue_active(&ws->wait)) {
> -			if (wake_index != atomic_read(&sbq->wake_index))
> -				atomic_set(&sbq->wake_index, wake_index);
> -			return ws;
> +			if (wake_index == old_wake_index)
> +				return ws;
> +
> +			if (atomic_cmpxchg(&sbq->wake_index, old_wake_index,
> +					   wake_index) == old_wake_index)
> +				return ws;
> +			goto again;
>  		}
>  
>  		wake_index = sbq_index_inc(wake_index);
> -- 
> 2.31.1
> 
-- 
Jan Kara <jack@suse.com>
SUSE Labs, CR

Re: [PATCH RFC v3 1/3] sbitmap: fix that same waitqueue can be woken up continuously

[Yu Kuai]

Hi!

在 2022/07/11 22:20, Jan Kara 写道:
> On Sun 10-07-22 12:21:58, Yu Kuai wrote:
>> From: Yu Kuai <yukuai3@huawei.com>
>>
>> __sbq_wake_up		__sbq_wake_up
>>   sbq_wake_ptr -> assume	0
>> 			 sbq_wake_ptr -> 0
>>   atomic_dec_return
>> 			atomic_dec_return
>>   atomic_cmpxchg -> succeed
>> 			 atomic_cmpxchg -> failed
>> 			  return true
>>
>> 			__sbq_wake_up
>> 			 sbq_wake_ptr
>> 			  atomic_read(&sbq->wake_index) -> still 0
>>   sbq_index_atomic_inc -> inc to 1
>> 			  if (waitqueue_active(&ws->wait))
>> 			   if (wake_index != atomic_read(&sbq->wake_index))
>> 			    atomic_set -> reset from 1 to 0
>>   wake_up_nr -> wake up first waitqueue
>> 			    // continue to wake up in first waitqueue
>>
>> Fix the problem by using atomic_cmpxchg() instead of atomic_set()
>> to update 'wake_index'.
>>
>> Fixes: 417232880c8a ("sbitmap: Replace cmpxchg with xchg")
>> Signed-off-by: Yu Kuai <yukuai3@huawei.com>
> 
> I don't think this patch is really needed after the following patches.  As
> I see it, wake_index is just a performance optimization (plus a fairness
> improvement) but in principle the code in sbq_wake_ptr() is always prone to
> races as the waitqueue it returns needn't have any waiters by the time we
> return. So for correctness the check-and-retry loop needs to happen at
> higher level than inside sbq_wake_ptr() and occasional wrong setting of
> wake_index will result only in a bit of unfairness and more scanning
> looking for suitable waitqueue but I don't think that really justifies the
> cost of atomic operations in cmpxchg loop...

It's right this patch just improve fairness. However, in hevyload tests
I found that the 'wrong setting of wake_index' can happen frequently,
for consequence, some waitqueue can be empty while some waitqueue have
a lot of waiters.

There shoud be lots of work to fix unfairness throughly, I can remove
this patch for now.

Thanks,
Kuai
> 
> 								Honza
>> ---
>>   lib/sbitmap.c | 15 ++++++++++-----
>>   1 file changed, 10 insertions(+), 5 deletions(-)
>>
>> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
>> index 29eb0484215a..b46fce1beb3a 100644
>> --- a/lib/sbitmap.c
>> +++ b/lib/sbitmap.c
>> @@ -579,19 +579,24 @@ EXPORT_SYMBOL_GPL(sbitmap_queue_min_shallow_depth);
>>   
>>   static struct sbq_wait_state *sbq_wake_ptr(struct sbitmap_queue *sbq)
>>   {
>> -	int i, wake_index;
>> +	int i, wake_index, old_wake_index;
>>   
>> +again:
>>   	if (!atomic_read(&sbq->ws_active))
>>   		return NULL;
>>   
>> -	wake_index = atomic_read(&sbq->wake_index);
>> +	old_wake_index = wake_index = atomic_read(&sbq->wake_index);
>>   	for (i = 0; i < SBQ_WAIT_QUEUES; i++) {
>>   		struct sbq_wait_state *ws = &sbq->ws[wake_index];
>>   
>>   		if (waitqueue_active(&ws->wait)) {
>> -			if (wake_index != atomic_read(&sbq->wake_index))
>> -				atomic_set(&sbq->wake_index, wake_index);
>> -			return ws;
>> +			if (wake_index == old_wake_index)
>> +				return ws;
>> +
>> +			if (atomic_cmpxchg(&sbq->wake_index, old_wake_index,
>> +					   wake_index) == old_wake_index)
>> +				return ws;
>> +			goto again;
>>   		}
>>   
>>   		wake_index = sbq_index_inc(wake_index);
>> -- 
>> 2.31.1
>>

[PATCH RFC v3 2/3] sbitmap: fix invalid wakeup on the wrong waitqueue

[Yu Kuai]

From: Yu Kuai <yukuai3@huawei.com>

For example, 2 * wake_batch tags are put, while only wake_batch threads
are woken:

__sbq_wake_up
 atomic_cmpxchg -> reset wait_cnt
			__sbq_wake_up -> decrease wait_cnt
			...
			__sbq_wake_up -> wait_cnt is decreased to 0 again
			 atomic_cmpxchg
			 sbq_index_atomic_inc -> increase wake_index
			 wake_up_nr -> wake up and waitqueue might be empty
 sbq_index_atomic_inc -> increase again, one waitqueue is skipped
 wake_up_nr -> invalid wake up because old wakequeue might be empty

To fix the problem, increasing 'wake_index' before resetting 'wait_cnt'.

Fixes: 88459642cba4 ("blk-mq: abstract tag allocation out into sbitmap library")
Signed-off-by: Yu Kuai <yukuai3@huawei.com>
---
 lib/sbitmap.c | 45 +++++++++++++++++++++++----------------------
 1 file changed, 23 insertions(+), 22 deletions(-)

diff --git a/lib/sbitmap.c b/lib/sbitmap.c
index b46fce1beb3a..57095dd88a33 100644
--- a/lib/sbitmap.c
+++ b/lib/sbitmap.c
@@ -616,32 +616,33 @@ static bool __sbq_wake_up(struct sbitmap_queue *sbq)
 		return false;
 
 	wait_cnt = atomic_dec_return(&ws->wait_cnt);
-	if (wait_cnt <= 0) {
-		int ret;
+	if (wait_cnt > 0)
+		return false;
 
-		wake_batch = READ_ONCE(sbq->wake_batch);
+	/*
+	 * For concurrent callers of this, callers should call this function
+	 * again to wakeup a new batch on a different 'ws'.
+	 */
+	if (wait_cnt < 0)
+		return true;
 
-		/*
-		 * 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();
+	wake_batch = READ_ONCE(sbq->wake_batch);
 
-		/*
-		 * For concurrent callers of this, the one that failed the
-		 * atomic_cmpxhcg() race should call this function again
-		 * to wakeup a new batch on a different 'ws'.
-		 */
-		ret = atomic_cmpxchg(&ws->wait_cnt, wait_cnt, wake_batch);
-		if (ret == wait_cnt) {
-			sbq_index_atomic_inc(&sbq->wake_index);
-			wake_up_nr(&ws->wait, wake_batch);
-			return false;
-		}
+	/*
+	 * 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();
 
-		return true;
-	}
+	/*
+	 * Increase wake_index before updating wait_cnt, otherwise concurrent
+	 * callers can see valid wait_cnt in old waitqueue, which can cause
+	 * invalid wakeup on the old waitqueue.
+	 */
+	sbq_index_atomic_inc(&sbq->wake_index);
+	atomic_set(&ws->wait_cnt, wake_batch);
+	wake_up_nr(&ws->wait, wake_batch);
 
 	return false;
 }
-- 
2.31.1

Re: [PATCH RFC v3 2/3] sbitmap: fix invalid wakeup on the wrong waitqueue

[Jan Kara]

On Sun 10-07-22 12:21:59, Yu Kuai wrote:
> From: Yu Kuai <yukuai3@huawei.com>
> 
> For example, 2 * wake_batch tags are put, while only wake_batch threads
> are woken:
> 
> __sbq_wake_up
>  atomic_cmpxchg -> reset wait_cnt
> 			__sbq_wake_up -> decrease wait_cnt
> 			...
> 			__sbq_wake_up -> wait_cnt is decreased to 0 again
> 			 atomic_cmpxchg
> 			 sbq_index_atomic_inc -> increase wake_index
> 			 wake_up_nr -> wake up and waitqueue might be empty
>  sbq_index_atomic_inc -> increase again, one waitqueue is skipped
>  wake_up_nr -> invalid wake up because old wakequeue might be empty
> 
> To fix the problem, increasing 'wake_index' before resetting 'wait_cnt'.
> 
> Fixes: 88459642cba4 ("blk-mq: abstract tag allocation out into sbitmap library")
> Signed-off-by: Yu Kuai <yukuai3@huawei.com>

This patch and the following one look sane to me but please merge them to a
single patch. They fix the same race of two concurrent wakers just with a
slightly different timing so there isn't a point in having two patches for
this (in particular changes in this patch are difficult to reason about
when we know the result is still buggy).

								Honza

> ---
>  lib/sbitmap.c | 45 +++++++++++++++++++++++----------------------
>  1 file changed, 23 insertions(+), 22 deletions(-)
> 
> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
> index b46fce1beb3a..57095dd88a33 100644
> --- a/lib/sbitmap.c
> +++ b/lib/sbitmap.c
> @@ -616,32 +616,33 @@ static bool __sbq_wake_up(struct sbitmap_queue *sbq)
>  		return false;
>  
>  	wait_cnt = atomic_dec_return(&ws->wait_cnt);
> -	if (wait_cnt <= 0) {
> -		int ret;
> +	if (wait_cnt > 0)
> +		return false;
>  
> -		wake_batch = READ_ONCE(sbq->wake_batch);
> +	/*
> +	 * For concurrent callers of this, callers should call this function
> +	 * again to wakeup a new batch on a different 'ws'.
> +	 */
> +	if (wait_cnt < 0)
> +		return true;
>  
> -		/*
> -		 * 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();
> +	wake_batch = READ_ONCE(sbq->wake_batch);
>  
> -		/*
> -		 * For concurrent callers of this, the one that failed the
> -		 * atomic_cmpxhcg() race should call this function again
> -		 * to wakeup a new batch on a different 'ws'.
> -		 */
> -		ret = atomic_cmpxchg(&ws->wait_cnt, wait_cnt, wake_batch);
> -		if (ret == wait_cnt) {
> -			sbq_index_atomic_inc(&sbq->wake_index);
> -			wake_up_nr(&ws->wait, wake_batch);
> -			return false;
> -		}
> +	/*
> +	 * 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();
>  
> -		return true;
> -	}
> +	/*
> +	 * Increase wake_index before updating wait_cnt, otherwise concurrent
> +	 * callers can see valid wait_cnt in old waitqueue, which can cause
> +	 * invalid wakeup on the old waitqueue.
> +	 */
> +	sbq_index_atomic_inc(&sbq->wake_index);
> +	atomic_set(&ws->wait_cnt, wake_batch);
> +	wake_up_nr(&ws->wait, wake_batch);
>  
>  	return false;
>  }
> -- 
> 2.31.1
> 
-- 
Jan Kara <jack@suse.com>
SUSE Labs, CR

Re: [PATCH RFC v3 2/3] sbitmap: fix invalid wakeup on the wrong waitqueue

[Yu Kuai]

Hi!

在 2022/07/11 22:26, Jan Kara 写道:
> On Sun 10-07-22 12:21:59, Yu Kuai wrote:
>> From: Yu Kuai <yukuai3@huawei.com>
>>
>> For example, 2 * wake_batch tags are put, while only wake_batch threads
>> are woken:
>>
>> __sbq_wake_up
>>   atomic_cmpxchg -> reset wait_cnt
>> 			__sbq_wake_up -> decrease wait_cnt
>> 			...
>> 			__sbq_wake_up -> wait_cnt is decreased to 0 again
>> 			 atomic_cmpxchg
>> 			 sbq_index_atomic_inc -> increase wake_index
>> 			 wake_up_nr -> wake up and waitqueue might be empty
>>   sbq_index_atomic_inc -> increase again, one waitqueue is skipped
>>   wake_up_nr -> invalid wake up because old wakequeue might be empty
>>
>> To fix the problem, increasing 'wake_index' before resetting 'wait_cnt'.
>>
>> Fixes: 88459642cba4 ("blk-mq: abstract tag allocation out into sbitmap library")
>> Signed-off-by: Yu Kuai <yukuai3@huawei.com>
> 
> This patch and the following one look sane to me but please merge them to a
> single patch. They fix the same race of two concurrent wakers just with a
> slightly different timing so there isn't a point in having two patches for
> this (in particular changes in this patch are difficult to reason about
> when we know the result is still buggy).

Ok, I'll merge them.

Thanks,
Kuai
> 
> 								Honza
> 
>> ---
>>   lib/sbitmap.c | 45 +++++++++++++++++++++++----------------------
>>   1 file changed, 23 insertions(+), 22 deletions(-)
>>
>> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
>> index b46fce1beb3a..57095dd88a33 100644
>> --- a/lib/sbitmap.c
>> +++ b/lib/sbitmap.c
>> @@ -616,32 +616,33 @@ static bool __sbq_wake_up(struct sbitmap_queue *sbq)
>>   		return false;
>>   
>>   	wait_cnt = atomic_dec_return(&ws->wait_cnt);
>> -	if (wait_cnt <= 0) {
>> -		int ret;
>> +	if (wait_cnt > 0)
>> +		return false;
>>   
>> -		wake_batch = READ_ONCE(sbq->wake_batch);
>> +	/*
>> +	 * For concurrent callers of this, callers should call this function
>> +	 * again to wakeup a new batch on a different 'ws'.
>> +	 */
>> +	if (wait_cnt < 0)
>> +		return true;
>>   
>> -		/*
>> -		 * 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();
>> +	wake_batch = READ_ONCE(sbq->wake_batch);
>>   
>> -		/*
>> -		 * For concurrent callers of this, the one that failed the
>> -		 * atomic_cmpxhcg() race should call this function again
>> -		 * to wakeup a new batch on a different 'ws'.
>> -		 */
>> -		ret = atomic_cmpxchg(&ws->wait_cnt, wait_cnt, wake_batch);
>> -		if (ret == wait_cnt) {
>> -			sbq_index_atomic_inc(&sbq->wake_index);
>> -			wake_up_nr(&ws->wait, wake_batch);
>> -			return false;
>> -		}
>> +	/*
>> +	 * 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();
>>   
>> -		return true;
>> -	}
>> +	/*
>> +	 * Increase wake_index before updating wait_cnt, otherwise concurrent
>> +	 * callers can see valid wait_cnt in old waitqueue, which can cause
>> +	 * invalid wakeup on the old waitqueue.
>> +	 */
>> +	sbq_index_atomic_inc(&sbq->wake_index);
>> +	atomic_set(&ws->wait_cnt, wake_batch);
>> +	wake_up_nr(&ws->wait, wake_batch);
>>   
>>   	return false;
>>   }
>> -- 
>> 2.31.1
>>

[PATCH RFC v3 3/3] sbitmap: fix that 'wait_cnt' can be decreased while waitqueue is empty

[Yu Kuai]

From: Yu Kuai <yukuai3@huawei.com>

As pointed out by Jan Kara, following race is still possible:

CPU1				CPU2
__sbq_wake_up			 __sbq_wake_up
 sbq_wake_ptr()			 sbq_wake_ptr() -> the same
 wait_cnt = atomic_dec_return()
 /* decreased to 0 */
 sbq_index_atomic_inc()
 /* move to next waitqueue */
 atomic_set()
 /* reset wait_cnt */
 wake_up_nr()
 /* wake up on the old waitqueue */
				 wait_cnt = atomic_dec_return()
				 /*
				  * decrease wait_cnt in the old
				  * waitqueue, while it can be
				  * empty.
				  */

Fix the problem by waking up before updating 'wake_index' and
'wait_cnt'.

With this patch, noted that 'wait_cnt' is still decreased in the old
empty waitqueue, however, the wakeup is redirected to a active waitqueue,
and the extra decrement on the old empty waitqueue is not handled.

Fixes: 88459642cba4 ("blk-mq: abstract tag allocation out into sbitmap library")
Signed-off-by: Yu Kuai <yukuai3@huawei.com>
---
 lib/sbitmap.c | 20 +++++++++++++++-----
 1 file changed, 15 insertions(+), 5 deletions(-)

diff --git a/lib/sbitmap.c b/lib/sbitmap.c
index 57095dd88a33..55826ebbe7db 100644
--- a/lib/sbitmap.c
+++ b/lib/sbitmap.c
@@ -616,22 +616,33 @@ static bool __sbq_wake_up(struct sbitmap_queue *sbq)
 		return false;
 
 	wait_cnt = atomic_dec_return(&ws->wait_cnt);
-	if (wait_cnt > 0)
-		return false;
-
 	/*
 	 * For concurrent callers of this, callers should call this function
 	 * again to wakeup a new batch on a different 'ws'.
 	 */
-	if (wait_cnt < 0)
+	if (wait_cnt < 0 || !waitqueue_active(&ws->wait))
 		return true;
 
+	if (wait_cnt > 0)
+		return false;
+
 	wake_batch = READ_ONCE(sbq->wake_batch);
 
+	/*
+	 * Wake up first in case that concurrent callers decrease wait_cnt
+	 * while waitqueue is empty.
+	 */
+	wake_up_nr(&ws->wait, wake_batch);
+
 	/*
 	 * Pairs with the memory barrier in sbitmap_queue_resize() to
 	 * ensure that we see the batch size update before the wait
 	 * count is reset.
+	 *
+	 * Also pairs with the implicit barrier between becrementing wait_cnt
+	 * and checking for waitqueue_active() to make sure waitqueue_active()
+	 * sees result of the wakeup if atomic_dec_return() has seen the result
+	 * of atomic_set().
 	 */
 	smp_mb__before_atomic();
 
@@ -642,7 +653,6 @@ static bool __sbq_wake_up(struct sbitmap_queue *sbq)
 	 */
 	sbq_index_atomic_inc(&sbq->wake_index);
 	atomic_set(&ws->wait_cnt, wake_batch);
-	wake_up_nr(&ws->wait, wake_batch);
 
 	return false;
 }
-- 
2.31.1