[PATCH 0/8] ublk: don't dispatch to canceled io commands

[PATCH 0/8] ublk: don't dispatch to canceled io commands

[Ming Lei]

Hi,

Josef reported three ways in which ublk dispatches a request to a
canceled io command, which oopses on the NULL io->cmd in
ublk_queue_cmd() [1]:

1) STOP_DEV, then START_DEV
2) partial FETCH: a task fetches some tags and exits, then another task
   fetches the rest
3) user recovery: one queue's task exits before all queues are ready

This series fixes them in a simpler way, without extra locking in the
I/O path:

- patch 1: a FETCH round which saw a cancel stays canceling until the
  server is gone (2, 3)
- patch 2-3: an io command is marked cancelable before it is published
- patch 4-6: the release resets the FETCH round, also without a disk
  and under ub->mutex; STOP_DEV holds a reference on /dev/ublkcN while
  it cancels, so no START_DEV, FETCH or new server can slip in (1)
- patch 7-8: selftest

Tested with all ublk selftests, plus KASAN and lockdep.

[1] https://lore.kernel.org/linux-block/20260928-b4-ublk-cancel-stop-v1-0-4a4360232a46@toxicpanda.com/

Thanks,
Ming

Ming Lei (8):
  ublk: keep a canceled FETCH round canceling until the server is gone
  ublk: mark the io command cancelable before publishing it
  ublk: mark the batch fetch command cancelable before linking it
  ublk: reset the FETCH round in release also without a disk
  ublk: reset the FETCH round under ub->mutex
  ublk: let STOP_DEV cancel the server's commands before its release
  selftests: ublk: move the control command helpers into ctrl.c
  selftests: ublk: add test for going live over canceled io commands

 drivers/block/ublk_drv.c                      | 206 +++-
 tools/testing/selftests/ublk/.gitignore       |   1 +
 tools/testing/selftests/ublk/Makefile         |   6 +-
 tools/testing/selftests/ublk/ctrl.c           | 268 +++++
 tools/testing/selftests/ublk/kublk.c          | 271 -----
 tools/testing/selftests/ublk/kublk.h          |  32 +
 .../testing/selftests/ublk/test_generic_18.sh |  37 +
 .../selftests/ublk/ublk_cancel_ready.c        | 984 ++++++++++++++++++
 8 files changed, 1484 insertions(+), 321 deletions(-)
 create mode 100644 tools/testing/selftests/ublk/ctrl.c
 create mode 100755 tools/testing/selftests/ublk/test_generic_18.sh
 create mode 100644 tools/testing/selftests/ublk/ublk_cancel_ready.c

-- 
2.55.0

[PATCH 1/8] ublk: keep a canceled FETCH round canceling until the server is gone

[Ming Lei]

A cancel completes a fetched io command and sets io->cmd to NULL, but
the io still counts as ready. Only ubq->canceling stops ublk_queue_rq()
from using the NULL io->cmd. Two paths clear that flag too early:

  partial FETCH round                 recovery, two queues
  -------------------                 --------------------
  task A: FETCH tags 0-2              q0 ready: q0->canceling = false
  task A exits:                       q0 task exits:
    start_cancel(): mark queues         start_cancel(): ub->canceling
    tags 0-2: io->cmd = NULL              is still set -> q0 not marked
  task B: FETCH tag 3                   q0: io->cmd = NULL
    queue ready: canceling = false    q1 ready
  START_DEV                           END_USER_RECOVERY
  read -> ublk_queue_cmd(NULL)        read on q0 -> ublk_queue_cmd(NULL)

A FETCH round runs from the open of /dev/ublkcN until
ublk_reset_ch_dev() resets the queues. A canceled command can't be
fetched again in the same round. So the ready check only needs one
fact: did this round see a cancel? ub->canceling already records it:
ublk_start_cancel() sets it under cancel_mutex before any command is
taken. It is only cleared too early, when all queues are ready.

Fix:

 - clear ub->canceling only in ublk_reset_ch_dev()
 - when a queue gets ready, clear ubq->canceling only if ub->canceling
   is not set, and check it under cancel_mutex

The check and the marking are ordered by cancel_mutex:

  check first:   queue cleared -> start_cancel() marks it again
                 -> command taken
  cancel first:  check sees ub->canceling -> queue stays canceling

While ub->canceling is set, no queue clears its flag. So the "all
queues marked" shortcut in ublk_start_cancel() is correct again.

UBLK_F_BATCH_IO follows the same rule. It does not dispatch through
io->cmd, so it never hit the oops, but a round which lost a fetch
command is treated the same way: it has to start over.

Fixes: 728cbac5fe21 ("ublk: move device reset into ublk_ch_release()")
Fixes: 3f3850785594 ("ublk: fix batch I/O recovery -ENODEV error")
Cc: stable@vger.kernel.org # v6.17+: needs cancel_mutex
Reported-by: Josef Bacik <josef@toxicpanda.com>
Closes: https://lore.kernel.org/linux-block/20260928-b4-ublk-cancel-stop-v1-0-4a4360232a46@toxicpanda.com/
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 54 ++++++++++++++++++++++++----------------
 1 file changed, 33 insertions(+), 21 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 66eb55e7162e..38ed7d0e3979 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -334,6 +334,11 @@ struct ublk_device {
 	u16			nr_queue_ready;
 	bool 			unprivileged_daemons;
 	struct mutex cancel_mutex;
+	/*
+	 * A cancel started in this FETCH round. Set by ublk_set_canceling(),
+	 * cleared only by ublk_reset_ch_dev() when a new round starts. While
+	 * it is set, no queue clears its ->canceling.
+	 */
 	bool canceling;
 	pid_t 	ublksrv_tgid;
 	struct delayed_work	exit_work;
@@ -2415,6 +2420,11 @@ static void ublk_reset_ch_dev(struct ublk_device *ub)
 		spin_unlock(&ubq->cancel_lock);
 	}
 
+	/* a new FETCH round starts, the queues stay canceling until ready */
+	mutex_lock(&ub->cancel_mutex);
+	ub->canceling = false;
+	mutex_unlock(&ub->cancel_mutex);
+
 	/* set to NULL, otherwise new tasks cannot mmap io_cmd_buf */
 	ub->mm = NULL;
 	ub->nr_queue_ready = 0;
@@ -3024,11 +3034,19 @@ static void ublk_reset_io_flags(struct ublk_queue *ubq, struct ublk_io *io)
 }
 
 /* reset per-queue io flags */
-static void ublk_queue_reset_io_flags(struct ublk_queue *ubq)
+static void ublk_queue_reset_io_flags(struct ublk_device *ub,
+				      struct ublk_queue *ubq)
 {
-	spin_lock(&ubq->cancel_lock);
-	ubq->canceling = false;
-	spin_unlock(&ubq->cancel_lock);
+	/*
+	 * A cancel in this FETCH round took a command which still counts as
+	 * ready, so the queue has to stay canceling. ub->canceling is set
+	 * under cancel_mutex before any command is taken: either we see it
+	 * here, or the cancel marks this queue again later.
+	 */
+	mutex_lock(&ub->cancel_mutex);
+	if (!ub->canceling)
+		ubq->canceling = false;
+	mutex_unlock(&ub->cancel_mutex);
 	ubq->fail_io = false;
 	ubq->force_abort = false;
 }
@@ -3051,24 +3069,17 @@ static void ublk_mark_io_ready(struct ublk_device *ub, u16 q_id,
 		ub->nr_queue_ready++;
 
 		/*
-		 * Reset queue flags as soon as this queue is ready.
-		 * This clears the canceling flag, allowing batch FETCH commands
-		 * to succeed during recovery without waiting for all queues.
+		 * Reset queue flags as soon as this queue is ready. Unless
+		 * this round saw a cancel, this clears the canceling flag,
+		 * allowing batch FETCH commands to succeed during recovery
+		 * without waiting for all queues.
 		 */
-		ublk_queue_reset_io_flags(ubq);
+		ublk_queue_reset_io_flags(ub, ubq);
 	}
 
-	/* Check if all queues are ready */
-	if (ublk_dev_ready(ub)) {
-		/*
-		 * All queues ready - clear device-level canceling flag
-		 * and wake ublk_dev_ready() waiters.
-		 */
-		mutex_lock(&ub->cancel_mutex);
-		ub->canceling = false;
-		mutex_unlock(&ub->cancel_mutex);
+	/* All queues ready - wake ublk_dev_ready() waiters */
+	if (ublk_dev_ready(ub))
 		wake_up_var(&ub->nr_queue_ready);
-	}
 }
 
 static inline int ublk_check_cmd_op(u32 cmd_op)
@@ -4433,9 +4444,10 @@ static bool ublk_validate_user_pid(struct ublk_device *ub, pid_t ublksrv_pid)
 
 /*
  * Wait until all queues have fetched their I/O commands, and return with
- * ub->mutex held and readiness guaranteed: then every queue's ->canceling
- * is cleared. Ready may regress between wakeup and mutex_lock() (F_BATCH
- * UNPREP, daemon death), so re-check it under the mutex and wait again.
+ * ub->mutex held and readiness guaranteed. The queues stay canceling if
+ * this round saw a cancel, see ublk_queue_reset_io_flags(). Ready may
+ * regress between wakeup and mutex_lock() (F_BATCH UNPREP, daemon death),
+ * so re-check it under the mutex and wait again.
  */
 static int ublk_wait_dev_ready_and_lock(struct ublk_device *ub)
 {
-- 
2.55.0

[PATCH 2/8] ublk: mark the io command cancelable before publishing it

[Ming Lei]

FETCH, COMMIT_AND_FETCH and NEED_GET_DATA publish the command in io->cmd
before marking it cancelable. A cancel from the control path (STOP_DEV,
QUIESCE_DEV) can complete it in between:

  issue path                       control-path cancel
  io->cmd = C
                                   take C, io_uring_cmd_done(C):
                                     C not marked, nothing to unlink
  ublk_prep_cancel(C)
    the completed C is linked on the cancelable list
  ring exit: the cancel walk hits it → GPF (KASAN, with a delay added)

Mark the command before ublk_fill_io_cmd() publishes it. The handlers
run with uring_lock held, as io_uring's cancel walk does, so marking
takes no lock and the walk can't see a marked command before it is
published. smp_wmb() orders the mark before the io->cmd store; the
cancel loads cmd with READ_ONCE(io->cmd) before reading cmd->flags.

A marked command has to be completed by io_uring_cmd_done(), so a failed
FETCH and the inline UBLK_IO_RES_OK of NEED_GET_DATA now do that. The CQE
is the same.

Fixes: 216c8f5ef0f2 ("ublk: replace monitor with cancelable uring_cmd")
Cc: stable@vger.kernel.org
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 27 +++++++++++++++++++++------
 1 file changed, 21 insertions(+), 6 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 38ed7d0e3979..6015fb2fb925 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2800,7 +2800,8 @@ static void ublk_cancel_cmd(struct ublk_queue *ubq, u16 tag,
 	done = !!(io->flags & UBLK_IO_FLAG_CANCELED);
 	if (!done) {
 		io->flags |= UBLK_IO_FLAG_CANCELED;
-		cmd = io->cmd;
+		/* dependency ordered against smp_wmb() in ublk_prep_cancel() */
+		cmd = READ_ONCE(io->cmd);
 		io->cmd = NULL;
 	}
 	spin_unlock(&ubq->cancel_lock);
@@ -3163,6 +3164,12 @@ ublk_fill_io_cmd(struct ublk_io *io, struct io_uring_cmd *cmd)
 	return req;
 }
 
+/*
+ * Call before ublk_fill_io_cmd() publishes @cmd in io->cmd: a control-path
+ * cancel may complete any command found there, and io_uring_cmd_done() only
+ * takes it off the cancelable list if it is marked already. The handlers
+ * hold uring_lock, so marking takes no lock.
+ */
 static inline void ublk_prep_cancel(struct io_uring_cmd *cmd,
 				    unsigned int issue_flags,
 				    struct ublk_queue *ubq, u16 tag)
@@ -3176,6 +3183,8 @@ static inline void ublk_prep_cancel(struct io_uring_cmd *cmd,
 	pdu->ubq = ubq;
 	pdu->tag = tag;
 	io_uring_cmd_mark_cancelable(cmd, issue_flags);
+	/* pairs with the cancel loading cmd from io->cmd, then cmd->flags */
+	smp_wmb();
 }
 
 static void ublk_io_release(void *priv)
@@ -3423,11 +3432,11 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 		ret = ublk_check_fetch_buf(ub, addr);
 		if (ret)
 			goto out;
+		/* before ublk_fetch() publishes io->cmd, see ublk_prep_cancel() */
+		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
 		ret = ublk_fetch(cmd, ub, io, addr, q_id);
 		if (ret)
-			goto out;
-
-		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
+			goto out_done;
 		return -EIOCBQUEUED;
 	}
 
@@ -3471,6 +3480,7 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 		if (ret)
 			goto out;
 		io->res = result;
+		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
 		req = ublk_fill_io_cmd(io, cmd);
 		ublk_apply_io_buf(ub, io, cmd, addr, &auto_buf, &buf_idx);
 		if (buf_idx != UBLK_INVALID_BUF_IDX)
@@ -3489,19 +3499,24 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 		 * uring_cmd active first and prepare for handling new requeued
 		 * request
 		 */
+		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
 		req = ublk_fill_io_cmd(io, cmd);
 		io->buf.addr = addr;
 		if (likely(ublk_get_data(ubq, io, req))) {
 			__ublk_prep_compl_io_cmd(io, req);
-			return UBLK_IO_RES_OK;
+			ret = UBLK_IO_RES_OK;
+			goto out_done;
 		}
 		break;
 	default:
 		goto out;
 	}
-	ublk_prep_cancel(cmd, issue_flags, ubq, tag);
 	return -EIOCBQUEUED;
 
+ out_done:
+	/* marked cancelable: complete through io_uring_cmd_done() */
+	io_uring_cmd_done(cmd, ret, issue_flags);
+	return -EIOCBQUEUED;
  out:
 	pr_devel("%s: complete: cmd op %d, tag %d ret %x io_flags %x\n",
 			__func__, cmd_op, tag, ret, io ? io->flags : 0);
-- 
2.55.0

[PATCH 3/8] ublk: mark the batch fetch command cancelable before linking it

[Ming Lei]

UBLK_U_IO_FETCH_IO_CMDS has the same order problem as the per-io
commands: ublk_batch_attach() links the fetch command into fcmd_head
and marks it cancelable only after dropping evts_lock. A cancel from
the control path (ublk_batch_cancel_queue()) can take it in between and
complete it, and the later mark puts a completed request on io_uring's
cancelable list.

Mark it before linking it. evts_lock orders the mark before the link,
which is where the cancel finds it.

Batch commands are not bounced to task work, so they can run from io-wq
without uring_lock, and io_uring's cancel walk can now find a fetch
command before it is linked. Two things make that safe:

 - initialize fcmd->node: it came from kzalloc(), so list_empty() saw a
   linked node and ublk_batch_cancel_cmd() would list_del_init() NULL
   pointers
 - on the -ENODEV path, complete the command with io_uring_cmd_done()
   before freeing fcmd: that removes it from the cancelable list under
   uring_lock, after which nothing can see fcmd

Also use data->cmd instead of fcmd->cmd after dropping evts_lock: once
fcmd is linked and not the active one, a control-path cancel may
complete and free it.

Fixes: a4d883755399 ("ublk: add UBLK_U_IO_FETCH_IO_CMDS for batch I/O processing")
Cc: stable@vger.kernel.org # v7.0+
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 24 ++++++++++++++++++------
 1 file changed, 18 insertions(+), 6 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 6015fb2fb925..5e37b8e9d9ac 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -815,6 +815,8 @@ ublk_batch_alloc_fcmd(struct io_uring_cmd *cmd)
 	if (fcmd) {
 		fcmd->cmd = cmd;
 		fcmd->buf_group = READ_ONCE(cmd->sqe->buf_index);
+		/* a cancel may look at it before it is linked */
+		INIT_LIST_HEAD(&fcmd->node);
 	}
 	return fcmd;
 }
@@ -3915,6 +3917,15 @@ static int ublk_batch_attach(struct ublk_queue *ubq,
 	bool free = false;
 	struct ublk_uring_cmd_pdu *pdu = ublk_get_uring_cmd_pdu(data->cmd);
 
+	/*
+	 * Mark it cancelable before linking it into fcmd_head, where a cancel
+	 * from the control path can take and complete it: see
+	 * ublk_prep_cancel(). evts_lock orders the mark before the link.
+	 */
+	pdu->ubq = ubq;
+	pdu->fcmd = fcmd;
+	io_uring_cmd_mark_cancelable(fcmd->cmd, data->issue_flags);
+
 	spin_lock(&ubq->evts_lock);
 	if (unlikely(ubq->force_abort || ubq->canceling)) {
 		free = true;
@@ -3925,14 +3936,12 @@ static int ublk_batch_attach(struct ublk_queue *ubq,
 	spin_unlock(&ubq->evts_lock);
 
 	if (unlikely(free)) {
+		/* off the cancelable list first, then nothing can see fcmd */
+		io_uring_cmd_done(data->cmd, -ENODEV, data->issue_flags);
 		ublk_batch_free_fcmd(fcmd);
-		return -ENODEV;
+		return -EIOCBQUEUED;
 	}
 
-	pdu->ubq = ubq;
-	pdu->fcmd = fcmd;
-	io_uring_cmd_mark_cancelable(fcmd->cmd, data->issue_flags);
-
 	if (!new_fcmd)
 		goto out;
 
@@ -3940,9 +3949,12 @@ static int ublk_batch_attach(struct ublk_queue *ubq,
 	 * If the two fetch commands are originated from same io_ring_ctx,
 	 * run batch dispatch directly. Otherwise, schedule task work for
 	 * doing it.
+	 *
+	 * Use data->cmd, not fcmd->cmd: once fcmd is linked and not active,
+	 * a cancel from the control path may complete and free it.
 	 */
 	if (io_uring_cmd_ctx_handle(new_fcmd->cmd) ==
-			io_uring_cmd_ctx_handle(fcmd->cmd)) {
+			io_uring_cmd_ctx_handle(data->cmd)) {
 		data->cmd = new_fcmd->cmd;
 		ublk_batch_dispatch(ubq, data, new_fcmd);
 	} else {
-- 
2.55.0

[PATCH 4/8] ublk: reset the FETCH round in release also without a disk

[Ming Lei]

The release of /dev/ublkcN resets the queues only when the device has a
disk. Without one it skips ublk_reset_ch_dev():

  server fetches, never starts;         old server exits
  or STOP_DEV removed the disk           release: no disk → skip reset
                                         ios stay ACTIVE / CANCELED,
                                         nr_io_ready stays full
  new server opens /dev/ublkcN           FETCH: -EBUSY (still ready), or
                                         -EINVAL after a partial round
                                         → the device can only be deleted

All uring_cmds are done when the release work runs, with or without a
disk, so the reset is safe either way. Do it in both cases. Then a new
server can fetch and start the device again, as the STOP_DEV entry in
Documentation/block/ublk.rst says ("ublk device is ready for the new
process").

Fixes: 82a8a30c581b ("ublk: improve detection and handling of ublk server exit")
Cc: stable@vger.kernel.org
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 5e37b8e9d9ac..8bf0739539c0 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2551,12 +2551,13 @@ static void ublk_ch_release_work_fn(struct work_struct *work)
 	}
 
 	/*
-	 * disk isn't attached yet, either device isn't live, or it has
-	 * been removed already, so we needn't to do anything
+	 * No disk: the device isn't live, or it has been removed already.
+	 * There are no requests to abort, but the round still has to be
+	 * reset, so that a new server can fetch and start the device.
 	 */
 	disk = ublk_get_disk(ub);
 	if (!disk)
-		goto out;
+		goto reset;
 
 	/*
 	 * All uring_cmd are done now, so abort any request outstanding to
@@ -2617,10 +2618,9 @@ static void ublk_ch_release_work_fn(struct work_struct *work)
 unlock:
 	mutex_unlock(&ub->mutex);
 	ublk_put_disk(disk);
-
+reset:
 	/* all uring_cmd has been done now, reset device & ubq */
 	ublk_reset_ch_dev(ub);
-out:
 	clear_bit(UB_STATE_OPEN, &ub->state);
 
 	/* put the reference grabbed in ublk_ch_release() */
-- 
2.55.0

[PATCH 5/8] ublk: reset the FETCH round under ub->mutex

[Ming Lei]

The release work resets the FETCH round without ub->mutex, while
START_DEV checks it under ub->mutex:

  START_DEV (ub->mutex)               release work
  all queues ready? yes
  ublksrv_tgid matches? yes
                                      ublk_reset_ch_dev():
                                        io->cmd = NULL, nr_queue_ready = 0
  go live: a request then reaches ublk_queue_cmd() with a NULL io->cmd

Do the reset under ub->mutex. Then START_DEV sees the round either ready
or reset. The work already takes ub->mutex when there is a disk, after
it has aborted the requests; nothing holding ub->mutex waits for it.

Fixes: 728cbac5fe21 ("ublk: move device reset into ublk_ch_release()")
Cc: stable@vger.kernel.org
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 8bf0739539c0..162a4d1a2e2a 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2556,8 +2556,10 @@ static void ublk_ch_release_work_fn(struct work_struct *work)
 	 * reset, so that a new server can fetch and start the device.
 	 */
 	disk = ublk_get_disk(ub);
-	if (!disk)
+	if (!disk) {
+		mutex_lock(&ub->mutex);
 		goto reset;
+	}
 
 	/*
 	 * All uring_cmd are done now, so abort any request outstanding to
@@ -2589,7 +2591,7 @@ static void ublk_ch_release_work_fn(struct work_struct *work)
 
 	/* double check after grabbing lock */
 	if (!ub->ub_disk)
-		goto unlock;
+		goto reset;
 
 	/*
 	 * Transition the device to the nosrv state. What exactly this
@@ -2615,12 +2617,14 @@ static void ublk_ch_release_work_fn(struct work_struct *work)
 				WRITE_ONCE(ublk_get_queue(ub, i)->fail_io, true);
 		}
 	}
-unlock:
-	mutex_unlock(&ub->mutex);
-	ublk_put_disk(disk);
 reset:
-	/* all uring_cmd has been done now, reset device & ubq */
+	/*
+	 * All uring_cmd has been done now, reset device & ubq. Under
+	 * ub->mutex, so START_DEV sees the round either ready or reset.
+	 */
 	ublk_reset_ch_dev(ub);
+	mutex_unlock(&ub->mutex);
+	ublk_put_disk(disk);
 	clear_bit(UB_STATE_OPEN, &ub->state);
 
 	/* put the reference grabbed in ublk_ch_release() */
-- 
2.55.0

[PATCH 6/8] ublk: let STOP_DEV cancel the server's commands before its release

[Ming Lei]

STOP_DEV cancels the fetched commands after dropping ub->mutex, so
START_DEV and FETCH can run in between:

  STOP_DEV                          START_DEV / FETCH
  lock ub->mutex
    ublk_stop_dev_unlocked()        (not started: does nothing)
  unlock ub->mutex
                                    START_DEV goes live, schedules scan
                                    FETCH publishes a new command
  cancel_work_sync(scan)            cancels the new disk's scan
  ublk_cancel_dev()                 takes the new command: NULL io->cmd
                                    on a queue which isn't canceling

A START_DEV after STOP_DEV also goes live over the canceled commands.

Hold a reference on the server's /dev/ublkcN while canceling:

  STOP_DEV
  lock ub->mutex
    ublk_stop_dev_unlocked()
    cancel_work_sync(scan)
    server attached: take a reference on its file, set STOPPING
  unlock ub->mutex
  ublk_cancel_dev()
  drop the reference

 - STOPPING makes FETCH and PREP get UBLK_IO_RES_ABORT and START_DEV
   -EBUSY; a START_DEV or END_USER_RECOVERY waiting for the device
   to get ready wakes up too; the reset in the server's release
   clears it
 - the release can't run before the reference is dropped, so no new
   server can attach, and the cancel only meets this server's commands
 - the cancel still runs after the unlock: io_uring_cmd_done() may take
   uring_lock, under which FETCH takes ub->mutex

ublk_ch_open() saves the file in ->ch_file and ublk_ch_release() clears
it, under cancel_mutex. The release runs after the last reference is
gone, so take ours with file_ref_get(). Drop it with __fput_sync(): if
it is the last one, fput() would defer the release to this task, which
DEL_DEV then makes wait for the device to be freed.

A server which opened the device but hasn't fetched is stopped too. The
cancel callback's io->cmd check allows NULL: the cancel may take the
command meanwhile.

Fixes: 85248d670b71 ("ublk: move ublk_cancel_dev() out of ub->mutex")
Cc: stable@vger.kernel.org
Reported-by: Josef Bacik <josef@toxicpanda.com>
Closes: https://lore.kernel.org/linux-block/20260928-b4-ublk-cancel-stop-v1-0-4a4360232a46@toxicpanda.com/
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 85 +++++++++++++++++++++++++++++++++++-----
 1 file changed, 76 insertions(+), 9 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 162a4d1a2e2a..f57d544c1da2 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -321,6 +321,8 @@ struct ublk_device {
 #define UB_STATE_OPEN		0
 #define UB_STATE_USED		1
 #define UB_STATE_DELETED	2
+/* STOP_DEV canceled the server's commands, until its release */
+#define UB_STATE_STOPPING	3
 	unsigned long		state;
 	int			ub_number;
 
@@ -334,6 +336,8 @@ struct ublk_device {
 	u16			nr_queue_ready;
 	bool 			unprivileged_daemons;
 	struct mutex cancel_mutex;
+	/* the open /dev/ublkcN, protected by cancel_mutex */
+	struct file *ch_file;
 	/*
 	 * A cancel started in this FETCH round. Set by ublk_set_canceling(),
 	 * cleared only by ublk_reset_ch_dev() when a new round starts. While
@@ -2406,6 +2410,9 @@ static int ublk_ch_open(struct inode *inode, struct file *filp)
 		return -EBUSY;
 	filp->private_data = ub;
 	ub->ublksrv_tgid = current->tgid;
+	mutex_lock(&ub->cancel_mutex);
+	ub->ch_file = filp;
+	mutex_unlock(&ub->cancel_mutex);
 	return 0;
 }
 
@@ -2432,6 +2439,7 @@ static void ublk_reset_ch_dev(struct ublk_device *ub)
 	ub->nr_queue_ready = 0;
 	ub->unprivileged_daemons = false;
 	ub->ublksrv_tgid = -1;
+	clear_bit(UB_STATE_STOPPING, &ub->state);
 }
 
 static struct gendisk *ublk_get_disk(struct ublk_device *ub)
@@ -2635,6 +2643,9 @@ static int ublk_ch_release(struct inode *inode, struct file *filp)
 {
 	struct ublk_device *ub = filp->private_data;
 
+	mutex_lock(&ub->cancel_mutex);
+	ub->ch_file = NULL;
+	mutex_unlock(&ub->cancel_mutex);
 	/*
 	 * Grab ublk device reference, so it won't be gone until we are
 	 * really released from work function.
@@ -2896,6 +2907,7 @@ static void ublk_uring_cmd_cancel_fn(struct io_uring_cmd *cmd,
 {
 	struct ublk_uring_cmd_pdu *pdu = ublk_get_uring_cmd_pdu(cmd);
 	struct ublk_queue *ubq = pdu->ubq;
+	struct io_uring_cmd *cur;
 	struct task_struct *task;
 	struct ublk_io *io;
 
@@ -2912,7 +2924,9 @@ static void ublk_uring_cmd_cancel_fn(struct io_uring_cmd *cmd,
 
 	ublk_start_cancel(ubq->dev);
 
-	WARN_ON_ONCE(io->cmd != cmd);
+	/* NULL if STOP_DEV's cancel took it meanwhile */
+	cur = READ_ONCE(io->cmd);
+	WARN_ON_ONCE(cur && cur != cmd);
 	ublk_cancel_cmd(ubq, pdu->tag, issue_flags);
 }
 
@@ -3023,13 +3037,54 @@ static void ublk_stop_dev_unlocked(struct ublk_device *ub)
 	put_disk(disk);
 }
 
+static struct file *ublk_get_ch_file(struct ublk_device *ub)
+{
+	struct file *file;
+
+	mutex_lock(&ub->cancel_mutex);
+	file = ub->ch_file;
+	if (file && !file_ref_get(&file->f_ref))
+		file = NULL;
+	mutex_unlock(&ub->cancel_mutex);
+	return file;
+}
+
 static void ublk_stop_dev(struct ublk_device *ub)
 {
+	struct file *file;
+
+	/*
+	 * FETCH, PREP and START_DEV take ub->mutex. If a server has
+	 * /dev/ublkcN open, set STOPPING, which turns it away, and hold a
+	 * reference on the file: the server's release, whose reset clears
+	 * STOPPING and lets a new server attach, can't run before we drop it.
+	 * So the cancel can run after the unlock and only meets this server's
+	 * commands; it has to: io_uring_cmd_done() may take uring_lock, under
+	 * which FETCH takes ub->mutex.
+	 */
 	mutex_lock(&ub->mutex);
 	ublk_stop_dev_unlocked(ub);
-	mutex_unlock(&ub->mutex);
 	cancel_work_sync(&ub->partition_scan_work);
+	file = ublk_get_ch_file(ub);
+	if (file) {
+		/*
+		 * The server has to close /dev/ublkcN before this device can
+		 * be started again: the reset in its release clears STOPPING.
+		 */
+		set_bit(UB_STATE_STOPPING, &ub->state);
+		/* for wake_up_var() below, see wake_up_bit() */
+		smp_mb__after_atomic();
+	}
+	mutex_unlock(&ub->mutex);
+
+	/* no server: nothing to cancel */
+	if (!file)
+		return;
+
+	/* wake a START_DEV waiting for the device to get ready */
+	wake_up_var(&ub->nr_queue_ready);
 	ublk_cancel_dev(ub);
+	__fput_sync(file);
 }
 
 static void ublk_reset_io_flags(struct ublk_queue *ubq, struct ublk_io *io)
@@ -3295,6 +3350,9 @@ static int ublk_check_fetch_buf(const struct ublk_device *ub, __u64 buf_addr)
 static int __ublk_fetch(struct io_uring_cmd *cmd, struct ublk_device *ub,
 			struct ublk_io *io, u16 q_id)
 {
+	if (test_bit(UB_STATE_STOPPING, &ub->state))
+		return UBLK_IO_RES_ABORT;
+
 	/* UBLK_IO_FETCH_REQ is only allowed before dev is setup */
 	if (ublk_dev_ready(ub))
 		return -EBUSY;
@@ -4473,22 +4531,27 @@ static bool ublk_validate_user_pid(struct ublk_device *ub, pid_t ublksrv_pid)
 	return ub->ublksrv_tgid == ublksrv_pid;
 }
 
+static bool ublk_dev_ready_or_stopping(const struct ublk_device *ub)
+{
+	return ublk_dev_ready(ub) || test_bit(UB_STATE_STOPPING, &ub->state);
+}
+
 /*
- * Wait until all queues have fetched their I/O commands, and return with
- * ub->mutex held and readiness guaranteed. The queues stay canceling if
- * this round saw a cancel, see ublk_queue_reset_io_flags(). Ready may
- * regress between wakeup and mutex_lock() (F_BATCH UNPREP, daemon death),
- * so re-check it under the mutex and wait again.
+ * Wait until all queues have fetched their I/O commands, or STOP_DEV set
+ * UB_STATE_STOPPING, and return with ub->mutex held. The queues stay
+ * canceling if this round saw a cancel, see ublk_queue_reset_io_flags().
+ * Ready may regress between wakeup and mutex_lock() (F_BATCH UNPREP,
+ * daemon death), so re-check it under the mutex and wait again.
  */
 static int ublk_wait_dev_ready_and_lock(struct ublk_device *ub)
 {
 	while (true) {
 		if (wait_var_event_interruptible(&ub->nr_queue_ready,
-						 ublk_dev_ready(ub)))
+						 ublk_dev_ready_or_stopping(ub)))
 			return -EINTR;
 
 		mutex_lock(&ub->mutex);
-		if (ublk_dev_ready(ub))
+		if (ublk_dev_ready_or_stopping(ub))
 			return 0;
 		mutex_unlock(&ub->mutex);
 	}
@@ -4588,6 +4651,10 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
 		ret = -EEXIST;
 		goto out_unlock;
 	}
+	if (test_bit(UB_STATE_STOPPING, &ub->state)) {
+		ret = -EBUSY;
+		goto out_unlock;
+	}
 
 	disk = blk_mq_alloc_disk(&ub->tag_set, &lim, NULL);
 	if (IS_ERR(disk)) {
-- 
2.55.0

[PATCH 7/8] selftests: ublk: move the control command helpers into ctrl.c

[Ming Lei]

Move the ublk_ctrl_*() helpers and ublk_setup_ring() out of kublk.c,
so a small test program can send control commands without copying
them. kublk still links every *.c file, ctrl.c included.

ublk_ctrl_deinit() now also exits the control ring, and ublk_ctrl_init()
checks the allocation and closes /dev/ublk-control when the ring setup
fails. kublk calls them only around its own lifetime, but a test which
opens a control handle per thread must not leak.

No functional change for kublk.

Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 tools/testing/selftests/ublk/ctrl.c  | 268 ++++++++++++++++++++++++++
 tools/testing/selftests/ublk/kublk.c | 271 ---------------------------
 tools/testing/selftests/ublk/kublk.h |  32 ++++
 3 files changed, 300 insertions(+), 271 deletions(-)
 create mode 100644 tools/testing/selftests/ublk/ctrl.c

diff --git a/tools/testing/selftests/ublk/ctrl.c b/tools/testing/selftests/ublk/ctrl.c
new file mode 100644
index 000000000000..53d54799f909
--- /dev/null
+++ b/tools/testing/selftests/ublk/ctrl.c
@@ -0,0 +1,268 @@
+// SPDX-License-Identifier: GPL-2.0
+
+/* ublk control commands, shared by kublk and ublk_cancel_ready */
+
+#include "kublk.h"
+
+static void ublk_ctrl_init_cmd(struct ublk_dev *dev,
+		struct io_uring_sqe *sqe,
+		struct ublk_ctrl_cmd_data *data)
+{
+	struct ublksrv_ctrl_dev_info *info = &dev->dev_info;
+	struct ublksrv_ctrl_cmd *cmd = (struct ublksrv_ctrl_cmd *)ublk_get_sqe_cmd(sqe);
+
+	sqe->fd = dev->ctrl_fd;
+	sqe->opcode = IORING_OP_URING_CMD;
+	sqe->ioprio = 0;
+
+	if (data->flags & CTRL_CMD_HAS_BUF) {
+		cmd->addr = data->addr;
+		cmd->len = data->len;
+	}
+
+	if (data->flags & CTRL_CMD_HAS_DATA)
+		cmd->data[0] = data->data[0];
+
+	cmd->dev_id = info->dev_id;
+	cmd->queue_id = -1;
+
+	ublk_set_sqe_cmd_op(sqe, data->cmd_op);
+
+	io_uring_sqe_set_data(sqe, cmd);
+}
+
+int __ublk_ctrl_cmd(struct ublk_dev *dev,
+		struct ublk_ctrl_cmd_data *data)
+{
+	struct io_uring_sqe *sqe;
+	struct io_uring_cqe *cqe;
+	int ret = -EINVAL;
+
+	sqe = io_uring_get_sqe(&dev->ring);
+	if (!sqe) {
+		ublk_err("%s: can't get sqe ret %d\n", __func__, ret);
+		return ret;
+	}
+
+	ublk_ctrl_init_cmd(dev, sqe, data);
+
+	ret = io_uring_submit(&dev->ring);
+	if (ret < 0) {
+		ublk_err("uring submit ret %d\n", ret);
+		return ret;
+	}
+
+	ret = io_uring_wait_cqe(&dev->ring, &cqe);
+	if (ret < 0) {
+		ublk_err("wait cqe: %s\n", strerror(-ret));
+		return ret;
+	}
+	io_uring_cqe_seen(&dev->ring, cqe);
+
+	return cqe->res;
+}
+
+void ublk_ctrl_deinit(struct ublk_dev *dev)
+{
+	io_uring_queue_exit(&dev->ring);
+	close(dev->ctrl_fd);
+	free(dev);
+}
+
+struct ublk_dev *ublk_ctrl_init(void)
+{
+	struct ublk_dev *dev = (struct ublk_dev *)calloc(1, sizeof(*dev));
+	struct ublksrv_ctrl_dev_info *info;
+	int ret;
+
+	if (!dev)
+		return NULL;
+	info = &dev->dev_info;
+	dev->ctrl_fd = open(CTRL_DEV, O_RDWR);
+	if (dev->ctrl_fd < 0) {
+		free(dev);
+		return NULL;
+	}
+
+	info->max_io_buf_bytes = UBLK_IO_MAX_BYTES;
+
+	ret = ublk_setup_ring(&dev->ring, UBLK_CTRL_RING_DEPTH,
+			UBLK_CTRL_RING_DEPTH, IORING_SETUP_SQE128);
+	if (ret < 0) {
+		ublk_err("queue_init: %s\n", strerror(-ret));
+		close(dev->ctrl_fd);
+		free(dev);
+		return NULL;
+	}
+	dev->nr_fds = 1;
+
+	return dev;
+}
+
+int ublk_ctrl_stop_dev(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_STOP_DEV,
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_try_stop_dev(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_TRY_STOP_DEV,
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_start_dev(struct ublk_dev *dev,
+		int daemon_pid)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_START_DEV,
+		.flags	= CTRL_CMD_HAS_DATA,
+	};
+
+	dev->dev_info.ublksrv_pid = data.data[0] = daemon_pid;
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_start_user_recovery(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_START_USER_RECOVERY,
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_end_user_recovery(struct ublk_dev *dev, int daemon_pid)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_END_USER_RECOVERY,
+		.flags	= CTRL_CMD_HAS_DATA,
+	};
+
+	dev->dev_info.ublksrv_pid = data.data[0] = daemon_pid;
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_add_dev(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_ADD_DEV,
+		.flags	= CTRL_CMD_HAS_BUF,
+		.addr = (__u64) (uintptr_t) &dev->dev_info,
+		.len = sizeof(struct ublksrv_ctrl_dev_info),
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_del_dev(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op = UBLK_U_CMD_DEL_DEV,
+		.flags = 0,
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_get_info(struct ublk_dev *dev)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_GET_DEV_INFO,
+		.flags	= CTRL_CMD_HAS_BUF,
+		.addr = (__u64) (uintptr_t) &dev->dev_info,
+		.len = sizeof(struct ublksrv_ctrl_dev_info),
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_set_params(struct ublk_dev *dev,
+		struct ublk_params *params)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_SET_PARAMS,
+		.flags	= CTRL_CMD_HAS_BUF,
+		.addr = (__u64) (uintptr_t) params,
+		.len = sizeof(*params),
+	};
+	params->len = sizeof(*params);
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_get_params(struct ublk_dev *dev,
+		struct ublk_params *params)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_GET_PARAMS,
+		.flags	= CTRL_CMD_HAS_BUF,
+		.addr = (__u64)params,
+		.len = sizeof(*params),
+	};
+
+	params->len = sizeof(*params);
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_get_features(struct ublk_dev *dev,
+		__u64 *features)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_GET_FEATURES,
+		.flags	= CTRL_CMD_HAS_BUF,
+		.addr = (__u64) (uintptr_t) features,
+		.len = sizeof(*features),
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_update_size(struct ublk_dev *dev,
+		__u64 nr_sects)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_UPDATE_SIZE,
+		.flags	= CTRL_CMD_HAS_DATA,
+	};
+
+	data.data[0] = nr_sects;
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_quiesce_dev(struct ublk_dev *dev, unsigned int timeout_ms)
+{
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op	= UBLK_U_CMD_QUIESCE_DEV,
+		.flags	= CTRL_CMD_HAS_DATA,
+	};
+
+	data.data[0] = timeout_ms;
+	return __ublk_ctrl_cmd(dev, &data);
+}
+
+int ublk_ctrl_reg_buf(struct ublk_dev *dev, void *addr, size_t size,
+		      __u32 flags)
+{
+	struct ublk_shmem_buf_reg buf_reg = {
+		.addr = (unsigned long)addr,
+		.len = size,
+		.flags = flags,
+	};
+	struct ublk_ctrl_cmd_data data = {
+		.cmd_op = UBLK_U_CMD_REG_BUF,
+		.flags = CTRL_CMD_HAS_BUF,
+		.addr = (unsigned long)&buf_reg,
+		.len = sizeof(buf_reg),
+	};
+
+	return __ublk_ctrl_cmd(dev, &data);
+}
diff --git a/tools/testing/selftests/ublk/kublk.c b/tools/testing/selftests/ublk/kublk.c
index 2400b4615766..15ba060ce7a3 100644
--- a/tools/testing/selftests/ublk/kublk.c
+++ b/tools/testing/selftests/ublk/kublk.c
@@ -37,203 +37,6 @@ static const struct ublk_tgt_ops *ublk_find_tgt(const char *name)
 	return NULL;
 }
 
-static inline int ublk_setup_ring(struct io_uring *r, int depth,
-		int cq_depth, unsigned flags)
-{
-	struct io_uring_params p;
-
-	memset(&p, 0, sizeof(p));
-	p.flags = flags | IORING_SETUP_CQSIZE;
-	p.cq_entries = cq_depth;
-
-	return io_uring_queue_init_params(depth, r, &p);
-}
-
-static void ublk_ctrl_init_cmd(struct ublk_dev *dev,
-		struct io_uring_sqe *sqe,
-		struct ublk_ctrl_cmd_data *data)
-{
-	struct ublksrv_ctrl_dev_info *info = &dev->dev_info;
-	struct ublksrv_ctrl_cmd *cmd = (struct ublksrv_ctrl_cmd *)ublk_get_sqe_cmd(sqe);
-
-	sqe->fd = dev->ctrl_fd;
-	sqe->opcode = IORING_OP_URING_CMD;
-	sqe->ioprio = 0;
-
-	if (data->flags & CTRL_CMD_HAS_BUF) {
-		cmd->addr = data->addr;
-		cmd->len = data->len;
-	}
-
-	if (data->flags & CTRL_CMD_HAS_DATA)
-		cmd->data[0] = data->data[0];
-
-	cmd->dev_id = info->dev_id;
-	cmd->queue_id = -1;
-
-	ublk_set_sqe_cmd_op(sqe, data->cmd_op);
-
-	io_uring_sqe_set_data(sqe, cmd);
-}
-
-static int __ublk_ctrl_cmd(struct ublk_dev *dev,
-		struct ublk_ctrl_cmd_data *data)
-{
-	struct io_uring_sqe *sqe;
-	struct io_uring_cqe *cqe;
-	int ret = -EINVAL;
-
-	sqe = io_uring_get_sqe(&dev->ring);
-	if (!sqe) {
-		ublk_err("%s: can't get sqe ret %d\n", __func__, ret);
-		return ret;
-	}
-
-	ublk_ctrl_init_cmd(dev, sqe, data);
-
-	ret = io_uring_submit(&dev->ring);
-	if (ret < 0) {
-		ublk_err("uring submit ret %d\n", ret);
-		return ret;
-	}
-
-	ret = io_uring_wait_cqe(&dev->ring, &cqe);
-	if (ret < 0) {
-		ublk_err("wait cqe: %s\n", strerror(-ret));
-		return ret;
-	}
-	io_uring_cqe_seen(&dev->ring, cqe);
-
-	return cqe->res;
-}
-
-static int ublk_ctrl_stop_dev(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_STOP_DEV,
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_try_stop_dev(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_TRY_STOP_DEV,
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_start_dev(struct ublk_dev *dev,
-		int daemon_pid)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_START_DEV,
-		.flags	= CTRL_CMD_HAS_DATA,
-	};
-
-	dev->dev_info.ublksrv_pid = data.data[0] = daemon_pid;
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_start_user_recovery(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_START_USER_RECOVERY,
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_end_user_recovery(struct ublk_dev *dev, int daemon_pid)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_END_USER_RECOVERY,
-		.flags	= CTRL_CMD_HAS_DATA,
-	};
-
-	dev->dev_info.ublksrv_pid = data.data[0] = daemon_pid;
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_add_dev(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_ADD_DEV,
-		.flags	= CTRL_CMD_HAS_BUF,
-		.addr = (__u64) (uintptr_t) &dev->dev_info,
-		.len = sizeof(struct ublksrv_ctrl_dev_info),
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_del_dev(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op = UBLK_U_CMD_DEL_DEV,
-		.flags = 0,
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_get_info(struct ublk_dev *dev)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_GET_DEV_INFO,
-		.flags	= CTRL_CMD_HAS_BUF,
-		.addr = (__u64) (uintptr_t) &dev->dev_info,
-		.len = sizeof(struct ublksrv_ctrl_dev_info),
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_set_params(struct ublk_dev *dev,
-		struct ublk_params *params)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_SET_PARAMS,
-		.flags	= CTRL_CMD_HAS_BUF,
-		.addr = (__u64) (uintptr_t) params,
-		.len = sizeof(*params),
-	};
-	params->len = sizeof(*params);
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_get_params(struct ublk_dev *dev,
-		struct ublk_params *params)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_GET_PARAMS,
-		.flags	= CTRL_CMD_HAS_BUF,
-		.addr = (__u64)params,
-		.len = sizeof(*params),
-	};
-
-	params->len = sizeof(*params);
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_get_features(struct ublk_dev *dev,
-		__u64 *features)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_GET_FEATURES,
-		.flags	= CTRL_CMD_HAS_BUF,
-		.addr = (__u64) (uintptr_t) features,
-		.len = sizeof(*features),
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
 static int parse_param_types(const char *arg, __u32 *types)
 {
 	char buf[128], *save = NULL, *tok;
@@ -283,30 +86,6 @@ static void ublk_init_params_from_ctx(const struct dev_ctx *ctx,
 	};
 }
 
-static int ublk_ctrl_update_size(struct ublk_dev *dev,
-		__u64 nr_sects)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_UPDATE_SIZE,
-		.flags	= CTRL_CMD_HAS_DATA,
-	};
-
-	data.data[0] = nr_sects;
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
-static int ublk_ctrl_quiesce_dev(struct ublk_dev *dev,
-				 unsigned int timeout_ms)
-{
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op	= UBLK_U_CMD_QUIESCE_DEV,
-		.flags	= CTRL_CMD_HAS_DATA,
-	};
-
-	data.data[0] = timeout_ms;
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
 static const char *ublk_dev_state_desc(struct ublk_dev *dev)
 {
 	switch (dev->dev_info.state) {
@@ -426,38 +205,6 @@ static void ublk_ctrl_dump(struct ublk_dev *dev)
 	fflush(stdout);
 }
 
-static void ublk_ctrl_deinit(struct ublk_dev *dev)
-{
-	close(dev->ctrl_fd);
-	free(dev);
-}
-
-static struct ublk_dev *ublk_ctrl_init(void)
-{
-	struct ublk_dev *dev = (struct ublk_dev *)calloc(1, sizeof(*dev));
-	struct ublksrv_ctrl_dev_info *info = &dev->dev_info;
-	int ret;
-
-	dev->ctrl_fd = open(CTRL_DEV, O_RDWR);
-	if (dev->ctrl_fd < 0) {
-		free(dev);
-		return NULL;
-	}
-
-	info->max_io_buf_bytes = UBLK_IO_MAX_BYTES;
-
-	ret = ublk_setup_ring(&dev->ring, UBLK_CTRL_RING_DEPTH,
-			UBLK_CTRL_RING_DEPTH, IORING_SETUP_SQE128);
-	if (ret < 0) {
-		ublk_err("queue_init: %s\n", strerror(-ret));
-		free(dev);
-		return NULL;
-	}
-	dev->nr_fds = 1;
-
-	return dev;
-}
-
 static size_t __ublk_queue_cmd_buf_sz(const struct ublk_queue *q, __u16 depth)
 {
 	size_t size = depth * (size_t)q->io_desc_size;
@@ -1283,24 +1030,6 @@ static void ublk_shmem_unregister_all(void)
 	shmem_count = 0;
 }
 
-static int ublk_ctrl_reg_buf(struct ublk_dev *dev, void *addr, size_t size,
-			     __u32 flags)
-{
-	struct ublk_shmem_buf_reg buf_reg = {
-		.addr = (unsigned long)addr,
-		.len = size,
-		.flags = flags,
-	};
-	struct ublk_ctrl_cmd_data data = {
-		.cmd_op = UBLK_U_CMD_REG_BUF,
-		.flags = CTRL_CMD_HAS_BUF,
-		.addr = (unsigned long)&buf_reg,
-		.len = sizeof(buf_reg),
-	};
-
-	return __ublk_ctrl_cmd(dev, &data);
-}
-
 /*
  * Handle one client connection: receive memfd, mmap it, register
  * the VA range with kernel, send back the assigned index.
diff --git a/tools/testing/selftests/ublk/kublk.h b/tools/testing/selftests/ublk/kublk.h
index d98f3d612d88..99b8ceff853c 100644
--- a/tools/testing/selftests/ublk/kublk.h
+++ b/tools/testing/selftests/ublk/kublk.h
@@ -294,6 +294,38 @@ struct ublk_dev {
 
 extern int ublk_queue_io_cmd(struct ublk_thread *t, struct ublk_io *io);
 
+static inline int ublk_setup_ring(struct io_uring *r, int depth,
+		int cq_depth, unsigned int flags)
+{
+	struct io_uring_params p;
+
+	memset(&p, 0, sizeof(p));
+	p.flags = flags | IORING_SETUP_CQSIZE;
+	p.cq_entries = cq_depth;
+
+	return io_uring_queue_init_params(depth, r, &p);
+}
+
+/* ctrl.c: control commands */
+struct ublk_dev *ublk_ctrl_init(void);
+void ublk_ctrl_deinit(struct ublk_dev *dev);
+int __ublk_ctrl_cmd(struct ublk_dev *dev, struct ublk_ctrl_cmd_data *data);
+int ublk_ctrl_add_dev(struct ublk_dev *dev);
+int ublk_ctrl_del_dev(struct ublk_dev *dev);
+int ublk_ctrl_get_info(struct ublk_dev *dev);
+int ublk_ctrl_set_params(struct ublk_dev *dev, struct ublk_params *params);
+int ublk_ctrl_get_params(struct ublk_dev *dev, struct ublk_params *params);
+int ublk_ctrl_get_features(struct ublk_dev *dev, __u64 *features);
+int ublk_ctrl_start_dev(struct ublk_dev *dev, int daemon_pid);
+int ublk_ctrl_stop_dev(struct ublk_dev *dev);
+int ublk_ctrl_try_stop_dev(struct ublk_dev *dev);
+int ublk_ctrl_start_user_recovery(struct ublk_dev *dev);
+int ublk_ctrl_end_user_recovery(struct ublk_dev *dev, int daemon_pid);
+int ublk_ctrl_update_size(struct ublk_dev *dev, __u64 nr_sects);
+int ublk_ctrl_quiesce_dev(struct ublk_dev *dev, unsigned int timeout_ms);
+int ublk_ctrl_reg_buf(struct ublk_dev *dev, void *addr, size_t size,
+		      __u32 flags);
+
 static inline int __ublk_use_batch_io(__u64 flags)
 {
 	return flags & UBLK_F_BATCH_IO;
-- 
2.55.0

[PATCH 8/8] selftests: ublk: add test for going live over canceled io commands

[Ming Lei]

Add ublk_cancel_ready, a small liburing program which sends control
commands with kublk's helpers from ctrl.c, and test_generic_18.sh,
which runs its modes:

  mode               what it does                     expected
  stop_start         STOP_DEV before START_DEV        START_DEV -EBUSY
  partial_fetch      task A fetches tags 0-2 and      START_DEV -ENODEV, or
                     dies, then task B fetches tag 3  live and reads -EIO
  recovery           in user recovery, queue 0's      END_USER_RECOVERY
                     task dies before queue 1 is      -ENODEV, or live and
                     ready                            reads -EIO
  stop_restart       STOP_DEV on a new device, then   START_DEV 0,
                     a server starts it               reads succeed
  stop_attached      STOP_DEV after open, before      FETCH ABORT and
                     FETCH, then a new server         START_DEV -EBUSY,
                                                      then a new one works
  stop_live_restart  STOP_DEV on a live device, then  FETCH and START_DEV
                     a new server starts it           0, reads succeed
  race_start         STOP_DEV and START_DEV at the    START_DEV 0 and
                     same time, 50 times              reads complete,
                                                      or -EBUSY
  race_fetch         STOP_DEV while a server opens    START_DEV 0 and
                     and fetches, then START_DEV,     reads succeed,
                     50 times                         or -EBUSY
  race_async_fetch   IOSQE_ASYNC FETCH while STOP_DEV no crash (KASAN
                     cancels, then close the ring     finds the bug)

If the device goes live, the test reads it, so an unfixed kernel shows
the oops in ublk_queue_cmd(). In stop_live_restart, an unfixed kernel
doesn't reset the FETCH round in ublk_ch_release() once the disk is
gone, so the new server's FETCH gets -EBUSY.

A dying task is a child process which fetches and then calls exec().
exec() cancels the task's uring_cmds before it returns, so the cancel
is done once the child is reaped, with no sleep. kublk can't be used:
its failure injection kills the whole process, which closes
/dev/ublkcN and starts a new round.

Link: https://lore.kernel.org/linux-block/20260928-b4-ublk-cancel-stop-v1-0-4a4360232a46@toxicpanda.com/
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
Assisted-by: LLM
---
 tools/testing/selftests/ublk/.gitignore       |   1 +
 tools/testing/selftests/ublk/Makefile         |   6 +-
 .../testing/selftests/ublk/test_generic_18.sh |  37 +
 .../selftests/ublk/ublk_cancel_ready.c        | 984 ++++++++++++++++++
 4 files changed, 1026 insertions(+), 2 deletions(-)
 create mode 100755 tools/testing/selftests/ublk/test_generic_18.sh
 create mode 100644 tools/testing/selftests/ublk/ublk_cancel_ready.c

diff --git a/tools/testing/selftests/ublk/.gitignore b/tools/testing/selftests/ublk/.gitignore
index e17bd28f27e0..d8e93fef7fcd 100644
--- a/tools/testing/selftests/ublk/.gitignore
+++ b/tools/testing/selftests/ublk/.gitignore
@@ -3,3 +3,4 @@
 /tools
 kublk
 metadata_size
+ublk_cancel_ready
diff --git a/tools/testing/selftests/ublk/Makefile b/tools/testing/selftests/ublk/Makefile
index 37883e9d50ec..fc64b8f02833 100644
--- a/tools/testing/selftests/ublk/Makefile
+++ b/tools/testing/selftests/ublk/Makefile
@@ -19,6 +19,7 @@ TEST_PROGS += test_generic_12.sh
 TEST_PROGS += test_generic_13.sh
 TEST_PROGS += test_generic_16.sh
 TEST_PROGS += test_generic_17.sh
+TEST_PROGS += test_generic_18.sh
 
 TEST_PROGS += test_batch_01.sh
 TEST_PROGS += test_batch_02.sh
@@ -76,13 +77,14 @@ TEST_FILES := settings
 TEST_FILES += test_common.sh
 TEST_FILES += trace
 
-TEST_GEN_PROGS_EXTENDED = kublk metadata_size
-STANDALONE_UTILS := metadata_size.c
+TEST_GEN_PROGS_EXTENDED = kublk metadata_size ublk_cancel_ready
+STANDALONE_UTILS := metadata_size.c ublk_cancel_ready.c
 
 LOCAL_HDRS += $(wildcard *.h)
 include ../lib.mk
 
 $(OUTPUT)/kublk: $(filter-out $(STANDALONE_UTILS),$(wildcard *.c))
+$(OUTPUT)/ublk_cancel_ready: ublk_cancel_ready.c ctrl.c
 
 check:
 	shellcheck -x -f gcc *.sh
diff --git a/tools/testing/selftests/ublk/test_generic_18.sh b/tools/testing/selftests/ublk/test_generic_18.sh
new file mode 100755
index 000000000000..223944dba149
--- /dev/null
+++ b/tools/testing/selftests/ublk/test_generic_18.sh
@@ -0,0 +1,37 @@
+#!/bin/bash
+# SPDX-License-Identifier: GPL-2.0
+
+. "$(cd "$(dirname "$0")" && pwd)"/test_common.sh
+
+ERR_CODE=0
+CANCEL_PROG="$(_ublk_test_top_dir)/ublk_cancel_ready"
+
+_prep_test "generic" "start device over canceled io commands"
+
+# the modes are described in ublk_cancel_ready.c
+for mode in stop_start partial_fetch recovery stop_restart stop_attached \
+		stop_live_restart race_start race_fetch race_async_fetch; do
+	dmesg_before=$(dmesg | wc -l)
+	timeout 60 "$CANCEL_PROG" "$mode" > "$UBLK_TMP" 2>&1
+	res=$?
+	msg=""
+
+	if dmesg | tail -n +"$((dmesg_before + 1))" | \
+			grep -q -e "BUG:" -e "Oops" -e "WARNING:"; then
+		msg="$mode: kernel oops/warning"
+		ERR_CODE=255
+	elif [ "$res" -eq "$UBLK_SKIP_CODE" ]; then
+		[ "$ERR_CODE" -eq 0 ] && ERR_CODE=$UBLK_SKIP_CODE
+	elif [ "$res" -ne 0 ]; then
+		msg="$mode: failed ($res)"
+		ERR_CODE=255
+	fi
+	# the output once: on failure, or always when not quiet
+	[ -n "$msg" ] && echo "$msg"
+	if [ -n "$msg" ] || [ "$UBLK_TEST_QUIET" -eq 0 ]; then
+		cat "$UBLK_TMP"
+	fi
+done
+
+_cleanup_test
+_show_result $TID $ERR_CODE
diff --git a/tools/testing/selftests/ublk/ublk_cancel_ready.c b/tools/testing/selftests/ublk/ublk_cancel_ready.c
new file mode 100644
index 000000000000..0ed0d01fa74a
--- /dev/null
+++ b/tools/testing/selftests/ublk/ublk_cancel_ready.c
@@ -0,0 +1,984 @@
+// SPDX-License-Identifier: GPL-2.0
+/*
+ * Bring a ublk device live while some of its fetched io commands are
+ * canceled.
+ *
+ * A cancel completes a fetched command and clears io->cmd, but the io
+ * still counts as ready. Only ubq->canceling keeps ublk_queue_rq() away
+ * from the NULL io->cmd. Each mode below loses ->canceling in another way:
+ *
+ * stop_start:    fetch every tag, STOP_DEV before START_DEV, START_DEV.
+ *                STOP_DEV cancels the commands of the attached server,
+ *                so START_DEV must get -EBUSY until that server is gone.
+ *                An unfixed kernel starts the device over them.
+ *
+ * partial_fetch: task A fetches tags 0..depth-2 and dies, so its
+ *                commands are canceled. Task B fetches the last tag, and
+ *                an unfixed kernel clears ->canceling when the queue gets
+ *                ready. /dev/ublkcN stays open all the time, so
+ *                ublk_ch_release() never resets the queue.
+ *
+ * recovery:      a UBLK_F_USER_RECOVERY device with two queues loses its
+ *                server. During recovery task Q0 fetches queue 0, which
+ *                clears q0->canceling, then dies. An unfixed kernel still
+ *                has ub->canceling set because queue 1 is not ready, so
+ *                ublk_start_cancel() does not mark queue 0 again. Reads
+ *                are issued on a CPU mapped to queue 0.
+ *
+ * In partial_fetch and recovery, START_DEV / END_USER_RECOVERY may refuse
+ * with -ENODEV, or bring the device live with its queue still canceling:
+ * then every read has to complete, with -EIO. An unfixed kernel oopses
+ * in ublk_queue_cmd() on a NULL io->cmd.
+ *
+ * stop_restart:  control mode. STOP_DEV on a new device, before any
+ *                server opened it, takes no command. A server started
+ *                afterwards has to start the device and serve I/O.
+ *
+ * stop_attached: STOP_DEV on a device whose server opened it but fetched
+ *                nothing stops that server too: FETCH gets ABORT and
+ *                START_DEV -EBUSY, until a new server opens the device.
+ *
+ * stop_live_restart: STOP_DEV on a live device; once its server is gone,
+ *                a new server has to fetch and start it again. An unfixed
+ *                ublk_ch_release() skips the reset once the disk is gone,
+ *                so the new FETCH gets -EBUSY.
+ *
+ * race_start:    STOP_DEV and START_DEV at the same time on a device whose
+ *                server fetched every tag. START_DEV either wins, and
+ *                reads complete (they fail once STOP_DEV removes the
+ *                disk), or gets -EBUSY. An unfixed kernel cancels the
+ *                commands of the live disk.
+ *
+ * race_fetch:    STOP_DEV while a server opens the device and fetches,
+ *                then START_DEV. If STOP_DEV came before the open, it must
+ *                not take any command, so START_DEV works and every read
+ *                succeeds; otherwise START_DEV gets -EBUSY. An unfixed
+ *                kernel takes commands fetched after its unlock and goes
+ *                live over them.
+ *
+ * race_async_fetch: FETCH with IOSQE_ASYNC while STOP_DEV cancels,
+ *                then close the ring. An unfixed FETCH marks its command
+ *                cancelable only after publishing it and dropping
+ *                ub->mutex; a cancel in between completes a command which
+ *                is then put on io_uring's cancelable list. Closing the
+ *                ring walks that list: KASAN reports a use after free.
+ *                Needs a KASAN kernel to see the bug; otherwise it must
+ *                just not crash.
+ *
+ * A dying task is a child process which fetches and then calls exec().
+ * exec() cancels the task's uring_cmds before it returns, so the cancel
+ * is done once the child is reaped. A thread exit does not cancel them,
+ * and closing the ring cancels them later, from io_ring_exit_work().
+ */
+#include <sched.h>
+
+#include "kublk.h"
+#include "../kselftest.h"
+
+#define NR_QUEUES	2
+#define DEPTH		4
+#define BUF_SIZE	(64 << 10)
+#define DEV_SECTORS	(64 << 11)	/* 64MB */
+#define NR_READS	(DEPTH * 2)
+#define SERVE_DELAY_US	200000
+#define RACE_LOOPS	50
+#define ASYNC_LOOPS	300
+
+/* one control handle per thread, the race modes send commands from two */
+static __thread struct ublk_dev *ctrl_dev;
+static int cdev_fd = -1;
+static int dev_id = -1;
+static int nr_queues;
+static void *bufs[NR_QUEUES][DEPTH];
+static const struct ublksrv_io_desc *iods[NR_QUEUES];
+static size_t iods_len;
+
+static struct io_uring_sqe *get_sqe(struct io_uring *ring)
+{
+	struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
+
+	if (!sqe) {
+		fprintf(stderr, "out of sqes\n");
+		exit(KSFT_FAIL);
+	}
+	return sqe;
+}
+
+static struct ublk_dev *ctrl(void)
+{
+	if (!ctrl_dev) {
+		ctrl_dev = ublk_ctrl_init();
+		if (!ctrl_dev) {
+			fprintf(stderr, "ublk_ctrl_init failed\n");
+			exit(KSFT_FAIL);
+		}
+	}
+	ctrl_dev->dev_info.dev_id = dev_id;
+	return ctrl_dev;
+}
+
+static void ctrl_put(void)
+{
+	if (ctrl_dev)
+		ublk_ctrl_deinit(ctrl_dev);
+	ctrl_dev = NULL;
+}
+
+static int dev_state(void)
+{
+	struct ublk_dev *dev = ctrl();
+	int ret = ublk_ctrl_get_info(dev);
+
+	return ret ? ret : dev->dev_info.state;
+}
+
+static int open_cdev(void)
+{
+	size_t max_len = UBLK_MAX_QUEUE_DEPTH * sizeof(struct ublksrv_io_desc);
+	int pg = getpagesize();
+	char path[64];
+
+	snprintf(path, sizeof(path), "/dev/ublkc%d", dev_id);
+	for (int i = 0; i < 100 && cdev_fd < 0; i++) {
+		cdev_fd = open(path, O_RDWR);
+		if (cdev_fd < 0)
+			usleep(50000);
+	}
+	if (cdev_fd < 0)
+		return -errno;
+
+	/* queue q's descriptors start at q * the size for the max depth */
+	max_len = (max_len + pg - 1) & ~(size_t)(pg - 1);
+	iods_len = (DEPTH * sizeof(struct ublksrv_io_desc) + pg - 1) &
+		~(size_t)(pg - 1);
+	for (int q = 0; q < nr_queues; q++) {
+		void *p = mmap(NULL, iods_len, PROT_READ,
+			       MAP_SHARED | MAP_POPULATE, cdev_fd,
+			       UBLKSRV_CMD_BUF_OFFSET + q * max_len);
+
+		if (p == MAP_FAILED)
+			return -errno;
+		iods[q] = p;
+	}
+	return 0;
+}
+
+/* the last reference to /dev/ublkcN runs ublk_ch_release() */
+static void close_cdev(void)
+{
+	for (int q = 0; q < nr_queues; q++) {
+		if (iods[q])
+			munmap((void *)iods[q], iods_len);
+		iods[q] = NULL;
+	}
+	if (cdev_fd >= 0)
+		close(cdev_fd);
+	cdev_fd = -1;
+}
+
+/* ADD_DEV and SET_PARAMS, without opening /dev/ublkcN */
+static int add_dev_noopen(int queues, __u64 flags)
+{
+	struct ublk_dev *dev = ctrl();
+	struct ublksrv_ctrl_dev_info info = {
+		.nr_hw_queues	= queues,
+		.queue_depth	= DEPTH,
+		.max_io_buf_bytes = BUF_SIZE,
+		.dev_id		= -1,
+		.flags		= UBLK_F_NO_AUTO_PART_SCAN | flags,
+	};
+	struct ublk_params p = {
+		.types	= UBLK_PARAM_TYPE_BASIC,
+		.basic	= {
+			.logical_bs_shift	= 9,
+			.physical_bs_shift	= 12,
+			.io_opt_shift		= 12,
+			.io_min_shift		= 9,
+			.max_sectors		= BUF_SIZE >> 9,
+			.dev_sectors		= DEV_SECTORS,
+		},
+	};
+	int ret;
+
+	nr_queues = queues;
+	dev->dev_info = info;
+	ret = ublk_ctrl_add_dev(dev);
+	if (ret)
+		return ret;
+	dev_id = dev->dev_info.dev_id;
+
+	ret = ublk_ctrl_set_params(ctrl(), &p);
+	if (ret)
+		return ret;
+
+	for (int q = 0; q < queues; q++)
+		for (int i = 0; i < DEPTH; i++)
+			if (!bufs[q][i] &&
+			    posix_memalign(&bufs[q][i], getpagesize(), BUF_SIZE))
+				return -ENOMEM;
+	return 0;
+}
+
+static int add_dev(int queues, __u64 flags)
+{
+	return add_dev_noopen(queues, flags) ?: open_cdev();
+}
+
+static void cleanup(void)
+{
+	if (dev_id < 0)
+		return;
+	ublk_ctrl_stop_dev(ctrl());
+	close_cdev();
+	ublk_ctrl_del_dev(ctrl());
+	dev_id = -1;
+}
+
+/* race_async_fetch sets IOSQE_ASYNC on the io commands */
+static int io_cmd_sqe_flags;
+
+static void queue_io_cmd(struct io_uring *ring, __u32 op, int q, int tag,
+			 int res)
+{
+	struct io_uring_sqe *sqe = get_sqe(ring);
+	struct ublksrv_io_cmd *cmd = (struct ublksrv_io_cmd *)sqe->cmd;
+
+	memset(sqe, 0, sizeof(*sqe));
+	sqe->fd = cdev_fd;
+	sqe->opcode = IORING_OP_URING_CMD;
+	sqe->flags = io_cmd_sqe_flags;
+	ublk_set_sqe_cmd_op(sqe, op);
+	cmd->q_id = q;
+	cmd->tag = tag;
+	cmd->result = res;
+	cmd->addr = (__u64)(uintptr_t)bufs[q][tag];
+	io_uring_sqe_set_data64(sqe, (q << 16) | tag);
+}
+
+struct async_arg {
+	pthread_barrier_t go, stopped;
+};
+
+/*
+ * FETCH every tag with IOSQE_ASYNC, racing STOP_DEV, then close the ring once
+ * STOP_DEV returned: that walks io_uring's list of cancelable commands.
+ */
+static void *race_async_fn(void *data)
+{
+	struct async_arg *a = data;
+	struct io_uring ring;
+	int ok = !io_uring_queue_init(DEPTH, &ring, 0);
+
+	if (ok)
+		for (int tag = 0; tag < DEPTH; tag++)
+			queue_io_cmd(&ring, UBLK_U_IO_FETCH_REQ, 0, tag, 0);
+	pthread_barrier_wait(&a->go);
+	if (ok)
+		io_uring_submit(&ring);
+	pthread_barrier_wait(&a->stopped);
+	if (ok)
+		io_uring_queue_exit(&ring);
+	return NULL;
+}
+
+/* reap @nr completions of canceled fetch commands, return how many */
+static int reap_aborts(struct io_uring *ring, int nr)
+{
+	struct __kernel_timespec ts = { .tv_sec = 2 };
+	struct io_uring_cqe *cqe;
+	int aborted = 0;
+
+	while (nr--) {
+		if (io_uring_wait_cqe_timeout(ring, &cqe, &ts))
+			break;
+		if (cqe->res == UBLK_IO_RES_ABORT)
+			aborted++;
+		io_uring_cqe_seen(ring, cqe);
+	}
+	return aborted;
+}
+
+/*
+ * A server thread fetches @nr_tags tags of queue @q, then completes each
+ * request after @delay_us, until its commands are aborted.
+ */
+struct server {
+	int q, first_tag, nr_tags, delay_us;
+	int failed;	/* result which ended the serve loop, 0 if none */
+	pthread_t thread;
+	pthread_barrier_t fetched;
+	/* race_fetch: wait for @go and @race_delay_us, open, fetch one by one */
+	pthread_barrier_t go;
+	int race, race_delay_us;
+};
+
+static void *server_fn(void *data)
+{
+	struct server *s = data;
+	struct io_uring ring;
+	struct io_uring_cqe *cqe;
+
+	if (io_uring_queue_init(DEPTH, &ring, 0)) {
+		pthread_barrier_wait(&s->fetched);
+		return NULL;
+	}
+	if (s->race) {
+		pthread_barrier_wait(&s->go);
+		usleep(s->race_delay_us);
+		if (open_cdev()) {
+			pthread_barrier_wait(&s->fetched);
+			io_uring_queue_exit(&ring);
+			return NULL;
+		}
+	}
+	for (int tag = s->first_tag; tag < s->first_tag + s->nr_tags; tag++) {
+		queue_io_cmd(&ring, UBLK_U_IO_FETCH_REQ, s->q, tag, 0);
+		if (s->race)
+			io_uring_submit(&ring);
+	}
+	io_uring_submit(&ring);
+	pthread_barrier_wait(&s->fetched);
+
+	while (!io_uring_wait_cqe(&ring, &cqe)) {
+		int tag = cqe->user_data & 0xffff;
+		const struct ublksrv_io_desc *iod = &iods[s->q][tag];
+		int res = cqe->res;
+
+		io_uring_cqe_seen(&ring, cqe);
+		if (res != UBLK_IO_RES_OK) {
+			__atomic_store_n(&s->failed, res, __ATOMIC_RELEASE);
+			break;
+		}
+		usleep(s->delay_us);
+		res = ublksrv_get_op(iod) <= UBLK_IO_OP_WRITE ?
+			iod->nr_sectors << 9 : 0;
+		queue_io_cmd(&ring, UBLK_U_IO_COMMIT_AND_FETCH_REQ, s->q, tag,
+			     res);
+		io_uring_submit(&ring);
+	}
+	io_uring_queue_exit(&ring);
+	return NULL;
+}
+
+/* returns once the fetch commands are issued; with @race, call race_go() */
+static void server_start(struct server *s)
+{
+	pthread_barrier_init(&s->fetched, NULL, 2);
+	pthread_barrier_init(&s->go, NULL, 2);
+	pthread_create(&s->thread, NULL, server_fn, s);
+	if (!s->race)
+		pthread_barrier_wait(&s->fetched);
+}
+
+/* wait up to 5s for the serve loop of @s to end, return its result */
+static int server_wait_failed(struct server *s)
+{
+	int res = 0;
+
+	for (int i = 0; i < 500 && !res; i++) {
+		res = __atomic_load_n(&s->failed, __ATOMIC_ACQUIRE);
+		if (!res)
+			usleep(10000);
+	}
+	return res;
+}
+
+/*
+ * A task fetches @nr_tags tags of queue @q and dies. Returns once its
+ * commands are canceled, see the top of this file.
+ */
+static int fetch_and_die(int q, int first_tag, int nr_tags)
+{
+	int status;
+	pid_t pid = fork();
+
+	if (pid < 0)
+		return -1;
+	if (!pid) {
+		struct io_uring ring;
+
+		if (io_uring_queue_init(DEPTH, &ring, 0))
+			_exit(1);
+		for (int tag = first_tag; tag < first_tag + nr_tags; tag++)
+			queue_io_cmd(&ring, UBLK_U_IO_FETCH_REQ, q, tag, 0);
+		if (io_uring_submit(&ring) != nr_tags)
+			_exit(1);
+		execlp("true", "true", NULL);
+		_exit(1);
+	}
+	if (waitpid(pid, &status, 0) != pid || !WIFEXITED(status) ||
+	    WEXITSTATUS(status))
+		return -1;
+	printf("q%d: task died with %d fetch cmds in flight\n", q, nr_tags);
+	return 0;
+}
+
+struct reads {
+	struct io_uring ring;
+	void *buf;
+	int fd, done, ok, eio, other;
+};
+
+/*
+ * Issue NR_READS reads at once, from @cpu if it is not negative: blk-mq
+ * maps the submitting CPU to the hw queue. A server holding its live
+ * tags for a while makes the other reads take the canceled tags.
+ */
+static int open_tries = 100;	/* 50ms each */
+
+static int reads_submit(struct reads *r, int cpu)
+{
+	char path[64];
+	cpu_set_t set, old;
+
+	snprintf(path, sizeof(path), "/dev/ublkb%d", dev_id);
+	r->fd = -1;
+	for (int i = 0; i < open_tries && r->fd < 0; i++) {
+		r->fd = open(path, O_RDONLY | O_DIRECT);
+		if (r->fd < 0)
+			usleep(50000);
+	}
+	if (r->fd < 0) {
+		fprintf(stderr, "open %s: %s\n", path, strerror(errno));
+		return -1;
+	}
+	if (posix_memalign(&r->buf, 4096, NR_READS * 4096) ||
+	    io_uring_queue_init(NR_READS, &r->ring, 0))
+		return -1;
+
+	for (int i = 0; i < NR_READS; i++)
+		io_uring_prep_read(get_sqe(&r->ring), r->fd,
+				   r->buf + i * 4096, 4096, i * 4096);
+
+	if (cpu >= 0) {
+		sched_getaffinity(0, sizeof(old), &old);
+		CPU_ZERO(&set);
+		CPU_SET(cpu, &set);
+		sched_setaffinity(0, sizeof(set), &set);
+	}
+	io_uring_submit(&r->ring);
+	if (cpu >= 0)
+		sched_setaffinity(0, sizeof(old), &old);
+	return 0;
+}
+
+static void reads_reap(struct reads *r, int timeout_s)
+{
+	struct __kernel_timespec ts = { .tv_sec = timeout_s };
+	struct io_uring_cqe *cqe;
+
+	while (r->done < NR_READS &&
+	       !io_uring_wait_cqe_timeout(&r->ring, &cqe, &ts)) {
+		if (cqe->res == 4096)
+			r->ok++;
+		else if (cqe->res == -EIO)
+			r->eio++;
+		else
+			r->other++;
+		r->done++;
+		io_uring_cqe_seen(&r->ring, cqe);
+	}
+}
+
+static void reads_put(struct reads *r)
+{
+	io_uring_queue_exit(&r->ring);
+	close(r->fd);
+	free(r->buf);
+}
+
+static int reads_result(struct reads *r)
+{
+	printf("reads: %d ok, %d -EIO, %d other, %d not completed\n",
+	       r->ok, r->eio, r->other, NR_READS - r->done);
+	reads_put(r);
+	return r->other || r->done < NR_READS ? KSFT_FAIL : KSFT_PASS;
+}
+
+static int start_dev(void)
+{
+	int ret = ublk_ctrl_start_dev(ctrl(), getpid());
+
+	printf("START_DEV: %d\n", ret);
+	return ret;
+}
+
+static int expect_err(const char *what, int ret, int want)
+{
+	if (ret == want)
+		return KSFT_PASS;
+	fprintf(stderr, "%s: %d, expected %d\n", what, ret, want);
+	return KSFT_FAIL;
+}
+
+/* START_DEV must fail with @want; if it went live, show what a read does */
+static int start_dev_expect(int want)
+{
+	struct reads r = {};
+	int ret = start_dev();
+
+	if (!ret && !reads_submit(&r, -1)) {
+		reads_reap(&r, 10);
+		reads_result(&r);
+	}
+	return expect_err("START_DEV", ret, want);
+}
+
+/* -ENODEV, or live over a canceling queue: then reads must complete */
+static int expect_enodev_or_reads(const char *what, int ret, struct reads *r)
+{
+	if (ret == -ENODEV)
+		return KSFT_PASS;
+	if (ret) {
+		fprintf(stderr, "%s: %d, expected 0 or %d\n", what, ret,
+			-ENODEV);
+		return KSFT_FAIL;
+	}
+	reads_reap(r, 10);
+	return reads_result(r);
+}
+
+static int test_stop_start(void)
+{
+	struct io_uring ring;
+	int ret;
+
+	if (add_dev(1, 0))
+		return KSFT_FAIL;
+	if (io_uring_queue_init(DEPTH, &ring, 0))
+		return KSFT_FAIL;
+	for (int tag = 0; tag < DEPTH; tag++)
+		queue_io_cmd(&ring, UBLK_U_IO_FETCH_REQ, 0, tag, 0);
+	io_uring_submit(&ring);
+
+	/* device is ready but not started: state is UBLK_S_DEV_DEAD */
+	ret = ublk_ctrl_stop_dev(ctrl());
+	printf("STOP_DEV: %d, canceled fetch cmds: %d/%d\n", ret,
+	       reap_aborts(&ring, DEPTH), DEPTH);
+
+	/* STOP_DEV canceled the attached server: no start until it exits */
+	ret = start_dev_expect(-EBUSY);
+	io_uring_queue_exit(&ring);
+	return ret;
+}
+
+static int test_partial_fetch(void)
+{
+	struct server b = { .q = 0, .first_tag = DEPTH - 1, .nr_tags = 1,
+			    .delay_us = SERVE_DELAY_US };
+	struct reads r = {};
+	int ret;
+
+	if (add_dev(1, 0) || fetch_and_die(0, 0, DEPTH - 1))
+		return KSFT_FAIL;
+
+	/* the last FETCH makes the queue ready and clears ->canceling */
+	server_start(&b);
+
+	ret = start_dev();
+	if (!ret && reads_submit(&r, -1))
+		ret = KSFT_FAIL;
+	else
+		ret = expect_enodev_or_reads("START_DEV", ret, &r);
+
+	cleanup();
+	pthread_join(b.thread, NULL);
+	return ret;
+}
+
+static int test_stop_restart(void)
+{
+	struct server s = { .q = 0, .nr_tags = DEPTH };
+	struct reads r = {};
+	int ret;
+
+	if (add_dev_noopen(1, 0))
+		return KSFT_FAIL;
+
+	/* no server is attached, so there is nothing to cancel */
+	ret = ublk_ctrl_stop_dev(ctrl());
+	printf("STOP_DEV: %d\n", ret);
+	if (open_cdev())
+		return KSFT_FAIL;
+
+	server_start(&s);
+	if (start_dev()) {
+		ret = KSFT_FAIL;
+	} else if (reads_submit(&r, -1)) {
+		ret = KSFT_FAIL;
+	} else {
+		reads_reap(&r, 10);
+		ret = reads_result(&r);
+		if (r.ok != NR_READS)
+			ret = KSFT_FAIL;
+	}
+
+	cleanup();
+	pthread_join(s.thread, NULL);
+	return ret;
+}
+
+struct start_arg {
+	pthread_barrier_t go;
+	int ret, reads_ok;
+};
+
+static void *race_start_fn(void *data)
+{
+	struct start_arg *a = data;
+	struct reads r = {};
+
+	pthread_barrier_wait(&a->go);
+	a->ret = ublk_ctrl_start_dev(ctrl(), getpid());
+	a->reads_ok = 1;
+	/*
+	 * The disk may be gone already if STOP_DEV came right after, and
+	 * reads may fail then: they only have to complete.
+	 */
+	if (!a->ret && !reads_submit(&r, -1)) {
+		reads_reap(&r, 5);
+		a->reads_ok = r.done == NR_READS;
+		if (a->reads_ok)
+			reads_put(&r);
+		else
+			reads_result(&r);
+	}
+	ctrl_put();
+	return NULL;
+}
+
+static int test_race_start(void)
+{
+	int live = 0, ebusy = 0;
+
+	open_tries = 4;
+	for (int i = 0; i < RACE_LOOPS; i++) {
+		struct server s = { .q = 0, .nr_tags = DEPTH };
+		struct start_arg a = {};
+		pthread_t t;
+
+		if (add_dev(1, 0))
+			return KSFT_FAIL;
+		server_start(&s);
+		pthread_barrier_init(&a.go, NULL, 2);
+		pthread_create(&t, NULL, race_start_fn, &a);
+		pthread_barrier_wait(&a.go);
+		ublk_ctrl_stop_dev(ctrl());
+		pthread_join(t, NULL);
+		cleanup();
+		pthread_join(s.thread, NULL);
+
+		if (a.ret == 0)
+			live++;
+		else if (a.ret == -EBUSY)
+			ebusy++;
+		if ((a.ret && a.ret != -EBUSY) || !a.reads_ok) {
+			fprintf(stderr, "loop %d: START_DEV %d, reads %s\n",
+				i, a.ret, a.reads_ok ? "ok" : "failed");
+			return KSFT_FAIL;
+		}
+	}
+	printf("%d loops: START_DEV won %d, got -EBUSY %d\n", RACE_LOOPS,
+	       live, ebusy);
+	return KSFT_PASS;
+}
+
+static int test_race_fetch(void)
+{
+	int live = 0, ebusy = 0;
+
+	for (int i = 0; i < RACE_LOOPS; i++) {
+		/* vary who goes first: STOP_DEV, or the server's open */
+		struct server s = { .q = 0, .nr_tags = DEPTH, .race = 1,
+				    .race_delay_us = (i % 10) * 50 };
+		struct reads r = {};
+		int ret;
+
+		if (add_dev_noopen(1, 0))
+			return KSFT_FAIL;
+		server_start(&s);
+		pthread_barrier_wait(&s.go);
+		ublk_ctrl_stop_dev(ctrl());
+		pthread_barrier_wait(&s.fetched);
+
+		ret = ublk_ctrl_start_dev(ctrl(), getpid());
+		if (!ret) {
+			/* nothing was taken, so every read has to succeed */
+			live++;
+			if (reads_submit(&r, -1))
+				return KSFT_FAIL;
+			reads_reap(&r, 5);
+			if (r.ok != NR_READS) {
+				fprintf(stderr, "loop %d: live, but ", i);
+				reads_result(&r);
+				return KSFT_FAIL;
+			}
+			reads_put(&r);
+		} else if (ret == -EBUSY) {
+			ebusy++;
+		} else {
+			fprintf(stderr, "loop %d: START_DEV %d\n", i, ret);
+			return KSFT_FAIL;
+		}
+		cleanup();
+		pthread_join(s.thread, NULL);
+	}
+	printf("%d loops: START_DEV worked %d, got -EBUSY %d\n", RACE_LOOPS,
+	       live, ebusy);
+	return KSFT_PASS;
+}
+
+static int test_race_async_fetch(void)
+{
+	io_cmd_sqe_flags = IOSQE_ASYNC;
+	for (int i = 0; i < ASYNC_LOOPS; i++) {
+		struct async_arg a;
+		pthread_t t;
+
+		if (add_dev(1, 0))
+			return KSFT_FAIL;
+		pthread_barrier_init(&a.go, NULL, 2);
+		pthread_barrier_init(&a.stopped, NULL, 2);
+		pthread_create(&t, NULL, race_async_fn, &a);
+		pthread_barrier_wait(&a.go);
+		/* vary where STOP_DEV lands relative to the FETCHes */
+		usleep((i % 5) * 1000);
+		ublk_ctrl_stop_dev(ctrl());
+		pthread_barrier_wait(&a.stopped);
+		pthread_join(t, NULL);
+		cleanup();
+	}
+	printf("%d loops done\n", ASYNC_LOOPS);
+	return KSFT_PASS;
+}
+
+/*
+ * The stopped server is gone: reopen /dev/ublkcN, start a new server and
+ * the device, and read from it.
+ */
+static int restart_server(void)
+{
+	struct server s = { .q = 0, .nr_tags = DEPTH };
+	struct reads r = {};
+	int ret;
+
+	close_cdev();
+	if (open_cdev())
+		return KSFT_FAIL;
+	server_start(&s);
+	usleep(200000);
+	if (s.failed) {
+		fprintf(stderr, "new server: FETCH failed %d\n", s.failed);
+		cleanup();
+		pthread_join(s.thread, NULL);
+		return KSFT_FAIL;
+	}
+	/* -EEXIST until the old disk is freed, e.g. after a udev probe */
+	for (int i = 0; i < 100; i++) {
+		ret = start_dev();
+		if (ret != -EEXIST)
+			break;
+		usleep(50000);
+	}
+	if (ret || reads_submit(&r, -1)) {
+		fprintf(stderr, "new server: START_DEV %d\n", ret);
+		ret = KSFT_FAIL;
+	} else {
+		reads_reap(&r, 10);
+		ret = reads_result(&r);
+		if (r.ok != NR_READS)
+			ret = KSFT_FAIL;
+	}
+	cleanup();
+	pthread_join(s.thread, NULL);
+	return ret;
+}
+
+/*
+ * stop_attached: STOP_DEV while a server has the device open but has
+ * fetched nothing stops that server too: its FETCH gets ABORT and
+ * START_DEV -EBUSY. Once it is gone, a new server can start the device.
+ */
+static int test_stop_attached(void)
+{
+	struct server s = { .q = 0, .nr_tags = DEPTH };
+	int ret;
+
+	if (add_dev(1, 0))
+		return KSFT_FAIL;
+	ret = ublk_ctrl_stop_dev(ctrl());
+	printf("STOP_DEV: %d\n", ret);
+
+	server_start(&s);
+	ret = server_wait_failed(&s);
+	printf("FETCH after STOP_DEV: %d\n", ret);
+	if (ret != UBLK_IO_RES_ABORT) {
+		cleanup();
+		pthread_join(s.thread, NULL);
+		return KSFT_FAIL;
+	}
+	pthread_join(s.thread, NULL);
+	if (expect_err("START_DEV", start_dev(), -EBUSY))
+		return KSFT_FAIL;
+	return restart_server();
+}
+
+/*
+ * stop_live_restart: STOP_DEV on a live device; once its server is gone,
+ * a new server has to fetch and start it again.
+ */
+static int test_stop_live_restart(void)
+{
+	struct server s = { .q = 0, .nr_tags = DEPTH };
+	int ret;
+
+	if (add_dev(1, 0))
+		return KSFT_FAIL;
+	server_start(&s);
+	if (expect_err("START_DEV", start_dev(), 0))
+		return KSFT_FAIL;
+	ret = ublk_ctrl_stop_dev(ctrl());
+	printf("STOP_DEV: %d\n", ret);
+	pthread_join(s.thread, NULL);
+	return restart_server();
+}
+
+/* first CPU which blk-mq maps to hw queue @q */
+static int queue_cpu(int q)
+{
+	char path[96];
+	FILE *f;
+	int cpu = -1;
+
+	snprintf(path, sizeof(path), "/sys/block/ublkb%d/mq/%d/cpu_list",
+		 dev_id, q);
+	for (int i = 0; i < 100 && !(f = fopen(path, "r")); i++)
+		usleep(50000);
+	if (!f)
+		return -1;
+	if (fscanf(f, "%d", &cpu) != 1)
+		cpu = -1;
+	fclose(f);
+	return cpu;
+}
+
+static int test_recovery(void)
+{
+	struct server q1 = { .q = 1, .nr_tags = DEPTH };
+	struct io_uring ring;
+	struct reads r = {};
+	int ret, cpu;
+
+	if (add_dev(NR_QUEUES, UBLK_F_USER_RECOVERY))
+		return KSFT_FAIL;
+
+	/* the first server: fetch everything, start, then die */
+	if (io_uring_queue_init(NR_QUEUES * DEPTH, &ring, 0))
+		return KSFT_FAIL;
+	for (int q = 0; q < NR_QUEUES; q++)
+		for (int tag = 0; tag < DEPTH; tag++)
+			queue_io_cmd(&ring, UBLK_U_IO_FETCH_REQ, q, tag, 0);
+	io_uring_submit(&ring);
+	if (start_dev())
+		return KSFT_FAIL;
+	cpu = queue_cpu(0);
+	if (cpu < 0) {
+		printf("no CPU maps to queue 0, can't aim the reads\n");
+		return KSFT_SKIP;
+	}
+
+	io_uring_queue_exit(&ring);
+	close_cdev();
+	for (int i = 0; i < 100 && dev_state() != UBLK_S_DEV_QUIESCED; i++)
+		usleep(50000);
+	printf("server exited, state %d (QUIESCED is %d)\n", dev_state(),
+	       UBLK_S_DEV_QUIESCED);
+
+	for (int i = 0; i < 100; i++) {
+		ret = ublk_ctrl_start_user_recovery(ctrl());
+		if (ret != -EBUSY)
+			break;
+		usleep(50000);
+	}
+	printf("START_USER_RECOVERY: %d\n", ret);
+	if (ret || open_cdev())
+		return KSFT_FAIL;
+
+	/* queue 0 gets ready, then its task dies before queue 1 is ready */
+	if (fetch_and_die(0, 0, DEPTH))
+		return KSFT_FAIL;
+
+	printf("reads on cpu %d, which maps to queue 0\n", cpu);
+	if (reads_submit(&r, cpu))
+		return KSFT_FAIL;
+
+	server_start(&q1);
+	ret = ublk_ctrl_end_user_recovery(ctrl(), getpid());
+	printf("END_USER_RECOVERY: %d\n", ret);
+	if (ret && ret != -ENODEV) {
+		fprintf(stderr, "END_USER_RECOVERY: %d, expected 0 or %d\n",
+			ret, -ENODEV);
+		ret = KSFT_FAIL;
+	} else {
+		ret = KSFT_PASS;
+	}
+
+	/*
+	 * After -ENODEV, requests held back on queue 0 wait for STOP_DEV to
+	 * fail them. Close the disk before DEL_DEV, which waits for its last
+	 * reference.
+	 */
+	reads_reap(&r, 1);
+	ublk_ctrl_stop_dev(ctrl());
+	reads_reap(&r, 5);
+	if (reads_result(&r))
+		ret = KSFT_FAIL;
+	cleanup();
+	pthread_join(q1.thread, NULL);
+	return ret;
+}
+
+static const struct {
+	const char *name;
+	int (*fn)(void);
+} modes[] = {
+	{ "stop_start",		test_stop_start },
+	{ "partial_fetch",	test_partial_fetch },
+	{ "recovery",		test_recovery },
+	{ "stop_restart",	test_stop_restart },
+	{ "race_start",		test_race_start },
+	{ "race_fetch",		test_race_fetch },
+	{ "race_async_fetch",	test_race_async_fetch },
+	{ "stop_attached",	test_stop_attached },
+	{ "stop_live_restart",	test_stop_live_restart },
+};
+
+int main(int argc, char **argv)
+{
+	int (*fn)(void) = NULL;
+	int ret;
+
+	for (int i = 0; argc == 2 && i < ARRAY_SIZE(modes); i++)
+		if (!strcmp(argv[1], modes[i].name))
+			fn = modes[i].fn;
+	if (!fn) {
+		fprintf(stderr, "usage: %s MODE, modes:", argv[0]);
+		for (int i = 0; i < ARRAY_SIZE(modes); i++)
+			fprintf(stderr, " %s", modes[i].name);
+		fprintf(stderr, "\n");
+		return KSFT_FAIL;
+	}
+
+	/* keep the output of a run that ends in an oops */
+	setvbuf(stdout, NULL, _IOLBF, 0);
+
+	if (access(CTRL_DEV, F_OK)) {
+		perror(CTRL_DEV);
+		return KSFT_SKIP;
+	}
+	printf("%s\n", argv[1]);
+	ret = fn();
+	cleanup();
+	ctrl_put();
+	return ret;
+}
-- 
2.55.0

[PATCH] ublk: refuse to go live after an io command was canceled

[Josef Bacik]

Since commit "ublk: keep a canceled FETCH round canceling until the
server is gone", a device whose FETCH round saw a cancel keeps its
queues canceling until the server goes away, but START_DEV and
END_USER_RECOVERY still bring it up. Without UBLK_F_USER_RECOVERY, or
with UBLK_F_USER_RECOVERY_FAIL_IO, every request of the new disk fails.
With UBLK_F_USER_RECOVERY, requests are requeued and never kicked: after
START_DEV the partition scan hangs under disk->open_mutex, and after
END_USER_RECOVERY every read parks while the command returned 0. The
server cannot fetch the canceled commands again, so the device can't
serve I/O until it restarts anyway.

Return -ENODEV from START_DEV and END_USER_RECOVERY while ub->canceling
is set. In ublk_ctrl_start_dev() check it and publish ub->ub_disk in one
cancel_mutex section, and have ublk_start_cancel() read the disk in its
cancel_mutex section. Today ublk_start_cancel() samples the disk before
taking the mutex, so a server dying during its own START_DEV can mark
the queues without quiescing a disk START_DEV published in between, with
its first I/O past the canceling check. Now either START_DEV sees the
cancel, or the cancel sees the disk and quiesces it before marking. The
END_USER_RECOVERY check is best effort: the disk exists there, and a
cancel after it is the ordinary death of the new server, which
ublk_start_cancel() handles by quiescing and marking.

Assisted-by: LLM
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
This applies on top of Ming's "[PATCH 0/8] ublk: don't dispatch to
canceled io commands" and needs patch 1 of it for ub->canceling to stay
set for the whole FETCH round. generic_18 still passes with it.

 Documentation/block/ublk.rst | 10 ++++++++--
 drivers/block/ublk_drv.c     | 38 ++++++++++++++++++++++++++++++++++--
 2 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/Documentation/block/ublk.rst b/Documentation/block/ublk.rst
index 28300fee22bf..b7875a3cf3fc 100644
--- a/Documentation/block/ublk.rst
+++ b/Documentation/block/ublk.rst
@@ -118,7 +118,11 @@ managing and controlling ublk devices with help of several control commands:
   After the server prepares userspace resources (such as creating I/O handler
   threads & io_uring for handling ublk IO), this command is sent to the
   driver for allocating & exposing ``/dev/ublkb*``. Parameters set via
-  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device.
+  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device. The command
+  fails with ``-ENODEV`` if an I/O command fetched by the current server
+  was canceled, because its io_uring is gone. The server can't fetch it
+  again, and the device can be started again once the server has closed
+  ``/dev/ublkc*``.
 
 - ``UBLK_CMD_STOP_DEV``
 
@@ -195,7 +199,9 @@ managing and controlling ublk devices with help of several control commands:
   command is accepted after ublk device is quiesced and a new process has
   opened ``/dev/ublkc*`` and get all ublk queues be ready. When this command
   returns, ublk device is unquiesced and new I/O requests are passed to the
-  new process.
+  new process. It fails with ``-ENODEV`` if an I/O command of the new
+  process was canceled already. The recovery can be started over once the
+  new process has closed ``/dev/ublkc*``.
 
 - user recovery feature description
 
diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index f57d544c1da2..39eb7775a351 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2759,9 +2759,11 @@ static void ublk_abort_queue(struct ublk_device *ub, struct ublk_queue *ubq)
 
 static void ublk_start_cancel(struct ublk_device *ub)
 {
-	struct gendisk *disk = ublk_get_disk(ub);
+	struct gendisk *disk;
 
+	/* sync with ublk_ctrl_start_dev() publishing the disk */
 	mutex_lock(&ub->cancel_mutex);
+	disk = ublk_get_disk(ub);
 	if (ub->canceling)
 		goto out;
 
@@ -4575,6 +4577,7 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
 		.dma_alignment		= 3,
 	};
 	struct gendisk *disk;
+	bool canceled;
 	int ret = -EINVAL;
 
 	if (ublksrv_pid <= 0)
@@ -4665,8 +4668,24 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
 	disk->fops = &ub_fops;
 	disk->private_data = ub;
 
+	/*
+	 * A command of this FETCH round was canceled and can't be fetched
+	 * again, don't bring up a disk over it.  Check and publish the disk
+	 * in one cancel_mutex section: either this sees ub->canceling, or
+	 * ublk_start_cancel() sees the disk and quiesces it before marking
+	 * the queues.
+	 */
+	mutex_lock(&ub->cancel_mutex);
+	canceled = ub->canceling;
+	if (!canceled)
+		ub->ub_disk = disk;
+	mutex_unlock(&ub->cancel_mutex);
+	if (canceled) {
+		put_disk(disk);
+		ret = -ENODEV;
+		goto out_unlock;
+	}
 	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
-	ub->ub_disk = disk;
 
 	ublk_apply_params(ub);
 
@@ -5238,6 +5257,7 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
 		const struct ublksrv_ctrl_cmd *header)
 {
 	int ublksrv_pid = (int)header->data[0];
+	bool canceled;
 	int ret = -EINVAL;
 
 	pr_devel("%s: Waiting for all FETCH_REQs, dev id %d...\n", __func__,
@@ -5261,6 +5281,20 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
 		ret = -EBUSY;
 		goto out_unlock;
 	}
+
+	/*
+	 * As in ublk_ctrl_start_dev(), a canceled command can't be fetched
+	 * again.  Best effort: the disk exists here, and a cancel after this
+	 * check is the ordinary death of the new server, which
+	 * ublk_start_cancel() handles by quiescing and marking.
+	 */
+	mutex_lock(&ub->cancel_mutex);
+	canceled = ub->canceling;
+	mutex_unlock(&ub->cancel_mutex);
+	if (canceled) {
+		ret = -ENODEV;
+		goto out_unlock;
+	}
 	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
 	ub->dev_info.state = UBLK_S_DEV_LIVE;
 	pr_devel("%s: new ublksrv_pid %d, dev id %d\n",
-- 
2.55.0

[PATCH v2] ublk: refuse to go live after an io command was canceled

[Josef Bacik]

Since commit "ublk: keep a canceled FETCH round canceling until the
server is gone", a device whose FETCH round saw a cancel keeps its
queues canceling until the server goes away, but START_DEV and
END_USER_RECOVERY still bring it up. Without UBLK_F_USER_RECOVERY, or
with UBLK_F_USER_RECOVERY_FAIL_IO, every request of the new disk fails.
With UBLK_F_USER_RECOVERY, requests are requeued and never kicked: after
START_DEV the partition scan hangs under disk->open_mutex, and after
END_USER_RECOVERY every read parks while the command returned 0. The
server cannot fetch the canceled commands again, so the device can't
serve I/O until it restarts anyway.

Return -EBUSY from START_DEV and END_USER_RECOVERY while ub->canceling
is set. In ublk_ctrl_start_dev() check it and publish ub->ub_disk in one
cancel_mutex section, and have ublk_start_cancel() read the disk in its
cancel_mutex section. Today ublk_start_cancel() samples the disk before
taking the mutex, so a server dying during its own START_DEV can mark
the queues without quiescing a disk START_DEV published in between, with
its first I/O past the canceling check. Now either START_DEV sees the
cancel, or the cancel sees the disk and quiesces it before marking. The
END_USER_RECOVERY check is best effort: the disk exists there, and a
cancel after it is the ordinary death of the new server, which
ublk_start_cancel() handles by quiescing and marking.

Assisted-by: LLM
Reviewed-by: Ming Lei <tom.leiming@gmail.com>
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
v2: -EBUSY instead of -ENODEV, the device isn't gone (Ming)

This applies on top of Ming's "[PATCH 0/8] ublk: don't dispatch to
canceled io commands" and needs patch 1 of it for ub->canceling to stay
set for the whole FETCH round.

 Documentation/block/ublk.rst | 10 ++++++++--
 drivers/block/ublk_drv.c     | 38 ++++++++++++++++++++++++++++++++++--
 2 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/Documentation/block/ublk.rst b/Documentation/block/ublk.rst
index 28300fee22bf..6c5536ebd987 100644
--- a/Documentation/block/ublk.rst
+++ b/Documentation/block/ublk.rst
@@ -118,7 +118,11 @@ managing and controlling ublk devices with help of several control commands:
   After the server prepares userspace resources (such as creating I/O handler
   threads & io_uring for handling ublk IO), this command is sent to the
   driver for allocating & exposing ``/dev/ublkb*``. Parameters set via
-  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device.
+  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device. The command
+  fails with ``-EBUSY`` if an I/O command fetched by the current server
+  was canceled, because its io_uring is gone. The server can't fetch it
+  again, and the device can be started again once the server has closed
+  ``/dev/ublkc*``.
 
 - ``UBLK_CMD_STOP_DEV``
 
@@ -195,7 +199,9 @@ managing and controlling ublk devices with help of several control commands:
   command is accepted after ublk device is quiesced and a new process has
   opened ``/dev/ublkc*`` and get all ublk queues be ready. When this command
   returns, ublk device is unquiesced and new I/O requests are passed to the
-  new process.
+  new process. It fails with ``-EBUSY`` if an I/O command of the new
+  process was canceled already. The recovery can be started over once the
+  new process has closed ``/dev/ublkc*``.
 
 - user recovery feature description
 
diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index f57d544c1da2..015aff07703c 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2759,9 +2759,11 @@ static void ublk_abort_queue(struct ublk_device *ub, struct ublk_queue *ubq)
 
 static void ublk_start_cancel(struct ublk_device *ub)
 {
-	struct gendisk *disk = ublk_get_disk(ub);
+	struct gendisk *disk;
 
+	/* sync with ublk_ctrl_start_dev() publishing the disk */
 	mutex_lock(&ub->cancel_mutex);
+	disk = ublk_get_disk(ub);
 	if (ub->canceling)
 		goto out;
 
@@ -4575,6 +4577,7 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
 		.dma_alignment		= 3,
 	};
 	struct gendisk *disk;
+	bool canceled;
 	int ret = -EINVAL;
 
 	if (ublksrv_pid <= 0)
@@ -4665,8 +4668,24 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
 	disk->fops = &ub_fops;
 	disk->private_data = ub;
 
+	/*
+	 * A command of this FETCH round was canceled and can't be fetched
+	 * again, don't bring up a disk over it.  Check and publish the disk
+	 * in one cancel_mutex section: either this sees ub->canceling, or
+	 * ublk_start_cancel() sees the disk and quiesces it before marking
+	 * the queues.
+	 */
+	mutex_lock(&ub->cancel_mutex);
+	canceled = ub->canceling;
+	if (!canceled)
+		ub->ub_disk = disk;
+	mutex_unlock(&ub->cancel_mutex);
+	if (canceled) {
+		put_disk(disk);
+		ret = -EBUSY;
+		goto out_unlock;
+	}
 	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
-	ub->ub_disk = disk;
 
 	ublk_apply_params(ub);
 
@@ -5238,6 +5257,7 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
 		const struct ublksrv_ctrl_cmd *header)
 {
 	int ublksrv_pid = (int)header->data[0];
+	bool canceled;
 	int ret = -EINVAL;
 
 	pr_devel("%s: Waiting for all FETCH_REQs, dev id %d...\n", __func__,
@@ -5261,6 +5281,20 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
 		ret = -EBUSY;
 		goto out_unlock;
 	}
+
+	/*
+	 * As in ublk_ctrl_start_dev(), a canceled command can't be fetched
+	 * again.  Best effort: the disk exists here, and a cancel after this
+	 * check is the ordinary death of the new server, which
+	 * ublk_start_cancel() handles by quiescing and marking.
+	 */
+	mutex_lock(&ub->cancel_mutex);
+	canceled = ub->canceling;
+	mutex_unlock(&ub->cancel_mutex);
+	if (canceled) {
+		ret = -EBUSY;
+		goto out_unlock;
+	}
 	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
 	ub->dev_info.state = UBLK_S_DEV_LIVE;
 	pr_devel("%s: new ublksrv_pid %d, dev id %d\n",
-- 
2.55.0

Re: [PATCH 0/8] ublk: don't dispatch to canceled io commands

[Josef Bacik]

On Thu, Oct 01, 2026 at 07:54:14AM -0500, Ming Lei wrote:
> Hi,
>
> Josef reported three ways in which ublk dispatches a request to a
> canceled io command, which oopses on the NULL io->cmd.

Thanks Ming. I ran 1-8 on top of for-next (d70609a2f68c) under QEMU with
KASAN and lockdep: my reproducers for the three routes plus the restart,
recovery and QUIESCE_DEV controls, the ublk selftests including your
generic_18, and my ADD vs STOP_DEV/DEL_DEV race script. No oops or
lockdep report anywhere, and generic_18 passes. A few things though.

1. kublk's batch daemon dies when PREP_IO_CMDS hits the new STOPPING
   check. With patch 6, __ublk_fetch() returns UBLK_IO_RES_ABORT while
   UB_STATE_STOPPING is set, and batch PREP goes through it, so a PREP
   racing STOP_DEV completes with -ENODEV. kublk asserts cqe->res == 0
   in ublk_batch_compl_commit_cmd(), the daemon aborts before it writes
   the start eventfd, and the parent of `kublk add` sits in
   eventfd_read() forever. My race script hangs on the first or second
   iteration with -b; the per-io servers handle the abort fine. The
   kernel side is fine, but kublk needs to fail the queue cleanly there,
   and generic_18 has no batch mode, which is why it doesn't see it.

2. Routes 2 and 3 still bring the device up. START_DEV returns 0 and
   every request on the new disk fails, and after END_USER_RECOVERY the
   device is LIVE with the reads on the dead task's queue parked forever
   with plain USER_RECOVERY (generic_18 accepts both, and turns off the
   partition scan, which is what parks under disk->open_mutex otherwise).
   I'll send a follow-up on top of your series that returns -ENODEV from
   START_DEV and END_USER_RECOVERY while ub->canceling is set; with it
   generic_18 still passes.

3. The same follow-up has ublk_start_cancel() read ub->ub_disk under
   cancel_mutex, with START_DEV publishing it in the same section. Today
   it samples the disk before taking the mutex, so a server dying during
   its own START_DEV can mark the queues without quiescing a disk that
   START_DEV published in between, with first I/O already past the
   canceling check. Narrow, and older than this series.

4. Patch 5: the release work decides there is no disk before it takes
   ub->mutex and doesn't look again, so a START_DEV that slips in with
   the exiting server's pid can go LIVE first and the work then resets
   a LIVE device. It needs START_DEV pending while the node is released,
   e.g. a separate control process, and nothing oopses (every queue is
   still canceling), but the device ends up LIVE with requests failing
   or parked until STOP_DEV/DEL_DEV.

5. Patch 6 changes STOP_DEV's contract: a server that still holds
   /dev/ublkcN after STOP_DEV gets ABORT on FETCH/PREP and -EBUSY on
   START_DEV until it closes the node, even if it had fetched nothing.
   That's fine by me, but it should go in Documentation/block/ublk.rst,
   since servers that STOP_DEV and restart on the same fd will see it.

For 1-6, with kublk fixed for 1:

Tested-by: Josef Bacik <josef@toxicpanda.com>

Thanks,
Josef

[PATCH 1/4] ublk: don't cancel commands in QUIESCE_DEV on a device that isn't live

[Josef Bacik]

QUIESCE_DEV on a device which is not LIVE returns 0 as "already in
expected state", but ublk_cancel_dev() still runs after ub->mutex is
dropped. A device which is QUIESCED because its server died may have a
new server fetching its commands for recovery at that point, and the
cancel completes them without marking anything:

    new server                       QUIESCE_DEV
    START_USER_RECOVERY
    FETCH part of the queue
                                     device is QUIESCED, ret = 0
                                     ublk_cancel_dev()
                                       io->cmd = NULL, command done
    FETCH the rest of the queue
    END_USER_RECOVERY
      device goes LIVE
                                     ublk_queue_rq()
                                       NULL io->cmd

Skip the cancel when the device was not LIVE. Its server is gone, so
there is nothing of the server QUIESCE_DEV is meant for left to cancel.

Fixes: b465ae7b2524 ("ublk: add feature UBLK_F_QUIESCE")
Assisted-by: LLM
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
 drivers/block/ublk_drv.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 39eb7775a351..0bd0b95b3217 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -5413,6 +5413,7 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 	/* zero means wait forever */
 	u64 timeout_ms = header->data[0];
 	struct gendisk *disk;
+	bool live = true;
 	int ret = -ENODEV;
 
 	if (!(ub->dev_info.flags & UBLK_F_QUIESCE))
@@ -5427,8 +5428,15 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 
 	ret = 0;
 	/* already in expected state */
-	if (ub->dev_info.state != UBLK_S_DEV_LIVE)
+	if (ub->dev_info.state != UBLK_S_DEV_LIVE) {
+		/*
+		 * Nothing to cancel either: the server this was meant for is
+		 * gone, and the commands of the next one, which may be fetching
+		 * them for recovery right now, are not ours to cancel.
+		 */
+		live = false;
 		goto put_disk;
+	}
 
 	/* Mark the device as canceling */
 	mutex_lock(&ub->cancel_mutex);
@@ -5447,7 +5455,7 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 	mutex_unlock(&ub->mutex);
 
 	/* Cancel pending uring_cmd */
-	if (!ret)
+	if (!ret && live)
 		ublk_cancel_dev(ub);
 	return ret;
 }
-- 
2.55.0

Re: [PATCH] ublk: refuse to go live after an io command was canceled

[Ming Lei]

On Mon, Oct 05, 2026 at 04:23:51PM +0000, Josef Bacik wrote:
> Since commit "ublk: keep a canceled FETCH round canceling until the
> server is gone", a device whose FETCH round saw a cancel keeps its
> queues canceling until the server goes away, but START_DEV and
> END_USER_RECOVERY still bring it up. Without UBLK_F_USER_RECOVERY, or
> with UBLK_F_USER_RECOVERY_FAIL_IO, every request of the new disk fails.
> With UBLK_F_USER_RECOVERY, requests are requeued and never kicked: after
> START_DEV the partition scan hangs under disk->open_mutex, and after
> END_USER_RECOVERY every read parks while the command returned 0. The
> server cannot fetch the canceled commands again, so the device can't
> serve I/O until it restarts anyway.
> 
> Return -ENODEV from START_DEV and END_USER_RECOVERY while ub->canceling
> is set. In ublk_ctrl_start_dev() check it and publish ub->ub_disk in one
> cancel_mutex section, and have ublk_start_cancel() read the disk in its
> cancel_mutex section. Today ublk_start_cancel() samples the disk before
> taking the mutex, so a server dying during its own START_DEV can mark
> the queues without quiescing a disk START_DEV published in between, with
> its first I/O past the canceling check. Now either START_DEV sees the
> cancel, or the cancel sees the disk and quiesces it before marking. The
> END_USER_RECOVERY check is best effort: the disk exists there, and a
> cancel after it is the ordinary death of the new server, which
> ublk_start_cancel() handles by quiescing and marking.
> 
> Assisted-by: LLM
> Signed-off-by: Josef Bacik <josef@toxicpanda.com>
> ---
> This applies on top of Ming's "[PATCH 0/8] ublk: don't dispatch to
> canceled io commands" and needs patch 1 of it for ub->canceling to stay
> set for the whole FETCH round. generic_18 still passes with it.
> 
>  Documentation/block/ublk.rst | 10 ++++++++--
>  drivers/block/ublk_drv.c     | 38 ++++++++++++++++++++++++++++++++++--
>  2 files changed, 44 insertions(+), 4 deletions(-)
> 
> diff --git a/Documentation/block/ublk.rst b/Documentation/block/ublk.rst
> index 28300fee22bf..b7875a3cf3fc 100644
> --- a/Documentation/block/ublk.rst
> +++ b/Documentation/block/ublk.rst
> @@ -118,7 +118,11 @@ managing and controlling ublk devices with help of several control commands:
>    After the server prepares userspace resources (such as creating I/O handler
>    threads & io_uring for handling ublk IO), this command is sent to the
>    driver for allocating & exposing ``/dev/ublkb*``. Parameters set via
> -  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device.
> +  ``UBLK_CMD_SET_PARAMS`` are applied for creating the device. The command
> +  fails with ``-ENODEV`` if an I/O command fetched by the current server
> +  was canceled, because its io_uring is gone. The server can't fetch it
> +  again, and the device can be started again once the server has closed
> +  ``/dev/ublkc*``.
>  
>  - ``UBLK_CMD_STOP_DEV``
>  
> @@ -195,7 +199,9 @@ managing and controlling ublk devices with help of several control commands:
>    command is accepted after ublk device is quiesced and a new process has
>    opened ``/dev/ublkc*`` and get all ublk queues be ready. When this command
>    returns, ublk device is unquiesced and new I/O requests are passed to the
> -  new process.
> +  new process. It fails with ``-ENODEV`` if an I/O command of the new
> +  process was canceled already. The recovery can be started over once the
> +  new process has closed ``/dev/ublkc*``.
>  
>  - user recovery feature description
>  
> diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
> index f57d544c1da2..39eb7775a351 100644
> --- a/drivers/block/ublk_drv.c
> +++ b/drivers/block/ublk_drv.c
> @@ -2759,9 +2759,11 @@ static void ublk_abort_queue(struct ublk_device *ub, struct ublk_queue *ubq)
>  
>  static void ublk_start_cancel(struct ublk_device *ub)
>  {
> -	struct gendisk *disk = ublk_get_disk(ub);
> +	struct gendisk *disk;
>  
> +	/* sync with ublk_ctrl_start_dev() publishing the disk */
>  	mutex_lock(&ub->cancel_mutex);
> +	disk = ublk_get_disk(ub);
>  	if (ub->canceling)
>  		goto out;
>  
> @@ -4575,6 +4577,7 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
>  		.dma_alignment		= 3,
>  	};
>  	struct gendisk *disk;
> +	bool canceled;
>  	int ret = -EINVAL;
>  
>  	if (ublksrv_pid <= 0)
> @@ -4665,8 +4668,24 @@ static int ublk_ctrl_start_dev(struct ublk_device *ub,
>  	disk->fops = &ub_fops;
>  	disk->private_data = ub;
>  
> +	/*
> +	 * A command of this FETCH round was canceled and can't be fetched
> +	 * again, don't bring up a disk over it.  Check and publish the disk
> +	 * in one cancel_mutex section: either this sees ub->canceling, or
> +	 * ublk_start_cancel() sees the disk and quiesces it before marking
> +	 * the queues.
> +	 */
> +	mutex_lock(&ub->cancel_mutex);
> +	canceled = ub->canceling;
> +	if (!canceled)
> +		ub->ub_disk = disk;
> +	mutex_unlock(&ub->cancel_mutex);
> +	if (canceled) {
> +		put_disk(disk);
> +		ret = -ENODEV;
> +		goto out_unlock;
> +	}
>  	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
> -	ub->ub_disk = disk;
>  
>  	ublk_apply_params(ub);
>  
> @@ -5238,6 +5257,7 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
>  		const struct ublksrv_ctrl_cmd *header)
>  {
>  	int ublksrv_pid = (int)header->data[0];
> +	bool canceled;
>  	int ret = -EINVAL;
>  
>  	pr_devel("%s: Waiting for all FETCH_REQs, dev id %d...\n", __func__,
> @@ -5261,6 +5281,20 @@ static int ublk_ctrl_end_recovery(struct ublk_device *ub,
>  		ret = -EBUSY;
>  		goto out_unlock;
>  	}
> +
> +	/*
> +	 * As in ublk_ctrl_start_dev(), a canceled command can't be fetched
> +	 * again.  Best effort: the disk exists here, and a cancel after this
> +	 * check is the ordinary death of the new server, which
> +	 * ublk_start_cancel() handles by quiescing and marking.
> +	 */
> +	mutex_lock(&ub->cancel_mutex);
> +	canceled = ub->canceling;
> +	mutex_unlock(&ub->cancel_mutex);
> +	if (canceled) {
> +		ret = -ENODEV;
> +		goto out_unlock;
> +	}
>  	ub->dev_info.ublksrv_pid = ub->ublksrv_tgid;
>  	ub->dev_info.state = UBLK_S_DEV_LIVE;
>  	pr_devel("%s: new ublksrv_pid %d, dev id %d\n",

Hi Josef,

Thanks for the follow-up. The check and the cancel_mutex ordering
look right to me, but I'd suggest -EBUSY instead of -ENODEV.

-ENODEV means the ublk device is gone, and it isn't true in the
START_DEV/END_RECOVERY cases.

With -EBUSY:

Reviewed-by: Ming Lei <tom.leiming@gmail.com>


thanks,
Ming

[PATCH 2/4] ublk: drop QUIESCE_DEV's wait for an idle command

[Josef Bacik]

ublk_wait_for_idle_io() is meant to wait until every queue has a command
whose request is not with the server, so that canceling it tells the
server about the quiesce. It never waits: blk_mq_tagset_busy_iter()
only calls ublk_count_busy_req() for started requests, and the callback
counts a request only when it is not started, so nr_busy is always 0
and every queue looks idle.

Making it count would not help either. It runs with ub->mutex held, so
a server which keeps every tag busy would hold up STOP_DEV, recovery
and its own release work for as long as the QUIESCE_DEV timeout, which
is forever by default. Drop it; nothing changes, since it never waited,
and a later patch has QUIESCE_DEV keep canceling until the server has
been told, without ub->mutex held.

The QUIESCE_DEV timeout in data[0] is unused until then.

Fixes: b465ae7b2524 ("ublk: add feature UBLK_F_QUIESCE")
Assisted-by: LLM
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
 drivers/block/ublk_drv.c | 75 ----------------------------------------
 1 file changed, 75 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 0bd0b95b3217..bd7126dcf92f 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -5338,80 +5338,9 @@ static int ublk_ctrl_set_size(struct ublk_device *ub, const struct ublksrv_ctrl_
 	return ret;
 }
 
-struct count_busy {
-	const struct ublk_queue *ubq;
-	u16 nr_busy;
-};
-
-static bool ublk_count_busy_req(struct request *rq, void *data)
-{
-	struct count_busy *idle = data;
-
-	if (!blk_mq_request_started(rq) && rq->mq_hctx->driver_data == idle->ubq)
-		idle->nr_busy += 1;
-	return true;
-}
-
-/* uring_cmd is guaranteed to be active if the associated request is idle */
-static bool ubq_has_idle_io(const struct ublk_queue *ubq)
-{
-	struct count_busy data = {
-		.ubq = ubq,
-	};
-
-	blk_mq_tagset_busy_iter(&ubq->dev->tag_set, ublk_count_busy_req, &data);
-	return data.nr_busy < ubq->q_depth;
-}
-
-/* Wait until each hw queue has at least one idle IO */
-static int ublk_wait_for_idle_io(struct ublk_device *ub,
-				 unsigned int timeout_ms)
-{
-	unsigned int elapsed = 0;
-	int ret;
-
-	/*
-	 * For UBLK_F_BATCH_IO ublk server can get notified with existing
-	 * or new fetch command, so needn't wait any more
-	 */
-	if (ublk_dev_support_batch_io(ub))
-		return 0;
-
-	while (elapsed < timeout_ms && !signal_pending(current)) {
-		u16 i, queues_cancelable = 0;
-
-		for (i = 0; i < ub->dev_info.nr_hw_queues; i++) {
-			struct ublk_queue *ubq = ublk_get_queue(ub, i);
-
-			queues_cancelable += !!ubq_has_idle_io(ubq);
-		}
-
-		/*
-		 * Each queue needs at least one active command for
-		 * notifying ublk server
-		 */
-		if (queues_cancelable == ub->dev_info.nr_hw_queues)
-			break;
-
-		msleep(UBLK_REQUEUE_DELAY_MS);
-		elapsed += UBLK_REQUEUE_DELAY_MS;
-	}
-
-	if (signal_pending(current))
-		ret = -EINTR;
-	else if (elapsed >= timeout_ms)
-		ret = -EBUSY;
-	else
-		ret = 0;
-
-	return ret;
-}
-
 static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 				 const struct ublksrv_ctrl_cmd *header)
 {
-	/* zero means wait forever */
-	u64 timeout_ms = header->data[0];
 	struct gendisk *disk;
 	bool live = true;
 	int ret = -ENODEV;
@@ -5445,10 +5374,6 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 	blk_mq_unquiesce_queue(disk->queue);
 	mutex_unlock(&ub->cancel_mutex);
 
-	if (!timeout_ms)
-		timeout_ms = UINT_MAX;
-	ret = ublk_wait_for_idle_io(ub, timeout_ms);
-
 put_disk:
 	ublk_put_disk(disk);
 unlock:
-- 
2.55.0

[PATCH 3/4] ublk: give the command back from COMMIT_AND_FETCH on a canceling queue

[Josef Bacik]

COMMIT_AND_FETCH and NEED_GET_DATA publish the server's next command in
io->cmd without a lock. A cancel which runs at the same time can see
the io active and still read the request pointer io->cmd shares its
storage with, or miss the command altogether. QUIESCE_DEV cancels while
the server still commits, and the command published right after its
cancel pass is never completed, which is what leaves the server waiting
for it forever.

Have the issuer decide instead: read ->canceling before publishing, and
on a canceling queue complete the committed request as usual, but give
the new command back with UBLK_IO_RES_ABORT instead of publishing it,
leaving the io canceled the way ublk_cancel_cmd() does. NEED_GET_DATA
sends its request back the way ublk_queue_rq() does on a canceling
queue.

The read and the publish are one RCU read section, so a cancel which
marks the queue and then calls synchronize_rcu() knows every command
published without seeing the mark is in place before it looks, and
every later one comes back from its issuer. That keeps the commit path
free of locks and barriers. ->canceling is read locklessly now, so write
it with WRITE_ONCE().

Assisted-by: LLM
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
 drivers/block/ublk_drv.c | 66 +++++++++++++++++++++++++++++++++++++---
 1 file changed, 61 insertions(+), 5 deletions(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index bd7126dcf92f..6717dabf3a23 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2492,6 +2492,9 @@ static void ublk_partition_scan_work(struct work_struct *work)
  * - there are no concurrent reads of ubq->canceling from the queue_rq
  *   path. This can be done by quiescing the queue, or through other
  *   means.
+ *
+ * ublk_commit_io_cmd() reads ubq->canceling locklessly, so it is written
+ * with WRITE_ONCE().
  */
 static void ublk_set_canceling(struct ublk_device *ub, bool canceling)
 	__must_hold(&ub->cancel_mutex)
@@ -2500,7 +2503,7 @@ static void ublk_set_canceling(struct ublk_device *ub, bool canceling)
 
 	ub->canceling = canceling;
 	for (i = 0; i < ub->dev_info.nr_hw_queues; i++)
-		ublk_get_queue(ub, i)->canceling = canceling;
+		WRITE_ONCE(ublk_get_queue(ub, i)->canceling, canceling);
 }
 
 static bool ublk_check_and_reset_active_ref(struct ublk_device *ub)
@@ -3109,7 +3112,7 @@ static void ublk_queue_reset_io_flags(struct ublk_device *ub,
 	 */
 	mutex_lock(&ub->cancel_mutex);
 	if (!ub->canceling)
-		ubq->canceling = false;
+		WRITE_ONCE(ubq->canceling, false);
 	mutex_unlock(&ub->cancel_mutex);
 	ubq->fail_io = false;
 	ubq->force_abort = false;
@@ -3227,6 +3230,49 @@ ublk_fill_io_cmd(struct ublk_io *io, struct io_uring_cmd *cmd)
 	return req;
 }
 
+/*
+ * Instead of ublk_fill_io_cmd() on a canceling queue: the server's new
+ * command is not published, it goes back with UBLK_IO_RES_ABORT, and the
+ * io is left canceled the way ublk_cancel_cmd() leaves it.  Returns the
+ * request the server owned.
+ */
+static struct request *ublk_cancel_io_cmd(struct ublk_queue *ubq,
+					  struct ublk_io *io)
+{
+	struct request *req = io->req;
+
+	spin_lock(&ubq->cancel_lock);
+	io->flags &= ~(UBLK_IO_FLAG_OWNED_BY_SRV | UBLK_IO_FLAG_NEED_GET_DATA);
+	io->flags |= UBLK_IO_FLAG_CANCELED;
+	spin_unlock(&ubq->cancel_lock);
+
+	return req;
+}
+
+/*
+ * Publish the command a COMMIT_AND_FETCH or NEED_GET_DATA brings, unless
+ * the queue is canceling, see ublk_quiesce_cancel().  The read of
+ * ->canceling and the publish are one RCU read section: a cancel which
+ * marks the queue after the read waits in synchronize_rcu() until the
+ * command is published, and claims it then.  Returns whether the command
+ * goes back to the server instead.
+ */
+static bool ublk_commit_io_cmd(struct ublk_queue *ubq, struct ublk_io *io,
+			       struct io_uring_cmd *cmd, struct request **req)
+{
+	bool canceling;
+
+	rcu_read_lock();
+	canceling = READ_ONCE(ubq->canceling);
+	if (likely(!canceling))
+		*req = ublk_fill_io_cmd(io, cmd);
+	else
+		*req = ublk_cancel_io_cmd(ubq, io);
+	rcu_read_unlock();
+
+	return canceling;
+}
+
 /*
  * Call before ublk_fill_io_cmd() publishes @cmd in io->cmd: a control-path
  * cancel may complete any command found there, and io_uring_cmd_done() only
@@ -3465,6 +3511,7 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 	u64 addr = READ_ONCE(ub_src->addr); /* unioned with zone_append_lba */
 	struct request *req;
 	int ret;
+	bool canceled;
 	bool compl;
 
 	WARN_ON_ONCE(issue_flags & IO_URING_F_UNLOCKED);
@@ -3547,7 +3594,7 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 			goto out;
 		io->res = result;
 		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
-		req = ublk_fill_io_cmd(io, cmd);
+		canceled = ublk_commit_io_cmd(ubq, io, cmd, &req);
 		ublk_apply_io_buf(ub, io, cmd, addr, &auto_buf, &buf_idx);
 		if (buf_idx != UBLK_INVALID_BUF_IDX)
 			io_buffer_unregister(cmd, buf_idx, issue_flags);
@@ -3557,6 +3604,10 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 			req->__sector = addr;
 		if (compl)
 			__ublk_complete_rq(req, io, ublk_dev_need_map_io(ub), NULL);
+		if (unlikely(canceled)) {
+			ret = UBLK_IO_RES_ABORT;
+			goto out_done;
+		}
 		break;
 	}
 	case UBLK_IO_NEED_GET_DATA:
@@ -3566,7 +3617,12 @@ static int ublk_ch_uring_cmd_local(struct io_uring_cmd *cmd,
 		 * request
 		 */
 		ublk_prep_cancel(cmd, issue_flags, ubq, tag);
-		req = ublk_fill_io_cmd(io, cmd);
+		if (unlikely(ublk_commit_io_cmd(ubq, io, cmd, &req))) {
+			/* as ublk_queue_rq() does on a canceling queue */
+			__ublk_abort_rq(ubq, req);
+			ret = UBLK_IO_RES_ABORT;
+			goto out_done;
+		}
 		io->buf.addr = addr;
 		if (likely(ublk_get_data(ubq, io, req))) {
 			__ublk_prep_compl_io_cmd(io, req);
@@ -3765,7 +3821,7 @@ static int ublk_batch_unprep_io(struct ublk_queue *ubq,
 	if (ublk_queue_ready(ubq)) {
 		data->ub->nr_queue_ready--;
 		spin_lock(&ubq->cancel_lock);
-		ubq->canceling = true;
+		WRITE_ONCE(ubq->canceling, true);
 		spin_unlock(&ubq->cancel_lock);
 	}
 	ubq->nr_io_ready--;
-- 
2.55.0

[PATCH 4/4] ublk: keep canceling in QUIESCE_DEV until the server's commands are taken

[Josef Bacik]

UBLK_CMD_QUIESCE_DEV can leave the server with a command nothing ever
completes. The server then never exits and the device stays LIVE:

    COMMIT_AND_FETCH                    QUIESCE_DEV
                                        queues marked canceling
                                        ublk_cancel_dev()
                                          request still with the
                                          server, io left alone
    ublk_fill_io_cmd()
      command armed again
    request completed

The same holds for a command whose request was dispatched before the
mark, and for the active fetch command of a UBLK_F_BATCH_IO queue, which
ublk_batch_cancel_queue() leaves to its dispatcher, and the dispatcher
only lets go of it. The kublk selftest server hangs this way within a
few quiesce and recover cycles under fio, on every kind of queue.

Since the previous patch a COMMIT_AND_FETCH on a canceling queue gives
its command back itself, once synchronize_rcu() has passed after the
mark. Keep taking the armed commands until no io of the server owes one
any more, and an active batch fetch command once its dispatcher has put
it back on the list. Stop on the QUIESCE_DEV timeout or a signal, with
-EBUSY or -EINTR. Leave ->force_abort of a batch queue alone, which
ublk_batch_cancel_queue() sets: requests of a recoverable device are
then requeued through ->canceling until the next server is ready, as on
a queue without UBLK_F_BATCH_IO, instead of failed.

The server may go away and a new one start fetching for recovery while
this runs, and its commands are not QUIESCE_DEV's to cancel. Count the
FETCH rounds in ub->fetch_round, which ublk_reset_ch_dev() bumps under
cancel_mutex when it clears ub->canceling, before a new server can
fetch. Each pass checks the round and takes the commands in one
cancel_mutex hold, and stops once the round has changed. The commands
are completed after cancel_mutex is dropped, since the ring's cancel
callback takes cancel_mutex under uring_lock.

Fixes: b465ae7b2524 ("ublk: add feature UBLK_F_QUIESCE")
Assisted-by: LLM
Signed-off-by: Josef Bacik <josef@toxicpanda.com>
---
 drivers/block/ublk_drv.c | 187 ++++++++++++++++++++++++++++++++++++++-
 1 file changed, 186 insertions(+), 1 deletion(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 6717dabf3a23..f8a5dd7e9404 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -129,6 +129,8 @@ struct ublk_uring_cmd_pdu {
 	union {
 		struct request *req;
 		struct request *req_list;
+		/* chains commands QUIESCE_DEV took, none has a request */
+		struct io_uring_cmd *next_claimed;
 	};
 
 	/*
@@ -296,6 +298,9 @@ struct ublk_queue {
 
 		/* Currently active fetch command (NULL = none active) */
 		struct ublk_batch_fetch_cmd  *active_fcmd;
+
+		/* fetch commands QUIESCE_DEV took, for it to complete */
+		struct list_head quiesce_fcmds;
 	}____cacheline_aligned_in_smp;
 
 	struct ublk_io ios[] __counted_by(q_depth);
@@ -344,6 +349,12 @@ struct ublk_device {
 	 * it is set, no queue clears its ->canceling.
 	 */
 	bool canceling;
+	/*
+	 * Counts FETCH rounds, bumped by ublk_reset_ch_dev() together with
+	 * clearing ->canceling, protected by cancel_mutex.  A cancel aimed at
+	 * one server checks it to leave the next server's commands alone.
+	 */
+	u32 fetch_round;
 	pid_t 	ublksrv_tgid;
 	struct delayed_work	exit_work;
 	struct work_struct	partition_scan_work;
@@ -2432,6 +2443,7 @@ static void ublk_reset_ch_dev(struct ublk_device *ub)
 	/* a new FETCH round starts, the queues stay canceling until ready */
 	mutex_lock(&ub->cancel_mutex);
 	ub->canceling = false;
+	ub->fetch_round++;
 	mutex_unlock(&ub->cancel_mutex);
 
 	/* set to NULL, otherwise new tasks cannot mmap io_cmd_buf */
@@ -4417,6 +4429,7 @@ static int ublk_init_queue(struct ublk_device *ub, u16 q_id)
 		if (ret)
 			goto fail;
 		INIT_LIST_HEAD(&ubq->fcmd_head);
+		INIT_LIST_HEAD(&ubq->quiesce_fcmds);
 	}
 	ub->queues[q_id] = ubq;
 	ubq->dev = ub;
@@ -5394,11 +5407,180 @@ static int ublk_ctrl_set_size(struct ublk_device *ub, const struct ublksrv_ctrl_
 	return ret;
 }
 
+/*
+ * Take the armed commands of a queue off their ios the way ublk_cancel_cmd()
+ * does, and chain them on @claimed.  Only after synchronize_rcu() in
+ * ublk_quiesce_cancel(): no COMMIT_AND_FETCH publishes a command on the
+ * canceling queue any more, see ublk_commit_io_cmd(), so an armed io is
+ * stable here.  Sets @left when an io still owes a command: its request is
+ * with the server, or its dispatch is pending and hands the request to the
+ * server, and either way the server's COMMIT_AND_FETCH gives the command
+ * back.
+ */
+static struct io_uring_cmd *ublk_quiesce_claim_queue(struct ublk_queue *ubq,
+						     struct io_uring_cmd *claimed,
+						     bool *left)
+	__must_hold(&ubq->dev->cancel_mutex)
+{
+	struct ublk_device *ub = ubq->dev;
+	u16 tag;
+
+	for (tag = 0; tag < ubq->q_depth; tag++) {
+		struct ublk_io *io = &ubq->ios[tag];
+		struct io_uring_cmd *cmd = NULL;
+		struct request *req;
+		bool started;
+
+		/* see ublk_cancel_cmd() */
+		req = blk_mq_tag_to_rq(ub->tag_set.tags[ubq->q_id], tag);
+		started = req && blk_mq_request_started(req) && req->tag == tag;
+
+		spin_lock(&ubq->cancel_lock);
+		if (!(io->flags & UBLK_IO_FLAG_CANCELED)) {
+			if ((io->flags & UBLK_IO_FLAG_ACTIVE) && !started) {
+				io->flags |= UBLK_IO_FLAG_CANCELED;
+				cmd = READ_ONCE(io->cmd);
+				io->cmd = NULL;
+			} else if (io->flags & (UBLK_IO_FLAG_ACTIVE |
+						UBLK_IO_FLAG_OWNED_BY_SRV)) {
+				*left = true;
+			}
+		}
+		spin_unlock(&ubq->cancel_lock);
+
+		if (cmd) {
+			ublk_get_uring_cmd_pdu(cmd)->next_claimed = claimed;
+			claimed = cmd;
+		}
+	}
+	return claimed;
+}
+
+/*
+ * The same for a UBLK_F_BATCH_IO queue: move its parked fetch commands to
+ * ->quiesce_fcmds.  The active one is left to its dispatcher, which puts it
+ * back on the list once it is done, and the next pass takes it.  Unlike
+ * ublk_batch_cancel_queue() this leaves ->force_abort alone: requests of a
+ * recoverable device are requeued through ->canceling until the next server
+ * is ready, as on a queue without UBLK_F_BATCH_IO.
+ */
+static void ublk_quiesce_claim_fcmds(struct ublk_queue *ubq, bool *left)
+	__must_hold(&ubq->dev->cancel_mutex)
+{
+	struct ublk_batch_fetch_cmd *fcmd;
+
+	spin_lock(&ubq->evts_lock);
+	list_splice_tail_init(&ubq->fcmd_head, &ubq->quiesce_fcmds);
+	fcmd = READ_ONCE(ubq->active_fcmd);
+	if (fcmd) {
+		list_move(&fcmd->node, &ubq->fcmd_head);
+		*left = true;
+	}
+	spin_unlock(&ubq->evts_lock);
+}
+
+/*
+ * The fetch commands stay linked, and the cancel callback of their ring may
+ * take one off the list first: whoever unlinks one under evts_lock
+ * completes it.
+ */
+static void ublk_quiesce_complete_fcmds(struct ublk_queue *ubq)
+{
+	struct ublk_batch_fetch_cmd *fcmd;
+
+	for (;;) {
+		spin_lock(&ubq->evts_lock);
+		fcmd = list_first_entry_or_null(&ubq->quiesce_fcmds,
+						struct ublk_batch_fetch_cmd, node);
+		if (fcmd)
+			list_del_init(&fcmd->node);
+		spin_unlock(&ubq->evts_lock);
+		if (!fcmd)
+			break;
+
+		io_uring_cmd_done(fcmd->cmd, UBLK_IO_RES_ABORT,
+				  IO_URING_F_UNLOCKED);
+		ublk_batch_free_fcmd(fcmd);
+	}
+}
+
+/*
+ * Cancel the server's commands for QUIESCE_DEV until none of the FETCH
+ * round it marked is left.  One pass is not enough: it has to skip a command
+ * whose request is with the server or whose dispatch is pending, and the
+ * server waits for every command before it exits.  After synchronize_rcu()
+ * a COMMIT_AND_FETCH gives its command back itself, and no new request is
+ * dispatched on a canceling queue, so each remaining command is either
+ * given back by its issuer or armed and taken by a later pass.
+ *
+ * Check the round and take the commands in one cancel_mutex hold, and stop
+ * once the round is over: ublk_reset_ch_dev() bumps it, under cancel_mutex,
+ * before the next server can fetch, and those commands are not ours to
+ * cancel.  Complete what was taken after cancel_mutex is dropped, the
+ * ring's cancel callback takes it under uring_lock.
+ */
+static int ublk_quiesce_cancel(struct ublk_device *ub, u32 round,
+			       unsigned int timeout_ms)
+{
+	unsigned int elapsed = 0;
+
+	/* see ublk_commit_io_cmd() */
+	synchronize_rcu();
+
+	for (;;) {
+		struct io_uring_cmd *claimed = NULL;
+		bool left = false;
+		u16 i;
+
+		mutex_lock(&ub->cancel_mutex);
+		if (ub->fetch_round != round) {
+			mutex_unlock(&ub->cancel_mutex);
+			return 0;
+		}
+		for (i = 0; i < ub->dev_info.nr_hw_queues; i++) {
+			struct ublk_queue *ubq = ublk_get_queue(ub, i);
+
+			if (ublk_support_batch_io(ubq))
+				ublk_quiesce_claim_fcmds(ubq, &left);
+			else
+				claimed = ublk_quiesce_claim_queue(ubq, claimed,
+								   &left);
+		}
+		mutex_unlock(&ub->cancel_mutex);
+
+		while (claimed) {
+			struct io_uring_cmd *cmd = claimed;
+
+			claimed = ublk_get_uring_cmd_pdu(cmd)->next_claimed;
+			io_uring_cmd_done(cmd, UBLK_IO_RES_ABORT,
+					  IO_URING_F_UNLOCKED);
+		}
+		for (i = 0; i < ub->dev_info.nr_hw_queues; i++) {
+			struct ublk_queue *ubq = ublk_get_queue(ub, i);
+
+			if (ublk_support_batch_io(ubq))
+				ublk_quiesce_complete_fcmds(ubq);
+		}
+
+		if (!left)
+			return 0;
+		if (signal_pending(current))
+			return -EINTR;
+		if (elapsed >= timeout_ms)
+			return -EBUSY;
+		msleep(UBLK_REQUEUE_DELAY_MS);
+		elapsed += UBLK_REQUEUE_DELAY_MS;
+	}
+}
+
 static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 				 const struct ublksrv_ctrl_cmd *header)
 {
+	/* zero means wait forever */
+	u64 timeout_ms = header->data[0];
 	struct gendisk *disk;
 	bool live = true;
+	u32 round;
 	int ret = -ENODEV;
 
 	if (!(ub->dev_info.flags & UBLK_F_QUIESCE))
@@ -5427,6 +5609,7 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 	mutex_lock(&ub->cancel_mutex);
 	blk_mq_quiesce_queue(disk->queue);
 	ublk_set_canceling(ub, true);
+	round = ub->fetch_round;
 	blk_mq_unquiesce_queue(disk->queue);
 	mutex_unlock(&ub->cancel_mutex);
 
@@ -5437,7 +5620,9 @@ static int ublk_ctrl_quiesce_dev(struct ublk_device *ub,
 
 	/* Cancel pending uring_cmd */
 	if (!ret && live)
-		ublk_cancel_dev(ub);
+		ret = ublk_quiesce_cancel(ub, round,
+					  timeout_ms ? min_t(u64, timeout_ms, UINT_MAX) :
+					  UINT_MAX);
 	return ret;
 }
 
-- 
2.55.0

[PATCH 0/4] ublk: fix UBLK_CMD_QUIESCE_DEV leaving commands behind

[Josef Bacik]

UBLK_CMD_QUIESCE_DEV has two problems the fixes for STOP_DEV and the
FETCH rounds don't touch.

Sent to a device that is not LIVE, it still cancels after it returns 0.
A device whose server died is QUIESCED, and a new server may be
fetching its commands for recovery at that point, so the cancel takes
them without marking anything and END_USER_RECOVERY brings the device
up over NULL io->cmd. Patch 1 makes it cancel nothing then.

On a LIVE device it cancels in one pass, which skips every command
whose request is with the server. The server's COMMIT_AND_FETCH arms
the command again right after, nothing ever completes it, and the
server, which waits for all its commands, never exits. The device stays
LIVE. Same for the active fetch command of a UBLK_F_BATCH_IO queue. The
kublk selftest server hangs this way within a few quiesce and recover
cycles under fio, on every kind of queue.

Patch 2 drops ublk_wait_for_idle_io(), which never waited and would
hold ub->mutex against a stalled server if it did. Patch 3 has
COMMIT_AND_FETCH and NEED_GET_DATA give their new command back on a
canceling queue instead of publishing it, deciding inside an RCU read
section, so the I/O path gains no lock or barrier. Patch 4 has
QUIESCE_DEV wait for that with synchronize_rcu() and then keep taking
the armed commands until the server owes none, bounded by its timeout,
and stop once the server's FETCH round is over, so the next server's
commands are left alone.

QUIESCE_DEV now returns -EBUSY or -EINTR when its timeout or a signal
ends that wait with commands still owed, where it returned 0 after one
pass before.

This applies on top of Ming's "[PATCH 0/8] ublk: don't dispatch to
canceled io commands" [1] and my "ublk: refuse to go live after an io
command was canceled" [2].

Tested under QEMU with KASAN and lockdep. Without the series, 20
quiesce and recover cycles under fio hang in every round on getdata,
zero copy and user copy devices and in some on batch ones, and the
quiesce-twice reproducer oopses. With it, 3 rounds of 20 cycles on each
kind of device pass, the reproducer is fine, and the ublk selftests
including generic_18 pass.

[1] https://lore.kernel.org/linux-block/20261001125422.1364260-1-tom.leiming@gmail.com/
[2] https://lore.kernel.org/linux-block/9b876f2c061abc401ec4b9b3c2529eda.josef@toxicpanda.com/

Thanks,
Josef

Josef Bacik (4):
  ublk: don't cancel commands in QUIESCE_DEV on a device that isn't live
  ublk: drop QUIESCE_DEV's wait for an idle command
  ublk: give the command back from COMMIT_AND_FETCH on a canceling queue
  ublk: keep canceling in QUIESCE_DEV until the server's commands are
    taken

 drivers/block/ublk_drv.c | 286 +++++++++++++++++++++++++++++++--------
 1 file changed, 230 insertions(+), 56 deletions(-)

-- 
2.55.0