Skip to main content

cubecl_wgpu/compute/
server.rs

1use std::collections::HashMap;
2use std::marker::PhantomData;
3
4use super::graph::WgpuGraph;
5use super::storage::{WgpuResource, WgpuStorage};
6use crate::WgpuCompiler;
7use crate::schedule::{BindingsResource, ScheduleTask, ScheduledWgpuBackend};
8use alloc::sync::Arc;
9use cubecl_common::pool::LeasePool;
10use cubecl_common::{
11    bytes::Bytes,
12    profile::{ProfileDuration, TimingMethod},
13};
14use cubecl_core::server::{BufferBinding, KernelResource, StreamErrorMode};
15use cubecl_core::zspace::Shape;
16use cubecl_core::{
17    MemoryConfiguration, WgpuCompilationOptions,
18    prelude::*,
19    server::{
20        CopyDescriptor, IoError, KernelArguments, LaunchError, ProfileError, ProfilingToken,
21        ServerCommunication, ServerError, ServerUtilities,
22    },
23    zspace::{Strides, strides},
24};
25use cubecl_environment::backtrace::BackTrace;
26use cubecl_environment::future::DynFut;
27#[cfg(feature = "spirv")]
28use cubecl_environment::persistence::Store;
29use cubecl_environment::stream::StreamId;
30use cubecl_ir::MemoryDeviceProperties;
31use cubecl_runtime::allocator::ContiguousMemoryLayoutPolicy;
32#[cfg(feature = "spirv")]
33use cubecl_runtime::compiler::{KernelCacheKey, compilation_store, store_compiled};
34use cubecl_runtime::memory_management::{
35    InstallMemoryPoolsError, ManagedMemoryHandle, MemoryReport, MemoryUsage, SharedMemoryBindings,
36};
37use cubecl_runtime::{
38    compiler::{CompilationCache, CubeTask},
39    config::{CubeClRuntimeConfig, RuntimeConfig},
40    dry_run::LaunchMode,
41    id::GraphId,
42    logging::ServerLogger,
43    memory_management::MemoryAllocationMode,
44    server::ComputeServer,
45    storage::ManagedResource,
46    stream::scheduler::{
47        SchedulerMultiStream, SchedulerMultiStreamOptions, SchedulerStrategy,
48        SchedulerStreamBackend,
49    },
50    validation::{validate_cube_dim, validate_units},
51};
52use wgpu::ComputePipeline;
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum ParamsTransfer {
56    Immediate,
57    Uniform,
58}
59
60/// Compiler kind and info used when compiling a specific kernel. Used to determine parameter passing strategies.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum CompilerInfo {
63    Vulkan { params_transfer: ParamsTransfer },
64    Metal,
65    WGSL,
66    None,
67}
68
69/// Wgpu compute server.
70#[derive(Debug)]
71pub struct WgpuServer<C: WgpuCompiler> {
72    pub(crate) device: wgpu::Device,
73    // A buffer that can be used to store stream id without extra allocations.
74    streams_pool: Vec<StreamId>,
75    /// The pipelines built so far, in front of the SPIR-V store when there is
76    /// one.
77    pipelines: CompilationCache<KernelId, (Arc<ComputePipeline>, CompilerInfo)>,
78    scheduler: SchedulerMultiStream<ScheduledWgpuBackend>,
79    #[cfg(feature = "spirv")]
80    pub(crate) spirv_cache: Option<Store<(u64, KernelCacheKey), cubecl_spirv::SpirvCacheEntry>>,
81    #[cfg(feature = "spirv")]
82    pub(crate) build_id: cubecl_common::hash::StableHash,
83    pub compilation_options: WgpuCompilationOptions,
84    pub(crate) backend: wgpu::Backend,
85    pub(crate) utilities: Arc<ServerUtilities<Self>>,
86    /// Reusable buffers for the cross-stream input bindings of each launch.
87    shared_bindings_pool: LeasePool<SharedMemoryBindings>,
88    /// Captured graphs owned by this server, keyed by the [`GraphId`] handed to
89    /// the client. `end_capture` inserts, `replay` looks up, `graph_destroy`
90    /// removes (dropping the [`WgpuGraph`] unpins the buffers it retained).
91    graphs: HashMap<GraphId, WgpuGraph>,
92    _compiler: PhantomData<C>,
93}
94
95impl<C: WgpuCompiler> ServerCommunication for WgpuServer<C> {
96    const SERVER_COMM_ENABLED: bool = false;
97}
98
99impl<C: WgpuCompiler> WgpuServer<C> {
100    /// Create a new server.
101    #[allow(clippy::too_many_arguments)]
102    pub fn new(
103        memory_properties: MemoryDeviceProperties,
104        memory_config: MemoryConfiguration,
105        compilation_options: WgpuCompilationOptions,
106        device: wgpu::Device,
107        queue: wgpu::Queue,
108        tasks_max: usize,
109        backend: wgpu::Backend,
110        timing_method: TimingMethod,
111        utilities: ServerUtilities<Self>,
112    ) -> Self {
113        #[cfg(feature = "spirv")]
114        let adapter_info = device.adapter_info();
115        let backend_scheduler = ScheduledWgpuBackend::new(
116            device.clone(),
117            queue.clone(),
118            memory_properties,
119            memory_config,
120            timing_method,
121            backend,
122            tasks_max,
123            utilities.logger.clone(),
124            compilation_options.supports_vulkan_compiler,
125        );
126
127        let config = CubeClRuntimeConfig::get();
128        let max_streams = config.streaming.max_streams;
129
130        #[cfg(feature = "spirv")]
131        let spirv_cache = compilation_store(
132            "vulkan",
133            format!("spirv_{}_{}", adapter_info.vendor, adapter_info.device),
134        );
135
136        // WGSL is compiled by the driver on every run, so without the SPIR-V
137        // store there is nothing persisted for a switch to invalidate.
138        #[cfg(feature = "spirv")]
139        let pipelines = CompilationCache::mirroring(&spirv_cache);
140        #[cfg(not(feature = "spirv"))]
141        let pipelines = CompilationCache::unbound();
142
143        Self {
144            compilation_options,
145            streams_pool: Vec::new(),
146            device,
147            pipelines,
148            scheduler: SchedulerMultiStream::new(
149                utilities.logger.clone(),
150                backend_scheduler,
151                SchedulerMultiStreamOptions {
152                    max_streams,
153                    max_tasks: tasks_max,
154                    strategy: SchedulerStrategy::Interleave,
155                },
156            ),
157            #[cfg(feature = "spirv")]
158            spirv_cache,
159            #[cfg(feature = "spirv")]
160            build_id: cubecl_runtime::compiler::build_id_hash(),
161            backend,
162            utilities: Arc::new(utilities),
163            shared_bindings_pool: LeasePool::with_capacity(tasks_max * max_streams as usize),
164            graphs: HashMap::new(),
165            _compiler: PhantomData,
166        }
167    }
168
169    fn prepare_bindings(
170        &mut self,
171        bindings: KernelArguments,
172        compiler_info: CompilerInfo,
173    ) -> Result<BindingsResource, IoError> {
174        // Store all the resources we'll be using. This could be eliminated if
175        // there was a way to tie the lifetime of the resource to the memory handle.
176        let mut resources = Vec::with_capacity(bindings.resources.len());
177
178        for resource in bindings.resources.into_iter() {
179            match resource {
180                KernelResource::Buffer(b) => {
181                    let stream = self.scheduler.stream(&b.stream);
182                    let resource = stream.mem_manage.get_resource(b)?;
183                    resources.push(resource);
184                }
185                KernelResource::TensorMap(_) => panic!("Tensor map not supported in wgpu"),
186            }
187        }
188
189        Ok(BindingsResource {
190            resources,
191            info: bindings.info,
192            compiler_info,
193        })
194    }
195
196    fn pipeline(
197        &mut self,
198        kernel: <Self as ComputeServer>::Kernel,
199        bindings: &KernelArguments,
200    ) -> Result<(Arc<ComputePipeline>, CompilerInfo), LaunchError> {
201        let kernel_id = kernel.id();
202        let mode = kernel_id.mode;
203
204        if let Some(pipeline) = self.pipelines.get(&kernel_id) {
205            return Ok(pipeline.clone());
206        }
207
208        let cached = self.load_cached_pipeline(&kernel_id, bindings, mode)?;
209
210        if let Some(Ok(pipeline)) = cached {
211            self.pipelines.insert(kernel_id, pipeline.clone());
212            return Ok(pipeline);
213        }
214
215        validate_cube_dim(&self.utilities.properties, &kernel_id)?;
216        validate_units(&self.utilities.properties, &kernel_id)?;
217
218        let definition = kernel.define();
219
220        let mut compiler = C::init(self.backend, &self.compilation_options);
221        let mut compiled = compiler.compile_kernel(self, kernel, definition)?;
222
223        if self.scheduler.logger.compilation_source_activated() {
224            compiled.debug_info = Some(DebugInformation::new(
225                compiler.lang_tag(),
226                kernel_id.clone(),
227            ));
228        }
229        self.scheduler.logger.log_compilation(&compiled);
230
231        compiler.validate_ir(&compiled.repr, &self.utilities.properties)?;
232        let (compiler_info, auto_repr) = compiler.normalize_repr(compiled.repr);
233        let repr = auto_repr.as_ref().map(|r| r.as_ref());
234
235        // /!\ Do not delete the following commented code.
236        // This is useful while working on the metal compiler.
237        // Also the errors are printed nicely which is not the case when this is the runtime
238        // that does it.
239        // {
240        //     // Write shader in metal file then compile it for error
241        //     std::fs::write("shader.metal", &compiled.source).expect("should write to file");
242        //     let status = std::process::Command::new("xcrun")
243        //         .args(vec![
244        //             "-sdk",
245        //             "macosx",
246        //             "metal",
247        //             "-o",
248        //             "shader.ir",
249        //             "-c",
250        //             "shader.metal",
251        //             "-w",
252        //         ])
253        //         .status()
254        //         .expect("should launch the command");
255        //     if !status.success() {
256        //         println!("SOURCE:\n{}", compiled.source);
257        //         std::process::exit(status.code().unwrap());
258        //     }
259        // }
260
261        let module = self.create_module(
262            &compiled.entrypoint_name,
263            kernel_id.cube_dim.into(),
264            repr,
265            &compiled.source,
266            mode,
267        )?;
268        let pipeline = self.create_pipeline(&compiled.entrypoint_name, repr, module, bindings);
269        self.pipelines
270            .insert(kernel_id.clone(), (pipeline.clone(), compiler_info));
271
272        #[cfg(feature = "spirv")]
273        if let Some(Err(key)) = cached
274            && let Some(crate::AutoRepresentation::SpirV(kernel)) = auto_repr
275        {
276            let cache = self.spirv_cache.as_mut().unwrap();
277            store_compiled(
278                cache,
279                key,
280                cubecl_spirv::SpirvCacheEntry::new(compiled.entrypoint_name, kernel),
281            );
282        }
283
284        Ok((pipeline, compiler_info))
285    }
286}
287
288impl<C: WgpuCompiler> ComputeServer for WgpuServer<C> {
289    type Kernel = Box<dyn CubeTask<C>>;
290    type Storage = WgpuStorage;
291    type MemoryLayoutPolicy = ContiguousMemoryLayoutPolicy;
292    type Info = wgpu::Backend;
293
294    fn logger(&self) -> Arc<ServerLogger> {
295        self.scheduler.logger.clone()
296    }
297
298    fn utilities(&self) -> Arc<ServerUtilities<Self>> {
299        self.utilities.clone()
300    }
301
302    fn staging(
303        &mut self,
304        _sizes: &[usize],
305        _stream_id: StreamId,
306    ) -> Result<Vec<Bytes>, ServerError> {
307        // TODO: Check if using a staging buffer is useful here.
308        Err(IoError::UnsupportedIoOperation {
309            backtrace: BackTrace::capture(),
310        }
311        .into())
312    }
313
314    fn initialize_memory(&mut self, memory: ManagedMemoryHandle, size: u64, stream_id: StreamId) {
315        let stream = self.scheduler.stream(&stream_id);
316        let reserved = stream
317            .empty(size)
318            .unwrap_or_else(|err| panic!("failed to reserve {size} bytes of device memory: {err}"));
319        stream.mem_manage.bind(reserved, memory);
320    }
321
322    fn read(
323        &mut self,
324        descriptors: Vec<CopyDescriptor>,
325        stream_id: StreamId,
326    ) -> DynFut<Result<Vec<Bytes>, ServerError>> {
327        // A read is a host sync: it cannot be recorded, and the recorded work
328        // has not executed, so there is nothing meaningful to read anyway.
329        if let Err(err) = self
330            .scheduler
331            .stream(&stream_id)
332            .reject_while_recording("read")
333        {
334            return Box::pin(async move { Err(err) });
335        }
336        let mut streams = vec![stream_id];
337        let mut resources = Vec::with_capacity(descriptors.len());
338        for desc in descriptors {
339            if contiguous_strides(&desc.shape) != desc.strides {
340                return Box::pin(async {
341                    Err(IoError::UnsupportedStrides {
342                        backtrace: BackTrace::capture(),
343                    }
344                    .into())
345                });
346            }
347            if !streams.contains(&desc.handle.stream) {
348                streams.push(desc.handle.stream);
349            }
350            let stream = self.scheduler.stream(&desc.handle.stream);
351            let resource = match stream.mem_manage.get_resource(desc.handle) {
352                Ok(val) => val,
353                Err(err) => return Box::pin(async move { Err(err.into()) }),
354            };
355            resources.push((resource, desc.shape, desc.elem_size));
356        }
357
358        self.scheduler.execute_streams(streams);
359
360        let stream = self.scheduler.stream(&stream_id);
361        stream.read_resources(resources)
362    }
363
364    fn write(&mut self, descriptors: Vec<(CopyDescriptor, Bytes)>, stream_id: StreamId) {
365        // Writes go on the queue, not the encoder — they cannot be recorded
366        // into a software graph (v1; CUDA records them as memcpy nodes).
367        // Reject them lazily so `end_capture` fails the capture instead of
368        // handing back a graph missing an operation.
369        {
370            let stream = self.scheduler.stream(&stream_id);
371            if let Err(err) = stream.reject_while_recording("write") {
372                stream.errors.push(err);
373                return;
374            }
375        }
376        for (desc, data) in descriptors {
377            let stream = self.scheduler.stream(&desc.handle.stream);
378
379            if contiguous_strides(&desc.shape) != desc.strides {
380                stream.error(ServerError::Io(IoError::UnsupportedStrides {
381                    backtrace: BackTrace::capture(),
382                }));
383                return;
384            }
385
386            let resource = match stream.mem_manage.get_resource(desc.handle) {
387                Ok(r) => r,
388                Err(err) => {
389                    stream.error(ServerError::Io(err));
390                    return;
391                }
392            };
393            let task = ScheduleTask::Write {
394                data,
395                buffer: resource,
396            };
397
398            self.scheduler.register(stream_id, task, &[]);
399        }
400    }
401
402    fn get_resource(
403        &mut self,
404        binding: BufferBinding,
405        stream_id: StreamId,
406    ) -> Result<ManagedResource<WgpuResource>, ServerError> {
407        let mut streams = vec![stream_id];
408        if binding.stream != stream_id {
409            streams.push(binding.stream);
410        }
411        self.scheduler.execute_streams(streams);
412        let stream = self.scheduler.stream(&binding.stream);
413        let memory = binding.memory.clone();
414        let resource = stream.mem_manage.get_resource(binding)?;
415
416        Ok(ManagedResource::new(memory, resource))
417    }
418
419    unsafe fn launch(
420        &mut self,
421        kernel: Self::Kernel,
422        count: CubeCount,
423        args: KernelArguments,
424        stream_id: StreamId,
425        launch_mode: LaunchMode,
426    ) {
427        let (pipeline, compiler_info) = match self.pipeline(kernel, &args) {
428            Ok(val) => val,
429            Err(err) => {
430                // We make the stream that would execute the kernel in error.
431                let stream = self.scheduler.stream(&stream_id);
432                stream.errors.push(ServerError::Launch(err));
433                return;
434            }
435        };
436
437        if launch_mode.is_skipped() {
438            return;
439        }
440
441        self.streams_pool.clear();
442        // Reuse a pooled buffer to avoid allocating on every launch; it returns to the pool
443        // automatically when the guard drops.
444        let mut shared_inputs = self.shared_bindings_pool.acquire();
445        // Pin the memory of every input that lives on another stream (released in `WgpuStream::flush`).
446        args.resources.iter().for_each(|resource| match resource {
447            KernelResource::Buffer(b) => {
448                self.streams_pool.push(b.stream);
449                if b.stream != stream_id {
450                    shared_inputs.push(b.memory.clone());
451                }
452            }
453            KernelResource::TensorMap(_) => {
454                panic!("Tensor maps not supported in WGPU")
455            }
456        });
457
458        let resources = match self.prepare_bindings(args, compiler_info) {
459            Ok(val) => val,
460            Err(err) => {
461                // We make the stream that would execute the kernel in error.
462                let stream = self.scheduler.stream(&stream_id);
463                stream.errors.push(ServerError::Io(err));
464                return;
465            }
466        };
467        let task = ScheduleTask::Execute {
468            pipeline,
469            count,
470            resources,
471            shared_inputs,
472        };
473
474        self.scheduler.register(stream_id, task, &self.streams_pool);
475    }
476
477    fn flush(&mut self, stream_id: StreamId) -> Result<(), ServerError> {
478        self.scheduler.execute_streams(vec![stream_id]);
479
480        let stream = self.scheduler.stream(&stream_id);
481
482        stream.flush(StreamErrorMode {
483            ignore: false,
484            flush: true,
485        })
486    }
487
488    /// Returns the total time of GPU work this sync completes.
489    fn sync(&mut self, stream_id: StreamId) -> DynFut<Result<(), ServerError>> {
490        if let Err(err) = self
491            .scheduler
492            .stream(&stream_id)
493            .reject_while_recording("sync")
494        {
495            return Box::pin(async move { Err(err) });
496        }
497        self.scheduler.execute_streams(vec![stream_id]);
498        let stream = self.scheduler.stream(&stream_id);
499
500        stream.sync()
501    }
502
503    fn start_profile(&mut self, stream_id: StreamId) -> Result<ProfilingToken, ServerError> {
504        // Recorded launches do not execute, so a profile of the window would
505        // measure nothing.
506        self.scheduler
507            .stream(&stream_id)
508            .reject_while_recording("start_profile")?;
509        self.scheduler.execute_streams(vec![stream_id]);
510        let stream = self.scheduler.stream(&stream_id);
511        stream.start_profile()
512    }
513
514    fn end_profile(
515        &mut self,
516        stream_id: StreamId,
517        token: ProfilingToken,
518    ) -> Result<ProfileDuration, ProfileError> {
519        self.scheduler.execute_streams(vec![stream_id]);
520        let stream = self.scheduler.stream(&stream_id);
521
522        stream.end_profile(token)
523    }
524
525    fn memory_usage(&mut self, stream_id: StreamId) -> Result<MemoryUsage, ServerError> {
526        self.scheduler.execute_streams(vec![stream_id]);
527        let stream = self.scheduler.stream(&stream_id);
528        Ok(stream.mem_manage.memory_usage())
529    }
530
531    fn memory_report(&mut self, stream_id: StreamId) -> Result<MemoryReport, ServerError> {
532        self.scheduler.execute_streams(vec![stream_id]);
533        let stream = self.scheduler.stream(&stream_id);
534        Ok(stream.mem_manage.memory_report())
535    }
536
537    fn stream_ids(&self) -> Vec<StreamId> {
538        self.scheduler.stream_ids().collect()
539    }
540
541    fn memory_cleanup(&mut self, stream_id: StreamId) {
542        self.scheduler.execute_streams(vec![stream_id]);
543        let stream = self.scheduler.stream(&stream_id);
544        // The info cache's buffers are live slices in the uniforms pool; an
545        // explicit cleanup exists to leave the pools empty, so every entry not
546        // pinned by a live graph goes too (entries are recreated on their next
547        // miss).
548        stream.info_cache.clear_unpinned();
549        stream.mem_manage.memory_cleanup(true);
550    }
551
552    fn allocation_mode(&mut self, mode: MemoryAllocationMode, stream_id: StreamId) {
553        self.scheduler.execute_streams(vec![stream_id]);
554        let stream = self.scheduler.stream(&stream_id);
555        stream.mem_manage.mode(mode);
556    }
557
558    fn install_memory_pools(
559        &mut self,
560        config: MemoryConfiguration,
561        stream_id: StreamId,
562    ) -> Result<(), InstallMemoryPoolsError> {
563        // Streams created from now on build their main pool with the new
564        // layout; memory is per stream, so already-created streams keep theirs.
565        self.scheduler
566            .backend_mut()
567            .factory()
568            .set_gpu_pools(config.clone());
569        let (_, props) = self.scheduler.backend_mut().factory().gpu_pools();
570
571        // The calling stream's pools are rebuilt in place, keeping the old
572        // layout when something is still live in them.
573        self.scheduler.execute_streams(vec![stream_id]);
574        let stream = self.scheduler.stream(&stream_id);
575        stream.mem_manage.install_memory_pools(config, &props)
576    }
577
578    fn graph_prepare(&mut self, stream_id: StreamId) -> Result<(), ServerError> {
579        // Drain queued tasks first so pre-capture work is not attributed to
580        // the capture window.
581        self.scheduler.execute_streams(vec![stream_id]);
582        let stream = self.scheduler.stream(&stream_id);
583
584        stream.capturing.prepare()?;
585
586        // Route every allocation from here until `end_capture` into the
587        // persistent pools and track the touched slices: warmup populates the
588        // pools with the capture run's full working set, the recorded run
589        // reuses those slices, and everything it touches is pinned to the
590        // graph at `end_capture`. The non-`NoCapture` state also isolates this
591        // stream in the scheduler (see `requires_isolation`).
592        stream.mem_manage.capture_begin();
593        Ok(())
594    }
595
596    fn begin_capture(&mut self, stream_id: StreamId) -> Result<(), ServerError> {
597        // Materialize the warmup work queued in the scheduler before the
598        // recording window opens.
599        self.scheduler.execute_streams(vec![stream_id]);
600        let stream = self.scheduler.stream(&stream_id);
601
602        stream.capturing.begin()?;
603
604        // Submit the warmup work and surface its queued errors now, so a
605        // warmup failure is reported here — where the diagnostic points at the
606        // cause — instead of failing `end_capture` later.
607        if let Err(err) = stream.flush(StreamErrorMode {
608            ignore: false,
609            flush: true,
610        }) {
611            // The capture never opened: disarm retention and return to
612            // `NoCapture`, so a failed `start_capture` leaves the stream fully
613            // usable and re-capturable.
614            stream.mem_manage.capture_end();
615            stream.info_cache.capture_discard();
616            stream.capturing.abort();
617            return Err(err);
618        }
619
620        // Warmup is over: release the slices it retained so the recorded run
621        // reuses them instead of growing the pools further.
622        stream.mem_manage.capture_priming_end();
623        Ok(())
624    }
625
626    fn end_capture(&mut self, stream_id: StreamId) -> Result<GraphId, ServerError> {
627        // Materialize the recorded launches still queued in the scheduler.
628        self.scheduler.execute_streams(vec![stream_id]);
629        let stream = self.scheduler.stream(&stream_id);
630
631        // The capture is over even on the failure path below, so an error here
632        // doesn't leave the stream stuck in capture/persistent state.
633        stream.capturing.end()?;
634        let recording = stream.take_recording();
635        let mut retained = stream.mem_manage.capture_end();
636
637        // An error queued during the window (a rejected write, a failed
638        // binding) means the recording is missing an operation: reject the
639        // capture rather than hand back a graph that silently skips work.
640        // `begin_capture` drained pre-window errors, so anything here arose
641        // inside the window.
642        let errors = stream.flush_errors_queue();
643        if !errors.is_empty() {
644            stream.info_cache.capture_discard();
645            return Err(ServerError::ServerUnhealthy {
646                errors,
647                backtrace: BackTrace::capture(),
648            });
649        }
650
651        let id = GraphId::new();
652        // Seal the info-cache entries this capture pinned under the graph's
653        // id, so `graph_destroy` can release them later.
654        stream.info_cache.capture_commit(id);
655        retained.extend(recording.uniform_pins);
656        self.graphs.insert(
657            id,
658            WgpuGraph {
659                tasks: recording.tasks,
660                _retained: retained,
661                _shared: recording.shared,
662            },
663        );
664        Ok(id)
665    }
666
667    fn replay(&mut self, graph: GraphId, stream_id: StreamId) {
668        // Order the replay after previously queued work on this stream.
669        self.scheduler.execute_streams(vec![stream_id]);
670
671        // Fire-and-forget like `launch`: on failure, push the error onto the
672        // stream's queue so it surfaces on the next flush/sync rather than
673        // blocking the caller here.
674        let Some(wgpu_graph) = self.graphs.get(&graph) else {
675            let stream = self.scheduler.stream(&stream_id);
676            stream.errors.push(ServerError::graph_state(
677                "replay was given an unknown or already-destroyed graph",
678            ));
679            return;
680        };
681        let stream = self.scheduler.stream(&stream_id);
682        if let Err(err) = stream.reject_while_recording("replay") {
683            stream.errors.push(err);
684            return;
685        }
686        stream.replay_graph(wgpu_graph);
687    }
688
689    fn graph_destroy(&mut self, graph: GraphId, stream_id: StreamId) {
690        // No-op for an unknown id (e.g. a double release). The graph is held
691        // until the end of this function, so its pins outlive the flush below.
692        let Some(wgpu_graph) = self.graphs.remove(&graph) else {
693            return;
694        };
695        let stream = self.scheduler.stream(&stream_id);
696        // Submit any replay still sitting in the encoder before the pins drop:
697        // a `queue.write_buffer` onto a reclaimed slice runs at the *next*
698        // submit, ahead of everything already in the encoder, so it would reach
699        // the GPU before the still-unsubmitted replay that reads it. The `Write`
700        // path flushes on its own account, so this covers the writes that do
701        // not — the uniform uploads in `create_uniform`/`info_uniform`. Once the
702        // replay is submitted, queue ordering makes releasing the slices safe
703        // with no host sync, unlike CUDA.
704        let _ = stream
705            .flush(StreamErrorMode {
706                ignore: true,
707                flush: false,
708            })
709            .ok();
710        // Release the info-cache entries this graph pinned; entries no other
711        // live graph still pins are dropped, freeing their buffers.
712        stream.info_cache.graph_release(graph);
713        drop(wgpu_graph);
714    }
715}
716
717pub(crate) fn contiguous_strides(shape: &Shape) -> Strides {
718    let rank = shape.len();
719    let mut strides = strides![1; rank];
720    for i in (0..rank - 1).rev() {
721        strides[i] = strides[i + 1] * shape[i + 1];
722    }
723    strides
724}