vello_cpu/dispatch/
multi_threaded.rs

1// Copyright 2025 the Vello Authors
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4use crate::coarse::CommandBucketer;
5use crate::coarse::depth::DepthBuffer;
6use crate::dispatch::Dispatcher;
7use crate::dispatch::multi_threaded::cost::{COST_THRESHOLD, estimate_render_task_cost};
8use crate::dispatch::multi_threaded::worker::Worker;
9use crate::filter::context::FilterContext;
10use crate::fine::{Fine, FineKernel, FineRenderParams, FineResources, rasterize_region};
11use crate::kurbo::{Affine, BezPath, PathEl, Point, Rect, Stroke};
12use crate::peniko::{BlendMode, Fill};
13use crate::record::RecordedFill;
14use crate::region::Regions;
15use crate::{CompositeMode, RasterizerSettings};
16use alloc::boxed::Box;
17use alloc::sync::Arc;
18use alloc::vec;
19use alloc::vec::Vec;
20use core::fmt::{Debug, Formatter};
21use crossbeam_channel::TryRecvError;
22use rayon::{ThreadPool, ThreadPoolBuilder};
23use std::cell::RefCell;
24use std::ops::Range;
25use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
26use std::sync::{Barrier, Mutex};
27use thread_local::ThreadLocal;
28use vello_common::clip::ClipContext;
29use vello_common::encode::EncodedPaint;
30use vello_common::fearless_simd::{Level, Simd, dispatch};
31use vello_common::filter::FilterData;
32use vello_common::geometry::RectU16;
33use vello_common::mask::Mask;
34use vello_common::paint::{ImageResolver, Paint};
35use vello_common::pixmap::PixmapMut;
36use vello_common::record::{CommandRecorder, LayerClip, LayerProps, PoppedLayer};
37use vello_common::strip::Strip;
38use vello_common::strip_generator::{GenerationMode, StripGenerator, StripStorage};
39
40mod cost;
41mod worker;
42
43type RenderTaskSender = crossbeam_channel::Sender<RenderTask>;
44type RecordedCommandSender = ordered_channel::Sender<RecordedCommandTask>;
45type RecordedCommandReceiver = ordered_channel::Receiver<RecordedCommandTask>;
46
47/// A dispatcher for multi-threaded rendering.
48///
49/// A small note for future contributors: Unfortunately, the logic of this dispatcher as well as
50/// the lifecycle of the different fields of the dispatcher can be a bit hard to grasp.
51/// The reason for this is that since we have to do a lot of communication across the thread boundary,
52/// we have to work with lots of `Option` and `core::mem::take` operations, to ensure that we are
53/// not needlessly cloning objects.
54///
55/// The below comments will hopefully help with understanding the overall structure and lifecycles
56/// a bit better.
57pub(crate) struct MultiThreadedDispatcher {
58    bucketer: Mutex<CommandBucketer>,
59    clip_context: ClipContext,
60    recorder: CommandRecorder<RecordedFill>,
61    strip_storage: StripStorage,
62    /// The thread pool that is used for dispatching tasks.
63    thread_pool: ThreadPool,
64    allocation_group: AllocationGroup,
65    /// The cost of the current batch.
66    batch_cost: f32,
67    /// The sender used to dispatch new rendering tasks from the main thread.
68    ///
69    /// This field will be set once we call the `init` method.
70    /// This field will be set back to `None` when running `flush` to drop the value and thus
71    /// indicate to receivers that no more rendering tasks will be dispatched from that point onward.
72    task_sender: Option<RenderTaskSender>,
73    /// Contains one worker object for each thread.
74    ///
75    /// The workers will be initialized once when building the multi-threaded dispatcher via
76    /// `MultiThreadedDispatcher::new`.
77    workers: Arc<ThreadLocal<RefCell<Worker>>>,
78    /// The receiver for commands generated by worker threads, used to record them on the main thread.
79    ///
80    /// Similarly to `task_sender`, this value is set to `None` initially, and will only be set once
81    /// we actually call the `init` method after registering a task.
82    recorded_command_receiver: Option<RecordedCommandReceiver>,
83    /// The storage for alpha values.
84    ///
85    /// Similarly to the single-threaded dispatcher, we want to be able to reuse the allocation holding
86    /// the alpha values across multiple runs of `reset`. However, we have the problem that during path
87    /// rendering, each thread needs to have its own allocation. We also need to be able to move
88    /// the allocation back and forth between the threads (during path rendering) and the main thread
89    /// (during fine rasterization). Because of this, we wrap it in this `MaybePresent` struct.
90    ///
91    /// During initialization, each thread will "take" the vector allocation out of its slot
92    /// (the vector has a length of `num_threads`, so each thread has a slot belonging to itself)
93    /// and will put it back to its slot after flushing. Then, during fine rasterization, we
94    /// take all slots out of the `MaybePresent` object so that we can easily access each buffer
95    /// when running the commands without having to go through the mutex. After fine rasterization,
96    /// the slots are put back into the `MaybePresent` object.
97    ///
98    alpha_storage: MaybePresent<Vec<Vec<u8>>>,
99    /// The task index that will be assigned to the next rendering task.
100    ///
101    /// Since we are rendering the paths on different threads, we need to make sure that they
102    /// come back in the right order. The `task_idx` is used to keep track of that order.
103    task_idx: u32,
104    /// The number of threads active in the thread pool.
105    num_threads: u16,
106    /// The strip generator for the main thread, used for clip path rasterization.
107    strip_generator: StripGenerator,
108    level: Level,
109    flushed: bool,
110    // So that we can reuse memory allocations across different runs.
111    allocations: Allocations,
112    layer_depth: usize,
113}
114
115impl MultiThreadedDispatcher {
116    pub(crate) fn new(width: u16, height: u16, num_threads: u16, level: Level) -> Self {
117        let thread_pool = ThreadPoolBuilder::new()
118            .num_threads(num_threads as usize)
119            .build()
120            .unwrap();
121        let alpha_storage = MaybePresent::new(vec![vec![]; usize::from(num_threads)]);
122        let workers = Arc::new(ThreadLocal::new());
123
124        {
125            let thread_ids = Arc::new(AtomicU8::new(0));
126            let workers = workers.clone();
127
128            // Create all workers once in `new`, so that later on we can just call`.get().unwrap()`.
129            thread_pool.spawn_broadcast(move |_| {
130                let thread_id = thread_ids.fetch_add(1, Ordering::SeqCst);
131                let worker = Worker::new(width, height, thread_id, level);
132
133                let _ = workers.get_or(|| RefCell::new(worker));
134            });
135        }
136
137        let task_idx = 0;
138        let batch_cost = 0.0;
139        let flushed = true;
140
141        Self {
142            bucketer: Mutex::new(CommandBucketer::from_wh(width, height)),
143            thread_pool,
144            allocations: Allocations::default(),
145            allocation_group: AllocationGroup::default(),
146            batch_cost,
147            task_idx,
148            flushed,
149            workers,
150            clip_context: ClipContext::new(),
151            recorder: CommandRecorder::new(width, height),
152            task_sender: None,
153            recorded_command_receiver: None,
154            strip_generator: StripGenerator::new(width, height, level),
155            strip_storage: StripStorage::new(GenerationMode::Append),
156            level,
157            alpha_storage,
158            num_threads,
159            layer_depth: 0,
160        }
161    }
162
163    #[cfg(feature = "f32_pipeline")]
164    fn rasterize_f32(
165        &self,
166        target: PixmapMut<'_>,
167        scene_width: u16,
168        scene_height: u16,
169        settings: RasterizerSettings,
170        encoded_paints: &[EncodedPaint],
171        image_resolver: &dyn ImageResolver,
172    ) {
173        use crate::fine::F32Kernel;
174        dispatch!(self.level, simd => self.rasterize_with::<_, F32Kernel>(simd, target, scene_width, scene_height, settings, encoded_paints, image_resolver));
175    }
176
177    #[cfg(feature = "u8_pipeline")]
178    fn rasterize_u8(
179        &self,
180        target: PixmapMut<'_>,
181        scene_width: u16,
182        scene_height: u16,
183        settings: RasterizerSettings,
184        encoded_paints: &[EncodedPaint],
185        image_resolver: &dyn ImageResolver,
186    ) {
187        use crate::fine::U8Kernel;
188        dispatch!(self.level, simd => self.rasterize_with::<_, U8Kernel>(simd, target, scene_width, scene_height, settings, encoded_paints, image_resolver));
189    }
190
191    fn init(&mut self) {
192        let (render_task_sender, render_task_receiver) = crossbeam_channel::unbounded();
193        let (recorded_command_sender, recorded_command_receiver) = ordered_channel::unbounded();
194        let workers = self.workers.clone();
195        let alpha_storage = self.alpha_storage.clone();
196
197        self.task_sender = Some(render_task_sender);
198        self.recorded_command_receiver = Some(recorded_command_receiver);
199
200        // Spawn the loop for the worker threads.
201        self.thread_pool.spawn_broadcast(move |_| {
202            let render_task_receiver = render_task_receiver.clone();
203            let mut recorded_command_sender = recorded_command_sender.clone();
204            let worker = workers.get().unwrap();
205            let mut worker = worker.borrow_mut();
206            let thread_id = worker.thread_id();
207
208            // Take out the allocation for alphas and store it in the worker.
209            alpha_storage
210                .with_inner(|alphas| worker.init(std::mem::take(&mut alphas[thread_id as usize])));
211
212            while let Ok(task) = render_task_receiver.recv() {
213                worker.run_render_task(task, &mut recorded_command_sender);
214            }
215
216            // If we reach this point, it means the `task_sender` has been dropped by the main thread
217            // and no more tasks are available (since we flushed).
218            // So we are done, and just need to place the alphas of the worker thread back into the
219            // vector.
220
221            alpha_storage.with_inner(|alphas| {
222                alphas[thread_id as usize] = worker.finalize();
223            });
224
225            // Then, we drop the `recorded_command_sender`. Once all worker threads have
226            // dropped their `recorded_command_sender`, the main thread knows that all workers are done
227            // and all alphas have been placed, so it's safe to proceed.
228            drop(recorded_command_sender);
229        });
230    }
231
232    fn register_task(&mut self, task: RenderTaskType) {
233        self.flushed = false;
234        if self.task_sender.is_none() {
235            self.init();
236        }
237
238        let cost = estimate_render_task_cost(&task, &self.allocation_group.path);
239        self.allocation_group.render_tasks.push(task);
240        self.batch_cost += cost;
241
242        if self.batch_cost > COST_THRESHOLD {
243            self.flush_tasks();
244        }
245    }
246
247    fn flush_tasks(&mut self) {
248        if self.allocation_group.render_tasks.is_empty() {
249            return;
250        }
251
252        self.send_pending_tasks();
253
254        self.batch_cost = 0.0;
255    }
256
257    fn bump_task_idx(&mut self) -> u32 {
258        let idx = self.task_idx;
259        self.task_idx += 1;
260        idx
261    }
262
263    fn send_pending_tasks(&mut self) {
264        let task_idx = self.bump_task_idx();
265        let allocation_group =
266            std::mem::replace(&mut self.allocation_group, self.allocations.get());
267        let task_sender = self.task_sender.as_mut().unwrap();
268        let clip_path = self.clip_context.get().map(|c| OwnedClip {
269            strips: c.strips.into(),
270            alphas: c.alphas.into(),
271            bbox: c.bbox,
272        });
273        let task = RenderTask {
274            idx: task_idx,
275            clip_path,
276            allocation_group,
277        };
278        task_sender.send(task).unwrap();
279        self.record_finished_commands(true);
280    }
281
282    // Currently, we record worker-generated commands in two phases:
283    //
284    // The first phase is when we are still processing new draw commands from the client. After each
285    // command, we check whether there are already any generated strips, and if so we register the
286    // recorded commands on the main thread. In this case, we want to abort in case there are no more
287    // commands available to process.
288    //
289    // The second phase is when we are flushing, in which case. even if the queue is empty, we only
290    // want to abort once all workers have closed the channel (and thus there won't be any more
291    // new recoded commands that will be generated).
292    //
293    // This is why we have the `abort_empty` flag.
294    fn append_strips(&mut self, strips: &[Strip]) -> Range<usize> {
295        // TODO: Maybe we shouldn't do this and just have each worker store their own
296        // strips.
297        let start = self.strip_storage.strips.len();
298        self.strip_storage.strips.extend_from_slice(strips);
299        start..self.strip_storage.strips.len()
300    }
301
302    fn record_finished_commands(&mut self, abort_empty: bool) {
303        loop {
304            match self.recorded_command_receiver.as_mut().unwrap().try_recv() {
305                Ok(mut task) => {
306                    let num_tasks = task.allocation_group.recorded_commands.len();
307                    for cmd in task.allocation_group.recorded_commands.drain(0..num_tasks) {
308                        match cmd {
309                            RecordedCommand::RenderPath {
310                                strips: strip_range,
311                                paint,
312                                blend_mode,
313                                thread_id,
314                                mask,
315                            } => {
316                                let strip_range = self.append_strips(
317                                    &task.allocation_group.strips
318                                        [strip_range.start as usize..strip_range.end as usize],
319                                );
320                                let strips =
321                                    &self.strip_storage.strips[strip_range.start..strip_range.end];
322                                let draw = RecordedFill::new(
323                                    thread_id,
324                                    strip_range,
325                                    paint.clone(),
326                                    blend_mode,
327                                    mask,
328                                );
329
330                                self.recorder.push_draw(draw, strips);
331                            }
332                            RecordedCommand::PushLayer {
333                                thread_id,
334                                clip_path,
335                                clip_bbox,
336                                blend_mode,
337                                mask,
338                                opacity,
339                            } => {
340                                let clip_path = clip_path.map(|strip_range| {
341                                    let strip_range = self.append_strips(
342                                        &task.allocation_group.strips
343                                            [strip_range.start as usize..strip_range.end as usize],
344                                    );
345                                    LayerClip {
346                                        strip_range,
347                                        thread_idx: thread_id,
348                                        bbox: clip_bbox.unwrap(),
349                                    }
350                                });
351
352                                self.recorder.push_layer(
353                                    LayerProps {
354                                        blend_mode,
355                                        opacity,
356                                        mask,
357                                        clip_path,
358                                    },
359                                    None,
360                                );
361                            }
362                            RecordedCommand::PopLayer => match self.recorder.pop_layer() {
363                                PoppedLayer::Regular => {}
364                                PoppedLayer::Filter => {
365                                    unreachable!("filters are not supported by MT")
366                                }
367                            },
368                        }
369                    }
370
371                    // Put the allocation group back so it can be reused in future iterations!
372                    self.allocations.put(task.allocation_group);
373                }
374                Err(e) => match e {
375                    TryRecvError::Empty => {
376                        if abort_empty {
377                            return;
378                        }
379                    }
380                    TryRecvError::Disconnected => return,
381                },
382            }
383        }
384    }
385
386    // No need to vectorize here, as vectorization happens in each of the
387    // functions that are called within.
388    fn rasterize_with<S: Simd, F: FineKernel<S>>(
389        &self,
390        simd: S,
391        mut target: PixmapMut<'_>,
392        scene_width: u16,
393        scene_height: u16,
394        settings: RasterizerSettings,
395        encoded_paints: &[EncodedPaint],
396        image_resolver: &dyn ImageResolver,
397    ) {
398        let mut bucketer = self.bucketer.lock().unwrap();
399        let filters = FilterContext::new(0);
400        bucketer.reset(RectU16::new(0, 0, scene_width, scene_height));
401        bucketer.bucket_commands(
402            &self.recorder.nodes,
403            &self.recorder.draws,
404            &self.recorder.layers,
405            &self.strip_storage.strips,
406            encoded_paints,
407            &filters,
408        );
409
410        let alpha_slots = self.alpha_storage.take();
411        {
412            let alpha_buffers = alpha_slots.iter().map(Vec::as_slice).collect::<Vec<_>>();
413            let use_src_over = settings.composite_mode == CompositeMode::SrcOver;
414            let resources = FineResources {
415                alpha_buffers: &alpha_buffers,
416                encoded_paints,
417                filter_paints: &bucketer.filter_paints,
418                image_resolver,
419            };
420            let params = FineRenderParams {
421                scene_size: (scene_width, scene_height),
422                target_offset: settings.offset,
423            };
424
425            let mut regions = Regions::new(
426                &mut target,
427                params.scene_size,
428                params.target_offset,
429                bucketer.rows().len(),
430            );
431            let fines = ThreadLocal::new();
432            self.thread_pool.install(|| {
433                regions.update_par(|region| {
434                    let mut fine = fines
435                        .get_or(|| {
436                            RefCell::new((
437                                Fine::<S, F>::new(simd, bucketer.width()),
438                                DepthBuffer::new(bucketer.width()),
439                            ))
440                        })
441                        .borrow_mut();
442                    let (fine, depth) = &mut *fine;
443
444                    rasterize_region::<S, F>(
445                        fine,
446                        depth,
447                        region,
448                        &bucketer,
449                        resources,
450                        use_src_over,
451                    );
452                });
453            });
454        }
455
456        self.alpha_storage.init(alpha_slots);
457    }
458}
459
460impl Dispatcher for MultiThreadedDispatcher {
461    fn has_layers(&self) -> bool {
462        self.layer_depth != 0
463    }
464
465    fn fill_path(
466        &mut self,
467        path: &BezPath,
468        fill_rule: Fill,
469        transform: Affine,
470        paint: Paint,
471        blend_mode: BlendMode,
472        aliasing_threshold: Option<u8>,
473        mask: Option<Mask>,
474    ) {
475        let start = self.allocation_group.path.len() as u32;
476        self.allocation_group.path.extend(path);
477        let end = self.allocation_group.path.len() as u32;
478        self.register_task(RenderTaskType::FillPath {
479            path_range: start..end,
480            transform,
481            paint,
482            fill_rule,
483            blend_mode,
484            aliasing_threshold,
485            mask,
486        });
487    }
488
489    fn stroke_path(
490        &mut self,
491        path: &BezPath,
492        stroke: &Stroke,
493        transform: Affine,
494        paint: Paint,
495        blend_mode: BlendMode,
496        aliasing_threshold: Option<u8>,
497        mask: Option<Mask>,
498    ) {
499        let start = self.allocation_group.path.len() as u32;
500        self.allocation_group.path.extend(path);
501        let end = self.allocation_group.path.len() as u32;
502        self.register_task(RenderTaskType::StrokePath {
503            path_range: start..end,
504            transform,
505            paint,
506            stroke: stroke.clone(),
507            blend_mode,
508            aliasing_threshold,
509            mask,
510        });
511    }
512
513    fn fill_rect_fast(
514        &mut self,
515        rect: &Rect,
516        paint: Paint,
517        blend_mode: BlendMode,
518        mask: Option<Mask>,
519    ) {
520        // For multi-threaded, fall back to path-based rendering.
521        // TODO: Implement optimized rect strip generation in worker threads.
522        let start = self.allocation_group.path.len() as u32;
523        self.allocation_group.path.extend([
524            PathEl::MoveTo(Point::new(rect.x0, rect.y0)),
525            PathEl::LineTo(Point::new(rect.x1, rect.y0)),
526            PathEl::LineTo(Point::new(rect.x1, rect.y1)),
527            PathEl::LineTo(Point::new(rect.x0, rect.y1)),
528            PathEl::ClosePath,
529        ]);
530        let end = self.allocation_group.path.len() as u32;
531        self.register_task(RenderTaskType::FillPath {
532            path_range: start..end,
533            transform: Affine::IDENTITY,
534            paint,
535            fill_rule: Fill::NonZero,
536            blend_mode,
537            aliasing_threshold: None,
538            mask,
539        });
540    }
541
542    fn push_layer(
543        &mut self,
544        clip_path: Option<&BezPath>,
545        fill_rule: Fill,
546        clip_transform: Affine,
547        blend_mode: BlendMode,
548        opacity: f32,
549        aliasing_threshold: Option<u8>,
550        mask: Option<Mask>,
551        filter_data: Option<FilterData>,
552    ) {
553        // TODO: Implement filter support in multi-threaded dispatcher.
554        // The single-threaded dispatcher has full filter support, but multi-threaded needs
555        // additional infrastructure for cross-thread layer coordination.
556        if filter_data.is_some() {
557            unimplemented!("Filter effects are not yet supported in multi-threaded rendering");
558        }
559
560        let clip_path = clip_path.map(|c| {
561            let start = self.allocation_group.path.len() as u32;
562            self.allocation_group.path.extend(c);
563            let end = self.allocation_group.path.len() as u32;
564
565            (start..end, clip_transform)
566        });
567
568        self.register_task(RenderTaskType::PushLayer {
569            clip_path,
570            blend_mode,
571            opacity,
572            mask,
573            fill_rule,
574            aliasing_threshold,
575        });
576        self.layer_depth += 1;
577    }
578
579    fn pop_layer(&mut self) {
580        self.register_task(RenderTaskType::PopLayer);
581        self.layer_depth = self
582            .layer_depth
583            .checked_sub(1)
584            .expect("layer stack underflow");
585    }
586
587    fn reset(&mut self, width: u16, height: u16) {
588        self.flush();
589
590        // Bucketer will be reset lazily during rasterization with the active viewport.
591        self.clip_context.reset();
592        self.recorder.reset(width, height);
593        self.strip_storage.clear();
594        self.allocation_group.clear();
595        self.batch_cost = 0.0;
596        self.task_idx = 0;
597        self.layer_depth = 0;
598        self.task_sender = None;
599        self.recorded_command_receiver = None;
600        self.strip_generator.reset(width, height);
601        self.alpha_storage.with_inner(|alphas| {
602            for alpha in alphas {
603                alpha.clear();
604            }
605        });
606
607        let workers = self.workers.clone();
608        // + 1 since we also wait on the main thread.
609        let barrier = Arc::new(Barrier::new(usize::from(self.num_threads) + 1));
610        let t_barrier = barrier.clone();
611
612        self.thread_pool.spawn_broadcast(move |_| {
613            let worker = workers.get().unwrap();
614            let mut borrowed = worker.borrow_mut();
615            borrowed.reset(width, height);
616            t_barrier.wait();
617        });
618
619        barrier.wait();
620    }
621
622    fn flush(&mut self) {
623        if self.flushed {
624            return;
625        }
626
627        self.flush_tasks();
628        let sender = core::mem::take(&mut self.task_sender);
629        // Note that dropping the sender will signal to the workers that no more new paths
630        // can arrive.
631        drop(sender);
632        self.record_finished_commands(false);
633
634        self.flushed = true;
635    }
636
637    fn rasterize(
638        &self,
639        target: PixmapMut<'_>,
640        scene_width: u16,
641        scene_height: u16,
642        settings: RasterizerSettings,
643        encoded_paints: &[EncodedPaint],
644        image_resolver: &dyn ImageResolver,
645    ) {
646        assert!(self.flushed, "attempted to rasterize before flushing");
647
648        // Only u8 pipeline enabled
649        #[cfg(all(feature = "u8_pipeline", not(feature = "f32_pipeline")))]
650        {
651            self.rasterize_u8(
652                target,
653                scene_width,
654                scene_height,
655                settings,
656                encoded_paints,
657                image_resolver,
658            );
659        }
660        // Only f32 pipeline enabled
661        #[cfg(all(feature = "f32_pipeline", not(feature = "u8_pipeline")))]
662        {
663            self.rasterize_f32(
664                target,
665                scene_width,
666                scene_height,
667                settings,
668                encoded_paints,
669                image_resolver,
670            );
671        }
672
673        // Both pipelines enabled
674        #[cfg(all(feature = "f32_pipeline", feature = "u8_pipeline"))]
675        match settings.render_mode {
676            crate::RenderMode::OptimizeSpeed => {
677                self.rasterize_u8(
678                    target,
679                    scene_width,
680                    scene_height,
681                    settings,
682                    encoded_paints,
683                    image_resolver,
684                );
685            }
686            crate::RenderMode::OptimizeQuality => {
687                self.rasterize_f32(
688                    target,
689                    scene_width,
690                    scene_height,
691                    settings,
692                    encoded_paints,
693                    image_resolver,
694                );
695            }
696        }
697    }
698
699    fn push_clip_path(
700        &mut self,
701        path: &BezPath,
702        fill_rule: Fill,
703        transform: Affine,
704        aliasing_threshold: Option<u8>,
705    ) {
706        self.flush_tasks();
707        self.clip_context.push_clip(
708            path.iter(),
709            &mut self.strip_generator,
710            fill_rule,
711            transform,
712            aliasing_threshold,
713        );
714    }
715
716    fn pop_clip_path(&mut self) {
717        self.flush_tasks();
718        self.clip_context.pop_clip();
719    }
720
721    fn is_multi_threaded(&self) -> bool {
722        true
723    }
724}
725
726impl Debug for MultiThreadedDispatcher {
727    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
728        f.write_str("MultiThreadedDispatcher { .. }")
729    }
730}
731
732impl Drop for MultiThreadedDispatcher {
733    fn drop(&mut self) {
734        self.flush();
735    }
736}
737
738#[derive(Debug)]
739pub(crate) struct OwnedClip {
740    strips: Box<[Strip]>,
741    alphas: Box<[u8]>,
742
743    /// A coarse bounding box of the clip path in pixel coordinates.
744    ///
745    /// These bounds have already been intersected with the viewport.
746    bbox: RectU16,
747}
748
749/// A structure that allows storing and fetching existing allocations.
750struct AllocationManager<T> {
751    entries: Vec<Vec<T>>,
752}
753
754impl<T> AllocationManager<T> {
755    /// Get a new vector allocation.
756    ///
757    /// The vector is guaranteed to have been cleared before.
758    fn get(&mut self) -> Vec<T> {
759        self.entries.pop().unwrap_or_default()
760    }
761
762    /// Insert a new allocation in the store.
763    fn put(&mut self, mut allocation: Vec<T>) {
764        allocation.clear();
765        self.entries.push(allocation);
766    }
767}
768
769impl<T> Default for AllocationManager<T> {
770    fn default() -> Self {
771        Self { entries: vec![] }
772    }
773}
774
775/// A structure to keep track of allocations that will be done while rendering with
776/// multi-threading.
777#[derive(Default)]
778struct Allocations {
779    /// The render tasks of a batch. They will be filled by the main thread as new fill/stroke commands
780    /// come in and consumed by worker threads as they process them.
781    render_tasks: AllocationManager<RenderTaskType>,
782    /// The path store of a batch. It will be filled by the main thread as new commands come in
783    /// and be used by the worker thread to generate the strips of a path.
784    paths: AllocationManager<PathEl>,
785    /// Stores allocations that are used by the worker thread to produce strips. They will be
786    /// sent back to the main thread which then records the corresponding commands.
787    strips: AllocationManager<Strip>,
788    /// The commands produced by a worker thread, which will be recorded by the main thread.
789    recorded_commands: AllocationManager<RecordedCommand>,
790}
791
792impl Allocations {
793    /// Return a new allocation group.
794    ///
795    /// The group is guaranteed to have been cleared.
796    fn get(&mut self) -> AllocationGroup {
797        let render_tasks = self.render_tasks.get();
798        let path = self.paths.get();
799        let strips = self.strips.get();
800        let recorded_commands = self.recorded_commands.get();
801
802        AllocationGroup {
803            path,
804            render_tasks,
805            recorded_commands,
806            strips,
807        }
808    }
809
810    fn put(&mut self, allocation: AllocationGroup) {
811        self.render_tasks.put(allocation.render_tasks);
812        self.paths.put(allocation.path);
813        self.strips.put(allocation.strips);
814        self.recorded_commands.put(allocation.recorded_commands);
815    }
816}
817
818#[derive(Default, Debug)]
819pub(crate) struct AllocationGroup {
820    pub(crate) path: Vec<PathEl>,
821    pub(crate) render_tasks: Vec<RenderTaskType>,
822    pub(crate) strips: Vec<Strip>,
823    pub(crate) recorded_commands: Vec<RecordedCommand>,
824}
825
826impl AllocationGroup {
827    fn clear(&mut self) {
828        self.path.clear();
829        self.render_tasks.clear();
830        self.strips.clear();
831        self.recorded_commands.clear();
832    }
833}
834
835#[derive(Debug)]
836pub(crate) struct RenderTask {
837    pub(crate) idx: u32,
838    pub(crate) clip_path: Option<OwnedClip>,
839    pub(crate) allocation_group: AllocationGroup,
840}
841
842#[derive(Debug, Clone)]
843pub(crate) enum RenderTaskType {
844    FillPath {
845        path_range: Range<u32>,
846        transform: Affine,
847        paint: Paint,
848        fill_rule: Fill,
849        blend_mode: BlendMode,
850        aliasing_threshold: Option<u8>,
851        mask: Option<Mask>,
852    },
853    StrokePath {
854        path_range: Range<u32>,
855        transform: Affine,
856        paint: Paint,
857        stroke: Stroke,
858        blend_mode: BlendMode,
859        aliasing_threshold: Option<u8>,
860        mask: Option<Mask>,
861    },
862    PushLayer {
863        clip_path: Option<(Range<u32>, Affine)>,
864        blend_mode: BlendMode,
865        opacity: f32,
866        mask: Option<Mask>,
867        fill_rule: Fill,
868        aliasing_threshold: Option<u8>,
869    },
870    PopLayer,
871}
872
873pub(crate) struct RecordedCommandTask {
874    allocation_group: AllocationGroup,
875}
876
877#[derive(Debug)]
878pub(crate) enum RecordedCommand {
879    RenderPath {
880        thread_id: u8,
881        strips: Range<u32>,
882        blend_mode: BlendMode,
883        paint: Paint,
884        mask: Option<Mask>,
885    },
886    PushLayer {
887        thread_id: u8,
888        clip_path: Option<Range<u32>>,
889        clip_bbox: Option<RectU16>,
890        blend_mode: BlendMode,
891        mask: Option<Mask>,
892        opacity: f32,
893    },
894    PopLayer,
895}
896
897/// An object that might hold a certain value (behind a mutex), and panics if we attempt
898/// to access it when it's not initialized.
899#[derive(Clone)]
900pub(crate) struct MaybePresent<T: Default> {
901    present: Arc<AtomicBool>,
902    value: Arc<Mutex<T>>,
903}
904
905impl<T: Default> MaybePresent<T> {
906    pub(crate) fn new(val: T) -> Self {
907        Self {
908            present: Arc::new(AtomicBool::new(true)),
909            value: Arc::new(Mutex::new(val)),
910        }
911    }
912
913    pub(crate) fn init(&self, value: T) {
914        let mut locked = self.value.lock().unwrap();
915        *locked = value;
916        self.present.store(true, Ordering::SeqCst);
917    }
918
919    pub(crate) fn with_inner(&self, mut func: impl FnMut(&mut T)) {
920        assert!(
921            self.present.load(Ordering::SeqCst),
922            "Tried to access `MaybePresent` before initialization."
923        );
924
925        let mut lock = self.value.lock().unwrap();
926        func(&mut lock);
927    }
928
929    pub(crate) fn take(&self) -> T {
930        assert!(
931            self.present.load(Ordering::SeqCst),
932            "Tried to access `MaybePresent` before initialization."
933        );
934
935        let mut locked = self.value.lock().unwrap();
936        self.present.store(false, Ordering::SeqCst);
937        std::mem::take(&mut *locked)
938    }
939}
940
941#[cfg(test)]
942mod tests {
943    use crate::Level;
944    use crate::color::palette::css::BLUE;
945    use crate::dispatch::Dispatcher;
946    use crate::dispatch::multi_threaded::MultiThreadedDispatcher;
947    use crate::kurbo::{Affine, Rect, Shape};
948    use crate::peniko::{BlendMode, Fill};
949    use vello_common::paint::{Paint, PremulColor};
950
951    /// Ensure we don't cause a memory leak.
952    #[test]
953    fn allocations() {
954        let mut dispatcher = MultiThreadedDispatcher::new(100, 100, 4, Level::new());
955        for _ in 0..20 {
956            dispatcher.fill_path(
957                &Rect::new(0.0, 0.0, 50.0, 50.0).to_path(0.1),
958                Fill::NonZero,
959                Affine::IDENTITY,
960                Paint::Solid(PremulColor::from_alpha_color(BLUE)),
961                BlendMode::default(),
962                None,
963                None,
964            );
965            dispatcher.flush();
966        }
967
968        assert_eq!(dispatcher.allocations.paths.entries.len(), 1);
969        assert_eq!(dispatcher.allocations.strips.entries.len(), 1);
970        assert_eq!(dispatcher.allocations.render_tasks.entries.len(), 1);
971        assert_eq!(dispatcher.allocations.recorded_commands.entries.len(), 1);
972    }
973}
vello_cpu/dispatch/multi_threaded.rs

vello_cpu/dispatch/
multi_threaded.rs
