vgsp: fail4 - device-survival chain for the large-buffer sequence + CE both-side op advance

Survival (the death cascade is gone for the full user sequence):
1. pool residue re-seed applies ABSOLUTE (cur=0) - the UVM assert
   (uvm_channel.c:204) that took the device down no longer fires
   (committed earlier in 67148bd8).
2. completion-hint window 10s->1.5s (67148bd8).
3. status-ring free-space discipline in gsp_rm_send_flags: the
   guest's consumption cursor (msgqRxHeader.readPtr at stat+rxHdrOff,
   the msgqTxGetFreeSpace source) gates EVENT posts; events drop
   freely when short, replies wait (bounded) and proceed. A guard
   rejects a cursor the guest cannot be behind (peer > our writePtr -
   the observed mirror tracks writePtr+1 and gating on it silenced
   every wake-up and wedged the clients).
Verified live: mail OK, nvidia-smi OK, victory_matmul OK,
vector_add completes (content still wrong) and victory_transformer
STATUS OK AFTER it - no ISR panic, no guest death, guest ssh alive.

Content: CE linear auto-advance now applied to BOTH offset registers
after each executed op (the driver re-sets one side per op and the
engine advances the other; H2D ops showed an explicit advancing dst
with a fixed src base: the device a held h_a[0..1MB) replicated).
vector_add content still FAIL (~66% bad) - the H2D operand stream
under retry/replay passes is the next lead (opadd rows show five dst
x two src combos with heavy retry multiplicity).

2 files changed, 105 insertions, 6 deletions

diff --git a/kernel/gsp_rm_emulation.c b/kernel/gsp_rm_emulation.c
index f6ca2c1..764ef05 100644
--- a/kernel/gsp_rm_emulation.c
+++ b/kernel/gsp_rm_emulation.c
@@ -552,6 +552,94 @@ static int gsp_rm_send_flags(struct gsp_emu_context *ctx, u32 function,
 		return -EOVERFLOW;
 	}
 
+	/*
+	 * fail4: free-space discipline. The guest's consumption cursor for
+	 * the status queue (msgqRxHeader.readPtr, at stat+rxHdrOff under the
+	 * queue's SWAP_RX layout - same location msgqTxGetFreeSpace polls)
+	 * gates this write. Overwriting unread records mixes entries in the
+	 * guest's RPC stream: under a completion burst the guest then runs
+	 * garbage RM callbacks (rm_run_rc_callback+0x0) and dies. Events are
+	 * wake-ups and drop freely; a reply waits briefly for the ring to
+	 * drain and is dropped (never overwritten) if it stays full.
+	 */
+	{
+		u32 peer_read = 0;
+		bool have_peer = false;
+		int w;
+
+		for (w = 0; w < 51; w++) {
+			u32 used, space, rxoff = 0;
+			u8 shdr[32];
+
+			if (gsp_rm_read_buf(ctx, rm->stat_off, shdr,
+					    sizeof(shdr)))
+				break;
+			memcpy(&rxoff, shdr + 24, 4);
+			if (!rxoff || rxoff >= 0x1000)
+				break;
+			if (gsp_rm_read_buf(ctx, rm->stat_off + rxoff,
+					    &peer_read, 4) ||
+			    peer_read >= rm->stat_msgcount)
+				break;	/* no usable cursor: blind write */
+			/*
+			 * fail4: only a cursor the guest is actually BEHIND
+			 * (peer <= our writePtr) is a consumption cursor.
+			 * The observed value at stat+rxHdrOff under this
+			 * queue's flags tracks writePtr+1 (a mirror), and
+			 * gating on it silenced every wake-up event and
+			 * wedged the clients. A mirror (peer > write) falls
+			 * back to the historical blind write.
+			 */
+			if (peer_read > rm->stat_write)
+				break;
+			have_peer = true;
+			used = (rm->stat_write + rm->stat_msgcount -
+				peer_read) % rm->stat_msgcount;
+			space = rm->stat_msgcount - used - 1u;
+			if (elem_count <= space)
+				break;
+			/*
+			 * Only event wake-ups drop freely; a reply waits
+			 * (bounded) and then proceeds - dropping replies
+			 * breaks the guest's RM init handshake (observed:
+			 * two fn=0x1 replies dropped at boot -> the device
+			 * never comes up).
+			 */
+			if (function == GSPRM_RPC_EVENT_POST_EVENT || w == 50)
+				break;
+			mutex_unlock(&rm->stat_lock);
+			msleep(1);
+			mutex_lock(&rm->stat_lock);
+		}
+		if (have_peer) {
+			u32 used = (rm->stat_write + rm->stat_msgcount -
+				    peer_read) % rm->stat_msgcount;
+			u32 space = rm->stat_msgcount - used - 1u;
+
+			if (elem_count > space &&
+			    function == GSPRM_RPC_EVENT_POST_EVENT) {
+				rm->events_dropped++;
+				if (rm->events_dropped <= 8 ||
+				    (rm->events_dropped % 1000) == 0)
+					nv_vgpu_info(ctx->vgpu,
+						     "vgsp-rm: status ring full (free=%u need=%u peer=%u write=%u mc=%u); event dropped, total=%llu",
+						     space, elem_count,
+						     peer_read, rm->stat_write,
+						     rm->stat_msgcount,
+						     rm->events_dropped);
+				kfree(buf);
+				mutex_unlock(&rm->stat_lock);
+				return -EAGAIN;
+			}
+			if (elem_count > space)
+				nv_vgpu_warn(ctx->vgpu,
+					     "vgsp-rm: status ring full for reply fn=0x%x (free=%u need=%u peer=%u write=%u mc=%u); writing anyway after wait",
+					     function, space, elem_count,
+					     peer_read, rm->stat_write,
+					     rm->stat_msgcount);
+		}
+	}
+
 	/* Write all elements to consecutive queue slots. Page-aware: the
 	 * queue buffer is non-contiguous, so translate per page. */
 	slot = rm->stat_write % rm->stat_msgcount;
diff --git a/kernel/gsp_sema_service.c b/kernel/gsp_sema_service.c
index 587fb96..6c75a9e 100644
--- a/kernel/gsp_sema_service.c
+++ b/kernel/gsp_sema_service.c
@@ -8341,13 +8341,24 @@ static void svc_decode_stream(struct nv_vgpu_device *vgpu, const u8 *buf,
 				struct svc_memset_op *lm = &ops[n_ops - 1];
 
 				/*
-				 * fail4 A/B note: a CE-style auto-advance of
-				 * st.offset_out per op was tried here and
-				 * reverted - it did not change the memset
-				 * coverage and the driver's per-op re-sets are
-				 * already decoded, so the executed dsts match
-				 * the stream.
+				 * fail4 CE linear auto-advance, BOTH sides: the
+				 * driver re-sets only ONE offset per op and the
+				 * copy engine advances the other by the transfer
+				 * size. H2D ops show an explicit advancing DST
+				 * with a FIXED SRC base (every 1 MB chunk read
+				 * the source base again: vector_add's device a
+				 * held h_a[0..1MB) x4 -> 75% mismatch; the
+				 * readbacks are the mirror image). Advance both
+				 * registers; an explicit re-set simply overrides.
 				 */
+				{
+					u64 adv = (lm->type == 1) ? lm->bytes :
+						  (u64)lm->elem * lm->count;
+
+					st.offset_out += adv;
+					if (lm->type == 1)
+						st.offset_in += adv;
+				}
 				if (svc.opadd_log < 2048) {
 					svc.opadd_log++;
 					pr_info("sema-opadd: h=0x%x type=%u dst=0x%llx src=0x%llx bytes=%llu elem=%u count=%u ap=%u src_ap=%u phys=%u src_phys=%u\n",