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).

1 files changed, 88 insertions, 0 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;