vgsp: read the status-queue consumption cursor from the command queue (SWAP_RX)

Under MSGQ_FLAGS_SWAP_RX (both sides set it; flags=1 in both headers) the
guest driver's msgqRxMarkConsumed writes its STATUS readPtr to
pOurRxHdr->readPtr, and the driver's pOurRxHdr is the COMMAND-queue buffer's
RX header (msgq.c msgqRxLink swap branch: pReadOutgoing = &pOurRxHdr->readPtr
with pOurRxHdr from msgqTxCreate's pBackingStore = pCommandQueue). The
status queue's own +rxHdrOff is the mirror we publish - our command-queue
read position.

The status free-space check read the mirror, so it tracked our own writes:
with the guest mid-drain the ring was declared full (free=0, peer=0,
write=62/63), replies were written into it anyway, unread records were
overwritten, and the guest RM read a newer record - GspMsgQueueReceiveStatus:
Bad sequence number. Expected 447 got 510. Possible memory corruption. -
after which every RPC failed (status 0x25) and the first client's cudaMalloc
returned device busy or unavailable. That is the boot lottery: on a lost
boot the first client (mail) FAILs and the boot is poisoned for every later
client.

Evidence (r330 vs r332): r330 mail FAILed at the first cudaMalloc and only
one channel unregistered in the whole boot (the guest RM never tore its
channels down after the status desync); with the fix r332 mail + matmul pass
and every client's channels unregister (79 unregisters, one session flush
per client).

Also adds bounded diagnostics: sema-d2hmap (per-page destination resolution
of >=1MB host-dest copies) and a raised DIVERGE witness budget.

1 files changed, 27 insertions, 15 deletions

diff --git a/kernel/gsp_rm_emulation.c b/kernel/gsp_rm_emulation.c
index 764ef05..cd9cbf2 100644
--- a/kernel/gsp_rm_emulation.c
+++ b/kernel/gsp_rm_emulation.c
@@ -571,27 +571,39 @@ static int gsp_rm_send_flags(struct gsp_emu_context *ctx, u32 function,
 			u32 used, space, rxoff = 0;
 			u8 shdr[32];
 
-			if (gsp_rm_read_buf(ctx, rm->stat_off, shdr,
-					    sizeof(shdr)))
+			/*
+			 * r332 SWAP_RX cursor fix: the guest's consumption
+			 * cursor for the STATUS queue lives in the COMMAND
+			 * queue's RX header, not in the status queue's own
+			 * rxHdrOff. Under MSGQ_FLAGS_SWAP_RX (both sides set
+			 * it: flags=1 in both headers) the driver's
+			 * msgqRxMarkConsumed writes its status readPtr to
+			 * pOurRxHdr->readPtr, and the driver's pOurRxHdr is
+			 * the COMMAND-queue buffer's RX header (msgq.c
+			 * msgqRxLink swap branch: pReadOutgoing =
+			 * &pOurRxHdr->readPtr with pOurRxHdr from
+			 * msgqTxCreate's pBackingStore = pCommandQueue). The
+			 * status queue's own +rxHdrOff is the mirror we
+			 * publish: OUR command-queue read position
+			 * (written below after every consumed command
+			 * record). Reading that mirror here made the
+			 * free-space math track our own writes: the ring was
+			 * declared "full" with the guest mid-drain, replies
+			 * were written into it anyway, unread records were
+			 * overwritten, and the guest RM read a newer record
+			 * ("Bad sequence number. Expected N got N+63") - the
+			 * first-client cudaMalloc then fails with "device
+			 * busy or unavailable" (the boot lottery) and every
+			 * later RPC fails until a fresh boot.
+			 */
+			if (gsp_rm_read_buf(ctx, rm->cmd_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) ||
+			if (gsp_rm_read_buf(ctx, rm->cmd_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;