zfs recv hangs if max recordsize is less than received recordsize

- Some optimizations for bqueue enqueue/dequeue.
- Added a fix to prevent deadlock when both bqueue_enqueue_impl()
and bqueue_dequeue() waits for signal to be triggered.

Reviewed-by: Alexander Motin <[email protected]>
Reviewed-by: Ryan Moeller <[email protected]>
Signed-off-by: Ameer Hamza <[email protected]>
Closes #13855

1 files changed, 13 insertions, 10 deletions

diff --git a/module/zfs/bqueue.c b/module/zfs/bqueue.c
index 22539efc4..ec5ce4388 100644
--- a/module/zfs/bqueue.c
+++ b/module/zfs/bqueue.c
@@ -42,8 +42,7 @@ obj2node(bqueue_t *q, void *data)
  * Return 0 on success, or -1 on failure.
  */
 int
-bqueue_init(bqueue_t *q, uint64_t fill_fraction, uint64_t size,
-    size_t node_offset)
+bqueue_init(bqueue_t *q, uint_t fill_fraction, size_t size, size_t node_offset)
 {
 	if (fill_fraction == 0) {
 		return (-1);
@@ -78,22 +77,26 @@ bqueue_destroy(bqueue_t *q)
 }
 
 static void
-bqueue_enqueue_impl(bqueue_t *q, void *data, uint64_t item_size,
-    boolean_t flush)
+bqueue_enqueue_impl(bqueue_t *q, void *data, size_t item_size, boolean_t flush)
 {
 	ASSERT3U(item_size, >, 0);
 	ASSERT3U(item_size, <=, q->bq_maxsize);
 	mutex_enter(&q->bq_lock);
 	obj2node(q, data)->bqn_size = item_size;
-	while (q->bq_size + item_size > q->bq_maxsize) {
+	while (q->bq_size && q->bq_size + item_size > q->bq_maxsize) {
+		/*
+		 * Wake up bqueue_dequeue() thread if already sleeping in order
+		 * to prevent the deadlock condition
+		 */
+		cv_signal(&q->bq_pop_cv);
 		cv_wait_sig(&q->bq_add_cv, &q->bq_lock);
 	}
 	q->bq_size += item_size;
 	list_insert_tail(&q->bq_list, data);
-	if (q->bq_size >= q->bq_maxsize / q->bq_fill_fraction)
-		cv_signal(&q->bq_pop_cv);
 	if (flush)
 		cv_broadcast(&q->bq_pop_cv);
+	else if (q->bq_size >= q->bq_maxsize / q->bq_fill_fraction)
+		cv_signal(&q->bq_pop_cv);
 	mutex_exit(&q->bq_lock);
 }
 
@@ -103,7 +106,7 @@ bqueue_enqueue_impl(bqueue_t *q, void *data, uint64_t item_size,
  * > 0.
  */
 void
-bqueue_enqueue(bqueue_t *q, void *data, uint64_t item_size)
+bqueue_enqueue(bqueue_t *q, void *data, size_t item_size)
 {
 	bqueue_enqueue_impl(q, data, item_size, B_FALSE);
 }
@@ -117,7 +120,7 @@ bqueue_enqueue(bqueue_t *q, void *data, uint64_t item_size)
  * destroy the condvar before the enqueuing thread is done.
  */
 void
-bqueue_enqueue_flush(bqueue_t *q, void *data, uint64_t item_size)
+bqueue_enqueue_flush(bqueue_t *q, void *data, size_t item_size)
 {
 	bqueue_enqueue_impl(q, data, item_size, B_TRUE);
 }
@@ -130,7 +133,7 @@ void *
 bqueue_dequeue(bqueue_t *q)
 {
 	void *ret = NULL;
-	uint64_t item_size;
+	size_t item_size;
 	mutex_enter(&q->bq_lock);
 	while (q->bq_size == 0) {
 		cv_wait_sig(&q->bq_pop_cv, &q->bq_lock);