Skip to main content

rivet/
gpu_pool.rs

1//! Process-wide GPU reservation pool.
2//!
3//! Each detected GPU is a slot. Callers `claim()` an available slot
4//! and hold the returned `GpuLease` for the duration of their work;
5//! `Drop` releases the slot back to the pool. The lease's
6//! `gpu_index` field is the device index the work should run on.
7//!
8//! Concurrency model: one variant per GPU at any time. With N GPUs
9//! and M waiters, the first N waiters get leases immediately and the
10//! remaining M−N park on the semaphore until a lease drops. This is
11//! the deliberate design decision from 2026-05-02 — concurrent
12//! NVENC sessions on the same CUDA context deadlocked at session
13//! ~5/5 init, GPU went idle, no frames encoded. One-encoder-per-GPU
14//! is the load-bearing invariant; the pool's role is to enforce it
15//! while still letting variants run in parallel ACROSS GPUs.
16//!
17//! CPU-only hosts (no GPUs detected): `claim()` returns `None`
18//! immediately — callers fall back to CPU encode without queuing.
19
20use std::sync::Arc;
21use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
22
23use codec::gpu::{GpuDevice, GpuVendor};
24use tokio::sync::{OwnedSemaphorePermit, Semaphore};
25
26pub struct GpuPool {
27    /// Per-slot GPU device index (`GpuDevice.index`, not vec position
28    /// — accommodates sparse `CUDA_VISIBLE_DEVICES` setups).
29    gpu_indices: Vec<u32>,
30    /// Per-slot vendor — load-bearing for the encoder factory's
31    /// vendor-aware dispatch. Without it, multi-vendor hosts (NVIDIA
32    /// + Intel Arc) ALWAYS picked NVENC because the factory tries
33    /// NVIDIA first and both vendors expose index 0; the Arc sat
34    /// idle even when the NVIDIA card was busy.
35    gpu_vendors: Vec<GpuVendor>,
36    /// Per-slot human-readable device name. Used by `snapshot_leases`
37    /// (Phase 2 worker_load reporting) so the backend's admin view
38    /// can label each GPU lease badge with the same string the hello
39    /// frame already advertised. Stays in lockstep with the hello
40    /// frame's `WsGpuInfo.name`.
41    gpu_names: Vec<String>,
42    /// Per-slot free flag. `true` = available; `false` = leased.
43    /// Atomic so the CAS-find-free-slot path under `claim()` is
44    /// lock-free; correctness is enforced by the semaphore counting
45    /// (see `claim`).
46    free: Arc<Vec<AtomicBool>>,
47    /// Semaphore with N permits (= number of GPUs). Acquiring a
48    /// permit guarantees at least one `free` slot exists, so the
49    /// CAS loop in `claim()` always succeeds without retry.
50    permits: Arc<Semaphore>,
51    /// Count of variant tasks currently blocked inside `claim()`'s
52    /// `acquire_owned().await`. Used by the LeaseArbiter (planned
53    /// 2026-05-10) to decide whether to dispatch a helper task: if
54    /// any variant is already waiting for a permit, that variant
55    /// must claim before the arbiter steals a permit for a helper.
56    /// Incremented immediately before `acquire_owned().await`;
57    /// decremented as soon as the await returns (success or
58    /// cancellation) via the `PendingClaimGuard` RAII helper.
59    ///
60    /// `try_claim()` does NOT touch this — helpers are not blocked
61    /// claimers in the spare-capacity sense.
62    pending_claimers: Arc<AtomicUsize>,
63}
64
65/// RAII guard that increments `pending_claimers` on construction
66/// and decrements on drop. Used inside `claim()` to bracket the
67/// `acquire_owned().await` so the counter stays accurate even when
68/// the awaiting task is cancelled mid-await (the future is dropped,
69/// guard drops, counter decrements).
70struct PendingClaimGuard {
71    counter: Arc<AtomicUsize>,
72}
73
74impl PendingClaimGuard {
75    fn new(counter: Arc<AtomicUsize>) -> Self {
76        counter.fetch_add(1, Ordering::AcqRel);
77        Self { counter }
78    }
79}
80
81impl Drop for PendingClaimGuard {
82    fn drop(&mut self) {
83        self.counter.fetch_sub(1, Ordering::AcqRel);
84    }
85}
86
87/// Snapshot of one GPU slot's lease state at a moment in time.
88/// Returned by [`GpuPool::snapshot_leases`] for Phase 2 worker_load
89/// reporting. Field shape matches `queue::WsGpuLeaseEntry` so the
90/// caller can map across without a wire-format-aware translation.
91#[derive(Debug, Clone)]
92pub struct GpuLeaseEntry {
93    pub vendor: GpuVendor,
94    pub name: String,
95    pub index: u32,
96    pub leased: bool,
97}
98
99/// RAII guard returned by `GpuPool::claim`. The slot is released
100/// (and the underlying semaphore permit dropped) when this value
101/// is dropped — typically at the end of the variant's encode task.
102pub struct GpuLease {
103    pub gpu_index: u32,
104    pub vendor: GpuVendor,
105    slot_idx: usize,
106    free: Arc<Vec<AtomicBool>>,
107    _permit: OwnedSemaphorePermit,
108}
109
110impl Drop for GpuLease {
111    fn drop(&mut self) {
112        self.free[self.slot_idx].store(true, Ordering::Release);
113    }
114}
115
116impl GpuPool {
117    /// Build a pool from the host's detected GPU inventory. An empty
118    /// inventory is permitted — the resulting pool always returns
119    /// `None` from `claim()` so CPU-only hosts work without
120    /// special-casing at the call site.
121    pub fn new(devices: &[GpuDevice]) -> Self {
122        let n = devices.len();
123        Self {
124            gpu_indices: devices.iter().map(|d| d.index).collect(),
125            gpu_vendors: devices.iter().map(|d| d.vendor).collect(),
126            gpu_names: devices.iter().map(|d| d.name.clone()).collect(),
127            free: Arc::new((0..n).map(|_| AtomicBool::new(true)).collect()),
128            // Semaphore::new(0) is valid but `acquire` would deadlock.
129            // We never acquire on the empty path because `claim()`
130            // returns `None` early on CPU-only hosts.
131            permits: Arc::new(Semaphore::new(n)),
132            pending_claimers: Arc::new(AtomicUsize::new(0)),
133        }
134    }
135
136    /// How many variant tasks are currently parked inside `claim()`
137    /// waiting for a permit. The LeaseArbiter consults this to decide
138    /// whether to dispatch a helper: when `pending_claimers() > 0`,
139    /// at least one variant task wants a GPU and the arbiter must
140    /// step back so the variant claims first (FIFO fairness).
141    ///
142    /// Reads with `Ordering::Acquire`. The result is momentary — by
143    /// the time the caller observes it, a claim may have resolved or
144    /// a new one parked. That's expected; the arbiter re-checks
145    /// before each dispatch decision.
146    pub fn pending_claimers(&self) -> usize {
147        self.pending_claimers.load(Ordering::Acquire)
148    }
149
150    /// How many GPUs this pool manages. Useful for pre-spawning
151    /// variants when fewer variants exist than GPUs (no point
152    /// over-claiming).
153    pub fn capacity(&self) -> usize {
154        self.gpu_indices.len()
155    }
156
157    /// Snapshot per-GPU lease state. Result preserves slot order
158    /// (matches the order [`GpuPool::new`] saw devices), so callers
159    /// stitching the result against the hello frame's `gpu_pool`
160    /// see consistent indices across both reports.
161    ///
162    /// Reads `free` slots with `Ordering::Acquire`. The result is a
163    /// momentary snapshot — by the time the caller observes it, a
164    /// claim or drop may have flipped any slot. That's expected;
165    /// load reporting is best-effort observability, not a
166    /// transactional view.
167    ///
168    /// Used by the worker's Phase 2 (2026-05-07) load-tick task to
169    /// build the `worker_load` frame's `gpu_pool` field.
170    pub fn snapshot_leases(&self) -> Vec<GpuLeaseEntry> {
171        self.gpu_indices
172            .iter()
173            .zip(self.gpu_vendors.iter())
174            .zip(self.gpu_names.iter())
175            .enumerate()
176            .map(|(slot_idx, ((index, vendor), name))| GpuLeaseEntry {
177                vendor: *vendor,
178                name: name.clone(),
179                index: *index,
180                leased: !self.free[slot_idx].load(Ordering::Acquire),
181            })
182            .collect()
183    }
184
185    /// Claim an available GPU. Awaits if every GPU is currently
186    /// leased. Returns `None` immediately on CPU-only hosts — the
187    /// caller should fall back to CPU encode.
188    pub async fn claim(self: &Arc<Self>) -> Option<GpuLease> {
189        if self.gpu_indices.is_empty() {
190            return None;
191        }
192        // Track "blocked waiting for a permit" for the LeaseArbiter's
193        // fairness check. Guard is scoped to the await: on success the
194        // guard drops at end-of-block (decrement); on cancellation the
195        // future is dropped mid-await, the guard drops, and the
196        // counter still decrements. Either way the count stays
197        // accurate.
198        let permit = {
199            let _pending = PendingClaimGuard::new(Arc::clone(&self.pending_claimers));
200            Arc::clone(&self.permits)
201                .acquire_owned()
202                .await
203                .expect("GpuPool semaphore should never be closed")
204        };
205        // The permit guarantees ≥1 free slot. None here means the
206        // semaphore count and free-flag count drifted apart — a bug
207        // (RAII Drop bypassed, wrong atomic ordering, etc.).
208        match self.assign_free_slot(permit) {
209            Some(lease) => Some(lease),
210            None => unreachable!(
211                "GpuPool: permit acquired but no free slot found — \
212                 semaphore count and free-flag count drifted apart"
213            ),
214        }
215    }
216
217    /// Try to claim a GPU without blocking. Returns `None` if every
218    /// GPU is currently leased OR if the host has no GPUs.
219    ///
220    /// Used by the LeaseArbiter (planned 2026-05-10) to grab a helper
221    /// lease without contending with blocked variant tasks. Tokio's
222    /// Semaphore preserves FIFO ordering for queued waiters — a
223    /// permit released while a variant task is parked in
224    /// `acquire_owned().await` is reserved for that waiter and is NOT
225    /// visible to `try_acquire_owned()`, so this method cannot steal
226    /// a permit out from under a queued variant.
227    ///
228    /// Does NOT increment `pending_claimers`; helpers are not blocked
229    /// claimers in the spare-capacity sense.
230    pub fn try_claim(self: &Arc<Self>) -> Option<GpuLease> {
231        if self.gpu_indices.is_empty() {
232            return None;
233        }
234        let permit = Arc::clone(&self.permits).try_acquire_owned().ok()?;
235        Some(self.assign_free_slot(permit).expect(
236            "GpuPool: try_acquire_owned succeeded but no free slot found — \
237             this would mean semaphore and free-flag counts drifted apart",
238        ))
239    }
240
241    /// Permit → lease conversion shared by `claim()` and `try_claim()`.
242    /// The permit guarantees ≥1 free slot exists; the CAS loop finds
243    /// the first slot we win the race for. With N ≤ 16 GPUs in
244    /// realistic deployments the linear scan is faster than any
245    /// index-tracking scheme.
246    ///
247    /// Returns `Some(lease)` on the only correct path. Returns `None`
248    /// only if the semaphore and free-flag counts drifted apart,
249    /// which the pool's invariants forbid (`claim` panics via
250    /// `unreachable!` in that case; `try_claim` propagates as a
251    /// distinguishable "this would never happen" via `expect`).
252    fn assign_free_slot(&self, permit: OwnedSemaphorePermit) -> Option<GpuLease> {
253        for (slot_idx, slot) in self.free.iter().enumerate() {
254            if slot
255                .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
256                .is_ok()
257            {
258                return Some(GpuLease {
259                    gpu_index: self.gpu_indices[slot_idx],
260                    vendor: self.gpu_vendors[slot_idx],
261                    slot_idx,
262                    free: Arc::clone(&self.free),
263                    _permit: permit,
264                });
265            }
266        }
267        None
268    }
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274    use codec::gpu::GpuVendor;
275
276    fn synth(index: u32) -> GpuDevice {
277        GpuDevice {
278            index,
279            vendor_index: index,
280            vendor: GpuVendor::Nvidia,
281            name: format!("synth-{index}"),
282            generation: "Synth".into(),
283            pci_id: String::new(),
284            vram_mib: 0,
285            serial: None,
286            host_pci_address: String::new(),
287            vendor_id_hex: String::new(),
288        }
289    }
290
291    fn synth_intel(index: u32) -> GpuDevice {
292        GpuDevice {
293            index,
294            vendor_index: index,
295            vendor: GpuVendor::Intel,
296            name: format!("intel-{index}"),
297            generation: "Synth".into(),
298            pci_id: String::new(),
299            vram_mib: 0,
300            serial: None,
301            host_pci_address: String::new(),
302            vendor_id_hex: String::new(),
303        }
304    }
305
306    #[tokio::test]
307    async fn empty_pool_returns_none() {
308        let pool = Arc::new(GpuPool::new(&[]));
309        assert!(pool.claim().await.is_none());
310        assert_eq!(pool.capacity(), 0);
311    }
312
313    #[tokio::test]
314    async fn single_gpu_serializes_claims() {
315        let pool = Arc::new(GpuPool::new(&[synth(0)]));
316        let lease1 = pool.claim().await.unwrap();
317        assert_eq!(lease1.gpu_index, 0);
318
319        // Second claim must wait — race it against a short timeout to
320        // assert it does NOT resolve while lease1 is held.
321        let pool_clone = Arc::clone(&pool);
322        let claim2 = tokio::spawn(async move { pool_clone.claim().await.unwrap() });
323
324        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
325        assert!(
326            !claim2.is_finished(),
327            "second claim resolved while lease held"
328        );
329
330        drop(lease1);
331        let lease2 = claim2.await.unwrap();
332        assert_eq!(lease2.gpu_index, 0);
333    }
334
335    #[tokio::test]
336    async fn two_gpus_concurrent_leases_distinct_indices() {
337        let pool = Arc::new(GpuPool::new(&[synth(0), synth(1)]));
338        let lease_a = pool.claim().await.unwrap();
339        let lease_b = pool.claim().await.unwrap();
340        assert_ne!(lease_a.gpu_index, lease_b.gpu_index);
341    }
342
343    #[tokio::test]
344    async fn third_claim_waits_until_one_drops() {
345        let pool = Arc::new(GpuPool::new(&[synth(0), synth(1)]));
346        let lease_a = pool.claim().await.unwrap();
347        let _lease_b = pool.claim().await.unwrap();
348
349        let pool_clone = Arc::clone(&pool);
350        let claim_c = tokio::spawn(async move { pool_clone.claim().await.unwrap() });
351
352        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
353        assert!(!claim_c.is_finished());
354
355        let dropped_idx = lease_a.gpu_index;
356        drop(lease_a);
357
358        let lease_c = claim_c.await.unwrap();
359        assert_eq!(lease_c.gpu_index, dropped_idx);
360    }
361
362    #[tokio::test]
363    async fn two_intel_arc_cards_both_get_intel_leases() {
364        // 2× Arc, 0× NVIDIA. Each card has its own per-vendor index.
365        // Both leases come back vendor=Intel and the indices are
366        // distinct so the encoder factory's pick_vendor_device(Intel,
367        // Some(0/1)) finds the right physical card per lease.
368        let pool = Arc::new(GpuPool::new(&[synth_intel(0), synth_intel(1)]));
369        let l1 = pool.claim().await.unwrap();
370        let l2 = pool.claim().await.unwrap();
371        assert_eq!(l1.vendor, GpuVendor::Intel);
372        assert_eq!(l2.vendor, GpuVendor::Intel);
373        let mut indices: Vec<u32> = vec![l1.gpu_index, l2.gpu_index];
374        indices.sort();
375        assert_eq!(indices, vec![0, 1]);
376    }
377
378    #[tokio::test]
379    async fn two_nvidia_cards_both_get_nvidia_leases() {
380        // 2× NVIDIA, 0× Arc. Same shape as the Intel-Intel case.
381        let pool = Arc::new(GpuPool::new(&[synth(0), synth(1)]));
382        let l1 = pool.claim().await.unwrap();
383        let l2 = pool.claim().await.unwrap();
384        assert_eq!(l1.vendor, GpuVendor::Nvidia);
385        assert_eq!(l2.vendor, GpuVendor::Nvidia);
386        let mut indices: Vec<u32> = vec![l1.gpu_index, l2.gpu_index];
387        indices.sort();
388        assert_eq!(indices, vec![0, 1]);
389    }
390
391    #[tokio::test]
392    async fn lease_carries_vendor_for_dispatch() {
393        // Multi-vendor host: NVIDIA at index 0 + Intel at index 0.
394        // Without vendor on the lease, the encoder factory's NVIDIA-
395        // first dispatch would have always picked NVENC. With vendor,
396        // each lease tells the factory which backend to use.
397        let pool = Arc::new(GpuPool::new(&[synth(0), synth_intel(0)]));
398        let l1 = pool.claim().await.unwrap();
399        let l2 = pool.claim().await.unwrap();
400        let mut vendors: Vec<GpuVendor> = vec![l1.vendor, l2.vendor];
401        // Order is non-deterministic between the two slots; both
402        // vendors must appear exactly once.
403        vendors.sort_by_key(|v| match v {
404            GpuVendor::Nvidia => 0,
405            GpuVendor::Amd => 1,
406            GpuVendor::Intel => 2,
407        });
408        assert_eq!(vendors, vec![GpuVendor::Nvidia, GpuVendor::Intel]);
409    }
410
411    #[tokio::test]
412    async fn snapshot_leases_reflects_current_state() {
413        // Phase 2 contract: snapshot returns one entry per slot in
414        // construction order; `leased` mirrors the live free-flag.
415        let pool = Arc::new(GpuPool::new(&[synth(0), synth_intel(1)]));
416
417        let snap0 = pool.snapshot_leases();
418        assert_eq!(snap0.len(), 2);
419        assert_eq!(snap0[0].index, 0);
420        assert_eq!(snap0[0].vendor, GpuVendor::Nvidia);
421        assert!(!snap0[0].leased);
422        assert_eq!(snap0[1].index, 1);
423        assert_eq!(snap0[1].vendor, GpuVendor::Intel);
424        assert!(!snap0[1].leased);
425
426        // Claim the NVIDIA slot → snapshot reflects it.
427        let lease = pool.claim().await.unwrap();
428        // Order in which slots get claimed isn't strictly tied to
429        // vec position, but with N=2 and one outstanding lease the
430        // snapshot must show exactly one `leased=true`.
431        let snap1 = pool.snapshot_leases();
432        let leased_count = snap1.iter().filter(|e| e.leased).count();
433        assert_eq!(leased_count, 1);
434
435        drop(lease);
436        let snap2 = pool.snapshot_leases();
437        assert!(snap2.iter().all(|e| !e.leased));
438    }
439
440    #[tokio::test]
441    async fn snapshot_leases_empty_for_cpu_host() {
442        let pool = Arc::new(GpuPool::new(&[]));
443        let snap = pool.snapshot_leases();
444        assert!(snap.is_empty());
445    }
446
447    #[tokio::test]
448    async fn snapshot_leases_carries_device_name() {
449        // The Phase 2 load-tick task reads .name straight off the
450        // snapshot to build the worker_load frame's gpu_pool entry,
451        // so the lookup must hit the real GpuDevice.name (not the
452        // stringified vendor).
453        let pool = Arc::new(GpuPool::new(&[synth(0)]));
454        let snap = pool.snapshot_leases();
455        assert_eq!(snap.len(), 1);
456        assert_eq!(snap[0].name, "synth-0");
457    }
458
459    // ---- pending_claimers + try_claim (2026-05-10) ----
460
461    #[tokio::test]
462    async fn pending_claimers_starts_at_zero() {
463        let pool = Arc::new(GpuPool::new(&[synth(0), synth(1)]));
464        assert_eq!(pool.pending_claimers(), 0);
465    }
466
467    #[tokio::test]
468    async fn pending_claimers_zero_after_unblocked_claim() {
469        // Single GPU, single immediate claim — never blocks, so the
470        // count should observe 0 both before AND after the claim.
471        let pool = Arc::new(GpuPool::new(&[synth(0)]));
472        assert_eq!(pool.pending_claimers(), 0);
473        let _lease = pool.claim().await.unwrap();
474        assert_eq!(pool.pending_claimers(), 0);
475    }
476
477    #[tokio::test]
478    async fn pending_claimers_increments_during_blocked_claim() {
479        // 1 GPU, take it; spawn a second claim → that task parks in
480        // `acquire_owned().await`; pending_claimers should observe 1.
481        let pool = Arc::new(GpuPool::new(&[synth(0)]));
482        let lease1 = pool.claim().await.unwrap();
483        assert_eq!(pool.pending_claimers(), 0);
484
485        let pool_clone = Arc::clone(&pool);
486        let claim2 = tokio::spawn(async move { pool_clone.claim().await.unwrap() });
487
488        // Give the spawned task a moment to enter the await.
489        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
490        assert_eq!(
491            pool.pending_claimers(),
492            1,
493            "blocked claimer should be counted",
494        );
495
496        // Release the lease → blocked claimer resumes.
497        drop(lease1);
498        let _lease2 = claim2.await.unwrap();
499        assert_eq!(
500            pool.pending_claimers(),
501            0,
502            "after resume, blocked count returns to 0",
503        );
504    }
505
506    #[tokio::test]
507    async fn pending_claimers_increments_for_multiple_blockers() {
508        // 1 GPU, 3 concurrent claimers (1 immediate, 2 blocked) →
509        // pending observes 2 while both are parked.
510        let pool = Arc::new(GpuPool::new(&[synth(0)]));
511        let lease1 = pool.claim().await.unwrap();
512
513        let pool_a = Arc::clone(&pool);
514        let _a = tokio::spawn(async move { pool_a.claim().await.unwrap() });
515        let pool_b = Arc::clone(&pool);
516        let _b = tokio::spawn(async move { pool_b.claim().await.unwrap() });
517
518        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
519        assert_eq!(pool.pending_claimers(), 2);
520
521        drop(lease1);
522        // First waiter resumes; second still parked → count goes 2→1.
523        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
524        assert_eq!(pool.pending_claimers(), 1);
525    }
526
527    #[tokio::test]
528    async fn pending_claimers_decrements_under_cancellation() {
529        // Park a claim, then abort the task before the await
530        // resolves. The PendingClaimGuard's Drop must still run and
531        // bring the count back to 0.
532        let pool = Arc::new(GpuPool::new(&[synth(0)]));
533        let _lease1 = pool.claim().await.unwrap();
534
535        let pool_clone = Arc::clone(&pool);
536        let task = tokio::spawn(async move { pool_clone.claim().await });
537        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
538        assert_eq!(pool.pending_claimers(), 1);
539
540        task.abort();
541        // Abort drops the future, which drops the PendingClaimGuard
542        // inside the await scope. Allow a scheduler tick to observe.
543        let _ = task.await; // resolves with JoinError(Cancelled)
544        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
545        assert_eq!(
546            pool.pending_claimers(),
547            0,
548            "cancelled claim must still decrement pending_claimers",
549        );
550    }
551
552    #[tokio::test]
553    async fn try_claim_returns_none_when_pool_full() {
554        // All permits taken → try_claim is None.
555        let pool = Arc::new(GpuPool::new(&[synth(0)]));
556        let _lease = pool.claim().await.unwrap();
557        assert!(pool.try_claim().is_none());
558    }
559
560    #[tokio::test]
561    async fn try_claim_returns_lease_when_capacity_available() {
562        let pool = Arc::new(GpuPool::new(&[synth(0), synth(1)]));
563        let lease1 = pool.try_claim().unwrap();
564        let lease2 = pool.try_claim().unwrap();
565        assert_ne!(lease1.gpu_index, lease2.gpu_index);
566        assert!(
567            pool.try_claim().is_none(),
568            "after both GPUs leased, third try_claim must be None",
569        );
570    }
571
572    #[tokio::test]
573    async fn try_claim_returns_none_on_cpu_only_host() {
574        let pool = Arc::new(GpuPool::new(&[]));
575        assert!(pool.try_claim().is_none());
576    }
577
578    #[tokio::test]
579    async fn try_claim_does_not_steal_from_blocked_claimer() {
580        // The contract the LeaseArbiter relies on: when a variant
581        // task is parked in `claim()`'s `acquire_owned().await` and a
582        // permit becomes available, that permit goes to the parked
583        // variant FIRST. A racing `try_claim()` must return None.
584        //
585        // Tokio's Semaphore is documented as FIFO for `acquire_owned`;
586        // released permits are reserved for queued waiters and are
587        // NOT visible to `try_acquire_owned()`. This test guards
588        // against an accidental regression (e.g. someone swapping in
589        // a non-fair semaphore) by verifying the behaviour
590        // empirically.
591        let pool = Arc::new(GpuPool::new(&[synth(0)]));
592        let lease1 = pool.claim().await.unwrap();
593
594        // Park a blocked claimer.
595        let pool_clone = Arc::clone(&pool);
596        let blocked = tokio::spawn(async move { pool_clone.claim().await.unwrap() });
597        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
598        assert_eq!(pool.pending_claimers(), 1);
599
600        // Release the lease. The released permit is now reserved for
601        // the parked claimer per Tokio's FIFO contract.
602        drop(lease1);
603
604        // Try to steal it from the blocked claimer — must fail.
605        assert!(
606            pool.try_claim().is_none(),
607            "try_claim must not steal a permit reserved for a queued claimer",
608        );
609
610        // The blocked claimer should still resolve.
611        let _lease2 = blocked.await.unwrap();
612    }
613
614    #[tokio::test]
615    async fn try_claim_lease_drop_releases_permit() {
616        // try_claim leases use the same RAII Drop path; verify the
617        // permit returns to the pool when the lease drops.
618        let pool = Arc::new(GpuPool::new(&[synth(0)]));
619        let lease = pool.try_claim().unwrap();
620        assert!(pool.try_claim().is_none());
621        drop(lease);
622        assert!(pool.try_claim().is_some(), "permit returned to pool after lease drop");
623    }
624
625    #[tokio::test]
626    async fn try_claim_does_not_affect_pending_claimers() {
627        // try_claim must not touch pending_claimers — helpers are
628        // opportunistic, not blocked claimers.
629        let pool = Arc::new(GpuPool::new(&[synth(0)]));
630        let _l1 = pool.try_claim().unwrap();
631        assert_eq!(pool.pending_claimers(), 0);
632        assert!(pool.try_claim().is_none());
633        assert_eq!(pool.pending_claimers(), 0);
634    }
635
636    #[tokio::test]
637    async fn sparse_indices_preserved() {
638        // CUDA_VISIBLE_DEVICES could expose only [0, 2, 5].
639        let pool = Arc::new(GpuPool::new(&[synth(0), synth(2), synth(5)]));
640        let l0 = pool.claim().await.unwrap();
641        let l1 = pool.claim().await.unwrap();
642        let l2 = pool.claim().await.unwrap();
643        let mut got: Vec<u32> = vec![l0.gpu_index, l1.gpu_index, l2.gpu_index];
644        got.sort();
645        assert_eq!(got, vec![0, 2, 5]);
646    }
647}