[PATCH v2 00/11] ublk: don't dispatch to canceled io commands

[PATCH v2 00/11] 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-5: the release resets the FETCH round, also without a disk
  and under ub->mutex
- patch 6: don't warn when the cancel callback finds io->cmd taken by a
  cancel from the control path
- patch 7: STOP_DEV holds a reference on /dev/ublkcN while it cancels,
  so no START_DEV, FETCH or new server can slip in (1)
- patch 8 (Josef): START_DEV and END_USER_RECOVERY refuse to go live
  after an io command was canceled (2, 3)
- patch 9-11: selftests

Tested with all ublk selftests under lockdep; v1 was also tested under
KASAN.

V2:
- add Josef's "ublk: refuse to go live after an io command was
  canceled", so routes 2 and 3 get -EBUSY instead of a live device
  failing every request
- split the cancel callback's WARN change out of the STOP_DEV patch
  (patch 6)
- document in ublk.rst what a server sees after STOP_DEV (Josef)
- kublk: stop a batch thread when PREP_IO_CMDS is aborted, instead of
  asserting and leaving `kublk add` waiting forever (Josef)
- selftest: partial_fetch and recovery expect -EBUSY; add a batch
  STOP_DEV race mode with kublk
- commit log fixes (patch 4, patch 7)
- add Josef's Tested-by

The release work deciding "no disk" before ub->mutex (Josef's point 4)
needs a START_DEV for a server which has already exited: no new server
can open /dev/ublkcN until that work is done, and patch 8 refuses such a
START_DEV with -EBUSY, so it gets no extra patch.

V1: https://lore.kernel.org/linux-block/20261001125422.1364260-1-tom.leiming@gmail.com/

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

Thanks,
Ming

Josef Bacik (1):
  ublk: refuse to go live after an io command was canceled

Ming Lei (10):
  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: allow a NULL io->cmd in the cancel callback
  ublk: let STOP_DEV cancel the server's commands before its release
  selftests: ublk: move the control command helpers into ctrl.c
  selftests: ublk: stop a batch thread when PREP_IO_CMDS fails
  selftests: ublk: add test for going live over canceled io commands

 Documentation/block/ublk.rst                  |  17 +-
 drivers/block/ublk_drv.c                      | 244 ++++-
 tools/testing/selftests/ublk/.gitignore       |   1 +
 tools/testing/selftests/ublk/Makefile         |   6 +-
 tools/testing/selftests/ublk/batch.c          |  13 +-
 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 |  69 ++
 .../selftests/ublk/ublk_cancel_ready.c        | 957 ++++++++++++++++++
 10 files changed, 1550 insertions(+), 328 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 v2 01/11] 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/
Tested-by: Josef Bacik <josef@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 v2 02/11] 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
Tested-by: Josef Bacik <josef@toxicpanda.com>
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 v2 03/11] 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+
Tested-by: Josef Bacik <josef@toxicpanda.com>
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 v2 04/11] 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
Tested-by: Josef Bacik <josef@toxicpanda.com>
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 v2 05/11] 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
Tested-by: Josef Bacik <josef@toxicpanda.com>
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 v2 06/11] ublk: allow a NULL io->cmd in the cancel callback

[Ming Lei]

A cancel from the control path, like STOP_DEV's ublk_cancel_dev() after
dropping ub->mutex, takes the command and clears io->cmd under
cancel_lock, then completes it with io_uring_cmd_done(), which takes
uring_lock to remove it from the cancelable list. The server's io_uring
cancel runs ublk_uring_cmd_cancel_fn() under uring_lock in between:

  control path cancel                    io_uring cancel (uring_lock)
  io->cmd = NULL, CANCELED
  io_uring_cmd_done()
    waits for uring_lock                 cmd still on the list
                                         ublk_uring_cmd_cancel_fn(cmd)
                                           io->cmd == NULL: WARN

ublk_cancel_cmd() then sees CANCELED and leaves the command alone, so
nothing is wrong; only warn if io->cmd holds another command.

Fixes: ab03a61c6614 ("ublk: have a per-io daemon instead of a per-queue daemon")
Tested-by: Josef Bacik <josef@toxicpanda.com>
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 drivers/block/ublk_drv.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 162a4d1a2e2a..2fd0c265ed20 100644
--- a/drivers/block/ublk_drv.c
+++ b/drivers/block/ublk_drv.c
@@ -2896,6 +2896,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 +2913,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 a cancel from the control path took it meanwhile */
+	cur = READ_ONCE(io->cmd);
+	WARN_ON_ONCE(cur && cur != cmd);
 	ublk_cancel_cmd(ubq, pdu->tag, issue_flags);
 }
 
-- 
2.55.0

[PATCH v2 07/11] 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.
Document what a server sees after STOP_DEV in ublk.rst.

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/
Tested-by: Josef Bacik <josef@toxicpanda.com>
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 Documentation/block/ublk.rst |  7 ++++
 drivers/block/ublk_drv.c     | 80 ++++++++++++++++++++++++++++++++----
 2 files changed, 79 insertions(+), 8 deletions(-)

diff --git a/Documentation/block/ublk.rst b/Documentation/block/ublk.rst
index 28300fee22bf..ed3285e361ad 100644
--- a/Documentation/block/ublk.rst
+++ b/Documentation/block/ublk.rst
@@ -126,6 +126,13 @@ managing and controlling ublk devices with help of several control commands:
   ublk server will release resources (such as destroying I/O handler threads &
   io_uring).
 
+  If a server has ``/dev/ublkc*`` open, its fetched I/O commands complete with
+  ``UBLK_IO_RES_ABORT``. Until it closes ``/dev/ublkc*``, new
+  ``UBLK_IO_FETCH_REQ`` and ``UBLK_U_IO_PREP_IO_CMDS`` commands complete with
+  ``UBLK_IO_RES_ABORT`` too, and ``UBLK_CMD_START_DEV`` fails with ``-EBUSY``,
+  even if the server hadn't fetched anything. The device can be started again
+  by a server that opens ``/dev/ublkc*`` after that.
+
 - ``UBLK_CMD_DEL_DEV``
 
   Remove ``/dev/ublkc*``. When this command returns, the allocated ublk device
diff --git a/drivers/block/ublk_drv.c b/drivers/block/ublk_drv.c
index 2fd0c265ed20..176fbf84737c 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.
@@ -3026,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)
@@ -3298,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;
@@ -4476,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);
 	}
@@ -4591,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 v2 08/11] ublk: refuse to go live after an io command was canceled

[Ming Lei]

From: Josef Bacik <josef@toxicpanda.com>

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>
---
 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 ed3285e361ad..f07611b0d489 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``
 
@@ -202,7 +206,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 176fbf84737c..ecadcf5e3e33 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

[PATCH v2 09/11] 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 v2 10/11] selftests: ublk: stop a batch thread when PREP_IO_CMDS fails

[Ming Lei]

Since STOP_DEV turns the server away until it closes /dev/ublkcN,
UBLK_U_IO_PREP_IO_CMDS can complete with UBLK_IO_RES_ABORT when it
races STOP_DEV. kublk asserts that PREP succeeds, so the daemon aborts
before it reports to `kublk add`, which then waits for the start event
forever.

Stop the thread like a failed FETCH_IO_CMDS does. The daemon's
START_DEV then fails, and `kublk add` gets the error.

Reported-by: Josef Bacik <josef@toxicpanda.com>
Closes: https://lore.kernel.org/linux-block/63d8149c093169706e7478e779116b61.josef@toxicpanda.com/
Signed-off-by: Ming Lei <tom.leiming@gmail.com>
---
 tools/testing/selftests/ublk/batch.c | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/tools/testing/selftests/ublk/batch.c b/tools/testing/selftests/ublk/batch.c
index d8d9ebed5979..dfc81ee17d07 100644
--- a/tools/testing/selftests/ublk/batch.c
+++ b/tools/testing/selftests/ublk/batch.c
@@ -410,9 +410,16 @@ static void ublk_batch_compl_commit_cmd(struct ublk_thread *t,
 {
 	unsigned short buf_idx = user_data_to_tag(cqe->user_data);
 
-	if (op == _IOC_NR(UBLK_U_IO_PREP_IO_CMDS))
-		ublk_assert(cqe->res == 0);
-	else if (op == _IOC_NR(UBLK_U_IO_COMMIT_IO_CMDS)) {
+	if (op == _IOC_NR(UBLK_U_IO_PREP_IO_CMDS)) {
+		/* e.g. UBLK_IO_RES_ABORT after STOP_DEV: stop like a FETCH */
+		if (cqe->res < 0) {
+			ublk_dbg(UBLK_DBG_IO_CMD, "%s: prep failed %d\n",
+				 __func__, cqe->res);
+			t->state |= UBLKS_T_STOPPING;
+		} else {
+			ublk_assert(cqe->res == 0);
+		}
+	} else if (op == _IOC_NR(UBLK_U_IO_COMMIT_IO_CMDS)) {
 		int nr_elem = user_data_to_tgt_data(cqe->user_data);
 
 		ublk_assert(cqe->res == t->commit_buf_elem_size * nr_elem);
-- 
2.55.0

[PATCH v2 11/11] 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 -EBUSY
                     dies, then task B fetches tag 3
  recovery           in user recovery, queue 0's      END_USER_RECOVERY
                     task dies before queue 1 is      -EBUSY
                     ready
  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)
  race_batch_stop    STOP_DEV while kublk's batch     kublk add returns,
                     server prepares and fetches,     started or not
                     20 times

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 |  69 ++
 .../selftests/ublk/ublk_cancel_ready.c        | 957 ++++++++++++++++++
 4 files changed, 1031 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..2dbb59e238a4
--- /dev/null
+++ b/tools/testing/selftests/ublk/test_generic_18.sh
@@ -0,0 +1,69 @@
+#!/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"
+
+# STOP_DEV while kublk's UBLK_F_BATCH_IO server prepares and fetches: the
+# server has to fail cleanly when PREP or FETCH is aborted, so `kublk add`
+# must return either way.
+race_batch_stop()
+{
+	local id=9999 i res started=0
+
+	_have_feature "BATCH_IO" || return "$UBLK_SKIP_CODE"
+	for i in $(seq 1 20); do
+		timeout 20 "$UBLK_PROG" add -t null -q 2 -b -n "$id" > /dev/null 2>&1 &
+		local pid=$!
+		# vary where STOP_DEV lands relative to the batch PREP and FETCH
+		sleep "0.$((i % 10))"
+		"$UBLK_PROG" stop -n "$id" > /dev/null 2>&1
+		wait "$pid"
+		res=$?
+		"$UBLK_PROG" del -n "$id" > /dev/null 2>&1
+		if [ "$res" -eq 124 ]; then
+			echo "run $i: kublk add -b hung after STOP_DEV"
+			return 255
+		fi
+		[ "$res" -eq 0 ] && started=$((started + 1))
+	done
+	echo "20 runs: started $started, stopped before start $((20 - started))"
+}
+
+# race_batch_stop is above, the other 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 \
+		race_batch_stop; do
+	dmesg_before=$(dmesg | wc -l)
+	if [ "$mode" = race_batch_stop ]; then
+		race_batch_stop > "$UBLK_TMP" 2>&1
+	else
+		timeout 60 "$CANCEL_PROG" "$mode" > "$UBLK_TMP" 2>&1
+	fi
+	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..23071babc0e4
--- /dev/null
+++ b/tools/testing/selftests/ublk/ublk_cancel_ready.c
@@ -0,0 +1,957 @@
+// 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 must refuse
+ * with -EBUSY. 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);
+}
+
+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 };
+	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_expect(-EBUSY);
+
+	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);
+	ret = expect_err("END_USER_RECOVERY", ret, -EBUSY);
+
+	/*
+	 * 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