1use std::cmp::{Ordering as CmpOrdering, Reverse};
4use std::collections::BinaryHeap;
5use std::hash::{Hash, Hasher};
6use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
7use std::sync::{Arc, LazyLock, Mutex, MutexGuard, OnceLock, Weak};
8use std::time::Instant;
9
10use dashmap::DashMap;
11use rustc_hash::{FxHashMap, FxHasher};
12use smallvec::SmallVec;
13use vyre_driver::{BackendError, ResidentHandle, ResidentOwner};
14
15use super::pool::PoolReturn;
16
17static NEXT_BUFFER_ID: AtomicU64 = AtomicU64::new(1);
18static RESIDENT_BUFFERS: OnceLock<DashMap<u64, Weak<GpuBufferInner>>> = OnceLock::new();
19const STAGING_BUFFER_POOL_CLASS_CAP: usize = 16;
20
21fn resident_buffers() -> &'static DashMap<u64, Weak<GpuBufferInner>> {
22 RESIDENT_BUFFERS.get_or_init(DashMap::new)
23}
24
25static RESIDENT_OWNER: LazyLock<Result<ResidentOwner, BackendError>> =
35 LazyLock::new(ResidentOwner::new);
36
37fn resident_owner() -> Result<ResidentOwner, BackendError> {
39 match &*RESIDENT_OWNER {
40 Ok(owner) => Ok(*owner),
41 Err(error) => Err(BackendError::new(format!(
42 "WGPU resident buffers have no namespace identity: {error} Fix: reduce the number of backend instances created in this process."
43 ))),
44 }
45}
46
47pub(crate) fn check_resident_owner(
52 handle: ResidentHandle,
53 context: &str,
54) -> Result<(), BackendError> {
55 resident_owner()?.resolve(handle, context)?;
56 Ok(())
57}
58
59fn pointer_identity_key<T>(ptr: *const T) -> u64 {
60 let mut hasher = FxHasher::default();
61 ptr.addr().hash(&mut hasher);
62 hasher.finish()
63}
64
65#[derive(Clone)]
71pub struct GpuBufferHandle {
72 inner: Arc<GpuBufferInner>,
73}
74
75struct GpuBufferInner {
76 id: u64,
77 buffer: Arc<wgpu::Buffer>,
78 byte_len: u64,
79 allocation_len: u64,
80 element_count: usize,
81 usage: wgpu::BufferUsages,
82 pool_return: Option<PoolReturn>,
83}
84
85#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
87pub struct StagingBufferPoolStats {
88 pub allocations: usize,
90 pub hits: usize,
92}
93
94#[derive(Clone, Default)]
102pub struct StagingBufferPool {
103 inner: Arc<Mutex<StagingBufferPoolInner>>,
104}
105
106impl std::fmt::Debug for StagingBufferPool {
107 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
108 f.debug_struct("StagingBufferPool").finish_non_exhaustive()
109 }
110}
111
112#[derive(Default)]
113struct StagingBufferPoolInner {
114 free: FxHashMap<(u64, u32), SmallVec<[wgpu::Buffer; STAGING_BUFFER_POOL_CLASS_CAP]>>,
115 allocations: usize,
116 hits: usize,
117}
118
119impl StagingBufferPool {
120 fn lock_inner(&self) -> MutexGuard<'_, StagingBufferPoolInner> {
121 self.inner.lock().unwrap_or_else(|error| {
122 tracing::error!(
123 "Vyre WGPU staging buffer pool lock was poisoned: {error}. Fix: discard the pool after a panic; continuing with recovered state."
124 );
125 error.into_inner()
126 })
127 }
128
129 #[must_use]
131 #[inline]
132 pub fn new() -> Self {
133 Self::default()
134 }
135
136 #[must_use]
138 pub fn stats(&self) -> StagingBufferPoolStats {
139 let inner = self.lock_inner();
140 StagingBufferPoolStats {
141 allocations: inner.allocations,
142 hits: inner.hits,
143 }
144 }
145
146 pub fn acquire(
151 &self,
152 device: &wgpu::Device,
153 size: u64,
154 usage: wgpu::BufferUsages,
155 ) -> wgpu::Buffer {
156 let key = (size, usage.bits());
157 let mut inner = self.lock_inner();
158 if let Some(buffers) = inner.free.get_mut(&key) {
159 if let Some(buffer) = buffers.pop() {
160 inner.hits += 1;
161 return buffer;
162 }
163 }
164 inner.allocations += 1;
165 drop(inner);
166 device.create_buffer(&wgpu::BufferDescriptor {
167 label: Some("vyre staging readback"),
168 size,
169 usage,
170 mapped_at_creation: false,
171 })
172 }
173
174 pub fn release(&self, buffer: wgpu::Buffer, size: u64, usage: wgpu::BufferUsages) {
179 let key = (size, usage.bits());
180 let mut inner = self.lock_inner();
181 let buffers = inner.free.entry(key).or_default();
182 if buffers.len() == STAGING_BUFFER_POOL_CLASS_CAP {
183 buffers.remove(0);
184 }
185 buffers.push(buffer);
186 }
187}
188
189pub(crate) enum PendingGpuBufferReadback {
191 Ready,
192 Mapping {
193 readback: wgpu::Buffer,
194 read_len: u64,
195 readback_usage: wgpu::BufferUsages,
196 pool: Option<StagingBufferPool>,
197 submission: wgpu::SubmissionIndex,
198 receiver: crossbeam_channel::Receiver<Result<(), wgpu::BufferAsyncError>>,
199 ready: Arc<std::sync::atomic::AtomicBool>,
200 trim_start: usize,
201 visible_len: usize,
202 },
203}
204
205impl PendingGpuBufferReadback {
206 pub(crate) fn is_ready(&self, device: &wgpu::Device) -> bool {
207 match self {
208 Self::Ready => true,
209 Self::Mapping { ready, .. } => {
210 if crate::runtime::device::poll_device_once(device).is_err() {
211 return false;
212 }
213 ready.load(Ordering::Acquire)
214 }
215 }
216 }
217
218 pub(crate) fn await_into(
219 self,
220 device: &wgpu::Device,
221 deadline: Option<Instant>,
222 out: &mut Vec<u8>,
223 ) -> Result<(), BackendError> {
224 let Self::Mapping {
225 readback,
226 read_len,
227 readback_usage,
228 pool,
229 submission,
230 receiver,
231 trim_start,
232 visible_len,
233 ..
234 } = self
235 else {
236 out.clear();
237 return Ok(());
238 };
239 let mapping = if let Some(deadline) = deadline {
240 let mut backoff = crate::wait_backoff::AdaptiveWaitBackoff::from_micros(64, 2, 50, 5);
241 loop {
242 crate::runtime::device::poll_device_once(device)?;
243 match receiver.try_recv() {
244 Ok(result) => break result,
245 Err(crossbeam_channel::TryRecvError::Empty) => {}
246 Err(crossbeam_channel::TryRecvError::Disconnected) => {
247 return Err(BackendError::new(
248 "persistent buffer readback channel closed before completion. Fix: keep the GPU device alive until readback completes.",
249 ));
250 }
251 }
252 let now = Instant::now();
253 if now >= deadline {
254 return Err(BackendError::new(
255 "dispatch cancelled after DispatchConfig.timeout before readback completed. Fix: raise DispatchConfig.timeout or split the program into smaller chunks.",
256 ));
257 }
258 backoff.idle_for(deadline.saturating_duration_since(now));
259 }
260 } else {
261 crate::runtime::device::poll_device_wait_for(device, submission)?;
262 receiver
263 .recv_timeout(std::time::Duration::from_secs(30))
264 .map_err(|source| {
265 BackendError::new(format!(
266 "persistent buffer readback callback did not complete after submission wait: {source}. Fix: keep the GPU device alive and inspect driver callback progress."
267 ))
268 })?
269 };
270 mapping.map_err(|source| {
271 BackendError::new(format!(
272 "persistent buffer readback mapping failed: {source:?}. Fix: use COPY_SRC handles and MAP_READ staging buffers."
273 ))
274 })?;
275 let slice = readback.slice(0..read_len);
276 let mapped = slice.get_mapped_range();
277 let trim_end = trim_start.checked_add(visible_len).ok_or_else(|| {
278 BackendError::new(format!(
279 "persistent buffer range trim overflows usize at offset {trim_start} len {visible_len}. Fix: split the buffer before readback."
280 ))
281 })?;
282 let visible = &mapped[trim_start..trim_end];
283 if out.len() == visible_len {
284 out.copy_from_slice(visible);
285 } else {
286 vyre_foundation::allocation::reserve_exact_cleared(out, visible_len).map_err(
287 |source| {
288 BackendError::new(format!(
289 "persistent buffer readback could not reserve {visible_len} output bytes exactly: {source}. Fix: lower max_output_bytes or stream readback in smaller shards."
290 ))
291 },
292 )?;
293 out.extend_from_slice(visible);
294 }
295 drop(mapped);
296 readback.unmap();
297 if let Some(pool) = pool {
298 pool.release(readback, read_len, readback_usage);
299 }
300 Ok(())
301 }
302}
303impl GpuBufferHandle {
304 pub fn upload(
313 device: &wgpu::Device,
314 queue: &wgpu::Queue,
315 bytes: &[u8],
316 usage: wgpu::BufferUsages,
317 ) -> Result<Self, BackendError> {
318 let allocation_len = aligned_len(bytes.len())?;
319 let final_usage = usage | wgpu::BufferUsages::COPY_DST;
320 if allocation_len > 0 {
333 let buffer = device.create_buffer(&wgpu::BufferDescriptor {
334 label: Some("vyre persistent upload"),
335 size: allocation_len,
336 usage: final_usage,
337 mapped_at_creation: true,
338 });
339 {
340 let mut mapped = buffer.slice(..).get_mapped_range_mut();
341 crate::padded_upload::write_padded_into_mapped(&mut mapped, bytes)?;
342 }
343 buffer.unmap();
344 let logical_len = u64::try_from(bytes.len()).map_err(|source| {
345 BackendError::new(format!(
346 "GPU upload logical byte length cannot fit u64: {source}. Fix: split the dispatch input."
347 ))
348 })?;
349 return Ok(Self::from_parts(
350 Arc::new(buffer),
351 logical_len,
352 allocation_len,
353 bytes.len(),
354 final_usage,
355 None,
356 ));
357 }
358 let buffer = device.create_buffer(&wgpu::BufferDescriptor {
362 label: Some("vyre persistent upload"),
363 size: allocation_len,
364 usage: final_usage,
365 mapped_at_creation: false,
366 });
367 write_padded(queue, &buffer, bytes, allocation_len)?;
368 Ok(Self::from_parts(
369 Arc::new(buffer),
370 0,
371 allocation_len,
372 bytes.len(),
373 final_usage,
374 None,
375 ))
376 }
377
378 pub fn alloc(
385 device: &wgpu::Device,
386 len: u64,
387 usage: wgpu::BufferUsages,
388 ) -> Result<Self, BackendError> {
389 let allocation_len = aligned_len_u64(len, "persistent GPU allocation length")?;
390 let buffer = device.create_buffer(&wgpu::BufferDescriptor {
391 label: Some("vyre persistent alloc"),
392 size: allocation_len,
393 usage,
394 mapped_at_creation: false,
395 });
396 let host_len = usize::try_from(len).map_err(|error| {
397 BackendError::new(format!(
398 "GpuBufferHandle::alloc received logical byte length {len} that does not fit usize on this host: {error}. Fix: shard the GPU buffer before allocating or run on a host with a wide enough address space."
399 ))
400 })?;
401 Ok(Self::from_parts(
402 Arc::new(buffer),
403 len,
404 allocation_len,
405 host_len,
406 usage,
407 None,
408 ))
409 }
410
411 pub fn readback(
421 &self,
422 device: &wgpu::Device,
423 queue: &wgpu::Queue,
424 out: &mut Vec<u8>,
425 ) -> Result<(), BackendError> {
426 self.readback_until(device, None, queue, out, None)
427 }
428
429 pub fn readback_prefix(
441 &self,
442 device: &wgpu::Device,
443 queue: &wgpu::Queue,
444 len: u64,
445 out: &mut Vec<u8>,
446 ) -> Result<(), BackendError> {
447 self.readback_prefix_until(device, None, queue, len, out, None)
448 }
449
450 pub fn readback_range(
460 &self,
461 device: &wgpu::Device,
462 queue: &wgpu::Queue,
463 byte_offset: u64,
464 len: u64,
465 out: &mut Vec<u8>,
466 ) -> Result<(), BackendError> {
467 self.readback_range_until(device, None, queue, byte_offset, len, out, None)
468 }
469
470 pub(crate) fn readback_until(
471 &self,
472 device: &wgpu::Device,
473 pool: Option<&StagingBufferPool>,
474 queue: &wgpu::Queue,
475 out: &mut Vec<u8>,
476 deadline: Option<Instant>,
477 ) -> Result<(), BackendError> {
478 self.readback_prefix_until(device, pool, queue, self.byte_len(), out, deadline)
479 }
480
481 pub(crate) fn readback_prefix_until(
482 &self,
483 device: &wgpu::Device,
484 pool: Option<&StagingBufferPool>,
485 queue: &wgpu::Queue,
486 len: u64,
487 out: &mut Vec<u8>,
488 deadline: Option<Instant>,
489 ) -> Result<(), BackendError> {
490 self.readback_range_until(device, pool, queue, 0, len, out, deadline)
491 }
492
493 pub(crate) fn readback_range_until(
494 &self,
495 device: &wgpu::Device,
496 pool: Option<&StagingBufferPool>,
497 queue: &wgpu::Queue,
498 byte_offset: u64,
499 len: u64,
500 out: &mut Vec<u8>,
501 deadline: Option<Instant>,
502 ) -> Result<(), BackendError> {
503 self.readback_range_async(device, pool, queue, byte_offset, len)?
504 .await_into(device, deadline, out)
505 }
506
507 pub(crate) fn readback_range_async(
508 &self,
509 device: &wgpu::Device,
510 pool: Option<&StagingBufferPool>,
511 queue: &wgpu::Queue,
512 byte_offset: u64,
513 len: u64,
514 ) -> Result<PendingGpuBufferReadback, BackendError> {
515 if !self.usage().contains(wgpu::BufferUsages::COPY_SRC) {
516 return Err(BackendError::new(
517 "GpuBufferHandle readback requires COPY_SRC usage. Fix: allocate terminal-output buffers with COPY_SRC.",
518 ));
519 }
520 let logical_end = byte_offset.checked_add(len).ok_or_else(|| {
521 BackendError::new(format!(
522 "GpuBufferHandle range readback overflows u64 at offset {byte_offset} len {len}. Fix: split the readback range before dispatch."
523 ))
524 })?;
525 if logical_end > self.byte_len() {
526 return Err(BackendError::new(format!(
527 "GpuBufferHandle range readback requested bytes [{byte_offset}..{logical_end}) from a {} byte buffer. Fix: clamp the requested range to the device-published count.",
528 self.byte_len()
529 )));
530 }
531 if len == 0 {
532 return Ok(PendingGpuBufferReadback::Ready);
533 }
534 let copy_start = byte_offset & !3;
535 let trim_start = byte_offset - copy_start;
536 let visible_copy_len = trim_start.checked_add(len).ok_or_else(|| {
537 BackendError::new(format!(
538 "GpuBufferHandle range readback copy length overflows u64 at trim {trim_start} len {len}. Fix: split the readback range before dispatch."
539 ))
540 })?;
541 let read_len = aligned_len_u64(visible_copy_len, "GPU readback visible copy length")?;
542 let copy_end = copy_start.checked_add(read_len).ok_or_else(|| {
543 BackendError::new(format!(
544 "GpuBufferHandle range readback aligned copy overflows u64 at start {copy_start} len {read_len}. Fix: split the readback range before dispatch."
545 ))
546 })?;
547 if copy_end > self.inner.allocation_len {
548 return Err(BackendError::new(format!(
549 "GpuBufferHandle range readback rounded bytes [{byte_offset}..{logical_end}) to aligned bytes [{copy_start}..{copy_end}), beyond allocation length {}. Fix: allocate buffers with 4-byte padding.",
550 self.inner.allocation_len
551 )));
552 }
553 let visible_len = usize::try_from(len).map_err(|source| {
554 BackendError::new(format!(
555 "persistent buffer prefix length {len} cannot fit usize: {source}. Fix: split the buffer before readback."
556 ))
557 })?;
558 let trim_start = usize::try_from(trim_start).map_err(|source| {
559 BackendError::new(format!(
560 "persistent buffer range trim offset {trim_start} cannot fit usize: {source}. Fix: split the buffer before readback."
561 ))
562 })?;
563 let readback_usage = wgpu::BufferUsages::COPY_DST | wgpu::BufferUsages::MAP_READ;
564 let readback = if let Some(pool) = pool {
565 pool.acquire(device, read_len, readback_usage)
566 } else {
567 device.create_buffer(&wgpu::BufferDescriptor {
568 label: Some("vyre persistent handle readback"),
569 size: read_len,
570 usage: readback_usage,
571 mapped_at_creation: false,
572 })
573 };
574 let mut encoder = device.create_command_encoder(&wgpu::CommandEncoderDescriptor {
575 label: Some("vyre persistent handle readback encoder"),
576 });
577 encoder.copy_buffer_to_buffer(self.buffer(), copy_start, &readback, 0, read_len);
578 let submission = queue.submit(std::iter::once(encoder.finish()));
579 let slice = readback.slice(0..read_len);
580 let (sender, receiver) = crossbeam_channel::bounded(1);
581 let ready = Arc::new(std::sync::atomic::AtomicBool::new(false));
582 let ready_callback = Arc::clone(&ready);
583 slice.map_async(wgpu::MapMode::Read, move |result| {
584 if let Err(error) = sender.send(result) {
585 tracing::error!(
586 ?error,
587 "persistent buffer readback map_async result was lost because the receiver dropped"
588 );
589 }
590 ready_callback.store(true, Ordering::Release);
591 });
592 Ok(PendingGpuBufferReadback::Mapping {
593 readback,
594 read_len,
595 readback_usage,
596 pool: pool.cloned(),
597 submission,
598 receiver,
599 ready,
600 trim_start,
601 visible_len,
602 })
603 }
604
605 #[must_use]
607 pub fn id(&self) -> u64 {
608 self.inner.id
609 }
610
611 #[must_use]
618 pub(crate) fn allocation_identity(&self) -> u64 {
619 pointer_identity_key(Arc::as_ptr(&self.inner.buffer))
620 }
621
622 #[must_use]
624 pub fn from_resident_id(id: u64) -> Option<Self> {
625 let registry = resident_buffers();
626 let entry = registry.get(&id)?;
627 let upgraded = entry.value().upgrade();
628 drop(entry);
629 match upgraded {
630 Some(inner) => Some(Self { inner }),
631 None => {
632 registry.remove(&id);
633 None
634 }
635 }
636 }
637
638 pub fn resident_handle(&self) -> Result<ResidentHandle, BackendError> {
645 Ok(resident_owner()?.handle(self.inner.id))
646 }
647
648 pub fn from_resident_handle(
658 handle: ResidentHandle,
659 context: &str,
660 ) -> Result<Option<Self>, BackendError> {
661 let id = resident_owner()?.resolve(handle, context)?;
662 Ok(Self::from_resident_id(id))
663 }
664
665 #[must_use]
667 pub fn buffer(&self) -> &wgpu::Buffer {
668 &self.inner.buffer
669 }
670
671 #[must_use]
675 pub fn buffer_arc(&self) -> Arc<wgpu::Buffer> {
676 Arc::clone(&self.inner.buffer)
677 }
678
679 #[must_use]
681 pub fn byte_len(&self) -> u64 {
682 self.inner.byte_len
683 }
684
685 #[must_use]
687 pub fn allocation_len(&self) -> u64 {
688 self.inner.allocation_len
689 }
690
691 #[must_use]
693 pub fn element_count(&self) -> usize {
694 self.inner.element_count
695 }
696
697 #[must_use]
699 pub fn usage(&self) -> wgpu::BufferUsages {
700 self.inner.usage
701 }
702
703 pub(crate) fn from_parts(
704 buffer: Arc<wgpu::Buffer>,
705 byte_len: u64,
706 allocation_len: u64,
707 element_count: usize,
708 usage: wgpu::BufferUsages,
709 pool_return: Option<PoolReturn>,
710 ) -> Self {
711 let inner = Arc::new(GpuBufferInner {
712 id: NEXT_BUFFER_ID.fetch_add(1, Ordering::Relaxed),
713 buffer,
714 byte_len,
715 allocation_len,
716 element_count,
717 usage,
718 pool_return,
719 });
720 resident_buffers().insert(inner.id, Arc::downgrade(&inner));
721 Self { inner }
722 }
723}
724
725impl std::fmt::Debug for GpuBufferHandle {
726 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
727 formatter
728 .debug_struct("GpuBufferHandle")
729 .field("id", &self.id())
730 .field("byte_len", &self.byte_len())
731 .field("allocation_len", &self.allocation_len())
732 .field("element_count", &self.element_count())
733 .field("usage", &self.usage())
734 .finish()
735 }
736}
737
738impl Drop for GpuBufferInner {
739 fn drop(&mut self) {
740 resident_buffers().remove(&self.id);
741 if let Some(pool_return) = self.pool_return.take() {
742 pool_return.release(
743 Arc::clone(&self.buffer),
744 self.byte_len,
745 self.allocation_len,
746 self.usage,
747 );
748 }
749 }
750}
751
752pub(crate) fn aligned_len(len: usize) -> Result<u64, BackendError> {
753 let padded = aligned_len_usize(len, "GPU buffer length")?;
754 u64::try_from(padded).map_err(|source| {
755 BackendError::new(format!(
756 "GPU buffer length {padded} cannot fit u64: {source}. Fix: split the dispatch input."
757 ))
758 })
759}
760
761fn aligned_len_u64(len: u64, label: &'static str) -> Result<u64, BackendError> {
762 crate::numeric::WGPU_NUMERIC.align_up_u64(len, 4, 4, label)
763}
764
765fn aligned_len_usize(len: usize, label: &'static str) -> Result<usize, BackendError> {
766 crate::numeric::WGPU_NUMERIC.align_up_usize(len, 4, 4, label)
767}
768
769pub(crate) fn write_padded(
770 queue: &wgpu::Queue,
771 buffer: &wgpu::Buffer,
772 bytes: &[u8],
773 allocation_len: u64,
774) -> Result<(), BackendError> {
775 crate::padded_upload::write_padded_and_zero_fill(queue, buffer, bytes, allocation_len)
776}
777
778const BIND_GROUP_CACHE_CAP: usize = 256;
780
781type BindGroupHandleKey = SmallVec<[u64; 16]>;
784
785#[derive(Clone)]
793pub struct BindGroupCache {
794 cache: Arc<Mutex<BindGroupCacheInner>>,
795 hits: Arc<AtomicUsize>,
796 misses: Arc<AtomicUsize>,
797 evictions: Arc<AtomicUsize>,
798}
799
800struct BindGroupCacheInner {
801 entries: FxHashMap<BindGroupCacheKey, BindGroupCacheEntry>,
802 lru: BinaryHeap<Reverse<BindGroupLruEntry>>,
803 cap: usize,
804 next_generation: u64,
805}
806
807struct BindGroupCacheEntry {
808 bind_group: Arc<wgpu::BindGroup>,
809 last_seen: u64,
810}
811
812#[derive(Clone, Debug, Eq, PartialEq)]
813struct BindGroupLruEntry {
814 last_seen: u64,
815 key: BindGroupCacheKey,
816}
817
818impl Ord for BindGroupLruEntry {
819 fn cmp(&self, other: &Self) -> CmpOrdering {
820 self.last_seen
821 .cmp(&other.last_seen)
822 .then_with(|| self.key.cmp(&other.key))
823 }
824}
825
826impl PartialOrd for BindGroupLruEntry {
827 fn partial_cmp(&self, other: &Self) -> Option<CmpOrdering> {
828 Some(self.cmp(other))
829 }
830}
831
832fn push_bind_group_handle_key(key: &mut BindGroupHandleKey, handle: &GpuBufferHandle) -> bool {
833 key.push(handle.allocation_identity());
834 let Ok(aligned_len) = aligned_len_u64(handle.byte_len(), "bind-group handle key byte length")
835 else {
836 key.pop();
837 return false;
838 };
839 key.push(aligned_len);
840 true
841}
842
843#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
844struct BindGroupCacheKey {
845 layout_id: usize,
846 handles: BindGroupHandleKey,
847}
848
849impl std::fmt::Debug for BindGroupCache {
850 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
851 f.debug_struct("BindGroupCache")
852 .field("hits", &self.hits.load(Ordering::Relaxed))
853 .field("misses", &self.misses.load(Ordering::Relaxed))
854 .field("evictions", &self.evictions.load(Ordering::Relaxed))
855 .field("entries", &self.lock_cache().entries.len())
856 .finish_non_exhaustive()
857 }
858}
859
860impl Default for BindGroupCache {
861 fn default() -> Self {
862 Self::new()
863 }
864}
865
866impl BindGroupCacheInner {
867 fn next_lru_generation(&mut self) -> u64 {
868 let generation = self.next_generation;
869 self.next_generation = self.next_generation.wrapping_add(1);
870 generation
871 }
872
873 fn touch_existing(&mut self, key: &BindGroupCacheKey) -> Option<Arc<wgpu::BindGroup>> {
874 let generation = self.next_lru_generation();
875 let bind_group = {
876 let entry = self.entries.get_mut(key)?;
877 entry.last_seen = generation;
878 Arc::clone(&entry.bind_group)
879 };
880 self.lru.push(Reverse(BindGroupLruEntry {
881 last_seen: generation,
882 key: key.clone(),
883 }));
884 self.compact_lru_if_needed();
885 Some(bind_group)
886 }
887
888 fn insert_entry(&mut self, key: BindGroupCacheKey, bind_group: Arc<wgpu::BindGroup>) {
889 let generation = self.next_lru_generation();
890 self.entries.insert(
891 key.clone(),
892 BindGroupCacheEntry {
893 bind_group,
894 last_seen: generation,
895 },
896 );
897 self.lru.push(Reverse(BindGroupLruEntry {
898 last_seen: generation,
899 key,
900 }));
901 self.compact_lru_if_needed();
902 }
903
904 fn evict_to_cap(&mut self, mut on_evict: impl FnMut()) {
905 while self.entries.len() > self.cap {
906 let Some(key) = self.pop_lru_key() else { break };
907 if self.entries.remove(&key).is_some() {
908 on_evict();
909 }
910 }
911 }
912
913 fn pop_lru_key(&mut self) -> Option<BindGroupCacheKey> {
914 while let Some(Reverse(entry)) = self.lru.pop() {
915 if self
916 .entries
917 .get(&entry.key)
918 .is_some_and(|current| current.last_seen == entry.last_seen)
919 {
920 return Some(entry.key);
921 }
922 }
923 None
924 }
925
926 fn compact_lru_if_needed(&mut self) {
927 let live = self.entries.len();
928 if let Some(limit) = stale_lru_limit(live) {
929 if self.lru.len() <= limit {
930 return;
931 }
932 }
933 self.lru.clear();
934 self.lru.extend(self.entries.iter().map(|(key, entry)| {
935 Reverse(BindGroupLruEntry {
936 last_seen: entry.last_seen,
937 key: key.clone(),
938 })
939 }));
940 }
941}
942
943fn stale_lru_limit(live: usize) -> Option<usize> {
944 live.checked_mul(4).map(|limit| limit.max(8))
945}
946
947impl BindGroupCache {
948 fn lock_cache(&self) -> MutexGuard<'_, BindGroupCacheInner> {
949 self.cache.lock().unwrap_or_else(|error| {
950 tracing::error!(
951 "Vyre WGPU bind-group cache lock was poisoned: {error}. Fix: discard the cache after a panic; continuing with recovered state."
952 );
953 error.into_inner()
954 })
955 }
956
957 #[must_use]
959 pub fn new() -> Self {
960 Self::with_cap(BIND_GROUP_CACHE_CAP)
961 }
962
963 #[must_use]
966 pub fn with_cap(cap: usize) -> Self {
967 Self {
968 cache: Arc::new(Mutex::new(BindGroupCacheInner {
969 entries: FxHashMap::default(),
970 lru: BinaryHeap::new(),
971 cap: cap.max(1),
972 next_generation: 0,
973 })),
974 hits: Arc::new(AtomicUsize::new(0)),
975 misses: Arc::new(AtomicUsize::new(0)),
976 evictions: Arc::new(AtomicUsize::new(0)),
977 }
978 }
979
980 pub fn get_or_create(
988 &self,
989 layout_id: usize,
990 handles: &[GpuBufferHandle],
991 factory: impl FnOnce() -> wgpu::BindGroup,
992 ) -> Arc<wgpu::BindGroup> {
993 let Some(key_part_count) = handles.len().checked_mul(2) else {
994 self.misses.fetch_add(1, Ordering::Relaxed);
995 return Arc::new(factory());
996 };
997 let mut key_parts = SmallVec::with_capacity(key_part_count);
998 for handle in handles {
999 if !push_bind_group_handle_key(&mut key_parts, handle) {
1000 self.misses.fetch_add(1, Ordering::Relaxed);
1001 return Arc::new(factory());
1002 }
1003 }
1004 self.get_or_create_by_ids(layout_id, key_parts, factory)
1005 }
1006
1007 pub(crate) fn get_or_create_by_ids(
1008 &self,
1009 layout_id: usize,
1010 handles: SmallVec<[u64; 16]>,
1011 factory: impl FnOnce() -> wgpu::BindGroup,
1012 ) -> Arc<wgpu::BindGroup> {
1013 let key = BindGroupCacheKey { layout_id, handles };
1014 {
1015 let mut cache = self.lock_cache();
1016 if let Some(existing) = cache.touch_existing(&key) {
1017 self.hits.fetch_add(1, Ordering::Relaxed);
1018 return existing;
1019 }
1020 }
1021 let bg = Arc::new(factory());
1022 let mut cache = self.lock_cache();
1023 cache.insert_entry(key, Arc::clone(&bg));
1024 cache.evict_to_cap(|| {
1025 self.evictions.fetch_add(1, Ordering::Relaxed);
1026 });
1027 self.misses.fetch_add(1, Ordering::Relaxed);
1028 bg
1029 }
1030
1031 pub(crate) fn get_by_ids(
1032 &self,
1033 layout_id: usize,
1034 handles: &[u64],
1035 ) -> Option<Arc<wgpu::BindGroup>> {
1036 let key = BindGroupCacheKey {
1037 layout_id,
1038 handles: SmallVec::from_slice(handles),
1039 };
1040 let mut cache = self.lock_cache();
1041 let existing = cache.touch_existing(&key)?;
1042 self.hits.fetch_add(1, Ordering::Relaxed);
1043 Some(existing)
1044 }
1045
1046 pub(crate) fn insert_by_ids(
1047 &self,
1048 layout_id: usize,
1049 handles: &[u64],
1050 bind_group: wgpu::BindGroup,
1051 ) -> Arc<wgpu::BindGroup> {
1052 let key = BindGroupCacheKey {
1053 layout_id,
1054 handles: SmallVec::from_slice(handles),
1055 };
1056 let mut cache = self.lock_cache();
1057 if let Some(existing) = cache.touch_existing(&key) {
1058 self.hits.fetch_add(1, Ordering::Relaxed);
1059 return existing;
1060 }
1061 let bg = Arc::new(bind_group);
1062 cache.insert_entry(key, Arc::clone(&bg));
1063 cache.evict_to_cap(|| {
1064 self.evictions.fetch_add(1, Ordering::Relaxed);
1065 });
1066 self.misses.fetch_add(1, Ordering::Relaxed);
1067 bg
1068 }
1069
1070 #[must_use]
1072 pub fn stats(&self) -> BindGroupCacheStats {
1073 BindGroupCacheStats {
1074 hits: self.hits.load(Ordering::Relaxed),
1075 misses: self.misses.load(Ordering::Relaxed),
1076 evictions: self.evictions.load(Ordering::Relaxed),
1077 entries: self.lock_cache().entries.len(),
1078 }
1079 }
1080}
1081
1082#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
1084pub struct BindGroupCacheStats {
1085 pub hits: usize,
1087 pub misses: usize,
1089 pub evictions: usize,
1091 pub entries: usize,
1093}
1094
1095#[cfg(test)]
1096mod tests {
1097 use super::*;
1098
1099 #[test]
1102 fn staging_pool_reuses_buffers_on_hot_readback_loop() {
1103 let arc = crate::runtime::cached_device()
1104 .expect("Fix: GPU device is required for staging pool test");
1105 let (device, queue) = &*arc;
1106
1107 let contents: Vec<u8> = vec![0xAB; 64];
1109 let handle =
1110 GpuBufferHandle::upload(device, queue, &contents, wgpu::BufferUsages::COPY_SRC)
1111 .expect("Fix: upload should succeed");
1112
1113 let pool = StagingBufferPool::new();
1114
1115 for _ in 0..100 {
1116 let mut out = Vec::new();
1117 handle
1118 .readback_until(device, Some(&pool), queue, &mut out, None)
1119 .expect("Fix: pooled readback should succeed");
1120 assert_eq!(out, contents, "readback bytes must match uploaded bytes");
1121 }
1122
1123 let stats = pool.stats();
1124 assert!(
1125 stats.allocations <= 2,
1126 "hot loop of 100 identical readbacks should allocate at most 2 staging buffers, got {} allocations and {} hits",
1127 stats.allocations,
1128 stats.hits
1129 );
1130 }
1131
1132 #[test]
1134 fn readback_without_pool_always_allocates() {
1135 let arc = crate::runtime::cached_device()
1136 .expect("Fix: GPU device is required for readback regression test");
1137 let (device, queue) = &*arc;
1138
1139 let contents: Vec<u8> = vec![0xCD; 32];
1140 let handle =
1141 GpuBufferHandle::upload(device, queue, &contents, wgpu::BufferUsages::COPY_SRC)
1142 .expect("Fix: upload should succeed");
1143
1144 for _ in 0..5 {
1145 let mut out = Vec::new();
1146 handle
1147 .readback(device, queue, &mut out)
1148 .expect("Fix: unpooled readback should succeed");
1149 assert_eq!(out, contents);
1150 }
1151 }
1152
1153 #[test]
1159 fn mapped_upload_roundtrips_exact_bytes_across_boundaries() {
1160 let arc = crate::runtime::cached_device()
1161 .expect("Fix: live GPU device required for mapped upload roundtrip test");
1162 let (device, queue) = &*arc;
1163 for &len in &[1usize, 3, 4, 5, 257, (1 << 20) + 3] {
1167 let contents: Vec<u8> = (0..len).map(|i| (i % 251) as u8).collect();
1168 let handle =
1169 GpuBufferHandle::upload(device, queue, &contents, wgpu::BufferUsages::COPY_SRC)
1170 .expect("Fix: mapped upload should succeed at every size");
1171 let mut out = Vec::new();
1172 handle
1173 .readback(device, queue, &mut out)
1174 .expect("Fix: readback should succeed");
1175 assert_eq!(
1176 out, contents,
1177 "mapped upload corrupted {len}-byte payload: readback != input"
1178 );
1179 }
1180 }
1181
1182 #[test]
1186 fn write_padded_into_mapped_zeroes_the_tail() {
1187 let mut mapped = [0xAAu8; 8];
1190 let bytes = [1u8, 2, 3, 4, 5];
1191 crate::padded_upload::write_padded_into_mapped(&mut mapped, &bytes)
1192 .expect("Fix: filling a large-enough mapped slice must succeed");
1193 assert_eq!(&mapped[..5], &bytes);
1194 assert_eq!(&mapped[5..], &[0u8, 0, 0], "alignment tail must be zeroed");
1195 let mut too_small = [0u8; 2];
1197 assert!(crate::padded_upload::write_padded_into_mapped(&mut too_small, &bytes).is_err());
1198 }
1199
1200 #[test]
1201 fn foreign_resident_handle_is_refused_not_resolved() {
1202 let foreign = ResidentOwner::new().expect("Fix: owner ids must be available");
1206 let native = resident_owner().expect("Fix: WGPU resident owner must be available");
1207 assert_ne!(foreign, native);
1208
1209 let error = GpuBufferHandle::from_resident_handle(foreign.handle(1), "refusal test")
1210 .expect_err("Fix: a foreign resident handle must never resolve to a WGPU buffer");
1211 let BackendError::InvalidProgram { fix } = error else {
1212 panic!("Fix: foreign resident handle refusal must be BackendError::InvalidProgram");
1213 };
1214 assert!(
1215 fix.contains("owned by backend instance") && fix.contains("Fix: "),
1216 "Fix: refusal must name the owning instance and carry actionable text, got {fix}"
1217 );
1218
1219 assert!(
1220 check_resident_owner(native.handle(u64::MAX), "refusal test").is_ok(),
1221 "Fix: a handle from this namespace must pass the owner check even when its buffer is gone"
1222 );
1223 }
1224
1225 #[test]
1226 fn resident_registry_handles_concurrent_lookup_and_drop() {
1227 let arc = crate::runtime::cached_device()
1228 .expect("Fix: GPU device is required for resident registry concurrency test");
1229 let (device, queue) = &*arc;
1230 let handle =
1231 GpuBufferHandle::upload(device, queue, &[1, 2, 3, 4], wgpu::BufferUsages::COPY_SRC)
1232 .expect("Fix: upload should register a resident buffer");
1233 let id = handle.id();
1234
1235 let readers = (0..8)
1239 .map(|_| {
1240 std::thread::spawn(move || {
1241 for _ in 0..1_000 {
1242 let resident = GpuBufferHandle::from_resident_id(id)
1243 .expect("Fix: resident id must resolve while handle is alive");
1244 assert_eq!(resident.id(), id);
1245 }
1246 })
1247 })
1248 .collect::<Vec<_>>();
1249 for reader in readers {
1250 reader
1251 .join()
1252 .expect("Fix: concurrent resident lookups must not panic");
1253 }
1254
1255 drop(handle);
1258 assert!(
1259 GpuBufferHandle::from_resident_id(id).is_none(),
1260 "dropped handles must be removed from the resident registry"
1261 );
1262 }
1263
1264 #[test]
1265 fn poisoned_staging_pool_lock_recovers_without_aborting_dispatch_path() {
1266 let pool = StagingBufferPool::new();
1267 let poisoned = pool.clone();
1268 let _ = std::thread::spawn(move || {
1269 let _guard = poisoned.lock_inner();
1270 panic!("poison staging buffer pool");
1271 })
1272 .join();
1273
1274 std::panic::catch_unwind(|| {
1275 let _ = pool.stats();
1276 })
1277 .expect("Fix: poisoned staging pool must recover so GPU readback pooling does not abort");
1278 }
1279
1280 #[test]
1281 fn poisoned_bind_group_cache_lock_recovers_without_aborting_dispatch_path() {
1282 let cache = BindGroupCache::new();
1283 let poisoned = cache.clone();
1284 let _ = std::thread::spawn(move || {
1285 let _guard = poisoned.lock_cache();
1286 panic!("poison bind group cache");
1287 })
1288 .join();
1289
1290 std::panic::catch_unwind(|| {
1291 let _ = cache.stats();
1292 })
1293 .expect("Fix: poisoned bind-group cache must recover so GPU dispatch does not abort");
1294 }
1295
1296 #[test]
1297 fn bind_group_cache_lru_heap_stays_capacity_scale() {
1298 let arc = crate::runtime::cached_device()
1299 .expect("Fix: GPU device is required for bind-group cache test");
1300 let (device, _) = &*arc;
1301 let layout = device.create_bind_group_layout(&wgpu::BindGroupLayoutDescriptor {
1302 label: Some("vyre bind-group cache lru test layout"),
1303 entries: &[wgpu::BindGroupLayoutEntry {
1304 binding: 0,
1305 visibility: wgpu::ShaderStages::COMPUTE,
1306 ty: wgpu::BindingType::Buffer {
1307 ty: wgpu::BufferBindingType::Storage { read_only: true },
1308 has_dynamic_offset: false,
1309 min_binding_size: wgpu::BufferSize::new(4),
1310 },
1311 count: None,
1312 }],
1313 });
1314 let buffer = device.create_buffer(&wgpu::BufferDescriptor {
1315 label: Some("vyre bind-group cache lru test buffer"),
1316 size: 4,
1317 usage: wgpu::BufferUsages::STORAGE,
1318 mapped_at_creation: false,
1319 });
1320 let bind_group = device.create_bind_group(&wgpu::BindGroupDescriptor {
1321 label: Some("vyre bind-group cache lru test bind group"),
1322 layout: &layout,
1323 entries: &[wgpu::BindGroupEntry {
1324 binding: 0,
1325 resource: buffer.as_entire_binding(),
1326 }],
1327 });
1328 let cache = BindGroupCache::with_cap(4);
1329
1330 for i in 0..64u64 {
1331 cache.insert_by_ids(1, &[i, 4], bind_group.clone());
1332 }
1333
1334 let inner = cache.lock_cache();
1335 assert_eq!(inner.entries.len(), 4);
1336 assert!(
1337 inner.lru.len() <= inner.entries.len().saturating_mul(4).max(8),
1338 "Fix: bind-group LRU heap must compact stale entries to cache-capacity scale"
1339 );
1340 }
1341
1342 #[test]
1361 fn bind_group_reuse_keys_on_buffer_identity_not_layout_alone() {
1362 let arc = crate::runtime::cached_device()
1363 .expect("Fix: GPU device is required for bind-group identity test");
1364 let (device, queue) = &*arc;
1365
1366 let layout = device.create_bind_group_layout(&wgpu::BindGroupLayoutDescriptor {
1367 label: Some("vyre bind-group identity test layout"),
1368 entries: &[wgpu::BindGroupLayoutEntry {
1369 binding: 0,
1370 visibility: wgpu::ShaderStages::COMPUTE,
1371 ty: wgpu::BindingType::Buffer {
1372 ty: wgpu::BufferBindingType::Storage { read_only: true },
1373 has_dynamic_offset: false,
1374 min_binding_size: wgpu::BufferSize::new(4),
1375 },
1376 count: None,
1377 }],
1378 });
1379
1380 let buffer_a =
1383 GpuBufferHandle::upload(device, queue, &[1u8; 4], wgpu::BufferUsages::STORAGE)
1384 .expect("Fix: upload of buffer a must succeed");
1385 let buffer_b =
1386 GpuBufferHandle::upload(device, queue, &[2u8; 4], wgpu::BufferUsages::STORAGE)
1387 .expect("Fix: upload of buffer b must succeed");
1388
1389 let cache = BindGroupCache::new();
1390 let layout_id = 1usize;
1391 let mut created = 0usize;
1392
1393 for handle in [
1396 &buffer_a, &buffer_a, &buffer_a, &buffer_b, &buffer_b, &buffer_b,
1397 ] {
1398 let slice = std::slice::from_ref(handle);
1399 cache.get_or_create(layout_id, slice, || {
1400 created += 1;
1401 device.create_bind_group(&wgpu::BindGroupDescriptor {
1402 label: Some("vyre bind-group identity test bind group"),
1403 layout: &layout,
1404 entries: &[wgpu::BindGroupEntry {
1405 binding: 0,
1406 resource: handle.buffer().as_entire_binding(),
1407 }],
1408 })
1409 });
1410 }
1411
1412 let stats = cache.stats();
1413 assert_eq!(
1414 created, 2,
1415 "six lookups over two distinct buffers must construct exactly two bind groups"
1416 );
1417 assert_eq!(
1418 stats.misses, 2,
1419 "each distinct buffer must miss exactly once. A layout-only key would \
1420 report 1 miss and share one bind group across both buffers, binding the \
1421 wrong resource on every dispatch against the second buffer."
1422 );
1423 assert_eq!(
1424 stats.hits, 4,
1425 "the two repeats of each buffer must reuse the cached bind group"
1426 );
1427 assert_eq!(stats.entries, 2, "one cached entry per distinct buffer");
1428
1429 let first = cache.get_or_create(layout_id, std::slice::from_ref(&buffer_a), || {
1432 panic!("Fix: buffer a is already cached and must not be rebuilt")
1433 });
1434 let again = cache.get_or_create(layout_id, std::slice::from_ref(&buffer_a), || {
1435 panic!("Fix: buffer a is already cached and must not be rebuilt")
1436 });
1437 assert!(
1438 Arc::ptr_eq(&first, &again),
1439 "repeated lookups for one buffer must hand back one bind-group instance"
1440 );
1441 }
1442}