sema-svc: consume rewind replays and sweep-delivered CE entries at once

Two stalls serialized the runtime client's init into one 8 s CE grace window
per push:

- a pin-batch rewind replays recently consumed GPFIFO slots; re-examining a
  consumed slot runs the rescue and then holds it for the full grace window
  again (observed: the same VA 'skipped after 8000 ms grace' twice, 38 s
  apart). The pump now records each consumed slot with the put value it was
  consumed at (32-entry per-channel replay ring, persisted across passes)
  and consumes a replayed slot whose producing put has not advanced
  immediately.
- the content sweep applies releases from RAM pages before the pump reads
  their entries, so the pool advances past the entry's payload while it is
  pending. entry_want (recorded on first examination) now detects that on
  both the unreadable-rescue and the readable-no-release paths and consumes
  the entry instead of waiting the grace window.

Evidence: real-release watcher cycle, pool semaphore advances no longer
gated at 8 s intervals once the replay skip engages.

1 files changed, 55 insertions, 1 deletions

diff --git a/kernel/gsp_sema_service.c b/kernel/gsp_sema_service.c
index 58778c2..7e15f44 100644
--- a/kernel/gsp_sema_service.c
+++ b/kernel/gsp_sema_service.c
@@ -157,6 +157,13 @@ struct svc_channel {
 				 * immediately instead of waiting out the grace
 				 * window (observed: one 8 s grace per push,
 				 * serializing the runtime init) */
+	u32	replay[32];	/* packed {put:16,slot:16} of recently consumed
+				 * entries. A pin-batch rewind replays entries;
+				 * re-processing one costs a full grace window
+				 * (observed: the same VA skipped twice, 38 s
+				 * apart), so a replayed slot whose producing
+				 * put has not advanced is consumed at once. */
+	u32	replay_next;
 	u64	last_rel_va;	/* last tracking-sem VA seen in a readable push */
 	u32	logged_expected;	/* last pool value logged (rate limit) */
 	unsigned long rescue_last;	/* rate limit for the full-RAM rescue scan */
@@ -4975,6 +4982,30 @@ static bool svc_fake_release(struct nv_vgpu_device *vgpu, struct svc_channel *ch
 	return true;
 }
 
+#define SVC_REPLAY_SLOTS 32
+
+/* Remember a consumed GPFIFO slot and the put it was consumed at. */
+static void svc_replay_note(struct svc_channel *ch, u32 idx, u32 put)
+{
+	ch->replay[ch->replay_next % SVC_REPLAY_SLOTS] =
+		((put & 0xffff) << 16) | (idx & 0xffff);
+	ch->replay_next++;
+}
+
+/* Was this slot consumed at this put? Then the entry is a rewind replay. */
+static bool svc_replay_seen(const struct svc_channel *ch, u32 idx, u32 put)
+{
+	u32 rec = ((put & 0xffff) << 16) | (idx & 0xffff);
+	u32 n = ch->replay_next < SVC_REPLAY_SLOTS ?
+		ch->replay_next : SVC_REPLAY_SLOTS;
+	u32 i;
+
+	for (i = 0; i < n; i++)
+		if (ch->replay[i] == rec)
+			return true;
+	return false;
+}
+
 static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 {
 	/* Only ever called from the service thread. */
@@ -5081,8 +5112,23 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 			 * consumed instead of waiting out the grace window -
 			 * otherwise every push costs one full grace period.
 			 */
-			if (!ch->defer_since)
+			if (!ch->defer_since) {
 				ch->entry_want = ch->expected_sem_payload;
+				/*
+				 * A pin-batch rewind replays recently consumed
+				 * entries. Re-processing one (rescue, grace)
+				 * costs seconds and delivers nothing: consume
+				 * it at once when the producing put has not
+				 * advanced since it was consumed.
+				 */
+				if (ch->engine_type &&
+				    svc_replay_seen(ch, idx, put)) {
+					ch->last_get = (idx + 1) %
+						ch->gpfifo_entries;
+					svc.gpfifo_skipped++;
+					continue;
+				}
+			}
 
 			if (rrc) {
 				stalled = true;	/* ring unreadable: sweep RAM */
@@ -5242,6 +5288,7 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 									    ch->entry_want) {
 										ch->defer_since = 0;
 										ch->rescue_fail = 0;
+										svc_replay_note(ch, idx, put);
 										ch->last_get = (idx + 1) %
 											ch->gpfifo_entries;
 										svc.gpfifo_skipped++;
@@ -5317,6 +5364,7 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 if (gsp_sema_fake_rel &&
 					    svc_fake_release(vgpu, ch)) {
 							ch->defer_since = 0;
+							svc_replay_note(ch, idx, put);
 							ch->last_get = (idx + 1) %
 								       ch->gpfifo_entries;
 							svc.gpfifo_skipped++;
@@ -5430,6 +5478,7 @@ if (gsp_sema_fake_rel &&
 								 */
 								ch->rescue_fail = 0;
 								ch->defer_since = 0;
+								svc_replay_note(ch, idx, put);
 								ch->last_get =
 									(idx + 1) %
 									ch->gpfifo_entries;
@@ -5546,6 +5595,7 @@ if (gsp_sema_fake_rel &&
 							}
 							ch->rescue_fail = 0;
 							ch->defer_since = 0;
+							svc_replay_note(ch, idx, put);
 							ch->last_get = (idx + 1) %
 								       ch->gpfifo_entries;
 							svc.gpfifo_skipped++;
@@ -5581,6 +5631,7 @@ if (gsp_sema_fake_rel &&
 				}
 			}
 
+			svc_replay_note(ch, idx, put);
 			ch->last_get = (idx + 1) % ch->gpfifo_entries;
 			ch->rescue_fail = 0;
 			ch->defer_since = 0;
@@ -5625,6 +5676,9 @@ if (gsp_sema_fake_rel &&
 				svc.chan[k].rescue_last = ch->rescue_last;
 				svc.chan[k].rescue_fail = ch->rescue_fail;
 				svc.chan[k].defer_since = ch->defer_since;
+				svc.chan[k].replay_next = ch->replay_next;
+				memcpy(svc.chan[k].replay, ch->replay,
+				       sizeof(ch->replay));
 				/*
 				 * Persist the counter slot autoreg resolved on the
 				 * snapshot: without this the next pass re-derives