Re: [PATCH V4 12/27] ublk: add io events fifo structure

Caleb Sander Mateos
2025-11-30T16:53:15Z

On Thu, Nov 20, 2025 at 5:59 PM Ming Lei <ming.lei@redhat.com> wrote:
>
> Add ublk io events fifo structure and prepare for supporting command
> batch, which will use io_uring multishot uring_cmd for fetching one
> batch of io commands each time.
>
> One nice feature of kfifo is to allow multiple producer vs single
> consumer. We just need lock the producer side, meantime the single
> consumer can be lockless.
>
> The producer is actually from ublk_queue_rq() or ublk_queue_rqs(), so
> lock contention can be eased by setting proper blk-mq nr_queues.
>
> Signed-off-by: Ming Lei <ming.lei@redhat.com>
> ---
> drivers/block/ublk_drv.c | 65 ++++++++++++++++++++++++++++++++++++----
> 1 file changed, 60 insertions(+), 5 deletions(-)
>
> diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
> index ea992366af5b..6ff284243630 100644
> --- a/drivers/block/ublk_drv.c
> +++ b/drivers/block/ublk_drv.c
> @@ -44,6 +44,7 @@
> #include <linux/task_work.h>
> #include <linux/namei.h>
> #include <linux/kref.h>
> +#include <linux/kfifo.h>
> #include <uapi/linux/ublk_cmd.h>
>
> #define UBLK_MINORS (1U << MINORBITS)
> @@ -217,6 +218,22 @@ struct ublk_queue {
> bool fail_io; /* copy of dev->state == UBLK_S_DEV_FAIL_IO */
> spinlock_t cancel_lock;
> struct ublk_device *dev;
> +
> + /*
> + * Inflight ublk request tag is saved in this fifo
> + *
> + * There are multiple writer from ublk_queue_rq() or ublk_queue_rqs(),
> + * so lock is required for storing request tag to fifo
> + *
> + * Make sure just one reader for fetching request from task work
> + * function to ublk server, so no need to grab the lock in reader
> + * side.

Can you clarify that this is only used for batch mode?

> + */
> + struct {
> + DECLARE_KFIFO_PTR(evts_fifo, unsigned short);
> + spinlock_t evts_lock;
> + }____cacheline_aligned_in_smp;
> +
> struct ublk_io ios[] __counted_by(q_depth);
> };
>
> @@ -282,6 +299,32 @@ static inline void ublk_io_unlock(struct ublk_io *io)
> spin_unlock(&io->lock);
> }
>
> +/* Initialize the queue */

"queue" -> "events queue"? Otherwise, it sounds like it's referring to
struct ublk_queue.

> +static inline int ublk_io_evts_init(struct ublk_queue *q, unsigned int size,
> + int numa_node)
> +{
> + spin_lock_init(&q->evts_lock);
> + return kfifo_alloc_node(&q->evts_fifo, size, GFP_KERNEL, numa_node);
> +}
> +
> +/* Check if queue is empty */
> +static inline bool ublk_io_evts_empty(const struct ublk_queue *q)
> +{
> + return kfifo_is_empty(&q->evts_fifo);
> +}
> +
> +/* Check if queue is full */
> +static inline bool ublk_io_evts_full(const struct ublk_queue *q)

Function is unused?

> +{
> + return kfifo_is_full(&q->evts_fifo);
> +}
> +
> +static inline void ublk_io_evts_deinit(struct ublk_queue *q)
> +{
> + WARN_ON_ONCE(!kfifo_is_empty(&q->evts_fifo));
> + kfifo_free(&q->evts_fifo);
> +}
> +
> static inline struct ublksrv_io_desc *
> ublk_get_iod(const struct ublk_queue *ubq, unsigned tag)
> {
> @@ -3038,6 +3081,9 @@ static void ublk_deinit_queue(struct ublk_device *ub, int q_id)
> if (ubq->io_cmd_buf)
> free_pages((unsigned long)ubq->io_cmd_buf, get_order(size));
>
> + if (ublk_dev_support_batch_io(ub))
> + ublk_io_evts_deinit(ubq);
> +
> kvfree(ubq);
> ub->queues[q_id] = NULL;
> }
> @@ -3062,7 +3108,7 @@ static int ublk_init_queue(struct ublk_device *ub, int q_id)
> struct ublk_queue *ubq;
> struct page *page;
> int numa_node;
> - int size, i;
> + int size, i, ret = -ENOMEM;
>
> /* Determine NUMA node based on queue's CPU affinity */
> numa_node = ublk_get_queue_numa_node(ub, q_id);
> @@ -3081,18 +3127,27 @@ static int ublk_init_queue(struct ublk_device *ub, int q_id)
>
> /* Allocate I/O command buffer on local NUMA node */
> page = alloc_pages_node(numa_node, gfp_flags, get_order(size));
> - if (!page) {
> - kvfree(ubq);
> - return -ENOMEM;
> - }
> + if (!page)
> + goto fail_nomem;
> ubq->io_cmd_buf = page_address(page);
>
> for (i = 0; i < ubq->q_depth; i++)
> spin_lock_init(&ubq->ios[i].lock);
>
> + if (ublk_dev_support_batch_io(ub)) {
> + ret = ublk_io_evts_init(ubq, ubq->q_depth, numa_node);
> + if (ret)
> + goto fail;
> + }
> ub->queues[q_id] = ubq;
> ubq->dev = ub;
> +
> return 0;
> +fail:
> + ublk_deinit_queue(ub, q_id);

This is a no-op since ub->queues[q_id] hasn't been assigned yet?

Best,
Caleb

> +fail_nomem:
> + kvfree(ubq);
> + return ret;
> }
>
> static void ublk_deinit_queues(struct ublk_device *ub)
> --
> 2.47.0
>