ieee802154: mcr20a: single lp->reg_msg reused from two unlocked contexts (xmit vs IRQ completion chain) double-submits the same spi_message and corrupts the SPI queue

ieee802154: mcr20a: single lp->reg_msg reused from two unlocked contexts (xmit vs IRQ completion chain) double-submits the same spi_message and corrupts the SPI queue

[徐晗翔]

[-- Attachment #1: Type: text/plain, Size: 6724 bytes --]

mcr20a reuses one struct spi_message (lp->reg_msg) as the transmit
vehicle for both the netdev xmit path and the IRQ-driven completion
chain, with no lock common to the two contexts, so a concurrent xmit
re-fills and re-submits a message that is still queued; the SPI core
accepts the double list_add_tail() without any check, and the resulting
queue corruption permanently hangs spi_sync() callers -- including the
RTNL-holding interface-close path -- or wedges the whole controller
queue.


The driver's actual mutual exclusion relies on chip-protocol causality
(the completion chain only re-arms itself after the previous message
finished) and on mac802154's TX queue hold, which only serializes TX
against TX. Nothing serializes the RX/IRQ completion chain (SPI
controller kworker context) against a newly started transmission
(ndo_start_xmit context): both sides fill lp->reg_msg's complete/
transfer/length fields and call spi_async() on the same object
(mcr20a.c:475-480 vs mcr20a.c:840-844/940-945/451-456), and
__spi_queued_transfer() does an unchecked list_add_tail(&msg->queue,
&ctlr->queue) (drivers/spi/spi.c:2479 in 7.3). Verified on mainline
7.3 with a function-for-function simulator of the real message flow
(reproducer attached): 201 double-submissions caught in 390 controlled
xmit attempts, and 1 fully natural collision in 8000 attempts without
any artificial delay. On a production kernel (no CONFIG_DEBUG_LIST)
the second list_add is not rejected but lands as real list corruption.
No special privileges are needed for the triggering socket traffic
once the interface is up; MCR20A hardware is required.


The detailed description:

* Test environment / affected version

Linux version 7.3.0 (gcc 13.3.0, GNU ld 2.42) #1 SMP PREEMPT_DYNAMIC
-- vanilla mainline 7.3 release tarball (kernel.org, unmodified),
rebuilt with CONFIG_SPI=y, CONFIG_REGMAP_SPI=y,
CONFIG_DEBUG_LIST=y, CONFIG_LIST_HARDENED=y, KASAN enabled, QEMU/KVM
x86_64 guest. drivers/net/ieee802154/mcr20a.c is byte-identical in
7.1.5 and 7.3; line numbers below are from 7.3 (= 7.1.5).

* Mechanism

One spi_message, four producers, two contexts, no common lock:

- netdev xmit path (process/softirq context):
mcr20a_xmit() fills lp->reg_msg (complete=NULL, transfer
command/data, lengths; mcr20a.c:475-480) and calls
spi_async(lp->spi, &lp->reg_msg).

- IRQ-driven completion chain (SPI controller kworker context):
the hard IRQ submits irq_msg; its completions chain reg_msg
re-submissions: write_tx_buf_complete() (mcr20a.c:451-456),
handle_rx() (mcr20a.c:840-844), irq_status_complete()
(mcr20a.c:940-945) each re-fill the SAME lp->reg_msg fields and
spi_async() it to derive the next chain step.


The serialization that exists does not cover this pair:

- mac802154's ieee802154_hold_queue/release_queue
(net/mac802154/util.c:67-85, phy->queue_lock + hold_txs) stops
the netdev queue around driver xmit/complete pairs -- TX vs TX
only; the RX chain never takes it.

- lp->is_tx itself is an unlocked cross-context flag; the chain
readers (mcr20a.c:897/909/921) race the xmit writer (473).

- SPI core: __spi_queued_transfer() appends
list_add_tail(&msg->queue, &ctlr->queue) with no
"already queued" check, and __spi_pump_messages() dequeues with
list_first_entry()+list_del_init(). A second enqueue of a node
still in the queue overwrites msg->next/msg->prev and one
neighbor's pointer while the other neighbor keeps pointing at
it.


Concrete corruption forms (first two are the ones the reproducer
catches, shape new==prev == queue-tail self-loop):

1. Queue-tail self-loop: the double add happens while reg_msg is
the last entry (new == prev in the DEBUG_LIST report):
reg_msg->next = reg_msg, detached from the head --
list_empty() stays false forever, the pump re-picks the same
message, every device on this controller stops making
progress; controller teardown polls and fails
(spi_stop_queue() timeout path).

2. Orphaned range: a mid-queue double add makes everything after
the old position unreachable. Any spi_sync() whose completion
lives in that range never fires: __spi_sync() blocks on a
stack completion forever. If the orphaned sync message is the
one mcr20a_stop() submits during interface close (called
under RTNL), the hang is a kernel-level networking deadlock.

3. Field tearing (no exact double-add needed): the two contexts
concurrently write reg_msg->complete/reg_cmd/reg_data while
the other side reads/submits them, so a message can be queued
with a mixed state -- e.g. complete overwritten to NULL after
the chain set it (chain breaks, chip IRQ left disabled,
netif queue never released) or a completion consuming the
wrong buffer.


* Trigger model


Interface up (brought up by an administrator once) + any RX traffic
keeping the completion chain active + a local socket sending
802.15.4 frames. The sending side needs no privileges; the chain
is active for tens-to-hundreds of microseconds after each received
frame, and a socket send landing in that window races it. Requires
MCR20A hardware. Consequence: hang/deadlock and controller-wide
I/O stall (DoS).


* Reproducer (attached: mcrsim.c) and results on 7.3

mcrsim.c is a self-contained module: a minimal software-loopback SPI
controller (spi_register_controller) plus a driver that mirrors the
real mcr20a message flow function-for-function (same reg_msg
reuse, same four producers across the same two contexts; simulated
chip IRQ + per-message controller delay to shape the window). No
kernel source modification; needs CONFIG_SPI=y and
CONFIG_REGMAP_SPI=y (the kernel I tested also had CONFIG_DEBUG_LIST=
y so the corruption is caught as a warning instead of landing).

# build
make -C <kernel-build-dir> M=$PWD modules
# controlled window (root):
insmod mcrsim.ko irq_period_ms=30 xfer_delay_ms=60
echo "xmit 400 2000" > /sys/module/mcrsim/parameters/action
echo status > /sys/module/mcrsim/parameters/action
# natural window (no artificial delay):
rmmod mcrsim; insmod mcrsim.ko irq_period_ms=10 xfer_delay_ms=0
echo "xmit 8000 200" > /sys/module/mcrsim/parameters/action


Results (7.3, 4 vCPU):

- controlled: 201 "list_add double add: new=... == prev=..."
catches during 390 xmit submissions (each catch is the exact
self-loop form 1; the rejections come from CONFIG_DEBUG_LIST --
without it these land as real corruption);
- natural: 1 collision in 8000 submissions, same new==prev
shape, with 18.4M messages otherwise processed cleanly -- the
race exists in the unmodified timing, only the rate is small.


The same simulator also demonstrates form 2 on request ("syncprobe"
action drives an spi_sync() whose completion gets orphaned by the
same double-submission).

Thanks for taking a look.

Regards,
Hanxiang Xu
xuhx@hust.edu.cn



[-- Attachment #2: mcrsim.c --]
[-- Type: text/plain, Size: 19234 bytes --]

// SPDX-License-Identifier: GPL-2.0
/*
 * mcrsim.c - dynamic verification for HV-2 / N-1 of the drivers/base
 * race-condition audit (mcr20a-family async SPI completion chains vs
 * teardown), against the KASAN+DEBUG_LIST kernel (build-kasan-spi).
 *
 * It reproduces the mcr20a structure faithfully at the API level, with
 * ZERO kernel-source modification:
 *
 *   - a fake, software-loopback SPI controller (queue-based,
 *     transfer_one_message + spi_finalize_current_message), which keeps
 *     processing its queue independently of slave-driver binding,
 *     exactly like a real controller.  An optional per-message delay
 *     (xfer_delay_ms) widens the natural windows for observation.
 *   - one fake slave device "mcrsim" bound to a driver whose private
 *     data (lp) is a plain kzalloc (NOT devm), freed directly in
 *     .remove just like ieee802154_free_hw() frees the single
 *     wpan_phy+local+lp allocation inside mcr20a_remove().
 *   - an mcr20a-shaped asynchronous completion chain driven by a
 *     periodic delayed_work that plays the role of the chip IRQ
 *     (initiator only; the chain itself survives on its own):
 *
 *       sim_irq(dwork) -> spi_async(&lp->irq_msg)
 *         -> irq_status_complete : regmap_update_bits_async(lp->regmap)
 *                                  + spi_async(&lp->reg_msg)
 *         -> irq_clean_complete  : reads lp->irq_data, dispatches
 *         -> handle_rx           : refills and resubmits &lp->reg_msg
 *         -> rx_read_len_complete: spi_async(&lp->rx_buf_msg)
 *         -> rx_buf_complete     : reads lp->rx_buf, then
 *                                  regmap_update_bits_async(lp->regmap)
 *
 *     All callbacks dereference the raw lp pointer; lp->reg_msg is
 *     shared by the chain AND by the "xmit" path (N-1).
 *   - regmap is devm_regmap_init_spi(): freed by devres AFTER
 *     .remove returns (authentic dd.c ordering), while chain callbacks
 *     keep calling regmap_update_bits_async (HV-2 step 6 = HV-1 wide
 *     window through the real regmap-spi path).
 *
 * Fidelity boundaries (documented in the audit report):
 *   - the chip IRQ is simulated by a periodic work (the initiator);
 *     the property under test is the chain's survival past unbind,
 *     not the IRQ itself;
 *   - the ieee802154/mac802154 upper layer is replaced by direct
 *     log-and-resubmit;
 *   - N-1's natural probabilistic collision is made near-certain by
 *     xfer_delay_ms (controlled window), plus a natural hammer mode.
 *
 * Actions (echo ... > /sys/module/mcrsim/parameters/action):
 *   run <ms>          : keep the chain running for ms (stats printed)
 *   xmit <n> <us>     : N-1: kthread loops the mcr20a_xmit-shaped body
 *                       (fills lp->reg_msg + spi_async) n times
 *   syncprobe         : N-1 outcome probe: one sync regmap op with a
 *                       5s hang watchdog (orphaned-message evidence)
 *   status            : statistics
 * Unbind for HV-2 is driven from the VM init script via
 *   echo spiB.C > /sys/bus/spi/drivers/mcrsim/unbind
 */
#define pr_fmt(fmt) "MCRSIM: " fmt

#include <linux/module.h>
#include <linux/init.h>
#include <linux/kernel.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/workqueue.h>
#include <linux/slab.h>
#include <linux/spi/spi.h>
#include <linux/regmap.h>
#include <linux/platform_device.h>
#include <linux/string.h>
#include <linux/ktime.h>
#include <linux/atomic.h>

/* ------------------------------------------------------------------ */
/* tunables                                                            */

static int irq_period_ms = 40;
module_param(irq_period_ms, int, 0600);
MODULE_PARM_DESC(irq_period_ms, "simulated chip IRQ period (chain initiator)");

static int xfer_delay_ms = 80;
module_param(xfer_delay_ms, int, 0600);
MODULE_PARM_DESC(xfer_delay_ms, "per-message delay inside the fake controller (controlled window)");

/* fake chip registers (mcr20a names kept for traceability) */
#define DAR_IRQ_STS1	0x02
#define DAR_PHY_CTRL1	0x10
#define DAR_RX_FRM_LEN	0x13

/* ------------------------------------------------------------------ */
/* fake software-loopback SPI controller                               */

#define SIM_CS_MAX	4

struct mcrsim_ctrl {
	struct platform_device	*pdev;
	struct spi_controller	*ctlr;
	u8			regfile[SIM_CS_MAX][128];
	u8			addr[SIM_CS_MAX];
	bool			rd[SIM_CS_MAX];
};

static struct mcrsim_ctrl *g_ctrl;
static struct spi_device *g_spidev;

static atomic_long_t stat_msgs = ATOMIC_LONG_INIT(0);
static atomic_long_t stat_delayed = ATOMIC_LONG_INIT(0);
static atomic_long_t stat_irq_submit = ATOMIC_LONG_INIT(0);
static atomic_long_t stat_chain_steps = ATOMIC_LONG_INIT(0);
static atomic_long_t stat_xmit = ATOMIC_LONG_INIT(0);
static atomic_long_t stat_async_calls = ATOMIC_LONG_INIT(0);

static void sim_handle_message(struct mcrsim_ctrl *c, struct spi_message *m)
{
	struct spi_transfer *xfer;
	u8 cs = m->spi->chip_select[0];
	unsigned int i;

	list_for_each_entry(xfer, &m->transfers, transfer_list) {
		const u8 *tx = xfer->tx_buf;
		u8 *rx = xfer->rx_buf;

		for (i = 0; i < xfer->len; i++) {
			if (tx) {
				u8 b = tx[i];

				if (i == 0) {
					c->addr[cs] = b & 0x7f;
					c->rd[cs] = !!(b & 0x80);
				} else if (!c->rd[cs]) {
					c->regfile[cs][c->addr[cs] & 0x7f] = b;
					c->addr[cs]++;
				}
			}
			if (rx)
				rx[i] = c->regfile[cs][c->addr[cs]++ & 0x7f];
		}
		m->actual_length += xfer->len;
	}
}

static int mcrsim_transfer_one_message(struct spi_controller *ctlr,
				       struct spi_message *m)
{
	atomic_long_inc(&stat_msgs);
	if (xfer_delay_ms > 0) {
		atomic_long_inc(&stat_delayed);
		msleep(xfer_delay_ms);
	}
	sim_handle_message(spi_controller_get_devdata(ctlr), m);
	m->status = 0;
	spi_finalize_current_message(ctlr);
	return 0;
}

/* ------------------------------------------------------------------ */
/* mcr20a-shaped slave driver                                          */

struct mcrsim_lp {
	struct spi_device	*spi;
	struct regmap		*regmap;

	struct spi_message	irq_msg;
	struct spi_transfer	irq_xhdr;
	struct spi_transfer	irq_xdata;
	u8			irq_hdr;
	u8			irq_data[2];

	struct spi_message	reg_msg;		/* SHARED by chain + xmit (N-1) */
	struct spi_transfer	reg_xcmd;
	struct spi_transfer	reg_xdata;
	u8			reg_cmd;
	u8			reg_data[4];

	struct spi_message	rx_buf_msg;
	struct spi_transfer	rx_xhdr;
	struct spi_transfer	rx_xbuf;
	struct spi_transfer	rx_xlqi;
	u8			rx_hdr;
	u8			rx_buf[32];
	u8			rx_lqi;

	struct delayed_work	sim_irq;	/* chip-IRQ initiator */
	int			alive;
	int			chain_busy;	/* IRQ-line gating analog: the real chip
					 * does not re-raise the IRQ while the
					 * chain owns it (ISR disables it) */
};

static void mcrsim_irq_status_complete(void *context);
static void mcrsim_rx_buf_complete(void *context);

static void setup_messages(struct mcrsim_lp *lp)
{
	spi_message_init(&lp->irq_msg);
	lp->irq_msg.context = lp;
	lp->irq_msg.complete = mcrsim_irq_status_complete;
	lp->irq_xhdr.len = 1;
	lp->irq_xhdr.tx_buf = &lp->irq_hdr;
	lp->irq_xhdr.rx_buf = &lp->irq_hdr;
	lp->irq_xdata.len = 2;
	lp->irq_xdata.rx_buf = lp->irq_data;
	spi_message_add_tail(&lp->irq_xhdr, &lp->irq_msg);
	spi_message_add_tail(&lp->irq_xdata, &lp->irq_msg);

	spi_message_init(&lp->reg_msg);
	lp->reg_msg.context = lp;
	lp->reg_xcmd.len = 1;
	lp->reg_xcmd.tx_buf = &lp->reg_cmd;
	lp->reg_xcmd.rx_buf = &lp->reg_cmd;
	lp->reg_xdata.rx_buf = lp->reg_data;
	lp->reg_xdata.tx_buf = lp->reg_data;
	spi_message_add_tail(&lp->reg_xcmd, &lp->reg_msg);
	spi_message_add_tail(&lp->reg_xdata, &lp->reg_msg);

	spi_message_init(&lp->rx_buf_msg);
	lp->rx_buf_msg.context = lp;
	lp->rx_buf_msg.complete = mcrsim_rx_buf_complete;
	lp->rx_xhdr.len = 1;
	lp->rx_xhdr.tx_buf = &lp->rx_hdr;
	lp->rx_xhdr.rx_buf = &lp->rx_hdr;
	lp->rx_xbuf.rx_buf = lp->rx_buf;
	lp->rx_xlqi.len = 1;
	lp->rx_xlqi.rx_buf = &lp->rx_lqi;
	spi_message_add_tail(&lp->rx_xhdr, &lp->rx_buf_msg);
	spi_message_add_tail(&lp->rx_xbuf, &lp->rx_buf_msg);
	spi_message_add_tail(&lp->rx_xlqi, &lp->rx_buf_msg);
}

/* ---- the chain (mirrors mcr20a.c callbacks one-to-one) ------------- */

static void mcrsim_request_rx(struct mcrsim_lp *lp)
{
	atomic_long_inc(&stat_async_calls);
	regmap_update_bits_async(lp->regmap, DAR_PHY_CTRL1, 0x1f, 0x05);
}

static void mcrsim_rx_buf_complete(void *context)
{
	struct mcrsim_lp *lp = context;

	atomic_long_inc(&stat_chain_steps);
	/* mcr20a: reads lp->reg_data[0] & lp->rx_buf ... */
	pr_info_ratelimited("CHAIN: rx_buf_complete derefs lp=%px rx_buf[0]=%02x lqi=%02x\n",
			    lp, lp->rx_buf[0], lp->rx_lqi);
	lp->chain_busy = 0;	/* mcr20a: chain end == IRQ re-enabled */
	mcrsim_request_rx(lp);
}

static void mcrsim_rx_read_len_complete(void *context)
{
	struct mcrsim_lp *lp = context;
	int ret;

	atomic_long_inc(&stat_chain_steps);
	lp->rx_xbuf.len = lp->reg_data[0] & 0x1f;
	ret = spi_async(lp->spi, &lp->rx_buf_msg);
	if (ret)
		pr_err_ratelimited("rx buf submit ret=%d\n", ret);
}

static void mcrsim_handle_rx(struct mcrsim_lp *lp)
{
	atomic_long_inc(&stat_chain_steps);
	lp->reg_msg.complete = mcrsim_rx_read_len_complete;
	lp->reg_cmd = DAR_RX_FRM_LEN;
	lp->reg_xdata.len = 1;
	spi_async(lp->spi, &lp->reg_msg);
}

static void mcrsim_irq_clean_complete(void *context)
{
	struct mcrsim_lp *lp = context;

	atomic_long_inc(&stat_chain_steps);
	/* mcr20a: reads lp->irq_data, enable_irq(), dispatches */
	pr_info_ratelimited("CHAIN: irq_clean_complete derefs lp=%px irq_data=%02x %02x\n",
			    lp, lp->irq_data[0], lp->irq_data[1]);
	mcrsim_handle_rx(lp);
}

static void mcrsim_irq_status_complete(void *context)
{
	struct mcrsim_lp *lp = context;
	int ret;

	atomic_long_inc(&stat_chain_steps);
	/* mcr20a_irq_status_complete: async regmap write + resubmit reg_msg */
	atomic_long_inc(&stat_async_calls);
	regmap_update_bits_async(lp->regmap, DAR_PHY_CTRL1, 0x1f, 0x00);

	lp->reg_msg.complete = mcrsim_irq_clean_complete;
	lp->reg_cmd = DAR_IRQ_STS1;
	lp->reg_data[0] = lp->irq_data[0];
	lp->reg_data[1] = lp->irq_data[1];
	lp->reg_xdata.len = 2;
	ret = spi_async(lp->spi, &lp->reg_msg);
	if (ret)
		pr_err_ratelimited("irq clean submit ret=%d\n", ret);
}

static void mcrsim_irq_workfn(struct work_struct *w)
{
	struct mcrsim_lp *lp = container_of(w, struct mcrsim_lp, sim_irq.work);

	if (!lp->alive)
		return;
	schedule_delayed_work(&lp->sim_irq, msecs_to_jiffies(irq_period_ms));
	if (lp->chain_busy)
		return;	/* hardware IRQ-line gating analog */
	lp->chain_busy = 1;
	lp->irq_hdr = 0x80 | DAR_IRQ_STS1;
	atomic_long_inc(&stat_irq_submit);
	if (spi_async(lp->spi, &lp->irq_msg))
		lp->chain_busy = 0;
}

/* ---- probe / remove: the mcr20a lifecycle, unmodified shape -------- */

static bool mcrsim_volatile(struct device *dev, unsigned int reg)
{
	/* mcr20a_dar_volatile(): only the IRQ status regs are volatile */
	switch (reg) {
	case DAR_IRQ_STS1:
		return true;
	default:
		return false;
	}
}

static const struct regmap_config mcrsim_regmap_cfg = {
	.reg_bits = 8,
	.val_bits = 8,
	.cache_type = REGCACHE_MAPLE,	/* mcr20a uses REGCACHE_MAPLE */
	.volatile_reg = mcrsim_volatile,
};

static int mcrsim_probe(struct spi_device *spi)
{
	struct mcrsim_lp *lp;

	lp = kzalloc(sizeof(*lp), GFP_KERNEL);
	if (!lp)
		return -ENOMEM;
	lp->spi = spi;
	lp->alive = 1;
	spi_set_drvdata(spi, lp);

	lp->regmap = devm_regmap_init_spi(spi, &mcrsim_regmap_cfg);
	if (IS_ERR(lp->regmap)) {
		kfree(lp);
		return PTR_ERR(lp->regmap);
	}

	/*
	 * mcr20a_start()/mcr20a_phy_init() write DAR_PHY_CTRL1 from process
	 * context before the IRQ chain ever runs; this also primes the
	 * regmap cache so the chain's update_bits_async() reads never fall
	 * back to a bus read from completion context.
	 */
	regmap_update_bits(lp->regmap, DAR_PHY_CTRL1, 0x1f, 0x00);

	setup_messages(lp);
	INIT_DELAYED_WORK(&lp->sim_irq, mcrsim_irq_workfn);

	pr_info("probe: spi=%px lp=%px regmap=%px (chain initiator period=%dms xfer_delay=%dms)\n",
		spi, lp, lp->regmap, irq_period_ms, xfer_delay_ms);
	schedule_delayed_work(&lp->sim_irq, msecs_to_jiffies(irq_period_ms));
	return 0;
}

static void mcrsim_remove(struct spi_device *spi)
{
	struct mcrsim_lp *lp = spi_get_drvdata(spi);

	pr_info("remove ENTER: lp=%px regmap=%px (mcr20a shape: sync write + stop initiator + kfree, NO chain drain)\n",
		lp, lp->regmap);

	/* mcr20a_stop analog: one SYNC regmap write (spi_sync) ... */
	regmap_update_bits(lp->regmap, DAR_PHY_CTRL1, 0x1f, 0x00);
	/* ... and stop the IRQ initiator (disable_irq analog) */
	lp->alive = 0;
	cancel_delayed_work_sync(&lp->sim_irq);

	/* mcr20a_remove analog: ieee802154_free_hw -> kfree of the single
	 * allocation.  The in-flight chain is NOT waited for (the bug). */
	kfree(lp);

	pr_info("remove DONE: lp freed; any queued/completing chain messages now dereference freed memory; devm regmap_exit follows via devres\n");
}

static const struct spi_device_id mcrsim_ids[] = {
	{ "mcrsim" },
	{ }
};
MODULE_DEVICE_TABLE(spi, mcrsim_ids);

static struct spi_driver mcrsim_driver = {
	.driver = {
		.name = "mcrsim",
	},
	.probe = mcrsim_probe,
	.remove = mcrsim_remove,
	.id_table = mcrsim_ids,
};

/* ------------------------------------------------------------------ */
/* actions                                                             */

static char action_cmd[64];

static int xmit_loops = 200;
static int xmit_interval_us = 1000;

static int xmit_thread(void *data)
{
	struct mcrsim_lp *lp = data;
	int i;

	pr_info("XMIT: kthread running mcr20a_xmit-shaped body %d times, interval=%dus, reg_msg=%px\n",
		xmit_loops, xmit_interval_us, &lp->reg_msg);
	for (i = 0; i < xmit_loops && !kthread_should_stop(); i++) {
		/* == mcr20a_xmit() body: unlocked refill + submit of the
		 * SAME lp->reg_msg that the IRQ chain also uses */
		lp->reg_msg.complete = NULL;
		lp->reg_cmd = DAR_PHY_CTRL1;
		lp->reg_data[0] = 0x00;		/* XCVSEQ_IDLE */
		lp->reg_xdata.len = 1;
		atomic_long_inc(&stat_xmit);
		spi_async(lp->spi, &lp->reg_msg);
		usleep_range(xmit_interval_us, xmit_interval_us + 100);
	}
	pr_info("XMIT: kthread done after %d submits\n", i);
	return 0;
}

static atomic_t syncprobe_done = ATOMIC_INIT(0);
static struct delayed_work syncprobe_wd;

static int syncprobe_run(void *data)
{
	struct mcrsim_lp *lp = data;

	pr_info("SYNCPROBE: issuing one SYNC regmap op (hang = orphaned-message evidence)\n");
	regmap_update_bits(lp->regmap, 0x11, 0xff, 0x55);
	atomic_set(&syncprobe_done, 1);
	pr_info("SYNCPROBE: COMPLETED (queue still functional)\n");
	return 0;
}

static void syncprobe_watchdog(struct work_struct *w)
{
	if (!atomic_read(&syncprobe_done))
		pr_err("SYNCPROBE-HUNG: sync regmap op did NOT complete within 5s -> message orphaned by queue corruption (N-1 outcome 1)\n");
	else
		pr_info("SYNCPROBE-WD: op finished in time\n");
}

static void action_workfn(struct work_struct *w)
{
	char cmd[64];
	char *p, *cp;
	long n = 0;

	strscpy(cmd, strstrip(action_cmd), sizeof(cmd));
	cp = cmd;
	p = strsep(&cp, " ");
	if (cp && cp[0] && kstrtol(cp, 0, &n))
		n = 0;

	if (!strcmp(p, "run")) {
		msleep(n ? n : 1000);
		pr_info("run: slept %ldms\n", n ? n : 1000);
	} else if (!strcmp(p, "xmit")) {
		struct task_struct *t;

		if (cp && cp[0]) {
			char *q;
			long v;

			if (!kstrtol(strsep(&cp, " "), 0, &v))
				xmit_loops = (int)v;
			q = cp;
			if (q && q[0] && !kstrtol(q, 0, &v))
				xmit_interval_us = (int)v;
		}
		if (!g_spidev) {
			pr_err("xmit: no spi device\n");
			goto out;
		}
		t = kthread_run(xmit_thread, spi_get_drvdata(g_spidev), "mcrsim_xmit");
		if (IS_ERR(t))
			pr_err("xmit: kthread failed\n");
	} else if (!strcmp(p, "syncprobe")) {
		if (!g_spidev) {
			pr_err("syncprobe: no spi device\n");
			goto out;
		}
		atomic_set(&syncprobe_done, 0);
		INIT_DELAYED_WORK(&syncprobe_wd, syncprobe_watchdog);
		schedule_delayed_work(&syncprobe_wd, msecs_to_jiffies(5000));
		{
			struct task_struct *t;

			t = kthread_run(syncprobe_run, spi_get_drvdata(g_spidev), "mcrsim_syncp");
			if (IS_ERR(t))
				pr_err("syncprobe: kthread failed\n");
		}
	} else if (!strcmp(p, "status")) {
		pr_info("status: msgs=%ld delayed_msgs(stalled)=%ld irq_submits=%ld chain_steps=%ld xmit_submits=%ld regmap_async=%ld\n",
			atomic_long_read(&stat_msgs),
			atomic_long_read(&stat_delayed),
			atomic_long_read(&stat_irq_submit),
			atomic_long_read(&stat_chain_steps),
			atomic_long_read(&stat_xmit),
			atomic_long_read(&stat_async_calls));
	} else {
		pr_info("unknown action '%s'\n", p);
	}
out:
	pr_info("ACTION_COMPLETE\n");
}

static DECLARE_WORK(action_work, action_workfn);

static int action_set(const char *val, const struct kernel_param *kp)
{
	int ret = param_set_copystring(val, kp);

	if (ret == 0)
		schedule_work(&action_work);
	return ret;
}

static const struct kernel_param_ops action_ops = {
	.set = action_set,
	.get = param_get_string,
};
static struct kparam_string action_param = {
	.maxlen = sizeof(action_cmd),
	.string = action_cmd,
};
module_param_cb(action, &action_ops, &action_param, 0200);
MODULE_PARM_DESC(action, "run <ms> | xmit <n> <us> | syncprobe | status");

/* ------------------------------------------------------------------ */
/* module init: fake controller + slave device                         */

static int __init mcrsim_init(void)
{
	struct mcrsim_ctrl *c;
	struct spi_board_info info;
	int ret;

	ret = spi_register_driver(&mcrsim_driver);
	if (ret)
		return ret;

	c = kzalloc(sizeof(*c), GFP_KERNEL);
	if (!c) {
		ret = -ENOMEM;
		goto err_drv;
	}
	c->pdev = platform_device_register_simple("mcrsim-ctrl", -1, NULL, 0);
	if (IS_ERR(c->pdev)) {
		ret = PTR_ERR(c->pdev);
		goto err_free;
	}

	c->ctlr = spi_alloc_host(&c->pdev->dev, 0);
	if (!c->ctlr) {
		ret = -ENOMEM;
		goto err_pdev;
	}
	c->ctlr->bus_num = -1;
	c->ctlr->num_chipselect = SIM_CS_MAX;
	c->ctlr->mode_bits = SPI_CPOL | SPI_CPHA | SPI_CS_HIGH | SPI_LSB_FIRST;
	c->ctlr->transfer_one_message = mcrsim_transfer_one_message;
	spi_controller_set_devdata(c->ctlr, c);

	ret = spi_register_controller(c->ctlr);
	if (ret)
		goto err_ctlr;
	g_ctrl = c;

	memset(&info, 0, sizeof(info));
	strscpy(info.modalias, "mcrsim", sizeof(info.modalias));
	info.bus_num = c->ctlr->bus_num;
	info.chip_select = 0;
	info.max_speed_hz = 1000000;
	info.mode = SPI_MODE_0;

	g_spidev = spi_new_device(c->ctlr, &info);
	if (!g_spidev) {
		pr_err("spi_new_device failed\n");
		ret = -ENODEV;
		goto err_ctrl_reg;
	}

	pr_info("init: controller bus_num=%d device=%s (unbind: echo %s > /sys/bus/spi/drivers/mcrsim/unbind)\n",
		c->ctlr->bus_num, dev_name(&g_spidev->dev), dev_name(&g_spidev->dev));
	return 0;

err_ctrl_reg:
	spi_unregister_controller(c->ctlr);
	g_ctrl = NULL;
err_ctlr:
	spi_controller_put(c->ctlr);
err_pdev:
	platform_device_unregister(c->pdev);
err_free:
	kfree(c);
err_drv:
	spi_unregister_driver(&mcrsim_driver);
	return ret;
}

static void __exit mcrsim_exit(void)
{
	if (g_spidev) {
		spi_unregister_device(g_spidev);
		g_spidev = NULL;
	}
	if (g_ctrl) {
		spi_unregister_controller(g_ctrl->ctlr);
		platform_device_unregister(g_ctrl->pdev);
		kfree(g_ctrl);
		g_ctrl = NULL;
	}
	spi_unregister_driver(&mcrsim_driver);
	pr_info("removed\n");
}

module_init(mcrsim_init);
module_exit(mcrsim_exit);
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("HV-2/N-1 verification: mcr20a-shaped async SPI chain vs unbind");