[PATCH 17/19] nvmet-rdma: don't run the wait list recursively

[Keith Busch]

From: Chris Mason <mason at kernel.org>

nvmet_rdma_release_rsp() returns the send queue slots of a response and
then runs the wait list, passing each waiting command to
nvmet_rdma_execute_command(). Many commands complete inside their
execute callback, and if the response can't be posted,
nvmet_rdma_queue_response() calls nvmet_rdma_release_rsp() right there,
which starts another run of the wait list inside the first:

    nvmet_rdma_process_wr_wait_list()
      nvmet_rdma_execute_command()
        nvmet_execute_admin_connect()
          nvmet_req_complete()
            nvmet_rdma_queue_response()
              ib_post_send() fails
              nvmet_rdma_release_rsp()
                nvmet_rdma_process_wr_wait_list()

Each waiting command adds a level while posts keep failing, and the host
decides how many commands wait: it sizes the receive queue, can fill
the send queue with commands that nvmet_req_init() rejects without
taking a slot, and over siw can make posts fail by holding back an
RDMA READ. 400 waiting Connects overflow the stack of the CQ worker.

Before the loop executes a command, point the response's new
wait_released at a flag on the loop's stack. A release that finds it
set means the loop hasn't returned from that command yet: it returns
the slots and releases the response, but sets the flag and leaves the
list to the loop instead of running it. The flag tells the loop not to
touch the response again.

Fixes: 8f000cac6e7a ("nvmet-rdma: add a NVMe over Fabrics RDMA target driver")
Cc: Christoph Hellwig <hch at lst.de>
Cc: Sagi Grimberg <sagi at grimberg.me>
Cc: Chaitanya Kulkarni <kch at nvidia.com>
Signed-off-by: Chris Mason <mason at kernel.org>
Signed-off-by: Jens Axboe <axboe at kernel.dk>
Signed-off-by: Keith Busch <kbusch at kernel.org>
---
 drivers/nvme/target/rdma.c | 28 +++++++++++++++++++++++++---
 1 file changed, 25 insertions(+), 3 deletions(-)

diff --git a/drivers/nvme/target/rdma.c b/drivers/nvme/target/rdma.c
index fced200678853..1e434cc451c28 100644
--- a/drivers/nvme/target/rdma.c
+++ b/drivers/nvme/target/rdma.c
@@ -77,6 +77,8 @@ struct nvmet_rdma_rsp {
 	u32			invalidate_rkey;
 
 	struct list_head	wait_list;
+	/* set under rsp_wr_wait_lock while the wait list's loop executes us */
+	bool			*wait_released;
 	int			tag;
 };
 
@@ -511,23 +513,42 @@ static int nvmet_rdma_post_recv(struct nvmet_rdma_device *ndev,
 	return ret;
 }
 
-static void nvmet_rdma_process_wr_wait_list(struct nvmet_rdma_queue *queue)
+static void nvmet_rdma_process_wr_wait_list(struct nvmet_rdma_queue *queue,
+					    struct nvmet_rdma_rsp *done)
 {
 	unsigned long flags;
 
 	spin_lock_irqsave(&queue->rsp_wr_wait_lock, flags);
+	if (done->wait_released) {
+		/*
+		 * The loop below has not come back from this command yet. It is
+		 * further up this stack if the command completed in place and
+		 * its response could not be posted. Otherwise it runs in
+		 * another context. It goes on with the list, so leave the list
+		 * to it.
+		 */
+		*done->wait_released = true;
+		WRITE_ONCE(done->wait_released, NULL);
+		spin_unlock_irqrestore(&queue->rsp_wr_wait_lock, flags);
+		return;
+	}
 	while (!list_empty(&queue->rsp_wr_wait_list)) {
 		struct nvmet_rdma_rsp *rsp;
+		bool released = false;
 		bool ret;
 
 		rsp = list_entry(queue->rsp_wr_wait_list.next,
 				struct nvmet_rdma_rsp, wait_list);
 		list_del(&rsp->wait_list);
+		rsp->wait_released = &released;
 
 		spin_unlock_irqrestore(&queue->rsp_wr_wait_lock, flags);
 		ret = nvmet_rdma_execute_command(rsp);
 		spin_lock_irqsave(&queue->rsp_wr_wait_lock, flags);
 
+		/* A released response can be freed, or in use again. */
+		if (!released)
+			WRITE_ONCE(rsp->wait_released, NULL);
 		if (!ret) {
 			list_add(&rsp->wait_list, &queue->rsp_wr_wait_list);
 			break;
@@ -680,8 +701,9 @@ static void nvmet_rdma_release_rsp(struct nvmet_rdma_rsp *rsp)
 
 	nvmet_rdma_free_rsp_resources(rsp);
 
-	if (unlikely(!list_empty_careful(&queue->rsp_wr_wait_list)))
-		nvmet_rdma_process_wr_wait_list(queue);
+	if (unlikely(READ_ONCE(rsp->wait_released) ||
+		     !list_empty_careful(&queue->rsp_wr_wait_list)))
+		nvmet_rdma_process_wr_wait_list(queue, rsp);
 
 	nvmet_rdma_put_rsp(rsp);
 }
-- 
2.52.0