[PATCHSET v2 0/9] Batched completions

[PATCHSET v2 0/9] Batched completions

[Jens Axboe]

Hi,

We now do decent batching of allocations for submit, but we still
complete requests individually. This costs a lot of CPU cycles.

This patchset adds support for collecting requests for completion,
and then completing them as a batch. This includes things like freeing
a batch of tags.

A few rough edges in this patchset, but sending it out for comments
so I can whip it into an upstreamable thing. As per patch 8, the
wins here are pretty massive - 8-10% improvement in the peak IOPS.

Changes since v1:
- Rename ib -> iob for the batch
- Add rq_list helpers
- Change array -> tags_array naming
- Drop nvme disposition patch
- Rework the batched completion on the nvme side to be a lot cleaner
- Add DEFINE_IO_BATCH helper
- Move batch into blkdev.h instead of in fs.h

-- 
Jens Axboe

[PATCH 1/9] block: define io_batch structure

[Jens Axboe]

This adds the io_batch structure, and a helper for defining one on the
stack. It's meant to be used for collecting requests for completion,
with a completion handler defined to be called at the end.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 include/linux/blkdev.h | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
index 2a8689e949b4..b39b19dbe7b6 100644
--- a/include/linux/blkdev.h
+++ b/include/linux/blkdev.h
@@ -1298,4 +1298,14 @@ int fsync_bdev(struct block_device *bdev);
 int freeze_bdev(struct block_device *bdev);
 int thaw_bdev(struct block_device *bdev);
 
+struct io_batch {
+	struct request *req_list;
+	void (*complete)(struct io_batch *);
+};
+
+#define DEFINE_IO_BATCH(name)			\
+	struct io_batch name = {		\
+		.req_list = NULL		\
+	}
+
 #endif /* _LINUX_BLKDEV_H */
-- 
2.33.0

Re: [PATCH 1/9] block: define io_batch structure

[Christoph Hellwig]

On Wed, Oct 13, 2021 at 10:54:08AM -0600, Jens Axboe wrote:
> This adds the io_batch structure, and a helper for defining one on the
> stack. It's meant to be used for collecting requests for completion,
> with a completion handler defined to be called at the end.

Isn't the name a little misleading given that it is all about
completions?

Also I wonder if this should be merged with the next patch as that's
sortof a logical unit.

Re: [PATCH 1/9] block: define io_batch structure

[Jens Axboe]

On 10/13/21 11:45 PM, Christoph Hellwig wrote:
> On Wed, Oct 13, 2021 at 10:54:08AM -0600, Jens Axboe wrote:
>> This adds the io_batch structure, and a helper for defining one on the
>> stack. It's meant to be used for collecting requests for completion,
>> with a completion handler defined to be called at the end.
> 
> Isn't the name a little misleading given that it is all about
> completions?

It is for completions, but I'd rather just keep it short for now.

> 
> Also I wonder if this should be merged with the next patch as that's
> sortof a logical unit.

Actually was like that before, I can fold them again.

-- 
Jens Axboe

Re: [PATCH 1/9] block: define io_batch structure

[Christoph Hellwig]

On Thu, Oct 14, 2021 at 09:50:10AM -0600, Jens Axboe wrote:
> On 10/13/21 11:45 PM, Christoph Hellwig wrote:
> > On Wed, Oct 13, 2021 at 10:54:08AM -0600, Jens Axboe wrote:
> >> This adds the io_batch structure, and a helper for defining one on the
> >> stack. It's meant to be used for collecting requests for completion,
> >> with a completion handler defined to be called at the end.
> > 
> > Isn't the name a little misleading given that it is all about
> > completions?
> 
> It is for completions, but I'd rather just keep it short for now.

I'd go for something like io_comp_batch or something like that at least.

> > Also I wonder if this should be merged with the next patch as that's
> > sortof a logical unit.
> 
> Actually was like that before, I can fold them again.

I'm fine either way.  Merged just seems a little more logical.

Re: [PATCH 1/9] block: define io_batch structure

[Jens Axboe]

On 10/14/21 10:06 AM, Christoph Hellwig wrote:
> On Thu, Oct 14, 2021 at 09:50:10AM -0600, Jens Axboe wrote:
>> On 10/13/21 11:45 PM, Christoph Hellwig wrote:
>>> On Wed, Oct 13, 2021 at 10:54:08AM -0600, Jens Axboe wrote:
>>>> This adds the io_batch structure, and a helper for defining one on the
>>>> stack. It's meant to be used for collecting requests for completion,
>>>> with a completion handler defined to be called at the end.
>>>
>>> Isn't the name a little misleading given that it is all about
>>> completions?
>>
>> It is for completions, but I'd rather just keep it short for now.
> 
> I'd go for something like io_comp_batch or something like that at least.

OK, made it io_comp_batch.

>>> Also I wonder if this should be merged with the next patch as that's
>>> sortof a logical unit.
>>
>> Actually was like that before, I can fold them again.
> 
> I'm fine either way.  Merged just seems a little more logical.

I just folded it.

-- 
Jens Axboe

[PATCH 2/9] block: add a struct io_batch argument to fops->iopoll()

[Jens Axboe]

struct io_batch contains a list head and a completion handler, which will
allow completions to more effciently completed batches of IO.

For now, no functional changes in this patch, we just add the argument to
the file_operations iopoll handler

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 block/blk-core.c              | 9 +++++----
 block/blk-exec.c              | 2 +-
 block/blk-mq.c                | 9 +++++----
 block/blk-mq.h                | 3 ++-
 block/fops.c                  | 4 ++--
 drivers/block/null_blk/main.c | 2 +-
 drivers/block/rnbd/rnbd-clt.c | 2 +-
 drivers/nvme/host/pci.c       | 4 ++--
 drivers/nvme/host/rdma.c      | 2 +-
 drivers/nvme/host/tcp.c       | 2 +-
 drivers/scsi/scsi_lib.c       | 2 +-
 fs/io_uring.c                 | 2 +-
 fs/iomap/direct-io.c          | 2 +-
 include/linux/blk-mq.h        | 2 +-
 include/linux/blkdev.h        | 5 +++--
 include/linux/fs.h            | 3 ++-
 mm/page_io.c                  | 2 +-
 17 files changed, 31 insertions(+), 26 deletions(-)

diff --git a/block/blk-core.c b/block/blk-core.c
index b199579c5f1f..67ec3f06c559 100644
--- a/block/blk-core.c
+++ b/block/blk-core.c
@@ -1002,7 +1002,7 @@ EXPORT_SYMBOL(submit_bio);
  * Note: the caller must either be the context that submitted @bio, or
  * be in a RCU critical section to prevent freeing of @bio.
  */
-int bio_poll(struct bio *bio, unsigned int flags)
+int bio_poll(struct bio *bio, struct io_batch *iob, unsigned int flags)
 {
 	struct request_queue *q = bio->bi_bdev->bd_disk->queue;
 	blk_qc_t cookie = READ_ONCE(bio->bi_cookie);
@@ -1020,7 +1020,7 @@ int bio_poll(struct bio *bio, unsigned int flags)
 	if (WARN_ON_ONCE(!queue_is_mq(q)))
 		ret = 0;	/* not yet implemented, should not happen */
 	else
-		ret = blk_mq_poll(q, cookie, flags);
+		ret = blk_mq_poll(q, cookie, iob, flags);
 	blk_queue_exit(q);
 	return ret;
 }
@@ -1030,7 +1030,8 @@ EXPORT_SYMBOL_GPL(bio_poll);
  * Helper to implement file_operations.iopoll.  Requires the bio to be stored
  * in iocb->private, and cleared before freeing the bio.
  */
-int iocb_bio_iopoll(struct kiocb *kiocb, unsigned int flags)
+int iocb_bio_iopoll(struct kiocb *kiocb, struct io_batch *iob,
+		    unsigned int flags)
 {
 	struct bio *bio;
 	int ret = 0;
@@ -1058,7 +1059,7 @@ int iocb_bio_iopoll(struct kiocb *kiocb, unsigned int flags)
 	rcu_read_lock();
 	bio = READ_ONCE(kiocb->private);
 	if (bio && bio->bi_bdev)
-		ret = bio_poll(bio, flags);
+		ret = bio_poll(bio, iob, flags);
 	rcu_read_unlock();
 
 	return ret;
diff --git a/block/blk-exec.c b/block/blk-exec.c
index 55f0cd34b37b..1b8b47f6e79b 100644
--- a/block/blk-exec.c
+++ b/block/blk-exec.c
@@ -77,7 +77,7 @@ static bool blk_rq_is_poll(struct request *rq)
 static void blk_rq_poll_completion(struct request *rq, struct completion *wait)
 {
 	do {
-		bio_poll(rq->bio, 0);
+		bio_poll(rq->bio, NULL, 0);
 		cond_resched();
 	} while (!completion_done(wait));
 }
diff --git a/block/blk-mq.c b/block/blk-mq.c
index 069837a020fe..6eac10fd244e 100644
--- a/block/blk-mq.c
+++ b/block/blk-mq.c
@@ -4229,7 +4229,7 @@ static bool blk_mq_poll_hybrid(struct request_queue *q, blk_qc_t qc)
 }
 
 static int blk_mq_poll_classic(struct request_queue *q, blk_qc_t cookie,
-		unsigned int flags)
+			       struct io_batch *iob, unsigned int flags)
 {
 	struct blk_mq_hw_ctx *hctx = blk_qc_to_hctx(q, cookie);
 	long state = get_current_state();
@@ -4240,7 +4240,7 @@ static int blk_mq_poll_classic(struct request_queue *q, blk_qc_t cookie,
 	do {
 		hctx->poll_invoked++;
 
-		ret = q->mq_ops->poll(hctx);
+		ret = q->mq_ops->poll(hctx, iob);
 		if (ret > 0) {
 			hctx->poll_success++;
 			__set_current_state(TASK_RUNNING);
@@ -4261,14 +4261,15 @@ static int blk_mq_poll_classic(struct request_queue *q, blk_qc_t cookie,
 	return 0;
 }
 
-int blk_mq_poll(struct request_queue *q, blk_qc_t cookie, unsigned int flags)
+int blk_mq_poll(struct request_queue *q, blk_qc_t cookie, struct io_batch *iob,
+		unsigned int flags)
 {
 	if (!(flags & BLK_POLL_NOSLEEP) &&
 	    q->poll_nsec != BLK_MQ_POLL_CLASSIC) {
 		if (blk_mq_poll_hybrid(q, cookie))
 			return 1;
 	}
-	return blk_mq_poll_classic(q, cookie, flags);
+	return blk_mq_poll_classic(q, cookie, iob, flags);
 }
 
 unsigned int blk_mq_rq_cpu(struct request *rq)
diff --git a/block/blk-mq.h b/block/blk-mq.h
index ceed0a001c76..924b537b2d52 100644
--- a/block/blk-mq.h
+++ b/block/blk-mq.h
@@ -38,7 +38,8 @@ struct blk_mq_ctx {
 } ____cacheline_aligned_in_smp;
 
 void blk_mq_submit_bio(struct bio *bio);
-int blk_mq_poll(struct request_queue *q, blk_qc_t cookie, unsigned int flags);
+int blk_mq_poll(struct request_queue *q, blk_qc_t cookie, struct io_batch *iob,
+		unsigned int flags);
 void blk_mq_exit_queue(struct request_queue *q);
 int blk_mq_update_nr_requests(struct request_queue *q, unsigned int nr);
 void blk_mq_wake_waiters(struct request_queue *q);
diff --git a/block/fops.c b/block/fops.c
index 551b71af6d90..89e1eae8f89a 100644
--- a/block/fops.c
+++ b/block/fops.c
@@ -105,7 +105,7 @@ static ssize_t __blkdev_direct_IO_simple(struct kiocb *iocb,
 		set_current_state(TASK_UNINTERRUPTIBLE);
 		if (!READ_ONCE(bio.bi_private))
 			break;
-		if (!(iocb->ki_flags & IOCB_HIPRI) || !bio_poll(&bio, 0))
+		if (!(iocb->ki_flags & IOCB_HIPRI) || !bio_poll(&bio, NULL, 0))
 			blk_io_schedule();
 	}
 	__set_current_state(TASK_RUNNING);
@@ -285,7 +285,7 @@ static ssize_t __blkdev_direct_IO(struct kiocb *iocb, struct iov_iter *iter,
 		if (!READ_ONCE(dio->waiter))
 			break;
 
-		if (!do_poll || !bio_poll(bio, 0))
+		if (!do_poll || !bio_poll(bio, NULL, 0))
 			blk_io_schedule();
 	}
 	__set_current_state(TASK_RUNNING);
diff --git a/drivers/block/null_blk/main.c b/drivers/block/null_blk/main.c
index 7ce911e2289d..161faf172527 100644
--- a/drivers/block/null_blk/main.c
+++ b/drivers/block/null_blk/main.c
@@ -1494,7 +1494,7 @@ static int null_map_queues(struct blk_mq_tag_set *set)
 	return 0;
 }
 
-static int null_poll(struct blk_mq_hw_ctx *hctx)
+static int null_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nullb_queue *nq = hctx->driver_data;
 	LIST_HEAD(list);
diff --git a/drivers/block/rnbd/rnbd-clt.c b/drivers/block/rnbd/rnbd-clt.c
index bd4a41afbbfc..5ccd6c602673 100644
--- a/drivers/block/rnbd/rnbd-clt.c
+++ b/drivers/block/rnbd/rnbd-clt.c
@@ -1176,7 +1176,7 @@ static blk_status_t rnbd_queue_rq(struct blk_mq_hw_ctx *hctx,
 	return ret;
 }
 
-static int rnbd_rdma_poll(struct blk_mq_hw_ctx *hctx)
+static int rnbd_rdma_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct rnbd_queue *q = hctx->driver_data;
 	struct rnbd_clt_dev *dev = q->dev;
diff --git a/drivers/nvme/host/pci.c b/drivers/nvme/host/pci.c
index 0dd4b44b59cd..9db6e23f41ef 100644
--- a/drivers/nvme/host/pci.c
+++ b/drivers/nvme/host/pci.c
@@ -1092,7 +1092,7 @@ static void nvme_poll_irqdisable(struct nvme_queue *nvmeq)
 	enable_irq(pci_irq_vector(pdev, nvmeq->cq_vector));
 }
 
-static int nvme_poll(struct blk_mq_hw_ctx *hctx)
+static int nvme_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nvme_queue *nvmeq = hctx->driver_data;
 	bool found;
@@ -1274,7 +1274,7 @@ static enum blk_eh_timer_return nvme_timeout(struct request *req, bool reserved)
 	 * Did we miss an interrupt?
 	 */
 	if (test_bit(NVMEQ_POLLED, &nvmeq->flags))
-		nvme_poll(req->mq_hctx);
+		nvme_poll(req->mq_hctx, NULL);
 	else
 		nvme_poll_irqdisable(nvmeq);
 
diff --git a/drivers/nvme/host/rdma.c b/drivers/nvme/host/rdma.c
index 40317e1b9183..f02c1411c8e9 100644
--- a/drivers/nvme/host/rdma.c
+++ b/drivers/nvme/host/rdma.c
@@ -2106,7 +2106,7 @@ static blk_status_t nvme_rdma_queue_rq(struct blk_mq_hw_ctx *hctx,
 	return ret;
 }
 
-static int nvme_rdma_poll(struct blk_mq_hw_ctx *hctx)
+static int nvme_rdma_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nvme_rdma_queue *queue = hctx->driver_data;
 
diff --git a/drivers/nvme/host/tcp.c b/drivers/nvme/host/tcp.c
index 3c1c29dd3020..a094920a6914 100644
--- a/drivers/nvme/host/tcp.c
+++ b/drivers/nvme/host/tcp.c
@@ -2429,7 +2429,7 @@ static int nvme_tcp_map_queues(struct blk_mq_tag_set *set)
 	return 0;
 }
 
-static int nvme_tcp_poll(struct blk_mq_hw_ctx *hctx)
+static int nvme_tcp_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nvme_tcp_queue *queue = hctx->driver_data;
 	struct sock *sk = queue->sock->sk;
diff --git a/drivers/scsi/scsi_lib.c b/drivers/scsi/scsi_lib.c
index 33fd9a01330c..0b382b2ad41b 100644
--- a/drivers/scsi/scsi_lib.c
+++ b/drivers/scsi/scsi_lib.c
@@ -1784,7 +1784,7 @@ static void scsi_mq_exit_request(struct blk_mq_tag_set *set, struct request *rq,
 }
 
 
-static int scsi_mq_poll(struct blk_mq_hw_ctx *hctx)
+static int scsi_mq_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct Scsi_Host *shost = hctx->driver_data;
 
diff --git a/fs/io_uring.c b/fs/io_uring.c
index e43e130a0e92..082ff64c1bcb 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -2412,7 +2412,7 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
 		if (READ_ONCE(req->iopoll_completed))
 			break;
 
-		ret = kiocb->ki_filp->f_op->iopoll(kiocb, poll_flags);
+		ret = kiocb->ki_filp->f_op->iopoll(kiocb, NULL, poll_flags);
 		if (unlikely(ret < 0))
 			return ret;
 		else if (ret)
diff --git a/fs/iomap/direct-io.c b/fs/iomap/direct-io.c
index 8efab177011d..83ecfba53abe 100644
--- a/fs/iomap/direct-io.c
+++ b/fs/iomap/direct-io.c
@@ -630,7 +630,7 @@ __iomap_dio_rw(struct kiocb *iocb, struct iov_iter *iter,
 				break;
 
 			if (!dio->submit.poll_bio ||
-			    !bio_poll(dio->submit.poll_bio, 0))
+			    !bio_poll(dio->submit.poll_bio, NULL, 0))
 				blk_io_schedule();
 		}
 		__set_current_state(TASK_RUNNING);
diff --git a/include/linux/blk-mq.h b/include/linux/blk-mq.h
index c05560524841..5106c4cc411a 100644
--- a/include/linux/blk-mq.h
+++ b/include/linux/blk-mq.h
@@ -563,7 +563,7 @@ struct blk_mq_ops {
 	/**
 	 * @poll: Called to poll for completion of a specific tag.
 	 */
-	int (*poll)(struct blk_mq_hw_ctx *);
+	int (*poll)(struct blk_mq_hw_ctx *, struct io_batch *);
 
 	/**
 	 * @complete: Mark the request as complete.
diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
index b39b19dbe7b6..f626ab192ea2 100644
--- a/include/linux/blkdev.h
+++ b/include/linux/blkdev.h
@@ -569,8 +569,9 @@ blk_status_t errno_to_blk_status(int errno);
 #define BLK_POLL_ONESHOT		(1 << 0)
 /* do not sleep to wait for the expected completion time */
 #define BLK_POLL_NOSLEEP		(1 << 1)
-int bio_poll(struct bio *bio, unsigned int flags);
-int iocb_bio_iopoll(struct kiocb *kiocb, unsigned int flags);
+int bio_poll(struct bio *bio, struct io_batch *iob, unsigned int flags);
+int iocb_bio_iopoll(struct kiocb *kiocb, struct io_batch *iob,
+			unsigned int flags);
 
 static inline struct request_queue *bdev_get_queue(struct block_device *bdev)
 {
diff --git a/include/linux/fs.h b/include/linux/fs.h
index f98a361a6752..0ba86433797b 100644
--- a/include/linux/fs.h
+++ b/include/linux/fs.h
@@ -48,6 +48,7 @@
 struct backing_dev_info;
 struct bdi_writeback;
 struct bio;
+struct io_batch;
 struct export_operations;
 struct fiemap_extent_info;
 struct hd_geometry;
@@ -2075,7 +2076,7 @@ struct file_operations {
 	ssize_t (*write) (struct file *, const char __user *, size_t, loff_t *);
 	ssize_t (*read_iter) (struct kiocb *, struct iov_iter *);
 	ssize_t (*write_iter) (struct kiocb *, struct iov_iter *);
-	int (*iopoll)(struct kiocb *kiocb, unsigned int flags);
+	int (*iopoll)(struct kiocb *kiocb, struct io_batch *, unsigned int flags);
 	int (*iterate) (struct file *, struct dir_context *);
 	int (*iterate_shared) (struct file *, struct dir_context *);
 	__poll_t (*poll) (struct file *, struct poll_table_struct *);
diff --git a/mm/page_io.c b/mm/page_io.c
index a68faab5b310..6010fb07f231 100644
--- a/mm/page_io.c
+++ b/mm/page_io.c
@@ -424,7 +424,7 @@ int swap_readpage(struct page *page, bool synchronous)
 		if (!READ_ONCE(bio->bi_private))
 			break;
 
-		if (!bio_poll(bio, 0))
+		if (!bio_poll(bio, NULL, 0))
 			blk_io_schedule();
 	}
 	__set_current_state(TASK_RUNNING);
-- 
2.33.0

[PATCH 3/9] sbitmap: add helper to clear a batch of tags

[Jens Axboe]

sbitmap currently only supports clearing tags one-by-one, add a helper
that allows the caller to pass in an array of tags to clear.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 include/linux/sbitmap.h | 11 +++++++++++
 lib/sbitmap.c           | 44 ++++++++++++++++++++++++++++++++++++++---
 2 files changed, 52 insertions(+), 3 deletions(-)

diff --git a/include/linux/sbitmap.h b/include/linux/sbitmap.h
index e30b56023ead..4a6ff274335a 100644
--- a/include/linux/sbitmap.h
+++ b/include/linux/sbitmap.h
@@ -528,6 +528,17 @@ void sbitmap_queue_min_shallow_depth(struct sbitmap_queue *sbq,
 void sbitmap_queue_clear(struct sbitmap_queue *sbq, unsigned int nr,
 			 unsigned int cpu);
 
+/**
+ * sbitmap_queue_clear_batch() - Free a batch of allocated bits
+ * &struct sbitmap_queue.
+ * @sbq: Bitmap to free from.
+ * @offset: offset for each tag in array
+ * @tags: array of tags
+ * @nr_tags: number of tags in array
+ */
+void sbitmap_queue_clear_batch(struct sbitmap_queue *sbq, int offset,
+				int *tags, int nr_tags);
+
 static inline int sbq_index_inc(int index)
 {
 	return (index + 1) & (SBQ_WAIT_QUEUES - 1);
diff --git a/lib/sbitmap.c b/lib/sbitmap.c
index f398e0ae548e..c6e2f1f2c4d2 100644
--- a/lib/sbitmap.c
+++ b/lib/sbitmap.c
@@ -628,6 +628,46 @@ void sbitmap_queue_wake_up(struct sbitmap_queue *sbq)
 }
 EXPORT_SYMBOL_GPL(sbitmap_queue_wake_up);
 
+static inline void sbitmap_update_cpu_hint(struct sbitmap *sb, int cpu, int tag)
+{
+	if (likely(!sb->round_robin && tag < sb->depth))
+		*per_cpu_ptr(sb->alloc_hint, cpu) = tag;
+}
+
+void sbitmap_queue_clear_batch(struct sbitmap_queue *sbq, int offset,
+				int *tags, int nr_tags)
+{
+	struct sbitmap *sb = &sbq->sb;
+	unsigned long *addr = NULL;
+	unsigned long mask = 0;
+	int i;
+
+	smp_mb__before_atomic();
+	for (i = 0; i < nr_tags; i++) {
+		const int tag = tags[i] - offset;
+		unsigned long *this_addr;
+
+		/* since we're clearing a batch, skip the deferred map */
+		this_addr = &sb->map[SB_NR_TO_INDEX(sb, tag)].word;
+		if (!addr) {
+			addr = this_addr;
+		} else if (addr != this_addr) {
+			atomic_long_andnot(mask, (atomic_long_t *) addr);
+			mask = 0;
+			addr = this_addr;
+		}
+		mask |= (1UL << SB_NR_TO_BIT(sb, tag));
+	}
+
+	if (mask)
+		atomic_long_andnot(mask, (atomic_long_t *) addr);
+
+	smp_mb__after_atomic();
+	sbitmap_queue_wake_up(sbq);
+	sbitmap_update_cpu_hint(&sbq->sb, raw_smp_processor_id(),
+					tags[nr_tags - 1] - offset);
+}
+
 void sbitmap_queue_clear(struct sbitmap_queue *sbq, unsigned int nr,
 			 unsigned int cpu)
 {
@@ -652,9 +692,7 @@ void sbitmap_queue_clear(struct sbitmap_queue *sbq, unsigned int nr,
 	 */
 	smp_mb__after_atomic();
 	sbitmap_queue_wake_up(sbq);
-
-	if (likely(!sbq->sb.round_robin && nr < sbq->sb.depth))
-		*per_cpu_ptr(sbq->sb.alloc_hint, cpu) = nr;
+	sbitmap_update_cpu_hint(&sbq->sb, cpu, nr);
 }
 EXPORT_SYMBOL_GPL(sbitmap_queue_clear);
 
-- 
2.33.0

[PATCH 4/9] sbitmap: test bit before calling test_and_set_bit()

[Jens Axboe]

If we come across bits that are already set, then it's quicker to test
that first and gate the test_and_set_bit() operation on the result of
the bit test.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 lib/sbitmap.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/sbitmap.c b/lib/sbitmap.c
index c6e2f1f2c4d2..11b244a8d00f 100644
--- a/lib/sbitmap.c
+++ b/lib/sbitmap.c
@@ -166,7 +166,7 @@ static int __sbitmap_get_word(unsigned long *word, unsigned long depth,
 			return -1;
 		}
 
-		if (!test_and_set_bit_lock(nr, word))
+		if (!test_bit(nr, word) && !test_and_set_bit_lock(nr, word))
 			break;
 
 		hint = nr + 1;
-- 
2.33.0

Re: [PATCH 4/9] sbitmap: test bit before calling test_and_set_bit()

[Hannes Reinecke]

On 10/13/21 6:54 PM, Jens Axboe wrote:
> If we come across bits that are already set, then it's quicker to test
> that first and gate the test_and_set_bit() operation on the result of
> the bit test.
> 
> Signed-off-by: Jens Axboe <axboe@kernel.dk>
> ---
>   lib/sbitmap.c | 2 +-
>   1 file changed, 1 insertion(+), 1 deletion(-)
> 
> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
> index c6e2f1f2c4d2..11b244a8d00f 100644
> --- a/lib/sbitmap.c
> +++ b/lib/sbitmap.c
> @@ -166,7 +166,7 @@ static int __sbitmap_get_word(unsigned long *word, unsigned long depth,
>   			return -1;
>   		}
>   
> -		if (!test_and_set_bit_lock(nr, word))
> +		if (!test_bit(nr, word) && !test_and_set_bit_lock(nr, word))
>   			break;
>   
>   		hint = nr + 1;
> 
Hah. Finally something to latch on.

I've seen this coding pattern quite a lot in the block layer, and, of 
course, mostly in the hot path.
(Kinda the point, I guess :-)

However, my question is this:

While 'test_and_set_bit()' is atomic, the combination of
'test_bit && !test_and_set_bit()' is not.

IE this change moves an atomic operation into a non-atomic one.
So how can we be sure that this patch doesn't introduce a race condition?
And if it doesn't, should we add some comment above the code why this is 
safe to do here?

Cheers,

Hannes
-- 
Dr. Hannes Reinecke                Kernel Storage Architect
hare@suse.de                              +49 911 74053 688
SUSE Software Solutions GmbH, Maxfeldstr. 5, 90409 Nürnberg
HRB 36809 (AG Nürnberg), Geschäftsführer: Felix Imendörffer

Re: [PATCH 4/9] sbitmap: test bit before calling test_and_set_bit()

[Jens Axboe]

On 10/14/21 1:20 AM, Hannes Reinecke wrote:
> On 10/13/21 6:54 PM, Jens Axboe wrote:
>> If we come across bits that are already set, then it's quicker to test
>> that first and gate the test_and_set_bit() operation on the result of
>> the bit test.
>>
>> Signed-off-by: Jens Axboe <axboe@kernel.dk>
>> ---
>>   lib/sbitmap.c | 2 +-
>>   1 file changed, 1 insertion(+), 1 deletion(-)
>>
>> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
>> index c6e2f1f2c4d2..11b244a8d00f 100644
>> --- a/lib/sbitmap.c
>> +++ b/lib/sbitmap.c
>> @@ -166,7 +166,7 @@ static int __sbitmap_get_word(unsigned long *word, unsigned long depth,
>>   			return -1;
>>   		}
>>   
>> -		if (!test_and_set_bit_lock(nr, word))
>> +		if (!test_bit(nr, word) && !test_and_set_bit_lock(nr, word))
>>   			break;
>>   
>>   		hint = nr + 1;
>>
> Hah. Finally something to latch on.
> 
> I've seen this coding pattern quite a lot in the block layer, and, of 
> course, mostly in the hot path.
> (Kinda the point, I guess :-)
> 
> However, my question is this:
> 
> While 'test_and_set_bit()' is atomic, the combination of
> 'test_bit && !test_and_set_bit()' is not.
> 
> IE this change moves an atomic operation into a non-atomic one.
> So how can we be sure that this patch doesn't introduce a race condition?
> And if it doesn't, should we add some comment above the code why this is 
> safe to do here?

If test_bit() returns true, we don't bother with the atomic
test_and_set. That's to avoid re-dirtying a cacheline when the operation
would be pointless. There's no race in this case, we just skip this bit.

The sequence of !test_bit && !test_and_set_bit is very much still
atomic, it's just gated on the fact that our optimistic first check said
that the bit is currently clear. It obviously has to be.

-- 
Jens Axboe

Re: [PATCH 4/9] sbitmap: test bit before calling test_and_set_bit()

[Bart Van Assche]

On 10/13/21 9:54 AM, Jens Axboe wrote:
> If we come across bits that are already set, then it's quicker to test
> that first and gate the test_and_set_bit() operation on the result of
> the bit test.
> 
> Signed-off-by: Jens Axboe <axboe@kernel.dk>
> ---
>   lib/sbitmap.c | 2 +-
>   1 file changed, 1 insertion(+), 1 deletion(-)
> 
> diff --git a/lib/sbitmap.c b/lib/sbitmap.c
> index c6e2f1f2c4d2..11b244a8d00f 100644
> --- a/lib/sbitmap.c
> +++ b/lib/sbitmap.c
> @@ -166,7 +166,7 @@ static int __sbitmap_get_word(unsigned long *word, unsigned long depth,
>   			return -1;
>   		}
>   
> -		if (!test_and_set_bit_lock(nr, word))
> +		if (!test_bit(nr, word) && !test_and_set_bit_lock(nr, word))
>   			break;
>   
>   		hint = nr + 1;

Hi Jens,

Are numbers available about how much this change improves performance? 
I'm asking this since my understanding is that this patch only helps if 
bit 'nr' is set after find_next_zero_bit() finished and before 
test_and_set_bit() is called. Isn't that a very short time window? Am I 
perhaps missing something?

Thanks,

Bart.

[PATCH 5/9] block: add support for blk_mq_end_request_batch()

[Jens Axboe]

Instead of calling blk_mq_end_request() on a single request, add a helper
that takes the new struct io_batch and completes any request stored in
there.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 block/blk-mq-tag.c     |  6 +++
 block/blk-mq-tag.h     |  1 +
 block/blk-mq.c         | 85 +++++++++++++++++++++++++++++++++++++-----
 include/linux/blk-mq.h | 13 +++++++
 4 files changed, 95 insertions(+), 10 deletions(-)

diff --git a/block/blk-mq-tag.c b/block/blk-mq-tag.c
index c43b97201161..b94c3e8ef392 100644
--- a/block/blk-mq-tag.c
+++ b/block/blk-mq-tag.c
@@ -207,6 +207,12 @@ void blk_mq_put_tag(struct blk_mq_tags *tags, struct blk_mq_ctx *ctx,
 	}
 }
 
+void blk_mq_put_tags(struct blk_mq_tags *tags, int *tag_array, int nr_tags)
+{
+	sbitmap_queue_clear_batch(&tags->bitmap_tags, tags->nr_reserved_tags,
+					tag_array, nr_tags);
+}
+
 struct bt_iter_data {
 	struct blk_mq_hw_ctx *hctx;
 	busy_iter_fn *fn;
diff --git a/block/blk-mq-tag.h b/block/blk-mq-tag.h
index 71c2f7d8e9b7..78ae2fb8e2a4 100644
--- a/block/blk-mq-tag.h
+++ b/block/blk-mq-tag.h
@@ -42,6 +42,7 @@ unsigned long blk_mq_get_tags(struct blk_mq_alloc_data *data, int nr_tags,
 			      unsigned int *offset);
 extern void blk_mq_put_tag(struct blk_mq_tags *tags, struct blk_mq_ctx *ctx,
 			   unsigned int tag);
+void blk_mq_put_tags(struct blk_mq_tags *tags, int *tag_array, int nr_tags);
 extern int blk_mq_tag_update_depth(struct blk_mq_hw_ctx *hctx,
 					struct blk_mq_tags **tags,
 					unsigned int depth, bool can_grow);
diff --git a/block/blk-mq.c b/block/blk-mq.c
index 6eac10fd244e..d603703cf272 100644
--- a/block/blk-mq.c
+++ b/block/blk-mq.c
@@ -603,19 +603,22 @@ void blk_mq_free_plug_rqs(struct blk_plug *plug)
 	}
 }
 
-inline void __blk_mq_end_request(struct request *rq, blk_status_t error)
+static inline void __blk_mq_end_request_acct(struct request *rq,
+					     blk_status_t error, u64 now)
 {
-	if (blk_mq_need_time_stamp(rq)) {
-		u64 now = ktime_get_ns();
+	if (rq->rq_flags & RQF_STATS) {
+		blk_mq_poll_stats_start(rq->q);
+		blk_stat_add(rq, now);
+	}
 
-		if (rq->rq_flags & RQF_STATS) {
-			blk_mq_poll_stats_start(rq->q);
-			blk_stat_add(rq, now);
-		}
+	blk_mq_sched_completed_request(rq, now);
+	blk_account_io_done(rq, now);
+}
 
-		blk_mq_sched_completed_request(rq, now);
-		blk_account_io_done(rq, now);
-	}
+inline void __blk_mq_end_request(struct request *rq, blk_status_t error)
+{
+	if (blk_mq_need_time_stamp(rq))
+		__blk_mq_end_request_acct(rq, error, ktime_get_ns());
 
 	if (rq->end_io) {
 		rq_qos_done(rq->q, rq);
@@ -848,6 +851,68 @@ void blk_mq_end_request(struct request *rq, blk_status_t error)
 }
 EXPORT_SYMBOL(blk_mq_end_request);
 
+#define TAG_COMP_BATCH		32
+#define TAG_SCHED_BATCH		(TAG_COMP_BATCH >> 1)
+
+static inline void blk_mq_flush_tag_batch(struct blk_mq_hw_ctx *hctx,
+					  int *tag_array, int nr_tags)
+{
+	struct request_queue *q = hctx->queue;
+
+	blk_mq_put_tags(hctx->tags, tag_array, nr_tags);
+	if (q->elevator)
+		blk_mq_put_tags(hctx->sched_tags, &tag_array[TAG_SCHED_BATCH],
+				nr_tags);
+	percpu_ref_put_many(&q->q_usage_counter, nr_tags);
+	blk_mq_sched_restart(hctx);
+}
+
+void blk_mq_end_request_batch(struct io_batch *iob)
+{
+	int tags[TAG_COMP_BATCH], nr_tags = 0, acct_tags = 0;
+	struct blk_mq_hw_ctx *last_hctx = NULL;
+	struct request *rq;
+	u64 now = 0;
+
+	while ((rq = rq_list_pop(&iob->req_list)) != NULL) {
+		if (!now && blk_mq_need_time_stamp(rq))
+			now = ktime_get_ns();
+		blk_update_request(rq, rq->status, blk_rq_bytes(rq));
+		__blk_mq_end_request_acct(rq, rq->status, now);
+
+		if (rq->q->elevator) {
+			blk_mq_free_request(rq);
+			continue;
+		}
+
+		if (!refcount_dec_and_test(&rq->ref))
+			continue;
+
+		blk_crypto_free_request(rq);
+		blk_pm_mark_last_busy(rq);
+		rq_qos_done(rq->q, rq);
+		WRITE_ONCE(rq->state, MQ_RQ_IDLE);
+
+		if (acct_tags == TAG_COMP_BATCH ||
+		    (last_hctx && last_hctx != rq->mq_hctx)) {
+			blk_mq_flush_tag_batch(last_hctx, tags, nr_tags);
+			acct_tags = nr_tags = 0;
+		}
+		tags[nr_tags] = rq->tag;
+		last_hctx = rq->mq_hctx;
+		if (last_hctx->queue->elevator) {
+			tags[nr_tags + TAG_SCHED_BATCH] = rq->internal_tag;
+			acct_tags++;
+		}
+		nr_tags++;
+		acct_tags++;
+	}
+
+	if (nr_tags)
+		blk_mq_flush_tag_batch(last_hctx, tags, nr_tags);
+}
+EXPORT_SYMBOL_GPL(blk_mq_end_request_batch);
+
 static void blk_complete_reqs(struct llist_head *list)
 {
 	struct llist_node *entry = llist_reverse_order(llist_del_all(list));
diff --git a/include/linux/blk-mq.h b/include/linux/blk-mq.h
index 5106c4cc411a..aea7d866a34c 100644
--- a/include/linux/blk-mq.h
+++ b/include/linux/blk-mq.h
@@ -183,9 +183,16 @@ struct request {
 	unsigned int timeout;
 	unsigned long deadline;
 
+	/*
+	 * csd is used for remote completions, fifo_time at scheduler time.
+	 * They are mutually exclusive. result is used at completion time
+	 * like csd, but for batched IO. Batched IO does not use IPI
+	 * completions.
+	 */
 	union {
 		struct __call_single_data csd;
 		u64 fifo_time;
+		blk_status_t status;
 	};
 
 	/*
@@ -570,6 +577,11 @@ struct blk_mq_ops {
 	 */
 	void (*complete)(struct request *);
 
+	/**
+	 * @complete_batch: Mark list of requests as complete
+	 */
+	void (*complete_batch)(struct io_batch *);
+
 	/**
 	 * @init_hctx: Called when the block layer side of a hardware queue has
 	 * been set up, allowing the driver to allocate/init matching
@@ -759,6 +771,7 @@ static inline void blk_mq_set_request_complete(struct request *rq)
 void blk_mq_start_request(struct request *rq);
 void blk_mq_end_request(struct request *rq, blk_status_t error);
 void __blk_mq_end_request(struct request *rq, blk_status_t error);
+void blk_mq_end_request_batch(struct io_batch *ib);
 
 void blk_mq_requeue_request(struct request *rq, bool kick_requeue_list);
 void blk_mq_kick_requeue_list(struct request_queue *q);
-- 
2.33.0

Re: [PATCH 5/9] block: add support for blk_mq_end_request_batch()

[Christoph Hellwig]

> +void blk_mq_end_request_batch(struct io_batch *iob)
> +{
> +	int tags[TAG_COMP_BATCH], nr_tags = 0, acct_tags = 0;
> +	struct blk_mq_hw_ctx *last_hctx = NULL;
> +	struct request *rq;
> +	u64 now = 0;
> +
> +	while ((rq = rq_list_pop(&iob->req_list)) != NULL) {
> +		if (!now && blk_mq_need_time_stamp(rq))
> +			now = ktime_get_ns();
> +		blk_update_request(rq, rq->status, blk_rq_bytes(rq));
> +		__blk_mq_end_request_acct(rq, rq->status, now);
> +
> +		if (rq->q->elevator) {
> +			blk_mq_free_request(rq);
> +			continue;
> +		}

So why do we even bother adding requests with an elevator to the batch
list?  

> +	/*
> +	 * csd is used for remote completions, fifo_time at scheduler time.
> +	 * They are mutually exclusive. result is used at completion time
> +	 * like csd, but for batched IO. Batched IO does not use IPI
> +	 * completions.
> +	 */
>  	union {
>  		struct __call_single_data csd;
>  		u64 fifo_time;
> +		blk_status_t status;
>  	};

The ->status field isn't needed any more now that error completions
aren't batched.

Re: [PATCH 5/9] block: add support for blk_mq_end_request_batch()

[Jens Axboe]

On 10/14/21 1:32 AM, Christoph Hellwig wrote:
>> +void blk_mq_end_request_batch(struct io_batch *iob)
>> +{
>> +	int tags[TAG_COMP_BATCH], nr_tags = 0, acct_tags = 0;
>> +	struct blk_mq_hw_ctx *last_hctx = NULL;
>> +	struct request *rq;
>> +	u64 now = 0;
>> +
>> +	while ((rq = rq_list_pop(&iob->req_list)) != NULL) {
>> +		if (!now && blk_mq_need_time_stamp(rq))
>> +			now = ktime_get_ns();
>> +		blk_update_request(rq, rq->status, blk_rq_bytes(rq));
>> +		__blk_mq_end_request_acct(rq, rq->status, now);
>> +
>> +		if (rq->q->elevator) {
>> +			blk_mq_free_request(rq);
>> +			continue;
>> +		}
> 
> So why do we even bother adding requests with an elevator to the batch
> list?  

You still get the benefit of amortized time keeping, and it's more
efficient to complete in batches rather than one-at-the-time. It's just
not as good as the non-elevator path.

>> +	/*
>> +	 * csd is used for remote completions, fifo_time at scheduler time.
>> +	 * They are mutually exclusive. result is used at completion time
>> +	 * like csd, but for batched IO. Batched IO does not use IPI
>> +	 * completions.
>> +	 */
>>  	union {
>>  		struct __call_single_data csd;
>>  		u64 fifo_time;
>> +		blk_status_t status;
>>  	};
> 
> The ->status field isn't needed any more now that error completions
> aren't batched.

Killed.

-- 
Jens Axboe

[PATCH 6/9] nvme: add support for batched completion of polled IO

[Jens Axboe]

Take advantage of struct io_batch, if passed in to the nvme poll handler.
If it's set, rather than complete each request individually inline, store
them in the io_batch list. We only do so for requests that will complete
successfully, anything else will be completed inline as before.

Add an mq_ops->complete_batch() handler to do the post-processing of
the io_batch list once polling is complete.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 drivers/nvme/host/core.c | 18 +++++++++++---
 drivers/nvme/host/nvme.h | 17 +++++++++++++
 drivers/nvme/host/pci.c  | 54 +++++++++++++++++++++++++++++++++++-----
 3 files changed, 80 insertions(+), 9 deletions(-)

diff --git a/drivers/nvme/host/core.c b/drivers/nvme/host/core.c
index c2c2e8545292..4b14258a3bac 100644
--- a/drivers/nvme/host/core.c
+++ b/drivers/nvme/host/core.c
@@ -346,15 +346,19 @@ static inline enum nvme_disposition nvme_decide_disposition(struct request *req)
 	return RETRY;
 }
 
-static inline void nvme_end_req(struct request *req)
+static inline void nvme_end_req_zoned(struct request *req)
 {
-	blk_status_t status = nvme_error_status(nvme_req(req)->status);
-
 	if (IS_ENABLED(CONFIG_BLK_DEV_ZONED) &&
 	    req_op(req) == REQ_OP_ZONE_APPEND)
 		req->__sector = nvme_lba_to_sect(req->q->queuedata,
 			le64_to_cpu(nvme_req(req)->result.u64));
+}
+
+static inline void nvme_end_req(struct request *req)
+{
+	blk_status_t status = nvme_error_status(nvme_req(req)->status);
 
+	nvme_end_req_zoned(req);
 	nvme_trace_bio_complete(req);
 	blk_mq_end_request(req, status);
 }
@@ -381,6 +385,14 @@ void nvme_complete_rq(struct request *req)
 }
 EXPORT_SYMBOL_GPL(nvme_complete_rq);
 
+void nvme_complete_batch_req(struct request *req)
+{
+	nvme_cleanup_cmd(req);
+	nvme_end_req_zoned(req);
+	req->status = BLK_STS_OK;
+}
+EXPORT_SYMBOL_GPL(nvme_complete_batch_req);
+
 /*
  * Called to unwind from ->queue_rq on a failed command submission so that the
  * multipathing code gets called to potentially failover to another path.
diff --git a/drivers/nvme/host/nvme.h b/drivers/nvme/host/nvme.h
index ed79a6c7e804..e0c079f704cf 100644
--- a/drivers/nvme/host/nvme.h
+++ b/drivers/nvme/host/nvme.h
@@ -638,6 +638,23 @@ static inline bool nvme_is_aen_req(u16 qid, __u16 command_id)
 }
 
 void nvme_complete_rq(struct request *req);
+void nvme_complete_batch_req(struct request *req);
+
+static __always_inline void nvme_complete_batch(struct io_batch *iob,
+						void (*fn)(struct request *rq))
+{
+	struct request *req;
+
+	req = rq_list_peek(&iob->req_list);
+	while (req) {
+		fn(req);
+		nvme_complete_batch_req(req);
+		req = rq_list_next(req);
+	}
+
+	blk_mq_end_request_batch(iob);
+}
+
 blk_status_t nvme_host_path_error(struct request *req);
 bool nvme_cancel_request(struct request *req, void *data, bool reserved);
 void nvme_cancel_tagset(struct nvme_ctrl *ctrl);
diff --git a/drivers/nvme/host/pci.c b/drivers/nvme/host/pci.c
index 9db6e23f41ef..ae253f6f5c80 100644
--- a/drivers/nvme/host/pci.c
+++ b/drivers/nvme/host/pci.c
@@ -959,7 +959,7 @@ static blk_status_t nvme_queue_rq(struct blk_mq_hw_ctx *hctx,
 	return ret;
 }
 
-static void nvme_pci_complete_rq(struct request *req)
+static __always_inline void nvme_pci_unmap_rq(struct request *req)
 {
 	struct nvme_iod *iod = blk_mq_rq_to_pdu(req);
 	struct nvme_dev *dev = iod->nvmeq->dev;
@@ -969,9 +969,19 @@ static void nvme_pci_complete_rq(struct request *req)
 			       rq_integrity_vec(req)->bv_len, rq_data_dir(req));
 	if (blk_rq_nr_phys_segments(req))
 		nvme_unmap_data(dev, req);
+}
+
+static void nvme_pci_complete_rq(struct request *req)
+{
+	nvme_pci_unmap_rq(req);
 	nvme_complete_rq(req);
 }
 
+static void nvme_pci_complete_batch(struct io_batch *iob)
+{
+	nvme_complete_batch(iob, nvme_pci_unmap_rq);
+}
+
 /* We read the CQE phase first to check if the rest of the entry is valid */
 static inline bool nvme_cqe_pending(struct nvme_queue *nvmeq)
 {
@@ -996,7 +1006,8 @@ static inline struct blk_mq_tags *nvme_queue_tagset(struct nvme_queue *nvmeq)
 	return nvmeq->dev->tagset.tags[nvmeq->qid - 1];
 }
 
-static inline void nvme_handle_cqe(struct nvme_queue *nvmeq, u16 idx)
+static inline void nvme_handle_cqe(struct nvme_queue *nvmeq,
+				   struct io_batch *iob, u16 idx)
 {
 	struct nvme_completion *cqe = &nvmeq->cqes[idx];
 	__u16 command_id = READ_ONCE(cqe->command_id);
@@ -1023,8 +1034,17 @@ static inline void nvme_handle_cqe(struct nvme_queue *nvmeq, u16 idx)
 	}
 
 	trace_nvme_sq(req, cqe->sq_head, nvmeq->sq_tail);
-	if (!nvme_try_complete_req(req, cqe->status, cqe->result))
-		nvme_pci_complete_rq(req);
+	if (!nvme_try_complete_req(req, cqe->status, cqe->result)) {
+		/*
+		 * Do normal inline completion if we don't have a batch
+		 * list, if we have an end_io handler, or if the status of
+		 * the request isn't just normal success.
+		 */
+		if (!iob || req->end_io || nvme_req(req)->status)
+			nvme_pci_complete_rq(req);
+		else
+			rq_list_add_tail(&iob->req_list, req);
+	}
 }
 
 static inline void nvme_update_cq_head(struct nvme_queue *nvmeq)
@@ -1050,7 +1070,7 @@ static inline int nvme_process_cq(struct nvme_queue *nvmeq)
 		 * the cqe requires a full read memory barrier
 		 */
 		dma_rmb();
-		nvme_handle_cqe(nvmeq, nvmeq->cq_head);
+		nvme_handle_cqe(nvmeq, NULL, nvmeq->cq_head);
 		nvme_update_cq_head(nvmeq);
 	}
 
@@ -1092,6 +1112,27 @@ static void nvme_poll_irqdisable(struct nvme_queue *nvmeq)
 	enable_irq(pci_irq_vector(pdev, nvmeq->cq_vector));
 }
 
+static inline int nvme_poll_cq(struct nvme_queue *nvmeq, struct io_batch *iob)
+{
+	int found = 0;
+
+	while (nvme_cqe_pending(nvmeq)) {
+		found++;
+		/*
+		 * load-load control dependency between phase and the rest of
+		 * the cqe requires a full read memory barrier
+		 */
+		dma_rmb();
+		nvme_handle_cqe(nvmeq, iob, nvmeq->cq_head);
+		nvme_update_cq_head(nvmeq);
+	}
+
+	if (found)
+		nvme_ring_cq_doorbell(nvmeq);
+	return found;
+}
+
+
 static int nvme_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nvme_queue *nvmeq = hctx->driver_data;
@@ -1101,7 +1142,7 @@ static int nvme_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 		return 0;
 
 	spin_lock(&nvmeq->cq_poll_lock);
-	found = nvme_process_cq(nvmeq);
+	found = nvme_poll_cq(nvmeq, iob);
 	spin_unlock(&nvmeq->cq_poll_lock);
 
 	return found;
@@ -1639,6 +1680,7 @@ static const struct blk_mq_ops nvme_mq_admin_ops = {
 static const struct blk_mq_ops nvme_mq_ops = {
 	.queue_rq	= nvme_queue_rq,
 	.complete	= nvme_pci_complete_rq,
+	.complete_batch = nvme_pci_complete_batch,
 	.commit_rqs	= nvme_commit_rqs,
 	.init_hctx	= nvme_init_hctx,
 	.init_request	= nvme_init_request,
-- 
2.33.0

Re: [PATCH 6/9] nvme: add support for batched completion of polled IO

[Christoph Hellwig]

On Wed, Oct 13, 2021 at 10:54:13AM -0600, Jens Axboe wrote:
> +void nvme_complete_batch_req(struct request *req)
> +{
> +	nvme_cleanup_cmd(req);
> +	nvme_end_req_zoned(req);
> +	req->status = BLK_STS_OK;
> +}
> +EXPORT_SYMBOL_GPL(nvme_complete_batch_req);
> +

I'd be tempted to just merge this helper into the only caller.
nvme_cleanup_cmd is exported anyway, so this would just add an export
for nvme_end_req_zoned.

> +static __always_inline void nvme_complete_batch(struct io_batch *iob,
> +						void (*fn)(struct request *rq))
> +{
> +	struct request *req;
> +
> +	req = rq_list_peek(&iob->req_list);
> +	while (req) {
> +		fn(req);
> +		nvme_complete_batch_req(req);
> +		req = rq_list_next(req);
> +	}
> +
> +	blk_mq_end_request_batch(iob);

Can we turn this into a normal for loop?

	for (req = rq_list_peek(&iob->req_list); req; req = rq_list_next(req)) {
		..
	}

> +	if (!nvme_try_complete_req(req, cqe->status, cqe->result)) {
> +		/*
> +		 * Do normal inline completion if we don't have a batch
> +		 * list, if we have an end_io handler, or if the status of
> +		 * the request isn't just normal success.
> +		 */
> +		if (!iob || req->end_io || nvme_req(req)->status)
> +			nvme_pci_complete_rq(req);
> +		else
> +			rq_list_add_tail(&iob->req_list, req);
> +	}

The check for the conditions where we can or cannot batch complete
really should go into a block layer helper.  Something like the
incremental patch below:

diff --git a/drivers/nvme/host/pci.c b/drivers/nvme/host/pci.c
index ce69e9666caac..57bef8229bfab 100644
--- a/drivers/nvme/host/pci.c
+++ b/drivers/nvme/host/pci.c
@@ -1034,17 +1034,9 @@ static inline void nvme_handle_cqe(struct nvme_queue *nvmeq,
 	}
 
 	trace_nvme_sq(req, cqe->sq_head, nvmeq->sq_tail);
-	if (!nvme_try_complete_req(req, cqe->status, cqe->result)) {
-		/*
-		 * Do normal inline completion if we don't have a batch
-		 * list, if we have an end_io handler, or if the status of
-		 * the request isn't just normal success.
-		 */
-		if (!iob || req->end_io || nvme_req(req)->status)
-			nvme_pci_complete_rq(req);
-		else
-			rq_list_add_tail(&iob->req_list, req);
-	}
+	if (!nvme_try_complete_req(req, cqe->status, cqe->result) &&
+	    !blk_mq_add_to_batch(req, iob, nvme_req(req)->status))
+		nvme_pci_complete_rq(req);
 }
 
 static inline void nvme_update_cq_head(struct nvme_queue *nvmeq)
diff --git a/include/linux/blk-mq.h b/include/linux/blk-mq.h
index aea7d866a34c6..383d887e32f6d 100644
--- a/include/linux/blk-mq.h
+++ b/include/linux/blk-mq.h
@@ -773,6 +773,19 @@ void blk_mq_end_request(struct request *rq, blk_status_t error);
 void __blk_mq_end_request(struct request *rq, blk_status_t error);
 void blk_mq_end_request_batch(struct io_batch *ib);
 
+/*
+ * Batched completions only work when there is no I/O error and not special
+ * ->end_io handler.
+ */
+static inline bool blk_mq_add_to_batch(struct request *req,
+		 struct io_batch *iob, bool ioerror)
+{
+	if (!iob || req->end_io || ioerror)
+		return false;
+	rq_list_add_tail(&iob->req_list, req);
+	return true;
+}
+
 void blk_mq_requeue_request(struct request *rq, bool kick_requeue_list);
 void blk_mq_kick_requeue_list(struct request_queue *q);
 void blk_mq_delay_kick_requeue_list(struct request_queue *q, unsigned long msecs);

Re: [PATCH 6/9] nvme: add support for batched completion of polled IO

[Jens Axboe]

On 10/14/21 1:43 AM, Christoph Hellwig wrote:
> On Wed, Oct 13, 2021 at 10:54:13AM -0600, Jens Axboe wrote:
>> +void nvme_complete_batch_req(struct request *req)
>> +{
>> +	nvme_cleanup_cmd(req);
>> +	nvme_end_req_zoned(req);
>> +	req->status = BLK_STS_OK;
>> +}
>> +EXPORT_SYMBOL_GPL(nvme_complete_batch_req);
>> +
> 
> I'd be tempted to just merge this helper into the only caller.
> nvme_cleanup_cmd is exported anyway, so this would just add an export
> for nvme_end_req_zoned.

Sure, I can do that.

>> +static __always_inline void nvme_complete_batch(struct io_batch *iob,
>> +						void (*fn)(struct request *rq))
>> +{
>> +	struct request *req;
>> +
>> +	req = rq_list_peek(&iob->req_list);
>> +	while (req) {
>> +		fn(req);
>> +		nvme_complete_batch_req(req);
>> +		req = rq_list_next(req);
>> +	}
>> +
>> +	blk_mq_end_request_batch(iob);
> 
> Can we turn this into a normal for loop?
> 
> 	for (req = rq_list_peek(&iob->req_list); req; req = rq_list_next(req)) {
> 		..
> 	}

If you prefer it that way for nvme, for me the while () setup is much
easier to read than a really long for line.

>> +	if (!nvme_try_complete_req(req, cqe->status, cqe->result)) {
>> +		/*
>> +		 * Do normal inline completion if we don't have a batch
>> +		 * list, if we have an end_io handler, or if the status of
>> +		 * the request isn't just normal success.
>> +		 */
>> +		if (!iob || req->end_io || nvme_req(req)->status)
>> +			nvme_pci_complete_rq(req);
>> +		else
>> +			rq_list_add_tail(&iob->req_list, req);
>> +	}
> 
> The check for the conditions where we can or cannot batch complete
> really should go into a block layer helper.  Something like the
> incremental patch below:

That's a good idea, I'll add that.

-- 
Jens Axboe

Re: [PATCH 6/9] nvme: add support for batched completion of polled IO

[Jens Axboe]

On 10/14/21 9:30 AM, Jens Axboe wrote:
> On 10/14/21 1:43 AM, Christoph Hellwig wrote:
>> On Wed, Oct 13, 2021 at 10:54:13AM -0600, Jens Axboe wrote:
>>> +void nvme_complete_batch_req(struct request *req)
>>> +{
>>> +	nvme_cleanup_cmd(req);
>>> +	nvme_end_req_zoned(req);
>>> +	req->status = BLK_STS_OK;
>>> +}
>>> +EXPORT_SYMBOL_GPL(nvme_complete_batch_req);
>>> +
>>
>> I'd be tempted to just merge this helper into the only caller.
>> nvme_cleanup_cmd is exported anyway, so this would just add an export
>> for nvme_end_req_zoned.
> 
> Sure, I can do that.

That'll turn it into two calls from the batch completion though, so I
skipped this change.

-- 
Jens Axboe

Re: [PATCH 6/9] nvme: add support for batched completion of polled IO

[Christoph Hellwig]

On Thu, Oct 14, 2021 at 09:30:57AM -0600, Jens Axboe wrote:
> > Can we turn this into a normal for loop?
> > 
> > 	for (req = rq_list_peek(&iob->req_list); req; req = rq_list_next(req)) {
> > 		..
> > 	}
> 
> If you prefer it that way for nvme, for me the while () setup is much
> easier to read than a really long for line.

I prefer the loop over the while loop.  My real preference would be
a helper macro and do:

	for_each_rq(req, &iob->req_list) {

as suggested last round.

Re: [PATCH 6/9] nvme: add support for batched completion of polled IO

[Jens Axboe]

On 10/14/21 10:07 AM, Christoph Hellwig wrote:
> On Thu, Oct 14, 2021 at 09:30:57AM -0600, Jens Axboe wrote:
>>> Can we turn this into a normal for loop?
>>>
>>> 	for (req = rq_list_peek(&iob->req_list); req; req = rq_list_next(req)) {
>>> 		..
>>> 	}
>>
>> If you prefer it that way for nvme, for me the while () setup is much
>> easier to read than a really long for line.
> 
> I prefer the loop over the while loop.  My real preference would be
> a helper macro and do:
> 
> 	for_each_rq(req, &iob->req_list) {
> 
> as suggested last round.

Sure, I can turn it into a helper and use that.

-- 
Jens Axboe

[PATCH 7/9] block: assign batch completion handler in blk_poll()

[Jens Axboe]

If an io_batch is passed in to blk_poll(), we need to assign the batch
handler associated with this queue. This allows callers to complete
an io_batch handler on by calling it.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 block/blk-mq.c | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/block/blk-mq.c b/block/blk-mq.c
index d603703cf272..deafd444761d 100644
--- a/block/blk-mq.c
+++ b/block/blk-mq.c
@@ -4302,6 +4302,19 @@ static int blk_mq_poll_classic(struct request_queue *q, blk_qc_t cookie,
 
 	hctx->poll_considered++;
 
+	/*
+	 * If batching is requested but the target doesn't support batched
+	 * completions, then just clear ib and completions will be handled
+	 * normally.
+	 */
+	if (iob) {
+		iob->complete = q->mq_ops->complete_batch;
+		if (!iob->complete) {
+			WARN_ON_ONCE(iob->req_list);
+			iob = NULL;
+		}
+	}
+
 	do {
 		hctx->poll_invoked++;
 
-- 
2.33.0

Re: [PATCH 7/9] block: assign batch completion handler in blk_poll()

[Christoph Hellwig]

On Wed, Oct 13, 2021 at 10:54:14AM -0600, Jens Axboe wrote:
> If an io_batch is passed in to blk_poll(), we need to assign the batch
> handler associated with this queue. This allows callers to complete
> an io_batch handler on by calling it.

blk_poll() is gone now :)

> +	if (iob) {
> +		iob->complete = q->mq_ops->complete_batch;
> +		if (!iob->complete) {
> +			WARN_ON_ONCE(iob->req_list);
> +			iob = NULL;
> +		}
> +	}

Assigning first and checking later looks a little strange, even if it
makes no actual difference.  Why not:

	if (iob && q->mq_ops->complete_batch)
	 	iob->complete = q->mq_ops->complete_batch;
	else
		iob = NULL;

Also I'd expect this patch in the series right after the other block
patches.

Re: [PATCH 7/9] block: assign batch completion handler in blk_poll()

[Jens Axboe]

On 10/14/21 1:48 AM, Christoph Hellwig wrote:
> On Wed, Oct 13, 2021 at 10:54:14AM -0600, Jens Axboe wrote:
>> If an io_batch is passed in to blk_poll(), we need to assign the batch
>> handler associated with this queue. This allows callers to complete
>> an io_batch handler on by calling it.
> 
> blk_poll() is gone now :)
> 
>> +	if (iob) {
>> +		iob->complete = q->mq_ops->complete_batch;
>> +		if (!iob->complete) {
>> +			WARN_ON_ONCE(iob->req_list);
>> +			iob = NULL;
>> +		}
>> +	}
> 
> Assigning first and checking later looks a little strange, even if it
> makes no actual difference.  Why not:
> 
> 	if (iob && q->mq_ops->complete_batch)
> 	 	iob->complete = q->mq_ops->complete_batch;
> 	else
> 		iob = NULL;

Done.

> Also I'd expect this patch in the series right after the other block
> patches.

Moved.

-- 
Jens Axboe

[PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Jens Axboe]

Wire up using an io_batch for f_op->iopoll(). If the lower stack supports
it, we can handle high rates of polled IO more efficiently.

This raises the single core efficiency on my system from ~6.1M IOPS to
~6.6M IOPS running a random read workload at depth 128 on two gen2
Optane drives.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 fs/io_uring.c | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/fs/io_uring.c b/fs/io_uring.c
index 082ff64c1bcb..5c031ab8f77f 100644
--- a/fs/io_uring.c
+++ b/fs/io_uring.c
@@ -2390,6 +2390,8 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
 {
 	struct io_wq_work_node *pos, *start, *prev;
 	unsigned int poll_flags = BLK_POLL_NOSLEEP;
+	struct file *file = NULL;
+	DEFINE_IO_BATCH(iob);
 	int nr_events = 0;
 
 	/*
@@ -2404,6 +2406,11 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
 		struct kiocb *kiocb = &req->rw.kiocb;
 		int ret;
 
+		if (!file)
+			file = kiocb->ki_filp;
+		else if (file != kiocb->ki_filp)
+			break;
+
 		/*
 		 * Move completed and retryable entries to our local lists.
 		 * If we find a request that requires polling, break out
@@ -2412,19 +2419,21 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
 		if (READ_ONCE(req->iopoll_completed))
 			break;
 
-		ret = kiocb->ki_filp->f_op->iopoll(kiocb, NULL, poll_flags);
+		ret = kiocb->ki_filp->f_op->iopoll(kiocb, &iob, poll_flags);
 		if (unlikely(ret < 0))
 			return ret;
 		else if (ret)
 			poll_flags |= BLK_POLL_ONESHOT;
 
 		/* iopoll may have completed current req */
-		if (READ_ONCE(req->iopoll_completed))
+		if (iob.req_list || READ_ONCE(req->iopoll_completed))
 			break;
 	}
 
-	if (!pos)
+	if (!pos && !iob.req_list)
 		return 0;
+	if (iob.req_list)
+		iob.complete(&iob);
 
 	prev = start;
 	wq_list_for_each_resume(pos, prev) {
-- 
2.33.0

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Christoph Hellwig]

On Wed, Oct 13, 2021 at 10:54:15AM -0600, Jens Axboe wrote:
> @@ -2404,6 +2406,11 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
>  		struct kiocb *kiocb = &req->rw.kiocb;
>  		int ret;
>  
> +		if (!file)
> +			file = kiocb->ki_filp;
> +		else if (file != kiocb->ki_filp)
> +			break;
> +

Can you explain why we now can only poll for a single file (independent
of the fact that batching is used)?

> +	if (!pos && !iob.req_list)
>  		return 0;
> +	if (iob.req_list)
> +		iob.complete(&iob);

Why not:

	if (iob.req_list)
		iob.complete(&iob);
	else if (!pos)
		return 0;

?

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Jens Axboe]

On 10/14/21 2:03 AM, Christoph Hellwig wrote:
> On Wed, Oct 13, 2021 at 10:54:15AM -0600, Jens Axboe wrote:
>> @@ -2404,6 +2406,11 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
>>  		struct kiocb *kiocb = &req->rw.kiocb;
>>  		int ret;
>>  
>> +		if (!file)
>> +			file = kiocb->ki_filp;
>> +		else if (file != kiocb->ki_filp)
>> +			break;
>> +
> 
> Can you explain why we now can only poll for a single file (independent
> of the fact that batching is used)?

Different file may be on a different backend, it's just playing it
safe and splitting it up. In practice it should not matter.

>> +	if (!pos && !iob.req_list)
>>  		return 0;
>> +	if (iob.req_list)
>> +		iob.complete(&iob);
> 
> Why not:
> 
> 	if (iob.req_list)
> 		iob.complete(&iob);
> 	else if (!pos)
> 		return 0;

That is cleaner, will adapt. Thanks!

-- 
Jens Axboe

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Christoph Hellwig]

On Thu, Oct 14, 2021 at 09:45:38AM -0600, Jens Axboe wrote:
> On 10/14/21 2:03 AM, Christoph Hellwig wrote:
> > On Wed, Oct 13, 2021 at 10:54:15AM -0600, Jens Axboe wrote:
> >> @@ -2404,6 +2406,11 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
> >>  		struct kiocb *kiocb = &req->rw.kiocb;
> >>  		int ret;
> >>  
> >> +		if (!file)
> >> +			file = kiocb->ki_filp;
> >> +		else if (file != kiocb->ki_filp)
> >> +			break;
> >> +
> > 
> > Can you explain why we now can only poll for a single file (independent
> > of the fact that batching is used)?
> 
> Different file may be on a different backend, it's just playing it
> safe and splitting it up. In practice it should not matter.

Well, with file systems even the same file can land on different
devices.  Maybe we need a cookie?

Either way this should be commented as right now it looks pretty
arbitrary.

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Jens Axboe]

On 10/14/21 10:08 AM, Christoph Hellwig wrote:
> On Thu, Oct 14, 2021 at 09:45:38AM -0600, Jens Axboe wrote:
>> On 10/14/21 2:03 AM, Christoph Hellwig wrote:
>>> On Wed, Oct 13, 2021 at 10:54:15AM -0600, Jens Axboe wrote:
>>>> @@ -2404,6 +2406,11 @@ static int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin)
>>>>  		struct kiocb *kiocb = &req->rw.kiocb;
>>>>  		int ret;
>>>>  
>>>> +		if (!file)
>>>> +			file = kiocb->ki_filp;
>>>> +		else if (file != kiocb->ki_filp)
>>>> +			break;
>>>> +
>>>
>>> Can you explain why we now can only poll for a single file (independent
>>> of the fact that batching is used)?
>>
>> Different file may be on a different backend, it's just playing it
>> safe and splitting it up. In practice it should not matter.
> 
> Well, with file systems even the same file can land on different
> devices.  Maybe we need a cookie?
> 
> Either way this should be commented as right now it looks pretty
> arbitrary.

I got rid of this and added a dev_id kind of cookie that gets matched
on batched addition.

-- 
Jens Axboe

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Christoph Hellwig]

On Thu, Oct 14, 2021 at 12:14:19PM -0600, Jens Axboe wrote:
> > Either way this should be commented as right now it looks pretty
> > arbitrary.
> 
> I got rid of this and added a dev_id kind of cookie that gets matched
> on batched addition.

Thinking about this a bit more.  I don't think we need to differenciate
devices, just complete handlers.  That is everything using the same
->complete_batch handler could go onto the same list.  Which means
different nvme devices can be polled together.  So I think we can
just check the complete/complete_batch handler and be done with it.
The big benefit besides saving a field is that this is pretty much
self-explaining.

Re: [PATCH 8/9] io_uring: utilize the io_batch infrastructure for more efficient polled IO

[Jens Axboe]

On 10/15/21 10:29 PM, Christoph Hellwig wrote:
> On Thu, Oct 14, 2021 at 12:14:19PM -0600, Jens Axboe wrote:
>>> Either way this should be commented as right now it looks pretty
>>> arbitrary.
>>
>> I got rid of this and added a dev_id kind of cookie that gets matched
>> on batched addition.
> 
> Thinking about this a bit more.  I don't think we need to differenciate
> devices, just complete handlers.  That is everything using the same
> ->complete_batch handler could go onto the same list.  Which means
> different nvme devices can be polled together.  So I think we can
> just check the complete/complete_batch handler and be done with it.
> The big benefit besides saving a field is that this is pretty much
> self-explaining.

Good idea, cleaner than a dev_id cookie too. I'll be sending out the
revised series soonish, but I incorporated that change.

-- 
Jens Axboe

[PATCH 9/9] nvme: wire up completion batching for the IRQ path

[Jens Axboe]

Trivial to do now, just need our own io_batch on the stack and pass that
in to the usual command completion handling.

I pondered making this dependent on how many entries we had to process,
but even for a single entry there's no discernable difference in
performance or latency. Running a sync workload over io_uring:

t/io_uring -b512 -d1 -s1 -c1 -p0 -F1 -B1 -n2 /dev/nvme1n1 /dev/nvme2n1

yields the below performance before the patch:

IOPS=254820, BW=124MiB/s, IOS/call=1/1, inflight=(1 1)
IOPS=251174, BW=122MiB/s, IOS/call=1/1, inflight=(1 1)
IOPS=250806, BW=122MiB/s, IOS/call=1/1, inflight=(1 1)

and the following after:

IOPS=255972, BW=124MiB/s, IOS/call=1/1, inflight=(1 1)
IOPS=251920, BW=123MiB/s, IOS/call=1/1, inflight=(1 1)
IOPS=251794, BW=122MiB/s, IOS/call=1/1, inflight=(1 1)

which definitely isn't slower, about the same if you factor in a bit of
variance. For peak performance workloads, benchmarking shows a 2%
improvement.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 drivers/nvme/host/pci.c | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/drivers/nvme/host/pci.c b/drivers/nvme/host/pci.c
index ae253f6f5c80..061f0b1cb0ec 100644
--- a/drivers/nvme/host/pci.c
+++ b/drivers/nvme/host/pci.c
@@ -1061,6 +1061,7 @@ static inline void nvme_update_cq_head(struct nvme_queue *nvmeq)
 
 static inline int nvme_process_cq(struct nvme_queue *nvmeq)
 {
+	DEFINE_IO_BATCH(iob);
 	int found = 0;
 
 	while (nvme_cqe_pending(nvmeq)) {
@@ -1070,12 +1071,15 @@ static inline int nvme_process_cq(struct nvme_queue *nvmeq)
 		 * the cqe requires a full read memory barrier
 		 */
 		dma_rmb();
-		nvme_handle_cqe(nvmeq, NULL, nvmeq->cq_head);
+		nvme_handle_cqe(nvmeq, &iob, nvmeq->cq_head);
 		nvme_update_cq_head(nvmeq);
 	}
 
-	if (found)
+	if (found) {
+		if (iob.req_list)
+			nvme_pci_complete_batch(&iob);
 		nvme_ring_cq_doorbell(nvmeq);
+	}
 	return found;
 }
 
-- 
2.33.0

Re: [PATCH 9/9] nvme: wire up completion batching for the IRQ path

[Christoph Hellwig]

Couldn't we do something like the incremental patch below to avoid
duplicating the CQ processing? With that this patch could probably go
away entirely and we could fold the change into the other nvme patch.

diff --git a/drivers/nvme/host/pci.c b/drivers/nvme/host/pci.c
index 061f0b1cb0ec3..ce69e9666caac 100644
--- a/drivers/nvme/host/pci.c
+++ b/drivers/nvme/host/pci.c
@@ -1059,9 +1059,8 @@ static inline void nvme_update_cq_head(struct nvme_queue *nvmeq)
 	}
 }
 
-static inline int nvme_process_cq(struct nvme_queue *nvmeq)
+static inline int nvme_poll_cq(struct nvme_queue *nvmeq, struct io_batch *iob)
 {
-	DEFINE_IO_BATCH(iob);
 	int found = 0;
 
 	while (nvme_cqe_pending(nvmeq)) {
@@ -1071,15 +1070,23 @@ static inline int nvme_process_cq(struct nvme_queue *nvmeq)
 		 * the cqe requires a full read memory barrier
 		 */
 		dma_rmb();
-		nvme_handle_cqe(nvmeq, &iob, nvmeq->cq_head);
+		nvme_handle_cqe(nvmeq, iob, nvmeq->cq_head);
 		nvme_update_cq_head(nvmeq);
 	}
 
-	if (found) {
-		if (iob.req_list)
-			nvme_pci_complete_batch(&iob);
+	if (found)
 		nvme_ring_cq_doorbell(nvmeq);
-	}
+	return found;
+}
+
+static inline int nvme_process_cq(struct nvme_queue *nvmeq)
+{
+	DEFINE_IO_BATCH(iob);
+	int found;
+
+	found = nvme_poll_cq(nvmeq, &iob);
+	if (iob.req_list)
+		nvme_pci_complete_batch(&iob);
 	return found;
 }
 
@@ -1116,27 +1123,6 @@ static void nvme_poll_irqdisable(struct nvme_queue *nvmeq)
 	enable_irq(pci_irq_vector(pdev, nvmeq->cq_vector));
 }
 
-static inline int nvme_poll_cq(struct nvme_queue *nvmeq, struct io_batch *iob)
-{
-	int found = 0;
-
-	while (nvme_cqe_pending(nvmeq)) {
-		found++;
-		/*
-		 * load-load control dependency between phase and the rest of
-		 * the cqe requires a full read memory barrier
-		 */
-		dma_rmb();
-		nvme_handle_cqe(nvmeq, iob, nvmeq->cq_head);
-		nvme_update_cq_head(nvmeq);
-	}
-
-	if (found)
-		nvme_ring_cq_doorbell(nvmeq);
-	return found;
-}
-
-
 static int nvme_poll(struct blk_mq_hw_ctx *hctx, struct io_batch *iob)
 {
 	struct nvme_queue *nvmeq = hctx->driver_data;

Re: [PATCH 9/9] nvme: wire up completion batching for the IRQ path

[Jens Axboe]

On 10/14/21 1:53 AM, Christoph Hellwig wrote:
> Couldn't we do something like the incremental patch below to avoid
> duplicating the CQ processing? With that this patch could probably go
> away entirely and we could fold the change into the other nvme patch.

Yeah should probably go into the first patch in some shape or form, I'll
take a look.

-- 
Jens Axboe