vgsp: retry deferred display completions; escalate the UVM init-slot release

Two live walls from the r1649/r1650 windowed-GZDoom runs:

1. Display completion wake-up. All NV01_EVENT registrations carry
   hSrcResource=0, so gsp_rm_post_channel_event() always takes the shotgun
   fallback, which skips events posted within the 250 ms hint dedup. When the
   display UPDATE is the channel's last push, the pump's retry was gated on
   new work (put != consumed) and never ran, so the X server's flip wait
   spun in _nv018351rm/os_schedule forever. Add
   gsp_rm_post_channel_event_force() (bypasses only the hint dedup; the
   event burst cap and status-ring free-space discipline still apply), a
   per-channel event_retry_at, and an idle-pass retry every 250 ms that
   logs "re-posted after deferral". Also move event_burst_at after the cap
   check so refused retries cannot keep the window fresh.

2. UVM channel init slot. r1651 gdb proved the channel's tracker entry waits
   for value 2 while the authored pool word reads 1; a manual write of 2
   cleared uvm_push_end_and_wait. The channel's control-entry + init pushes
   are boot-time configuration, so keep the rank slot moving for 10 s after
   first submission: one attempt per 250 ms per rank, escalating 1->2->3->4
   through every published VAS translation. Values beyond that need the
   real CE push decode.

Evidence: backups/evidence-20261002-victory/r1649/ (X spin perf, event
journal), r1650/ (re-post lines), r1651/ (tracker entry value=2, manual
write clears the spin).

4 files changed, 135 insertions, 31 deletions

diff --git a/kernel/gsp_display_engine.c b/kernel/gsp_display_engine.c
index f96fc6a..c7c9dca 100644
--- a/kernel/gsp_display_engine.c
+++ b/kernel/gsp_display_engine.c
@@ -101,6 +101,11 @@
 #define GSP_DISP_NOTIFIER_CTRL_NOTIFY	BIT(12)
 /* The CONTROL offset field is in 16-byte units; expose it in bytes. */
 #define GSP_DISP_NOTIFIER_CTRL_OFFSET(c) ((((c) >> 4) & 0xff) * 16)
+/* Coalesce a refused completion wake-up, then re-post it.  Must be at least
+ * the RM event hint dedup window (HZ / 4 = 250 ms) so the retry is not
+ * coalesced into the same post that starved it; at most four attempts/s per
+ * waiting channel. */
+#define GSP_DISP_EVENT_RETRY_MS		250
 
 /* Display instance memory (v03_00 layout). */
 #define GSP_DISP_INST_HASH_OFF		0x0000
@@ -136,6 +141,7 @@ struct gsp_disp_chan {
 	u32 notifier_writes;
 	u32 notifier_posts;
 	bool event_pending;	/* completion event send failed; retry in pump */
+	unsigned long event_retry_at;	/* next forced re-post attempt */
 	u32 event_drop_logs;
 	u32 update_logs;
 
@@ -1115,6 +1121,8 @@ static void gsp_disp_method(struct nv_vgpu_device *vgpu,
 				 * pump instead of giving up.
 				 */
 				ch->event_pending = true;
+				ch->event_retry_at = jiffies +
+					msecs_to_jiffies(GSP_DISP_EVENT_RETRY_MS);
 				if (ch->event_drop_logs < 8) {
 					ch->event_drop_logs++;
 					pr_info("h=0x%x UPDATE completion event deferred rc=%d (hObj=0x%x)\n",
@@ -1353,39 +1361,51 @@ void gsp_disp_engine_pump(struct nv_vgpu_device *vgpu)
 				s->hclass, s->instance, put, s->consumed,
 				s->ctl_off);
 		}
-		if (put == s->consumed)
-			continue;
-
-		gsp_disp_run(vgpu, s, put);
+		if (put != s->consumed) {
+			gsp_disp_run(vgpu, s, put);
 
-		/*
-		 * Publish our read pointer; the driver's nvEvoMakeRoom and
-		 * EvoCoreReadGet rely on it to reuse the ring.
-		 */
-		gsp_disp_bar0_write(vgpu, s->ctl_off + GSP_DISP_REG_GET,
-				    s->consumed);
-		get = gsp_disp_bar0_read(vgpu,
-					 s->ctl_off + GSP_DISP_REG_GET);
-
-		if (disp.log_budget < 4096) {
-			u64 age = (u64)(now - disp.boot_jiffies) *
-				  (NSEC_PER_SEC / HZ);
-
-			disp.log_budget++;
-			pr_info("pump h=0x%x inst=%u put=0x%x get=0x%x consumed=0x%x at %llu ms\n",
-				s->hclass, s->instance, put, get, s->consumed,
-				div_u64(age, 1000000));
+			/*
+			 * Publish our read pointer; the driver's nvEvoMakeRoom
+			 * and EvoCoreReadGet rely on it to reuse the ring.
+			 */
+			gsp_disp_bar0_write(vgpu, s->ctl_off + GSP_DISP_REG_GET,
+					    s->consumed);
+			get = gsp_disp_bar0_read(vgpu,
+						 s->ctl_off + GSP_DISP_REG_GET);
+
+			if (disp.log_budget < 4096) {
+				u64 age = (u64)(now - disp.boot_jiffies) *
+					  (NSEC_PER_SEC / HZ);
+
+				disp.log_budget++;
+				pr_info("pump h=0x%x inst=%u put=0x%x get=0x%x consumed=0x%x at %llu ms\n",
+					s->hclass, s->instance, put, get,
+					s->consumed, div_u64(age, 1000000));
+			}
 		}
 
 		/*
-		 * Retry a completion event whose send was refused (burst
-		 * cap / transient queue state).  Runs outside disp.lock:
-		 * the post takes the RM status-queue lock.
+		 * Retry a completion event whose send was refused (burst cap /
+		 * transient queue state, or the hint dedup coalescing it into a
+		 * post the waiter never saw).  Runs outside disp.lock (the post
+		 * takes the RM status-queue lock) and on idle passes too: the
+		 * fatal case is an UPDATE that is the channel's last push, so a
+		 * retry gated on new work never runs (r1649).  The forced post
+		 * ignores the per-event hint dedup; the retry window bounds it
+		 * to one attempt per 250 ms per waiting channel while the burst
+		 * cap still guards the status ring.
 		 */
 		if (s->event_pending && s->h_object && gsp_disp_events &&
-		    !gsp_rm_post_channel_event(vgpu, s->h_object)) {
-			s->event_pending = false;
-			s->notifier_posts++;
+		    time_after_eq(jiffies, s->event_retry_at)) {
+			if (!gsp_rm_post_channel_event_force(vgpu, s->h_object)) {
+				s->event_pending = false;
+				s->notifier_posts++;
+				pr_info("h=0x%x UPDATE completion event re-posted after deferral (posts=%u)\n",
+					s->hclass, s->notifier_posts);
+			} else {
+				s->event_retry_at = jiffies +
+					msecs_to_jiffies(GSP_DISP_EVENT_RETRY_MS);
+			}
 		}
 
 		/* Write the bookkeeping back if the slot still names this
@@ -1412,6 +1432,7 @@ void gsp_disp_engine_pump(struct nv_vgpu_device *vgpu)
 				c->notifier_writes = s->notifier_writes;
 				c->notifier_posts = s->notifier_posts;
 				c->event_pending = s->event_pending;
+				c->event_retry_at = s->event_retry_at;
 				c->event_drop_logs = s->event_drop_logs;
 				c->update_logs = s->update_logs;
 				c->logged_update = s->logged_update;
diff --git a/kernel/gsp_rm_emulation.c b/kernel/gsp_rm_emulation.c
index c1b98f3..37b2200 100644
--- a/kernel/gsp_rm_emulation.c
+++ b/kernel/gsp_rm_emulation.c
@@ -512,11 +512,14 @@ static int gsp_rm_send_flags(struct gsp_emu_context *ctx, u32 function,
 				   : GSPRM_EVENT_BURST_MAX;
 
 		/* m171: expire the budget by wall time so a long-ago
-		 * flood can never starve a fresh completion post. */
+		 * flood can never starve a fresh completion post.  Only an
+		 * admitted record moves the window: a refused attempt that
+		 * retried (the display completion retry runs every 250 ms)
+		 * must not keep the window perpetually fresh, or the budget
+		 * would never expire and the retry would livelock. */
 		if (time_after(jiffies, rm->event_burst_at +
 			       GSPRM_EVENT_BURST_WINDOW))
 			rm->event_burst = 0;
-		rm->event_burst_at = jiffies;
 		if (rm->event_burst >= cap) {
 			rm->events_dropped++;
 			if (rm->events_dropped <= 8 ||
@@ -529,6 +532,7 @@ static int gsp_rm_send_flags(struct gsp_emu_context *ctx, u32 function,
 			kfree(buf);
 			return -EAGAIN;
 		}
+		rm->event_burst_at = jiffies;
 		rm->event_burst++;
 	} else {
 		rm->event_burst = 0;
@@ -850,7 +854,8 @@ static u32 gsp_rm_ev_function(void)
  * Safe to call from the progress service's thread: the status queue write is
  * serialized by rm->stat_lock and the guest is woken with MSI-X vector 0.
  */
-int gsp_rm_post_channel_event(struct nv_vgpu_device *vgpu, u32 h_src)
+static int gsp_rm_post_channel_event_common(struct nv_vgpu_device *vgpu,
+					    u32 h_src, bool force)
 {
 	struct gsp_emu_context *ctx = vgpu ? vgpu->gsp_comm_ctx : NULL;
 	struct gsp_rm_emu *rm;
@@ -908,7 +913,8 @@ int gsp_rm_post_channel_event(struct nv_vgpu_device *vgpu, u32 h_src)
 			 * client's dispatcher re-reads channel state on any
 			 * event - and the 0.5 s hint re-posts late waiters.
 			 */
-			if (time_before(jiffies, rm->events[i].last_post +
+			if (!force &&
+			    time_before(jiffies, rm->events[i].last_post +
 					GSPRM_EVENT_HINT_DEDUP))
 				continue;
 			gsp_rm_fill_ev_payload(rm, i, payload);
@@ -940,6 +946,28 @@ int gsp_rm_post_channel_event(struct nv_vgpu_device *vgpu, u32 h_src)
 	return posted ? 0 : -ENOENT;
 }
 
+int gsp_rm_post_channel_event(struct nv_vgpu_device *vgpu, u32 h_src)
+{
+	return gsp_rm_post_channel_event_common(vgpu, h_src, false);
+}
+
+/*
+ * Force a completion wake-up past the hint dedup.  The 250 ms dedup assumes a
+ * recent post already woke the client, but a waiter that arrived after that
+ * post misses it, and a busy event stream keeps refreshing every event's
+ * stamp so the coalesced completion is never delivered (r1649: the X server's
+ * display UPDATE completion was refused - the coincident hint returned
+ * -EAGAIN because every event was inside the window - and the flip wait spun
+ * in the nvidia ioctl forever).  Callers use this only for a refused
+ * completion edge, rate-limited by their own retry window; the event burst
+ * cap and the status-ring free-space discipline still apply inside
+ * gsp_rm_send_flags().
+ */
+int gsp_rm_post_channel_event_force(struct nv_vgpu_device *vgpu, u32 h_src)
+{
+	return gsp_rm_post_channel_event_common(vgpu, h_src, true);
+}
+
 /*
  * r623 experiment: deliver a synthetic OS_ERROR_LOG (0x1006) record to the
  * guest's RM queue - the error report a real GSP sends into the client's
diff --git a/kernel/gsp_rm_emulation.h b/kernel/gsp_rm_emulation.h
index c0eecf8..db5abdd 100644
--- a/kernel/gsp_rm_emulation.h
+++ b/kernel/gsp_rm_emulation.h
@@ -233,6 +233,7 @@ extern unsigned gsp_rm_ev_func;
 int  gsp_rm_emu_init(struct nv_vgpu_device *vgpu, struct gsp_emu_context *ctx);
 void gsp_rm_emu_fini(struct gsp_emu_context *ctx);
 int  gsp_rm_post_channel_event(struct nv_vgpu_device *vgpu, u32 h_src);
+int  gsp_rm_post_channel_event_force(struct nv_vgpu_device *vgpu, u32 h_src);
 int  gsp_rm_force_post(struct nv_vgpu_device *vgpu, u32 idx);
 /*
  * r623: deliver a synthetic OS_ERROR_LOG (0x1006) record to the guest's RM
diff --git a/kernel/gsp_sema_service.c b/kernel/gsp_sema_service.c
index 11c4540..c9abc71 100644
--- a/kernel/gsp_sema_service.c
+++ b/kernel/gsp_sema_service.c
@@ -969,6 +969,9 @@ static unsigned long svc_ring_done;	/* CE ranks whose ring is pinned (session st
  * channel registration and cleared by svc_session_flush.
  */
 static unsigned long svc_ce_initrel_done;
+static unsigned long svc_ce_initrel_until[32];	/* re-fan-out window per rank */
+static unsigned long svc_ce_initrel_next[32];	/* next retry per rank */
+static u32 svc_ce_initrel_val[32];		/* last value authored per rank */
 static u32 svc_ce_initrel_log;
 
 /* Per-guest session records (one per open mdev; see SVC_MAX_CHANNELS for the channel cap). */
@@ -2019,6 +2022,9 @@ static void svc_chan_rearm_rank(struct svc_channel *ch)
 		clear_bit(ch->ce_index, &svc_ring_done);
 		clear_bit(ch->ce_index, &svc_pb_done);
 		clear_bit(ch->ce_index, &svc_ce_initrel_done);
+		svc_ce_initrel_until[ch->ce_index] = 0;
+		svc_ce_initrel_next[ch->ce_index] = 0;
+		svc_ce_initrel_val[ch->ce_index] = 0;
 	} else {
 		clear_bit(ch->gr_index, &svc_gr_ring_done);
 		/*
@@ -22644,6 +22650,9 @@ static void svc_session_flush(void)
 	memset(svc_ce_hold_since, 0, sizeof(svc_ce_hold_since));
 	memset(svc_ce_firstsub_off, 0, sizeof(svc_ce_firstsub_off));
 	svc_ce_initrel_done = 0;
+	memset(svc_ce_initrel_until, 0, sizeof(svc_ce_initrel_until));
+	memset(svc_ce_initrel_next, 0, sizeof(svc_ce_initrel_next));
+	memset(svc_ce_initrel_val, 0, sizeof(svc_ce_initrel_val));
 	svc_ce_initrel_log = 0;
 	svc_fb_claim_reset();
 	svc_userd_sweep_log = 0;
@@ -25546,6 +25555,16 @@ static void svc_ce_init_rel_author(struct nv_vgpu_device *vgpu,
 		return;
 	if (test_and_set_bit(ch->ce_index, &svc_ce_initrel_done))
 		return;
+	/*
+	 * r1650: keep re-fanning the rank slot for a short window. The UVM
+	 * pool page can move (and a per-client VAS directory can be
+	 * discovered) seconds after the first submission, so a single
+	 * one-shot write can race the translation; the counter is
+	 * forward-only, so the pump's periodic re-write is harmless.
+	 */
+	svc_ce_initrel_until[ch->ce_index] = jiffies + 10 * HZ;
+	svc_ce_initrel_next[ch->ce_index] =
+		jiffies + msecs_to_jiffies(250);
 
 	/*
 	 * UVM's per-channel tracking semaphore is pool slot rank (allocation
@@ -25555,6 +25574,7 @@ static void svc_ce_init_rel_author(struct nv_vgpu_device *vgpu,
 	tva = SVC_UVM_POOL_MAP_BASE + 4ULL * (u64)ch->ce_index;
 	svc.cur_ch = ch->h_object;
 	wrc = svc_sem_write(vgpu, ch->h_object, tva, 1);
+	svc_ce_initrel_val[ch->ce_index] = 1;
 	pr_info("sema-ceinit: h=0x%x rank=%u tracker=0x%llx val=1 %s (%s; init push release)\n",
 		ch->h_object, ch->ce_index, tva,
 		wrc == 0 ? "ok" : (wrc == -ENOENT ? "queued" : "failed"), why);
@@ -40431,6 +40451,40 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 		}
 
 		/*
+		 * r1650: keep the UVM rank-slot release moving for a short
+		 * window. Two effects need it: the pool page can move / a
+		 * per-client VAS directory can be discovered seconds later
+		 * (r1649: the one-shot write landed on a stale page before
+		 * the live directory existed), and the channel's init
+		 * sequence queues more than one push (r1651 gdb: the rank-0
+		 * tracker entry waited for value 2 while the authored word
+		 * read 1; a manual write of 2 cleared uvm_push_end_and_wait).
+		 * Forward-only and idempotent, one attempt per 250 ms per
+		 * rank for 10 s, escalating 1 -> 2 -> 3 -> 4: the channel's
+		 * control-entry + init pushes are boot-time configuration,
+		 * so completing them host-side is the same shortcut the
+		 * first release already uses. Values beyond that need the
+		 * real CE push decode and are not authored here.
+		 */
+		if (ch->engine_type && ch->ce_index < 32 &&
+		    test_bit(ch->ce_index, &svc_ce_initrel_done) &&
+		    time_before(jiffies, svc_ce_initrel_until[ch->ce_index]) &&
+		    time_after_eq(jiffies,
+				  svc_ce_initrel_next[ch->ce_index])) {
+			u64 tva2 = SVC_UVM_POOL_MAP_BASE +
+				   4ULL * (u64)ch->ce_index;
+
+			svc_ce_initrel_next[ch->ce_index] =
+				jiffies + msecs_to_jiffies(250);
+			if (svc_ce_initrel_val[ch->ce_index] < 4) {
+				svc_ce_initrel_val[ch->ce_index]++;
+				svc_pool_write_all_vas(vgpu, ch->h_object,
+						       tva2,
+						       svc_ce_initrel_val[ch->ce_index]);
+			}
+		}
+
+		/*
 		 * P6c C3: a latched fault notifier is released when the guest
 		 * submits more work (GPFIFO put advanced past the value at
 		 * delivery), alongside the guest-ack read in