gsp-phys: freeze-safe release, boot-proof READY, and guest-wiring P2/P3/P5-wave1

Host-freeze fix bundle (validated 2/2 supervised cycles):
- power unlock now latches D0/D3cold by default (new gsp_phys_repower,
  default 0); releasing the claim can no longer power-gate the board and
  hard-freeze the host on resume
- bounded post-rebind readiness gate before dropping the locks; bounded D0
  settle/verify before the unbind
- READY means a live GSP-RM session: the async worker and the module-owned
  join only expose READY when init-done is seen and no unload was sent;
  otherwise the claim is released fail-open
- FWSEC reads DMEM0/IMEM back before STARTCPU and fails fast when the
  staging is not visible
- the claim harness leaves the board latched by default and takes
  --restore-power to opt back into the old restore

Guest-wiring:
- P2: generation-keyed fn65 static-config snapshot cache plus a default-off
  identity overlay (new gsp_phys_static_live) for name/short-name/VBIOS;
  the probe cycle runs once per backend generation at READY
- P3: default-off read-only RM_CONTROL forwarding whitelist (new
  gsp_phys_ctrl_forward) with emulated-handle translation to the adopted
  server handles and full emulation fallback
- P5 wave 1: default-off shadow execution of guest CE pushes on the ladder
  channel (new gsp_phys_work_forward) via a dedicated ordered work item;
  the ladder stage-4 submission is refactored into a shared bounded helper
  and the software release path stays authoritative

Live: module-owned boot to init-done, guest smoke ALL OK 64/64 with P2/P3
enabled, 514 shadow copies observed on the physical CE; all cycles released
cleanly with WPR2 down and no reboot.

1 files changed, 538 insertions, 91 deletions

diff --git a/kernel/gsp_phys_booter.c b/kernel/gsp_phys_booter.c
index 255d79e..f9c0947 100644
--- a/kernel/gsp_phys_booter.c
+++ b/kernel/gsp_phys_booter.c
@@ -955,6 +955,25 @@ static struct gsp_phys_boot_stage {
 	struct gsp_phys_boot_chunk logrm;	/* LOGRM task log */
 } boot_stage;
 
+/*
+ * Live-session accessor for the P1 fail-open check: the real GSP-RM session
+ * is usable only after stage 3e saw the init-done event and before the
+ * UNLOADING_GUEST_DRIVER RPC has been published in this claim.
+ */
+bool gsp_phys_boot_rm_ready(void)
+{
+	return boot_stage.rm_init_done && !boot_stage.rm_unload_submitted;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_boot_rm_ready);
+
+/* Companion detail for the fail-open log line (see gsp_phys_backend_start_work). */
+void gsp_phys_boot_rm_flags(u32 *init_done, u32 *unload_submitted)
+{
+	*init_done = boot_stage.rm_init_done;
+	*unload_submitted = boot_stage.rm_unload_submitted;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_boot_rm_flags);
+
 static void *gsp_phys_chunk_page_va(const struct gsp_phys_boot_chunk *c,
 				    u32 page)
 {
@@ -4163,6 +4182,15 @@ static void gsp_phys_sci_string(char *dst, size_t dst_len, const u8 *p,
 }
 
 /*
+ * Decoded function-65 snapshot cache (P2). Filled by gsp_phys_sci_decode()
+ * under gsp_phys_rm_lock, served by gsp_phys_backend_static_get() only while
+ * the cache generation still matches the backend session's and the session is
+ * READY, and dropped by gsp_phys_backend_static_invalidate() when the backend
+ * leaves READY.
+ */
+static struct gsp_phys_backend_static_config gsp_phys_backend_static;
+
+/*
  * Decode the GspStaticConfigInfo fields this module needs. Every field is
  * bounds-checked against the reply payload length, so a short or older reply
  * logs only what it carries. The struct itself never lands on the stack.
@@ -4243,6 +4271,84 @@ static void gsp_phys_sci_decode(const u8 *p, u32 len)
 			boot_stage.rm_probe_internal_valid = true;
 		}
 	}
+
+	/*
+	 * Fill the backend static-config snapshot from this decode. Every
+	 * decoded field is copied (fb geometry/PDE bases are kept for later
+	 * phases), and valid mirrors the all-three-handles-nonzero test above
+	 * so the guest overlay and the rm-call stage-5 adoption agree. The
+	 * generation key ties the cache to the backend session; the getter
+	 * refuses it once the session leaves READY.
+	 */
+	gsp_phys_backend_static.valid = false;
+	gsp_phys_backend_static.generation = gsp_phys_backend_generation();
+	gsp_phys_backend_static.fw_crc32 = gsp_phys_gpu_fw_crc32();
+	gsp_phys_backend_static.payload_len = len;
+	gsp_phys_backend_static.h_internal_client =
+		boot_stage.rm_probe_h_internal_client;
+	gsp_phys_backend_static.h_internal_device =
+		boot_stage.rm_probe_h_internal_device;
+	gsp_phys_backend_static.h_internal_subdevice =
+		boot_stage.rm_probe_h_internal_subdevice;
+	gsp_phys_backend_static.bar1_pde_base =
+		gsp_phys_sci_u64(p, len, GSP_PHYS_SCI_BAR1_PDE_BASE_OFF);
+	gsp_phys_backend_static.bar2_pde_base =
+		gsp_phys_sci_u64(p, len, GSP_PHYS_SCI_BAR2_PDE_BASE_OFF);
+	gsp_phys_backend_static.fb_length =
+		gsp_phys_sci_u64(p, len, GSP_PHYS_SCI_FB_LENGTH_OFF);
+	gsp_phys_backend_static.fbio_mask =
+		gsp_phys_sci_u64(p, len, GSP_PHYS_SCI_FBIO_MASK_OFF);
+	gsp_phys_backend_static.fbp_mask =
+		gsp_phys_sci_u64(p, len, GSP_PHYS_SCI_FBP_MASK_OFF);
+	gsp_phys_backend_static.fb_bus_width =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_FB_BUS_WIDTH_OFF);
+	gsp_phys_backend_static.fb_ram_type =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_FB_RAM_TYPE_OFF);
+	gsp_phys_backend_static.l2_cache_size =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_L2_CACHE_SIZE_OFF);
+	gsp_phys_backend_static.engine_caps[0] =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_ENGINE_CAPS_OFF);
+	gsp_phys_backend_static.engine_caps[1] =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_ENGINE_CAPS_OFF + 4);
+	gsp_phys_backend_static.engine_caps[2] =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_ENGINE_CAPS_OFF + 8);
+	strscpy(gsp_phys_backend_static.gpu_name, name,
+		sizeof(gsp_phys_backend_static.gpu_name));
+	strscpy(gsp_phys_backend_static.gpu_short_name, short_name,
+		sizeof(gsp_phys_backend_static.gpu_short_name));
+	gsp_phys_backend_static.sriov_max_gfid =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_SRIOV_MAX_GFID_OFF);
+	gsp_phys_backend_static.vbios_sub_vendor =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_VBIOS_SUB_VENDOR_OFF);
+	gsp_phys_backend_static.vbios_sub_device =
+		gsp_phys_sci_u32(p, len, GSP_PHYS_SCI_VBIOS_SUB_DEVICE_OFF);
+	gsp_phys_backend_static.b_vbios_valid =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_VBIOS_VALID_OFF);
+	gsp_phys_backend_static.b_page_retirement =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_PAGE_RETIREMENT_OFF);
+	gsp_phys_backend_static.b_split_vas =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_SPLIT_VAS_OFF);
+	gsp_phys_backend_static.b_self_hosted =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_SELF_HOSTED_OFF);
+	gsp_phys_backend_static.b_ats_supported =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_ATS_SUPPORTED_OFF);
+	gsp_phys_backend_static.b_poison_fuse =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_POISON_FUSE_OFF);
+	gsp_phys_backend_static.b_gpu_uefi =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_GPU_UEFI_OFF);
+	gsp_phys_backend_static.b_efi_init =
+		gsp_phys_sci_u8(p, len, GSP_PHYS_SCI_EFI_INIT_OFF);
+	gsp_phys_backend_static.valid = boot_stage.rm_probe_internal_valid;
+
+	pr_info("gsp-phys-boot: rm-probe: static snapshot cached: generation=%u fw_crc32=0x%08x payload=%u name='%s' handles=0x%08x/0x%08x/0x%08x valid=%u\n",
+		gsp_phys_backend_static.generation,
+		gsp_phys_backend_static.fw_crc32,
+		gsp_phys_backend_static.payload_len,
+		gsp_phys_backend_static.gpu_name,
+		gsp_phys_backend_static.h_internal_client,
+		gsp_phys_backend_static.h_internal_device,
+		gsp_phys_backend_static.h_internal_subdevice,
+		gsp_phys_backend_static.valid);
 }
 
 /*
@@ -4447,8 +4553,6 @@ static int gsp_phys_rm_probe(void __iomem *bar0)
  * Backend session transport API (P0 skeleton, inert)
  * ============================================================ */
 
-static struct gsp_phys_backend_static_config gsp_phys_backend_static;
-
 int gsp_phys_backend_rpc(u32 function, const void *req, u32 req_len,
 			 void **reply, u32 *reply_len, u32 timeout_ms)
 {
@@ -4549,7 +4653,77 @@ out:
 	return ret;
 }
 
-int gsp_phys_backend_static_snapshot(struct gsp_phys_backend_static_config *out)
+/*
+ * Bounded wait for the snapshot cache to become valid (P2). Polls in 10 ms
+ * sleeps and never holds gsp_phys_rm_lock across a sleep (each poll takes and
+ * releases it), so it is safe from the emulation thread. Fails fast with
+ * -EAGAIN when no session can produce a snapshot (OFF/STOPPING); returns
+ * -ENODATA when the session is live but the cache did not arrive in time.
+ * The caller fails open to the emulated model.
+ */
+#define GSP_PHYS_STATIC_WAIT_POLL_MS	10
+
+int gsp_phys_backend_static_wait(struct gsp_phys_backend_static_config *out,
+				 u32 timeout_ms)
+{
+	unsigned long deadline = jiffies + msecs_to_jiffies(timeout_ms);
+	int state;
+	int ret;
+
+	for (;;) {
+		ret = gsp_phys_backend_static_get(out);
+		if (!ret)
+			return 0;
+
+		state = gsp_phys_backend_state();
+		if (state != GSP_PHYS_BACKEND_READY &&
+		    state != GSP_PHYS_BACKEND_CLAIMING &&
+		    state != GSP_PHYS_BACKEND_BOOTING)
+			return -EAGAIN;
+
+		if (time_after_eq(jiffies, deadline))
+			return -ENODATA;
+		msleep(GSP_PHYS_STATIC_WAIT_POLL_MS);
+	}
+}
+EXPORT_SYMBOL_GPL(gsp_phys_backend_static_wait);
+
+/*
+ * Drop the cached static-config snapshot. Called when the backend session
+ * leaves READY (or a new claim starts): the cache generation key is bumped
+ * past the current backend generation so a concurrent getter cannot serve the
+ * previous session's decode; the next successful fn65 decode re-fills and
+ * re-keys it. Takes gsp_phys_rm_lock to serialise against the decode.
+ */
+void gsp_phys_backend_static_invalidate(void)
+{
+	mutex_lock(&gsp_phys_rm_lock);
+	gsp_phys_backend_static.valid = false;
+	gsp_phys_backend_static.generation = gsp_phys_backend_generation() + 1;
+	mutex_unlock(&gsp_phys_rm_lock);
+}
+
+/*
+ * Run one fn1+fn65 probe on a READY backend to fill the static snapshot
+ * cache. Takes gsp_phys_rm_lock inside gsp_phys_rm_probe(), so the caller
+ * must not hold it. Non-fatal: a failed probe leaves the cache as it was and
+ * the caller keeps the emulation path.
+ */
+int gsp_phys_backend_static_probe(void)
+{
+	void __iomem *bar0;
+
+	if (gsp_phys_backend_state() != GSP_PHYS_BACKEND_READY)
+		return -EAGAIN;
+
+	bar0 = gsp_phys_gpu_bar0();
+	if (!bar0)
+		return -ENODEV;
+
+	return gsp_phys_rm_probe(bar0);
+}
+
+int gsp_phys_backend_static_get(struct gsp_phys_backend_static_config *out)
 {
 	u32 generation;
 
@@ -4566,15 +4740,18 @@ int gsp_phys_backend_static_snapshot(struct gsp_phys_backend_static_config *out)
 		mutex_unlock(&gsp_phys_rm_lock);
 		return -EAGAIN;
 	}
-	if (!gsp_phys_backend_static.valid) {
+	if (!gsp_phys_backend_static.valid ||
+	    gsp_phys_backend_static.generation != generation) {
 		mutex_unlock(&gsp_phys_rm_lock);
-		pr_debug("gsp-phys-boot: backend static snapshot: cache not populated yet\n");
+		pr_debug("gsp-phys-boot: backend static snapshot: no valid cache for generation %u\n",
+			 generation);
 		return -ENODATA;
 	}
 	*out = gsp_phys_backend_static;
 	mutex_unlock(&gsp_phys_rm_lock);
 	return 0;
 }
+EXPORT_SYMBOL_GPL(gsp_phys_backend_static_get);
 
 /* ============================================================
  * Host RM client ladder (R1): gsp-phys-rm-call
@@ -5780,18 +5957,20 @@ static u32 *gsp_phys_rmcall_push(u32 *p, u32 method, u32 data)
 }
 
 /*
- * CE method block for one 4 KB sysmem-to-sysmem pitch copy with a release
- * semaphore, in the all-virtual user-channel shape: the semaphore methods
- * always carry a GPU VA (no mode bit; uvm_pascal_ce.c:42-63) and the src/dst
- * VAs translate through the channel PDB, so no SET_SRC/DST_PHYS_MODE pair is
- * pushed (uvm_maxwell_ce.c:179-205). The method order mirrors
+ * CE method block for one bytes-long sysmem-to-sysmem pitch copy with a
+ * release semaphore, in the all-virtual user-channel shape: the semaphore
+ * methods always carry a GPU VA (no mode bit; uvm_pascal_ce.c:42-63) and the
+ * src/dst VAs translate through the channel PDB, so no SET_SRC/DST_PHYS_MODE
+ * pair is pushed (uvm_maxwell_ce.c:179-205). The method order mirrors
  * _ceChannelPushMethodsBlock_GM107 (gpu/mem_mgr/arch/maxwell/
  * mem_utils_gm107.c:1968-2118) and the semaphore payload block
  * (channel_utils.c:812-820). Returns the block size in dwords (13 method
- * pairs = 26 dwords, <= GPFIFO entry LENGTH field, bits 30:10).
+ * pairs = 26 dwords, <= GPFIFO entry LENGTH field, bits 30:10). The ladder
+ * stage-4 call passes GSP_PHYS_RMCALL_COPY_SIZE; the P5 shadow submit passes
+ * the caller's size (both within one 4 KB ladder page).
  */
 static u32 gsp_phys_rmcall_build_methods(u32 *pb, u64 sem_va, u64 src_va,
-					 u64 dst_va)
+					 u64 dst_va, u32 bytes)
 {
 	u32 *p = pb;
 
@@ -5807,16 +5986,13 @@ static u32 gsp_phys_rmcall_build_methods(u32 *pb, u64 sem_va, u64 src_va,
 				 (u32)(dst_va >> 32));
 	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_OFFSET_OUT_LOWER,
 				 (u32)dst_va);
-	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_PITCH_OUT,
-				 GSP_PHYS_RMCALL_COPY_SIZE);
+	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_PITCH_OUT, bytes);
 	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_OFFSET_IN_UPPER,
 				 (u32)(src_va >> 32));
 	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_OFFSET_IN_LOWER,
 				 (u32)src_va);
-	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_PITCH_IN,
-				 GSP_PHYS_RMCALL_COPY_SIZE);
-	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_LINE_LENGTH_IN,
-				 GSP_PHYS_RMCALL_COPY_SIZE);
+	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_PITCH_IN, bytes);
+	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_LINE_LENGTH_IN, bytes);
 	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_LINE_COUNT, 1);
 	p = gsp_phys_rmcall_push(p, GSP_PHYS_RMCALL_CE_LAUNCH_DMA,
 				 GSP_PHYS_RMCALL_LAUNCH_DMA);
@@ -5824,6 +6000,326 @@ static u32 gsp_phys_rmcall_build_methods(u32 *pb, u64 sem_va, u64 src_va,
 }
 
 /*
+ * Result of one ladder CE submit, valid only when the submit helper returned 0.
+ * Kept for the caller's log line and the GP_GET/GPPut context.
+ */
+struct gsp_phys_rmcall_ce_submit {
+	u32 words;
+	u32 slot;
+	u32 gpput;
+	u32 gpget;
+	u32 e0;
+	u32 e1;
+	u32 doorbell;
+	u32 runlist_id;
+	u32 ch_id;
+	u64 pb_dma;
+	u64 sem_dma;
+	u64 src_dma;
+	u64 dst_dma;
+};
+
+/*
+ * One CE copy block on the ladder channel: clear the module semaphore, build
+ * the methods for bytes into the module pushbuffer, publish the GPFIFO entry
+ * and USERD GP_PUT and ring the doorbell. The caller must hold
+ * gsp_phys_rm_lock and must poll separately (stage 4 keeps its historical
+ * lock-across-poll; the P5 work API polls without the lock). Every guard of
+ * the original stage-4 submission body is preserved here: server-memory
+ * refusal, BAR0 presence/length, GP_PUT bounds, GPFIFO GET encode and the
+ * token check. The semaphore is cleared before the doorbell so a stale release
+ * from a previous run cannot be observed as this submission's completion.
+ */
+static int gsp_phys_rmcall_ce_submit_locked(void __iomem *bar0, u64 src_va,
+					    u64 dst_va, u64 sem_va, u32 bytes,
+					    struct gsp_phys_rmcall_ce_submit *out)
+{
+	struct gsp_phys_boot_chunk *pb =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_PB];
+	struct gsp_phys_boot_chunk *ring =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_GPFIFO];
+	struct gsp_phys_boot_chunk *userd_c =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_USERD];
+	struct gsp_phys_boot_chunk *sem =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SEM];
+	struct gsp_phys_boot_chunk *src =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SRC];
+	struct gsp_phys_boot_chunk *dst =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_DST];
+	resource_size_t bar0_len;
+	u32 *pb_va, *userd, *ring_va;
+	u32 words, gpput, gpget, slot, e0, e1;
+	u32 runlist_id, ch_id, doorbell;
+	u64 pb_gpu_va;
+
+	if (boot_stage.rmcall_channel_server_mem)
+		return -EOPNOTSUPP;
+	if (!boot_stage.pdev)
+		return -EINVAL;
+	bar0_len = pci_resource_len(boot_stage.pdev, 0);
+	if (bar0_len < GSP_PHYS_RMCALL_DOORBELL_OFF + 4)
+		return -EIO;
+	if (!bar0 || !boot_stage.rmcall_token || !bytes ||
+	    bytes > GSP_PHYS_RMCALL_COPY_SIZE)
+		return -EINVAL;
+	pb_gpu_va = boot_stage.rmcall_va_pb;
+	if (!pb_gpu_va || (pb_gpu_va & 3))
+		return -EIO;
+
+	pb_va = gsp_phys_chunk_page_va(pb, 0);
+	ring_va = gsp_phys_chunk_page_va(ring, 0);
+	userd = gsp_phys_chunk_page_va(userd_c, 0);
+	if (!pb_va || !ring_va || !userd ||
+	    !gsp_phys_chunk_page_va(sem, 0) ||
+	    !gsp_phys_chunk_page_va(src, 0) ||
+	    !gsp_phys_chunk_page_va(dst, 0))
+		return -EIO;
+
+	/* A stale release from a previous submission must not be observed. */
+	WRITE_ONCE(*(u32 *)gsp_phys_chunk_page_va(sem, 0), 0);
+	wmb();
+
+	words = gsp_phys_rmcall_build_methods(pb_va, sem_va, src_va, dst_va,
+					      bytes);
+
+	gpput = get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPPUT_OFF);
+	gpget = get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPGET_OFF);
+	if (gpput >= GSP_PHYS_RMCALL_GPFIFO_ENTRIES) {
+		pr_err("gsp-phys-boot: rm-call: USERD GP_PUT=%u is outside the %u-entry ring; refusing to submit\n",
+		       gpput, GSP_PHYS_RMCALL_GPFIFO_ENTRIES);
+		return -EIO;
+	}
+	slot = gpput;
+
+	/*
+	 * e0: bits 31:2 hold the GET field, which itself is (fetch address
+	 * >> 2); the engine reconstructs (field << 2) = (e0 & ~3), so for a
+	 * 4 KB-aligned pushbuffer VA the raw word is LO32(VA) with bits 1:0
+	 * clear. Storing LO32(VA) >> 2 here shifts a second time in the
+	 * engine (the 0x40000 fetch for pb VA 0x101000). FETCH/NO_CONTEXT_
+	 * SWITCH 0 (class/cl906f.h:136-142); e1: GET_HI 7:0 = the VA high byte
+	 * and LENGTH 30:10 in dwords, exactly the reference RM's own submission
+	 * (mem_utils_gm107.c:1858-1868: get = pbGpuVA + offset, NvU64_HI32(get)
+	 * in GET_HI, NvU64_LO32(get) >> 2 in GET, PRIV USER and LEVEL MAIN).
+	 * The engine translates this VA through the channel PDB built in
+	 * gsp_phys_rmcall_pdb_build.
+	 */
+	e0 = (u32)pb_gpu_va & ~3u;
+	if (unlikely(e0 != (u32)((pb_gpu_va >> 2) << 2)))
+		pr_err("gsp-phys-boot: rm-call: GPFIFO GET encode mismatch: e0=0x%08x decodes to fetch 0x%llx, expected pb VA 0x%llx\n",
+		       e0, (u64)(e0 & ~3u), pb_gpu_va);
+	e1 = (u32)((pb_gpu_va >> 32) & 0xffu) | (words << 10);
+	put_unaligned_le32(e0, (u8 *)ring_va + slot * 8);
+	put_unaligned_le32(e1, (u8 *)ring_va + slot * 8 + 4);
+	dma_wmb();
+	/*
+	 * GP_PUT advances on the client sysmem USERD chunk, which is the
+	 * channel's hardware USERD because the params carry that same chunk;
+	 * the host write is therefore what the engine polls (cycle 21).
+	 */
+	put_unaligned_le32((gpput + 1) % GSP_PHYS_RMCALL_GPFIFO_ENTRIES,
+			   (u8 *)userd + GSP_PHYS_RMCALL_USERD_GPPUT_OFF);
+	/*
+	 * Doorbell value: workSubmitToken = ((runlistId << 16) & 0x7f00000) |
+	 * (chId & 0xfff) (kernel_fifo_tu102.c:118-121). The server-generated
+	 * token already is that value; log its runlist/ChID decomposition and
+	 * write the formula's result (identical unless the token carries stray
+	 * high bits, which is logged).
+	 */
+	runlist_id = (boot_stage.rmcall_token >> 16) & 0x7fu;
+	ch_id = boot_stage.rmcall_token & 0xfffu;
+	doorbell = ((runlist_id << 16) & 0x7f00000u) | (ch_id & 0xfffu);
+	if (doorbell != boot_stage.rmcall_token)
+		pr_warn("gsp-phys-boot: rm-call: token 0x%x carries bits outside runlistId[22:16]|chId[11:0]; writing the recomposed doorbell 0x%x\n",
+			boot_stage.rmcall_token, doorbell);
+	wmb();
+	writel(doorbell, bar0 + GSP_PHYS_RMCALL_DOORBELL_OFF);
+
+	if (out) {
+		out->words = words;
+		out->slot = slot;
+		out->gpput = gpput;
+		out->gpget = gpget;
+		out->e0 = e0;
+		out->e1 = e1;
+		out->doorbell = doorbell;
+		out->runlist_id = runlist_id;
+		out->ch_id = ch_id;
+		out->pb_dma = (u64)gsp_phys_chunk_page_dma(pb, 0);
+		out->sem_dma = (u64)gsp_phys_chunk_page_dma(sem, 0);
+		out->src_dma = (u64)gsp_phys_chunk_page_dma(src, 0);
+		out->dst_dma = (u64)gsp_phys_chunk_page_dma(dst, 0);
+	}
+	return 0;
+}
+
+/*
+ * P5 wave-1 shadow work API (see gsp_phys_booter.h). The counters are plain
+ * increments from the service thread and the debugfs paths; only logging and
+ * the debugfs state file read them.
+ */
+#define GSP_PHYS_BACKEND_WORK_POLL_MAX_MS	200u
+
+static u64 gsp_phys_work_submitted;
+static u64 gsp_phys_work_released;
+static u64 gsp_phys_work_timed_out;
+static u64 gsp_phys_work_refused;
+
+bool gsp_phys_backend_work_ready(void)
+{
+	if (!gsp_phys_work_forward_enabled())
+		return false;
+	if (gsp_phys_backend_state() != GSP_PHYS_BACKEND_READY)
+		return false;
+	if (!gsp_phys_gpu_claimed() || !boot_stage.rm_init_done ||
+	    boot_stage.rm_unload_submitted)
+		return false;
+	if (!boot_stage.rmcall_channel || boot_stage.rmcall_channel_server_mem ||
+	    !boot_stage.rmcall_token)
+		return false;
+	if (!boot_stage.rmcall_va_pb || !boot_stage.rmcall_va_gpfifo ||
+	    !boot_stage.rmcall_va_src || !boot_stage.rmcall_va_dst ||
+	    !boot_stage.rmcall_va_sem || !boot_stage.rmcall_pdb_root)
+		return false;
+	return true;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_backend_work_ready);
+
+bool gsp_phys_backend_work_operands(u64 *src_va, u64 *dst_va, u64 *sem_va,
+				    u32 *bytes)
+{
+	if (!gsp_phys_backend_work_ready())
+		return false;
+	if (src_va)
+		*src_va = boot_stage.rmcall_va_src;
+	if (dst_va)
+		*dst_va = boot_stage.rmcall_va_dst;
+	if (sem_va)
+		*sem_va = boot_stage.rmcall_va_sem;
+	if (bytes)
+		*bytes = GSP_PHYS_RMCALL_COPY_SIZE;
+	return true;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_backend_work_operands);
+
+int gsp_phys_backend_ce_copy_submit(u64 src_va, u64 dst_va, u64 sem_va,
+				    u32 bytes, u32 poll_ms)
+{
+	struct gsp_phys_rmcall_ce_submit sub;
+	struct gsp_phys_boot_chunk *sem =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SEM];
+	struct gsp_phys_boot_chunk *src =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SRC];
+	struct gsp_phys_boot_chunk *dst =
+		&boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_DST];
+	void __iomem *bar0;
+	unsigned long deadline;
+	u32 poll;
+	int ret;
+
+	if (!gsp_phys_backend_work_ready()) {
+		gsp_phys_work_refused++;
+		return -EAGAIN;
+	}
+	if (!bytes || bytes > GSP_PHYS_RMCALL_COPY_SIZE) {
+		gsp_phys_work_refused++;
+		return -EINVAL;
+	}
+	if (poll_ms > GSP_PHYS_BACKEND_WORK_POLL_MAX_MS)
+		poll_ms = GSP_PHYS_BACKEND_WORK_POLL_MAX_MS;
+
+	bar0 = gsp_phys_gpu_bar0();
+	if (!bar0) {
+		gsp_phys_work_refused++;
+		return -EAGAIN;
+	}
+
+	gsp_phys_work_submitted++;
+	mutex_lock(&gsp_phys_rm_lock);
+	/*
+	 * Re-check under the lock: the release may have committed between the
+	 * readiness gate and the lock, and the ladder state may have changed.
+	 * No device write happens on any of these paths.
+	 */
+	if (!gsp_phys_gpu_claimed() || !boot_stage.rm_init_done ||
+	    boot_stage.rm_unload_submitted || !boot_stage.rmcall_channel ||
+	    boot_stage.rmcall_channel_server_mem || !boot_stage.rmcall_token) {
+		mutex_unlock(&gsp_phys_rm_lock);
+		gsp_phys_work_refused++;
+		if ((gsp_phys_work_refused & 63) == 1)
+			pr_info("gsp-phys-work: shadow submit refused before the doorbell (backend changed) refused=%llu\n",
+				gsp_phys_work_refused);
+		return -EAGAIN;
+	}
+	ret = gsp_phys_rmcall_ce_submit_locked(bar0, src_va, dst_va, sem_va,
+					       bytes, &sub);
+	mutex_unlock(&gsp_phys_rm_lock);
+	if (ret) {
+		gsp_phys_work_refused++;
+		if ((gsp_phys_work_refused & 63) == 1)
+			pr_info("gsp-phys-work: shadow submit refused (ret %d) refused=%llu\n",
+				ret, gsp_phys_work_refused);
+		return ret;
+	}
+
+	/*
+	 * Poll the module semaphore WITHOUT gsp_phys_rm_lock: the release is
+	 * written by the engine, never by a host path, so no lock is needed
+	 * and holding it across the poll would stall the RPC path (plan §1.2).
+	 */
+	deadline = jiffies + msecs_to_jiffies(poll_ms);
+	for (;;) {
+		poll = READ_ONCE(*(u32 *)gsp_phys_chunk_page_va(sem, 0));
+		if (poll == GSP_PHYS_RMCALL_SEM_PAYLOAD)
+			break;
+		if (time_after_eq(jiffies, deadline)) {
+			gsp_phys_work_timed_out++;
+			if ((gsp_phys_work_timed_out & 63) == 1)
+				pr_info("gsp-phys-work: shadow copy timeout after %u ms (sem=0x%08x want=0x%08x) timeouts=%llu\n",
+					poll_ms, poll,
+					GSP_PHYS_RMCALL_SEM_PAYLOAD,
+					gsp_phys_work_timed_out);
+			return -ETIMEDOUT;
+		}
+		msleep(GSP_PHYS_RMCALL_POLL_MS);
+	}
+
+	/*
+	 * Only the ladder operands are verified: the engine copied within the
+	 * module-owned src/dst pages, so a mismatch is a real engine/encode
+	 * failure. A caller with different VAs (none in wave 1) is not
+	 * verified against the ladder pages.
+	 */
+	if (src_va == boot_stage.rmcall_va_src &&
+	    dst_va == boot_stage.rmcall_va_dst &&
+	    memcmp(gsp_phys_chunk_page_va(src, 0),
+		   gsp_phys_chunk_page_va(dst, 0), bytes) != 0) {
+		gsp_phys_work_refused++;
+		pr_info("gsp-phys-work: semaphore observed but the destination does not match the source after %u bytes (refused=%llu)\n",
+			bytes, gsp_phys_work_refused);
+		return -EIO;
+	}
+
+	gsp_phys_work_released++;
+	return 0;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_backend_ce_copy_submit);
+
+void gsp_phys_backend_work_counters(u64 *submitted, u64 *released,
+				    u64 *timed_out, u64 *refused)
+{
+	if (submitted)
+		*submitted = gsp_phys_work_submitted;
+	if (released)
+		*released = gsp_phys_work_released;
+	if (timed_out)
+		*timed_out = gsp_phys_work_timed_out;
+	if (refused)
+		*refused = gsp_phys_work_refused;
+}
+EXPORT_SYMBOL_GPL(gsp_phys_backend_work_counters);
+
+/*
  * Stage 4: the corrected token sequence - BIND (COPY0) -> GPFIFO_SCHEDULE ->
  * TSG_GPFIFO_SCHEDULE -> SET_WORK_SUBMIT_TOKEN_NOTIF_INDEX ->
  * GET_WORK_SUBMIT_TOKEN - then the first real work item: write the CE method
@@ -5848,20 +6344,18 @@ static u32 gsp_phys_rmcall_build_methods(u32 *pb, u64 sem_va, u64 src_va,
  */
 static int gsp_phys_rmcall_stage4(void __iomem *bar0)
 {
-	struct gsp_phys_boot_chunk *pb = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_PB];
-	struct gsp_phys_boot_chunk *ring = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_GPFIFO];
 	struct gsp_phys_boot_chunk *userd_c = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_USERD];
 	struct gsp_phys_boot_chunk *sem = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SEM];
 	struct gsp_phys_boot_chunk *src = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_SRC];
 	struct gsp_phys_boot_chunk *dst = &boot_stage.rmcall_chunks[GSP_PHYS_RMCALL_BUF_DST];
+	struct gsp_phys_rmcall_ce_submit sub;
 	u32 bind_params = GSP_PHYS_RMCALL_ENGINE_COPY0;	/* NVA06F engineType */
 	u32 notif_index = GSP_PHYS_RMCALL_NOTIF_INDEX;	/* NVC36F index */
 	u8 sched_params[2] = { 1, 0 };	/* NVA06F bEnable=1, bSkipSubmit=0 */
 	resource_size_t bar0_len;
-	u32 *pb_va, *userd, *ring_va;
-	u32 words, gpput, gpget, slot, e0, e1, poll;
-	u32 token, rpc_result, status, runlist_id, ch_id, doorbell;
-	u64 pb_dma, sem_dma, src_dma, dst_dma;
+	u32 *userd;
+	u32 gpput, gpget, poll;
+	u32 token, rpc_result, status;
 	unsigned long deadline;
 	int ret;
 
@@ -5992,79 +6486,32 @@ static int gsp_phys_rmcall_stage4(void __iomem *bar0)
 	}
 	boot_stage.rmcall_token = token;
 
-	pb_va = gsp_phys_chunk_page_va(pb, 0);
-	ring_va = gsp_phys_chunk_page_va(ring, 0);
 	userd = gsp_phys_chunk_page_va(userd_c, 0);
-	pb_dma = (u64)gsp_phys_chunk_page_dma(pb, 0);
-	sem_dma = (u64)gsp_phys_chunk_page_dma(sem, 0);
-	src_dma = (u64)gsp_phys_chunk_page_dma(src, 0);
-	dst_dma = (u64)gsp_phys_chunk_page_dma(dst, 0);
 
-	words = gsp_phys_rmcall_build_methods(pb_va, GSP_PHYS_RMCALL_VA_SEM,
-					      GSP_PHYS_RMCALL_VA_SRC,
-					      GSP_PHYS_RMCALL_VA_DST);
-
-	gpput = get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPPUT_OFF);
-	gpget = get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPGET_OFF);
-	if (gpput >= GSP_PHYS_RMCALL_GPFIFO_ENTRIES) {
-		pr_err("gsp-phys-boot: rm-call: USERD GP_PUT=%u is outside the %u-entry ring; refusing to submit\n",
-		       gpput, GSP_PHYS_RMCALL_GPFIFO_ENTRIES);
-		return -EIO;
-	}
-	slot = gpput;
-
-	/*
-	 * e0: bits 31:2 hold the GET field, which itself is (fetch address
-	 * >> 2); the engine reconstructs (field << 2) = (e0 & ~3), so for a
-	 * 4 KB-aligned pushbuffer VA the raw word is LO32(VA) with bits 1:0
-	 * clear. Storing LO32(VA) >> 2 here shifts a second time in the
-	 * engine (the 0x40000 fetch for pb VA 0x101000). FETCH/NO_CONTEXT_
-	 * SWITCH 0 (class/cl906f.h:136-142); e1: GET_HI 7:0 = the VA high byte
-	 * and LENGTH 30:10 in dwords, exactly the reference RM's own submission
-	 * (mem_utils_gm107.c:1858-1868: get = pbGpuVA + offset, NvU64_HI32(get)
-	 * in GET_HI, NvU64_LO32(get) >> 2 in GET, PRIV USER and LEVEL MAIN).
-	 * The engine translates this VA through the channel PDB built in
-	 * gsp_phys_rmcall_pdb_build.
-	 */
-	BUILD_BUG_ON(GSP_PHYS_RMCALL_VA_PB & 3);
-	e0 = (u32)GSP_PHYS_RMCALL_VA_PB & ~3u;
-	if (unlikely(e0 != (u32)((GSP_PHYS_RMCALL_VA_PB >> 2) << 2)))
-		pr_err("gsp-phys-boot: rm-call: GPFIFO GET encode mismatch: e0=0x%08x decodes to fetch 0x%llx, expected pb VA 0x%llx\n",
-		       e0, (u64)(e0 & ~3u), (u64)GSP_PHYS_RMCALL_VA_PB);
-	e1 = (u32)((GSP_PHYS_RMCALL_VA_PB >> 32) & 0xffu) | (words << 10);
-	put_unaligned_le32(e0, (u8 *)ring_va + slot * 8);
-	put_unaligned_le32(e1, (u8 *)ring_va + slot * 8 + 4);
-	dma_wmb();
 	/*
-	 * GP_PUT advances on the client sysmem USERD chunk, which is the
-	 * channel's hardware USERD because the params carry that same chunk;
-	 * the host write is therefore what the engine polls (cycle 21).
+	 * The submission body is shared with the P5 shadow API
+	 * (gsp_phys_rmcall_ce_submit_locked): method block, GPFIFO entry,
+	 * USERD GP_PUT and doorbell, with the same guards. Stage 4 keeps its
+	 * historical lock-across-poll: gsp_phys_rmcall_run() holds
+	 * gsp_phys_rm_lock for the whole stage.
 	 */
-	put_unaligned_le32((gpput + 1) % GSP_PHYS_RMCALL_GPFIFO_ENTRIES,
-			   (u8 *)userd + GSP_PHYS_RMCALL_USERD_GPPUT_OFF);
-	/*
-	 * Doorbell value: workSubmitToken = ((runlistId << 16) & 0x7f00000) |
-	 * (chId & 0xfff) (kernel_fifo_tu102.c:118-121). The server-generated
-	 * token already is that value; log its runlist/ChID decomposition and
-	 * write the formula's result (identical unless the token carries stray
-	 * high bits, which is logged).
-	 */
-	runlist_id = (boot_stage.rmcall_token >> 16) & 0x7fu;
-	ch_id = boot_stage.rmcall_token & 0xfffu;
-	doorbell = ((runlist_id << 16) & 0x7f00000u) | (ch_id & 0xfffu);
-	if (doorbell != boot_stage.rmcall_token)
-		pr_warn("gsp-phys-boot: rm-call: token 0x%x carries bits outside runlistId[22:16]|chId[11:0]; writing the recomposed doorbell 0x%x\n",
-			boot_stage.rmcall_token, doorbell);
-	wmb();
-	writel(doorbell, bar0 + GSP_PHYS_RMCALL_DOORBELL_OFF);
+	ret = gsp_phys_rmcall_ce_submit_locked(bar0, GSP_PHYS_RMCALL_VA_SRC,
+					       GSP_PHYS_RMCALL_VA_DST,
+					       GSP_PHYS_RMCALL_VA_SEM,
+					       GSP_PHYS_RMCALL_COPY_SIZE, &sub);
+	if (ret) {
+		pr_err("gsp-phys-boot: rm-call: stage 4 aborted before the doorbell: submission refused (%d)\n",
+		       ret);
+		return ret;
+	}
 
 	pr_info("gsp-phys-boot: rm-call: submitted: pb=0x%llx pbVa=0x%llx gpFifoVa=0x%llx vaSrc=0x%llx vaDst=0x%llx vaSem=0x%llx userd=sysmem (host-writable) words=%u slot=%u (was GPGet=%u GPPut=%u) e0=0x%08x e1=0x%08x token=0x%08x (runlistId=%u chId=0x%x) doorbell=0x%08x @0x%x; polling sem=0x%llx for 0x%08x (<=%u ms)\n",
-		pb_dma, GSP_PHYS_RMCALL_VA_PB, GSP_PHYS_RMCALL_VA_GPFIFO,
+		sub.pb_dma, GSP_PHYS_RMCALL_VA_PB, GSP_PHYS_RMCALL_VA_GPFIFO,
 		GSP_PHYS_RMCALL_VA_SRC, GSP_PHYS_RMCALL_VA_DST,
 		GSP_PHYS_RMCALL_VA_SEM,
-		words, slot, gpget, gpput, e0, e1,
-		boot_stage.rmcall_token, runlist_id, ch_id, doorbell,
-		GSP_PHYS_RMCALL_DOORBELL_OFF, sem_dma,
+		sub.words, sub.slot, sub.gpget, sub.gpput, sub.e0, sub.e1,
+		boot_stage.rmcall_token, sub.runlist_id, sub.ch_id, sub.doorbell,
+		GSP_PHYS_RMCALL_DOORBELL_OFF, sub.sem_dma,
 		GSP_PHYS_RMCALL_SEM_PAYLOAD, GSP_PHYS_RMCALL_TIMEOUT_MS);
 
 	deadline = jiffies + msecs_to_jiffies(GSP_PHYS_RMCALL_TIMEOUT_MS);
@@ -6103,9 +6550,9 @@ static int gsp_phys_rmcall_stage4(void __iomem *bar0)
 
 	gpget = get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPGET_OFF);
 	pr_info("gsp-phys-boot: rm-call: REAL COPY OK sem=0x%08x src=0x%llx dst=0x%llx bytes=%u GPGet=%u GPPut=%u words=%u token=0x%08x\n",
-		poll, src_dma, dst_dma, GSP_PHYS_RMCALL_COPY_SIZE, gpget,
+		poll, sub.src_dma, sub.dst_dma, GSP_PHYS_RMCALL_COPY_SIZE, gpget,
 		get_unaligned_le32((u8 *)userd + GSP_PHYS_RMCALL_USERD_GPPUT_OFF),
-		words, boot_stage.rmcall_token);
+		sub.words, boot_stage.rmcall_token);
 	return 0;
 }