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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum CompilerInfo {
63 Vulkan { params_transfer: ParamsTransfer },
64 Metal,
65 WGSL,
66 None,
67}
68
69#[derive(Debug)]
71pub struct WgpuServer<C: WgpuCompiler> {
72 pub(crate) device: wgpu::Device,
73 streams_pool: Vec<StreamId>,
75 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 shared_bindings_pool: LeasePool<SharedMemoryBindings>,
88 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 #[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 #[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 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 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 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 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 {
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 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 let mut shared_inputs = self.shared_bindings_pool.acquire();
445 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 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 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 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 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 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 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 self.scheduler.execute_streams(vec![stream_id]);
582 let stream = self.scheduler.stream(&stream_id);
583
584 stream.capturing.prepare()?;
585
586 stream.mem_manage.capture_begin();
593 Ok(())
594 }
595
596 fn begin_capture(&mut self, stream_id: StreamId) -> Result<(), ServerError> {
597 self.scheduler.execute_streams(vec![stream_id]);
600 let stream = self.scheduler.stream(&stream_id);
601
602 stream.capturing.begin()?;
603
604 if let Err(err) = stream.flush(StreamErrorMode {
608 ignore: false,
609 flush: true,
610 }) {
611 stream.mem_manage.capture_end();
615 stream.info_cache.capture_discard();
616 stream.capturing.abort();
617 return Err(err);
618 }
619
620 stream.mem_manage.capture_priming_end();
623 Ok(())
624 }
625
626 fn end_capture(&mut self, stream_id: StreamId) -> Result<GraphId, ServerError> {
627 self.scheduler.execute_streams(vec![stream_id]);
629 let stream = self.scheduler.stream(&stream_id);
630
631 stream.capturing.end()?;
634 let recording = stream.take_recording();
635 let mut retained = stream.mem_manage.capture_end();
636
637 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 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 self.scheduler.execute_streams(vec![stream_id]);
670
671 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 let Some(wgpu_graph) = self.graphs.remove(&graph) else {
693 return;
694 };
695 let stream = self.scheduler.stream(&stream_id);
696 let _ = stream
705 .flush(StreamErrorMode {
706 ignore: true,
707 flush: false,
708 })
709 .ok();
710 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}