perf(net): skip empty cache deadline clock reads

quest/m1/release-perf/README.md

@@ -46,6 +46,14 @@ noise.
   with those wires; the earlier under-0.01% estimate omitted their walk.
   This remaining per-session and per-connection sampling is accepted. Deleting
   per-label bytes was rejected.
+- `track_gc/idle` is accepted at about 20 ns, flat across track counts, after
+  [#6291](https://github.com/moq-dev/moq/pull/6291) skipped the clock read
+  with no wall deadline and reused the bench's waiter (36.7 ns before). The
+  remaining ~8 ns over release's 11.6 ns is the lock and driver registration
+  that the wall-clock aging from
+  [#5548](https://github.com/moq-dev/moq/pull/5548) needs. An earliest-deadline
+  atomic cannot prove a replacement driver is registered, and a lock-free
+  notification primitive is not worth ~8 ns.
 
 ### Release as a baseline
 
@@ -64,7 +72,7 @@ usually pays for a feature.
 | `track_gap_flush/512` | 1.8 us | 48 us | 27x | same |
 | `track_gap_flush/64` | 1.5 us | 6.2 us | 4.1x | same |
 | `stats/session-report/1000x10` | 11.0 us | 86.8 us | 7.9x | accepted (above) |
-| `track_gc/idle/16384` | 11.6 ns | 73.7 ns | 6.3x | [Idle GC](/quest/m1/release-perf/gc-idle.md) |
+| `track_gc/idle/16384` | 11.6 ns | 73.7 ns | 6.3x | accepted (above) |
 | `origin/resume_overlap/sequence/1r_64g_32c` | 93 us | 426 us | 4.6x | fixed in [moq-dev/moq#6292](https://github.com/moq-dev/moq/pull/6292) |
 | `session_delivery_group_cost/moq-transport-22/.../viewers=64/.../fetch` | 1.19 ms | 3.86 ms | 3.2x | accepted (above) |
 | `group_write_frames/single/8192` | 424 us | 486 us | 1.17x | [Re-check](/quest/m1/release-perf/recheck.md) |
@@ -98,7 +106,6 @@ outside the worktree to keep the next one.
 
 - [FETCH fan-out](/quest/m1/release-perf/fetch-fanout.md) - a relay serves later viewers' lite FETCHes of a group from its cache, so per-viewer cost falls with viewers again
 - [Sparse gaps](/quest/m1/release-perf/sparse-gaps.md) - an arrival and a gap flush cost O(log gaps), not O(gaps)
-- [Idle GC](/quest/m1/release-perf/gc-idle.md) - an idle track GC skips the wall-clock pass when nothing is due
 - [Re-check](/quest/m1/release-perf/recheck.md) - the screen-only moves are re-measured with instruction counts, and the ones that hold are fixed or accepted
 - [JS frames](/quest/m1/release-perf/js-frames.md) - the js/net frame path is no slower than release in a browser, or the gap is accepted
 

quest/m2/performance-survey/cache-metadata.md

@@ -22,8 +22,8 @@ Correct the accounting explanation with the chosen implementation.
 
 Coordinate with [group storage](/quest/m2/group-storage-bench.md) and
 [log adoption](/quest/m2/perf/kio-channel-contention/log-adoption.md) if their
-representation has changed. [Idle GC](/quest/m1/release-perf/gc-idle.md) owns
-unnecessary clock/lock work, not retained queue backing.
+representation has changed. Idle GC clock and lock work was settled by
+[moq-dev/moq#6291](https://github.com/moq-dev/moq/pull/6291); this owns retained queue backing.
 
 Follow the [survey benchmark contract](/quest/m2/performance-survey/README.md):
 baseline first, paired evidence, equivalent work and nightly coverage. A

rs/moq-net/benches/track.rs

@@ -431,7 +431,8 @@ fn bench_wall(c: &mut Criterion) {
 			group.throughput(Throughput::Elements(1));
 			group.bench_function(BenchmarkId::new("idle", &id), |b| {
 				let aged = Aged::new(tracks, groups);
-				b.iter(|| black_box(aged.pool.gc(aged.now, &kio::Waiter::noop())));
+				let waiter = kio::Waiter::noop();
+				b.iter(|| black_box(aged.pool.gc(aged.now, &waiter)));
 			});
 			// Re-polling one parked read re-arms its deadline.
 			group.bench_function(BenchmarkId::new("repoll", &id), |b| {
@@ -470,17 +471,29 @@ fn bench_gc(c: &mut Criterion) {
 		group.throughput(Throughput::Elements(1));
 		group.bench_with_input(BenchmarkId::new("idle", tracks), &tracks, |b, &tracks| {
 			let swept = Swept::new(tracks);
-			b.iter(|| black_box(swept.pool.gc(swept.now, &kio::Waiter::noop())));
+			let waiter = kio::Waiter::noop();
+			b.iter(|| black_box(swept.pool.gc(swept.now, &waiter)));
+		});
+		// Also advance supplied time, so reusing an identical instant cannot hide clock work.
+		group.bench_with_input(BenchmarkId::new("advancing", tracks), &tracks, |b, &tracks| {
+			let swept = Swept::new(tracks);
+			let mut now = swept.now;
+			let waiter = kio::Waiter::noop();
+			b.iter(|| {
+				now += Duration::from_nanos(1);
+				black_box(swept.pool.gc(now, &waiter))
+			});
 		});
 		// A due pass, stepping one sweep interval per call.
 		group.throughput(Throughput::Elements(tracks as u64));
 		group.bench_with_input(BenchmarkId::new("sweep", tracks), &tracks, |b, &tracks| {
 			let swept = Swept::new(tracks);
 			let interval = cache::DEFAULT_EXPIRY / 2;
 			let mut now = swept.now;
+			let waiter = kio::Waiter::noop();
 			b.iter(|| {
 				now += interval;
-				black_box(swept.pool.gc(now, &kio::Waiter::noop()))
+				black_box(swept.pool.gc(now, &waiter))
 			});
 		});
 	}

rs/moq-net/src/model/cache.rs

@@ -391,7 +391,8 @@ impl Pool {
 			first = self.inner.wall_epoch.get().is_none();
 			let epoch = *self.inner.wall_epoch.get_or_init(|| now);
 			let micros = u64::try_from(now.duration_since(epoch).as_micros()).unwrap_or(u64::MAX);
-			self.inner.wall_micros.fetch_max(micros, Ordering::Relaxed);
+			// The clock lock already orders drivers and clamps their supplied time.
+			self.inner.wall_micros.store(micros, Ordering::Relaxed);
 			if self.sweep_interval().is_none() {
 				clock.sweep = None;
 			} else if sweep && clock.sweep.is_none_or(|at| at <= now) {
@@ -404,36 +405,45 @@ impl Pool {
 			let mut undriven = self.inner.wall.lock().undriven.take();
 			undriven.wake();
 		}
-		let wall = self.release_due(self.now_wall(), driver);
+		let wall = self.release_due(driver);
 		next.into_iter().chain(wall).min()
 	}
 
-	/// Wake every read parked on a wall-clock deadline at or before `now`, returning the
+	/// Wake every read whose wall-clock deadline has passed, returning the
 	/// earliest left and registering `driver` to hear of an earlier one.
-	fn release_due(&self, now: crate::time::Instant, driver: Option<&kio::Waiter>) -> Option<crate::time::Instant> {
+	fn release_due(&self, driver: Option<&kio::Waiter>) -> Option<crate::time::Instant> {
+		let mut wall = self.inner.wall.lock();
+		if wall.due.is_empty() {
+			if let Some(driver) = driver {
+				driver.register(&mut wall.drivers);
+			}
+			return None;
+		}
+		// With no parked read there is nothing to age. Otherwise the ambient clock
+		// still matters, even when a caller supplies an instant behind its deadline.
+		// Keep one sample across tracks that rearm while this pass releases them.
+		let now = self.now_wall();
 		loop {
-			let due: Vec<_> = {
-				let mut wall = self.inner.wall.lock();
-				let mut due = Vec::new();
-				while let Some(entry) = wall.due.first_entry()
-					&& entry.key().0 <= now
-				{
-					due.push(entry.remove());
-				}
-				if due.is_empty() {
-					if let Some(driver) = driver {
-						driver.register(&mut wall.drivers);
-					}
-					return wall.due.first_key_value().map(|((at, _), _)| *at);
+			let mut due = Vec::new();
+			while let Some(entry) = wall.due.first_entry()
+				&& entry.key().0 <= now
+			{
+				due.push(entry.remove());
+			}
+			if due.is_empty() {
+				if let Some(driver) = driver {
+					driver.register(&mut wall.drivers);
 				}
-				due
-			};
+				return wall.due.first_key_value().map(|((at, _), _)| *at);
+			}
+			drop(wall);
 			// Outside the lock: a track re-arms its next deadline here.
 			for track in due.iter().filter_map(Weak::upgrade) {
 				if let Some(next) = track.wakes.release_wall(now) {
 					self.schedule(next, None, &track);
 				}
 			}
+			wall = self.inner.wall.lock();
 		}
 	}
 
@@ -1339,6 +1349,52 @@ mod test {
 		assert!(read.0.load(Ordering::SeqCst), "the read was never released");
 	}
 
+	/// A fresh driver must hear deadlines armed after an empty collection, including
+	/// after the previous driver was dropped or a prior deadline drained its wake list.
+	#[test]
+	fn replacement_drivers_hear_new_deadlines() {
+		let pool = Pool::unbounded();
+		let now = super::super::clock::wall();
+		pool.gc(now, &kio::Waiter::noop());
+		let track = Track::new(pool.clone(), Default::default(), kio::Weak::new());
+		let (read, reading) = flag();
+		let (driver, driving) = flag();
+		assert_eq!(pool.gc(now, &driving), None);
+		let at = now + Duration::from_secs(1);
+		track.watch_wall(at, &reading);
+		assert!(driver.0.swap(false, Ordering::SeqCst));
+		assert_eq!(pool.gc(at, &driving), None);
+		assert!(read.0.swap(false, Ordering::SeqCst));
+		track.watch_wall(at + Duration::from_secs(1), &reading);
+		assert!(driver.0.load(Ordering::SeqCst), "the drained list was never rearmed");
+	}
+
+	/// Wall time can lead a lagging driver's sample, while another driver can supply
+	/// future time. Neither a new driver nor a backward sample may rewind expiration.
+	#[moq_clock::test]
+	async fn wall_deadlines_use_the_later_of_ambient_and_driven_time() {
+		let pool = Pool::unbounded();
+		let now = moq_clock::now();
+		let waiter = kio::Waiter::noop();
+		pool.gc(now, &waiter);
+		let track = Track::new(pool.clone(), Default::default(), kio::Weak::new());
+		let (read, reading) = flag();
+		let at = now + Duration::from_secs(1);
+		track.watch_wall(at, &reading);
+		moq_clock::sleep(Duration::from_secs(1)).await;
+		assert_eq!(pool.gc(now, &waiter), None);
+		assert!(read.0.swap(false, Ordering::SeqCst), "ambient time was ignored");
+
+		let future = at + Duration::from_secs(1);
+		track.watch_wall(future, &reading);
+		assert_eq!(pool.gc(future, &waiter), None);
+		assert!(read.0.swap(false, Ordering::SeqCst), "driven time was ignored");
+		let tick = pool.now();
+		pool.gc(now, &waiter);
+		assert_eq!(pool.now_wall(), future);
+		assert_eq!(pool.now(), tick);
+	}
+
 	/// A track indexes one wall-clock deadline, its earliest, and takes it along when dropped.
 	#[test]
 	fn a_dropped_track_leaves_the_wall_index() {