Skip to main content

ferroui_base/rendering/composition/
compositor.rs

1use super::server::{BatchQueue, CompositorClock, ServerCompositionVisual, ServerCompositor, ServerObjectId};
2use super::transport::{
3    BatchMarker, BatchObject, BatchStreamWriter, CommittedBatch, CompositionBatch, ServerJob, ServerObjectFactory, ServerObjectJob,
4};
5use super::{CompositionOptions, CompositionVisual, ICompositorSerializable};
6use crate::animation::easings::{IEasing, SplineEasing};
7use crate::animation::KeySpline;
8use crate::media::imaging::Bitmap;
9use crate::platform::IPlatformGraphics;
10use crate::reactive::{Disposable, IDisposable};
11use crate::rendering::composition::server::IServerObject;
12use crate::rendering::{IRenderLoop, IRenderLoopTask};
13use crate::threading::{Dispatcher, DispatcherPriority};
14use crate::utilities::HandlerList;
15use crate::{FerroLocator, LocatorExtensions};
16use std::cell::{Cell, RefCell};
17use std::collections::{HashMap, HashSet, VecDeque};
18use std::rc::{Rc, Weak};
19use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
20use std::sync::{Arc, Mutex};
21use std::time::Duration;
22
23/// Receives the request to commit a compositor's pending changes; the media
24/// context implements it and schedules the commit with the next frame.
25pub trait ICompositorScheduler {
26    fn commit_requested(&self, compositor: &Rc<Compositor>);
27}
28
29static NEXT_COMPOSITOR_KEY: AtomicU64 = AtomicU64::new(1);
30
31crate::ferro_thread_local! {
32    /// The compositors of this thread by key: lets `Send` callbacks that
33    /// come back to the thread find their compositor.
34    static COMPOSITORS: RefCell<HashMap<u64, Weak<Compositor>>> = RefCell::new(HashMap::new());
35}
36
37fn find_compositor(key: u64) -> Option<Rc<Compositor>> {
38    COMPOSITORS.with(|compositors| compositors.borrow().get(&key).and_then(Weak::upgrade))
39}
40
41/// The compositor: manages the composition objects of the UI thread and
42/// commits their changes to the server compositor in batches.
43pub struct Compositor {
44    this: Weak<Compositor>,
45    key: u64,
46    render_loop: Arc<dyn IRenderLoop>,
47    loop_task: Arc<dyn IRenderLoopTask>,
48    use_ui_thread_for_synchronous_commits: bool,
49    /// The server compositor, behind the compositor lock.
50    server: Arc<super::server::LockedServerCompositor>,
51    /// Whether the render loop renders on the thread that ticks it (the
52    /// render-thread mode); otherwise a tick is marshalled to the thread of
53    /// this compositor (the dispatcher-thread mode).
54    render_thread: bool,
55    /// The loop task of the render-thread mode when the server compositor is
56    /// confined to the render thread (see
57    /// [`is_confined_to_render_thread`](Self::is_confined_to_render_thread)):
58    /// the compositor asks it for the release of the server graph.
59    confined_loop_task: Option<Arc<RenderThreadLoopTask>>,
60    /// Whether a job of the render thread is on its way to fill the cache
61    /// of the features of the render interface.
62    /// The readback of the server: what the two threads share besides the
63    /// queue of batches.
64    readback: Arc<super::server::ReadbackIndices>,
65    batches: Arc<BatchQueue>,
66    clock: CompositorClock,
67    next_commit: RefCell<Option<Arc<CompositionBatch>>>,
68    object_serialization_queue: RefCell<VecDeque<Rc<dyn ICompositorSerializable>>>,
69    object_serialization_hash_set: RefCell<HashSet<*const ()>>,
70    invoke_before_commit_write: RefCell<VecDeque<Box<dyn FnOnce()>>>,
71    invoke_before_commit_read: RefCell<VecDeque<Box<dyn FnOnce()>>>,
72    dispose_on_next_batch: RefCell<Vec<PendingDisposal>>,
73    pending_creations: RefCell<Vec<(ServerObjectId, ServerObjectFactory)>>,
74    next_server_object_id: Cell<u32>,
75    /// The ids that can be handed out again: each was released by a batch
76    /// that is committed, so the server drops its object before it reads a
77    /// creation under the same id, which can only come with a later batch.
78    free_server_object_ids: RefCell<Vec<u32>>,
79    /// By id, whether the id is handed out: an id is returned to the free
80    /// list once, however often its release is asked for.
81    server_object_id_in_use: RefCell<Vec<bool>>,
82    pending_batch: Arc<Mutex<Option<Arc<CompositionBatch>>>>,
83    pending_server_compositor_jobs: RefCell<Vec<PendingServerJob>>,
84    pending_server_compositor_post_target_jobs: RefCell<Vec<PendingServerJob>>,
85    scheduler: Weak<dyn ICompositorScheduler>,
86    dispatcher: Arc<Dispatcher>,
87    after_commit: HandlerList<dyn Fn()>,
88    default_easing: Rc<dyn IEasing>,
89    /// The open scoped batches and the batch of the current frame (WinUI
90    /// `CompositionScopedBatch` and `CompositionCommitBatch`).
91    animation_batches: super::composition_batch::CompositorBatches,
92}
93
94/// The render loop task of a compositor. The loop may tick on any thread;
95/// the server compositor is confined to the thread of its compositor, so a
96/// tick from elsewhere is marshalled there.
97struct ServerCompositorLoopTask {
98    key: u64,
99    dispatcher: Arc<Dispatcher>,
100    marshalled: Arc<AtomicBool>,
101    last_result: Arc<AtomicBool>,
102}
103
104impl IRenderLoopTask for ServerCompositorLoopTask {
105    fn render(&self) -> bool {
106        if let Some(compositor) = find_compositor(self.key) {
107            let result = compositor.server().render();
108            self.last_result.store(result, Ordering::SeqCst);
109            return result;
110        }
111        if !self.marshalled.swap(true, Ordering::SeqCst) {
112            let key = self.key;
113            let marshalled = self.marshalled.clone();
114            let last_result = self.last_result.clone();
115            self.dispatcher.post(
116                move || {
117                    marshalled.store(false, Ordering::SeqCst);
118                    if let Some(compositor) = find_compositor(key) {
119                        last_result.store(compositor.server().render(), Ordering::SeqCst);
120                    }
121                },
122                DispatcherPriority::RENDER,
123            );
124        }
125        self.last_result.load(Ordering::SeqCst)
126    }
127}
128
129/// The clock of a compositor whose server runs on the render thread: read
130/// by both threads.
131pub type SharedCompositorClock = Arc<dyn Fn() -> Duration + Send + Sync>;
132
133/// The render loop task of a compositor in the render-thread mode: the
134/// thread that ticks renders, under the compositor lock.
135struct RenderThreadLoopTask {
136    this: std::sync::Weak<RenderThreadLoopTask>,
137    server: Arc<super::server::LockedServerCompositor>,
138    /// Whether the thread that ticks is the only one that renders: the
139    /// server graph is then confined to it, and released by it.
140    confined: bool,
141    /// Set by the compositor when it is dropped, where `confined`: the next
142    /// tick releases the server graph instead of rendering.
143    release_requested: AtomicBool,
144    removal_posted: AtomicBool,
145    render_loop: std::sync::Weak<dyn IRenderLoop>,
146    dispatcher: Arc<Dispatcher>,
147}
148
149impl RenderThreadLoopTask {
150    /// Asks the render thread to release the server graph with its next
151    /// tick. From then on no frame is rendered.
152    fn request_release(&self) {
153        self.release_requested.store(true, Ordering::SeqCst);
154    }
155
156    /// Takes the task out of the render loop once the graph is released. A
157    /// task is added and removed by the thread of the compositor, so the
158    /// removal is posted there.
159    fn remove_from_loop(&self) {
160        if self.removal_posted.swap(true, Ordering::SeqCst) {
161            return;
162        }
163        let Some(this) = self.this.upgrade() else { return };
164        let render_loop = self.render_loop.clone();
165        self.dispatcher.post(
166            move || {
167                if let Some(render_loop) = render_loop.upgrade() {
168                    let task: Arc<dyn IRenderLoopTask> = this;
169                    render_loop.remove(&task);
170                }
171            },
172            DispatcherPriority::SEND,
173        );
174    }
175}
176
177impl IRenderLoopTask for RenderThreadLoopTask {
178    fn render(&self) -> bool {
179        if self.confined {
180            self.server.confine_to_current_thread();
181        }
182        // The request is read inside the lock: the compositor sets it before
183        // it enters the lock to take its handles back, so a frame either
184        // ends before that or does not start.
185        let rendered = self
186            .server
187            .try_with(|server| (!self.release_requested.load(Ordering::SeqCst)).then(|| server.render()));
188        match rendered {
189            Some(Some(wants_next_tick)) => wants_next_tick,
190            Some(None) => {
191                // The last job of the render thread: the graph, and with it
192                // every render target and the graphics context, is dropped
193                // here. What it drew with is released first, while its own
194                // graphics context is current: a thread that renders several
195                // compositors has several contexts, and between two frames
196                // any of them, or none, is the current one.
197                self.server.try_with(|server| server.release_gpu_resources());
198                self.server.release();
199                self.remove_from_loop();
200                false
201            }
202            // Nothing to render once the compositor is gone.
203            None => false,
204        }
205    }
206}
207
208impl Compositor {
209    /// Creates a compositor in the render-thread mode: the thread that ticks
210    /// `render_loop` renders the frames.
211    ///
212    /// The server compositor is confined to the compositor lock, not to a
213    /// thread: with `use_ui_thread_for_synchronous_commits` the thread of
214    /// this compositor renders too, at the synchronous points (a resize, the
215    /// first show), as the reference does on platforms that ask for it.
216    ///
217    /// Without it, and with a loop that runs in the background, the server
218    /// compositor is also confined to the thread that ticks (see
219    /// [`is_confined_to_render_thread`](Self::is_confined_to_render_thread)).
220    /// The loop says where it runs when the compositor is created.
221    ///
222    /// `gpu` is shared with the caller, who may keep its handle: the server
223    /// side holds one of its own and the thread that renders asks it for
224    /// the graphics context.
225    pub fn with_render_thread(
226        render_loop: Arc<dyn IRenderLoop>,
227        gpu: Option<Arc<dyn IPlatformGraphics>>,
228        use_ui_thread_for_synchronous_commits: bool,
229        scheduler: &Rc<dyn ICompositorScheduler>,
230        dispatcher: Arc<Dispatcher>,
231        options: Option<CompositionOptions>,
232        clock: Option<SharedCompositorClock>,
233    ) -> Rc<Compositor> {
234        let options = options
235            .or_else(|| FerroLocator::current().get_service::<CompositionOptions>().map(|o| *o))
236            .unwrap_or_default();
237        // The clock is read by both threads. Each side has a handle of its
238        // own to it: the server graph may be released by the render thread.
239        let clock = clock.unwrap_or_else(|| {
240            let origin = std::time::Instant::now();
241            Arc::new(move || origin.elapsed())
242        });
243        let server_clock: CompositorClock = {
244            let clock = clock.clone();
245            Rc::new(move || clock())
246        };
247        let clock: CompositorClock = Rc::new(move || clock());
248        let batches = Arc::new(BatchQueue::default());
249        let readback = Arc::new(super::server::ReadbackIndices::new());
250        let server = Arc::new(super::server::LockedServerCompositor::new(ServerCompositor::new(
251            gpu,
252            options,
253            batches.clone(),
254            readback.clone(),
255            server_clock,
256        )));
257        let key = NEXT_COMPOSITOR_KEY.fetch_add(1, Ordering::SeqCst);
258        // The condition of the reference for a thread of the compositor that
259        // never renders (`UseUiThreadForSynchronousCommits: false`,
260        // `Loop.RunsInBackground: true`).
261        let confined = !use_ui_thread_for_synchronous_commits && render_loop.runs_in_background();
262        let task = Arc::new_cyclic(|this| RenderThreadLoopTask {
263            this: this.clone(),
264            server: server.clone(),
265            confined,
266            release_requested: AtomicBool::new(false),
267            removal_posted: AtomicBool::new(false),
268            render_loop: Arc::downgrade(&render_loop),
269            dispatcher: dispatcher.clone(),
270        });
271        let loop_task: Arc<dyn IRenderLoopTask> = task.clone();
272        Self::build(
273            render_loop,
274            loop_task,
275            confined.then_some(task),
276            use_ui_thread_for_synchronous_commits,
277            server,
278            true,
279            batches,
280            readback,
281            clock,
282            scheduler,
283            dispatcher,
284            key,
285        )
286    }
287
288    /// Creates a compositor that renders on the render loop registered in
289    /// the service locator and commits through the media context.
290    ///
291    /// The threads follow the render loop, as in the reference: with a loop
292    /// that runs in the background (the thread of the platform's render
293    /// timer) that thread renders the frames, under the compositor lock (the
294    /// render-thread mode); with a loop of the UI thread the UI thread
295    /// renders. `FERROUI_RENDER_THREAD=0` in the environment keeps the
296    /// rendering on the UI thread with a background loop too (each tick is
297    /// marshalled to it): a way back while the mode is young.
298    pub fn new(gpu: Option<Arc<dyn IPlatformGraphics>>, use_ui_thread_for_synchronous_commits: bool) -> Rc<Compositor> {
299        let render_loop = FerroLocator::current().get_required_service::<Arc<dyn IRenderLoop>>();
300        let scheduler: Rc<dyn ICompositorScheduler> = crate::media::MediaContext::instance().scheduler();
301        let render_thread_disabled = std::env::var("FERROUI_RENDER_THREAD").is_ok_and(|value| value == "0");
302        if render_loop.runs_in_background() && !render_thread_disabled {
303            return Self::with_render_thread(
304                (*render_loop).clone(),
305                gpu,
306                use_ui_thread_for_synchronous_commits,
307                &scheduler,
308                Dispatcher::ui_thread(),
309                None,
310                None,
311            );
312        }
313        Self::with_scheduler(
314            (*render_loop).clone(),
315            gpu,
316            use_ui_thread_for_synchronous_commits,
317            &scheduler,
318            Dispatcher::ui_thread(),
319            None,
320            None,
321        )
322    }
323
324    /// Creates a compositor with every collaborator given explicitly.
325    ///
326    /// `options` defaults to the options registered in the service locator;
327    /// `clock` defaults to the dispatcher's clock.
328    pub fn with_scheduler(
329        render_loop: Arc<dyn IRenderLoop>,
330        gpu: Option<Arc<dyn IPlatformGraphics>>,
331        use_ui_thread_for_synchronous_commits: bool,
332        scheduler: &Rc<dyn ICompositorScheduler>,
333        dispatcher: Arc<Dispatcher>,
334        options: Option<CompositionOptions>,
335        clock: Option<CompositorClock>,
336    ) -> Rc<Compositor> {
337        let options = options
338            .or_else(|| FerroLocator::current().get_service::<CompositionOptions>().map(|o| *o))
339            .unwrap_or_default();
340        let clock = clock.unwrap_or_else(|| {
341            let platform_impl = dispatcher.platform_impl();
342            let origin = platform_impl.now();
343            Rc::new(move || Duration::from_millis((platform_impl.now() - origin).max(0) as u64))
344        });
345        let batches = Arc::new(BatchQueue::default());
346        let readback = Arc::new(super::server::ReadbackIndices::new());
347        let server = Arc::new(super::server::LockedServerCompositor::new(ServerCompositor::new(
348            gpu,
349            options,
350            batches.clone(),
351            readback.clone(),
352            clock.clone(),
353        )));
354        let key = NEXT_COMPOSITOR_KEY.fetch_add(1, Ordering::SeqCst);
355        let loop_task: Arc<dyn IRenderLoopTask> = Arc::new(ServerCompositorLoopTask {
356            key,
357            dispatcher: dispatcher.clone(),
358            marshalled: Arc::new(AtomicBool::new(false)),
359            last_result: Arc::new(AtomicBool::new(true)),
360        });
361        Self::build(
362            render_loop,
363            loop_task,
364            None,
365            use_ui_thread_for_synchronous_commits,
366            server,
367            false,
368            batches,
369            readback,
370            clock,
371            scheduler,
372            dispatcher,
373            key,
374        )
375    }
376
377    #[allow(clippy::too_many_arguments)]
378    fn build(
379        render_loop: Arc<dyn IRenderLoop>,
380        loop_task: Arc<dyn IRenderLoopTask>,
381        confined_loop_task: Option<Arc<RenderThreadLoopTask>>,
382        use_ui_thread_for_synchronous_commits: bool,
383        server: Arc<super::server::LockedServerCompositor>,
384        render_thread: bool,
385        batches: Arc<BatchQueue>,
386        readback: Arc<super::server::ReadbackIndices>,
387        clock: CompositorClock,
388        scheduler: &Rc<dyn ICompositorScheduler>,
389        dispatcher: Arc<Dispatcher>,
390        key: u64,
391    ) -> Rc<Compositor> {
392        let compositor = Rc::new_cyclic(|this| Compositor {
393            this: this.clone(),
394            key,
395            render_loop: render_loop.clone(),
396            loop_task: loop_task.clone(),
397            use_ui_thread_for_synchronous_commits,
398            readback,
399            server,
400            render_thread,
401            confined_loop_task,
402            batches,
403            clock,
404            next_commit: RefCell::new(None),
405            object_serialization_queue: RefCell::new(VecDeque::new()),
406            object_serialization_hash_set: RefCell::new(HashSet::new()),
407            invoke_before_commit_write: RefCell::new(VecDeque::new()),
408            invoke_before_commit_read: RefCell::new(VecDeque::new()),
409            dispose_on_next_batch: RefCell::new(Vec::new()),
410            pending_creations: RefCell::new(Vec::new()),
411            next_server_object_id: Cell::new(0),
412            free_server_object_ids: RefCell::new(Vec::new()),
413            server_object_id_in_use: RefCell::new(Vec::new()),
414            pending_batch: Arc::new(Mutex::new(None)),
415            pending_server_compositor_jobs: RefCell::new(Vec::new()),
416            pending_server_compositor_post_target_jobs: RefCell::new(Vec::new()),
417            scheduler: Rc::downgrade(scheduler),
418            dispatcher,
419            after_commit: HandlerList::new(),
420            default_easing: Rc::new(SplineEasing::with_key_spline(KeySpline::with_points(0.25, 0.1, 0.25, 1.0))),
421            animation_batches: super::composition_batch::CompositorBatches::new(),
422        });
423        COMPOSITORS.with(|compositors| {
424            let mut compositors = compositors.borrow_mut();
425            compositors.retain(|_, c| c.strong_count() > 0);
426            compositors.insert(key, Rc::downgrade(&compositor));
427        });
428        render_loop.add(loop_task);
429        compositor
430    }
431
432    /// The easing of the key frames that are inserted without one.
433    pub(crate) fn default_easing(&self) -> Rc<dyn IEasing> {
434        self.default_easing.clone()
435    }
436
437    /// The open scoped batches and the batch of the current frame.
438    pub(crate) fn animation_batches(&self) -> &super::composition_batch::CompositorBatches {
439        &self.animation_batches
440    }
441
442    /// The render loop the server compositor is driven by.
443    pub fn render_loop(&self) -> &Arc<dyn IRenderLoop> {
444        &self.render_loop
445    }
446
447    pub fn use_ui_thread_for_synchronous_commits(&self) -> bool {
448        self.use_ui_thread_for_synchronous_commits
449    }
450
451    /// The server compositor. It belongs to the render thread: UI-thread
452    /// code may only use it to drive a frame when the render loop runs on
453    /// the UI thread, and in tests.
454    pub fn server(&self) -> &Rc<ServerCompositor> {
455        assert!(
456            !self.render_thread,
457            "in the render-thread mode the server compositor is reached under its lock: with_server"
458        );
459        self.server.same_thread()
460    }
461
462    /// Runs `f` with the server compositor under the compositor lock: the
463    /// way in from this thread in either mode. What `f` gets must not leave
464    /// it.
465    pub fn with_server<R>(&self, f: impl FnOnce(&Rc<ServerCompositor>) -> R) -> R {
466        self.server.with(f)
467    }
468
469    /// The server compositor behind its lock, as a handle that may cross to
470    /// the thread that renders: for work of that thread that cannot be a job
471    /// of a batch, because it has to run whether or not this compositor
472    /// commits again and whether or not its graphics are ready (the release
473    /// of a surface that is closed). The work enters the graph with
474    /// [`try_with`](super::server::LockedServerCompositor::try_with), which
475    /// answers `None` once the graph is released.
476    ///
477    /// Not from upstream, where such work captures the server compositor
478    /// itself.
479    pub fn locked_server(&self) -> Arc<super::server::LockedServerCompositor> {
480        self.server.clone()
481    }
482
483    /// Binds `value` to the compositor lock: an object of the server side
484    /// that this thread asks for and jobs of the render thread use (see
485    /// [`LockBound`](super::server::LockBound)). `server` is what
486    /// [`with_server`](Self::with_server) or a job hands out.
487    pub(crate) fn bind_to_lock<T>(&self, server: &ServerCompositor, value: T) -> super::server::LockBound<T> {
488        super::server::LockBound::new(&self.server, server, value)
489    }
490
491    /// Renders a frame on this thread, under the compositor lock: what the
492    /// synchronous points do (`Server.Render` upstream).
493    ///
494    /// # Panics
495    ///
496    /// When the compositor is confined to its render thread: this thread
497    /// waits for a frame there, it never renders one.
498    pub fn render_on_this_thread(&self) -> bool {
499        assert!(
500            !self.is_confined_to_render_thread(),
501            "a compositor that is confined to its render thread is not rendered by the thread of the compositor"
502        );
503        self.server.with(|server| server.render())
504    }
505
506    /// Whether the thread that ticks the render loop renders the frames,
507    /// not the thread of this compositor.
508    pub fn renders_on_render_thread(&self) -> bool {
509        self.render_thread
510    }
511
512    /// Whether the thread that ticks the render loop is the only one that
513    /// renders: the render-thread mode without
514    /// `use_ui_thread_for_synchronous_commits`, on a loop that runs in the
515    /// background (the condition under which the reference waits for the
516    /// render thread at the synchronous points).
517    ///
518    /// The render targets, the graphics context and everything else that
519    /// belongs to the thread that draws are then created, used and dropped
520    /// by the render thread alone. This thread still enters the compositor
521    /// lock ([`with_server`](Self::with_server)), for what does not render:
522    /// it commits batches, posts jobs and waits for them.
523    pub fn is_confined_to_render_thread(&self) -> bool {
524        self.confined_loop_task.is_some()
525    }
526
527    /// The readback indices the server writes and this side reads.
528    pub fn readback(&self) -> &Arc<super::server::ReadbackIndices> {
529        &self.readback
530    }
531
532    /// The dispatcher of the thread the compositor belongs to.
533    pub fn dispatcher(&self) -> &Arc<Dispatcher> {
534        &self.dispatcher
535    }
536
537    /// The time elapsed on the compositor clock.
538    pub fn clock_elapsed(&self) -> Duration {
539        (self.clock)()
540    }
541
542    /// Subscribes to the notification raised after each commit.
543    pub fn after_commit(&self, handler: impl Fn() + 'static) -> Rc<dyn IDisposable> {
544        let token = self.after_commit.add(Rc::new(handler));
545        let weak = self.this.clone();
546        Disposable::create(move || {
547            if let Some(this) = weak.upgrade() {
548                this.after_commit.remove(token);
549            }
550        })
551    }
552
553    pub(crate) fn this_handle(&self) -> Rc<Compositor> {
554        self.this()
555    }
556
557    fn this(&self) -> Rc<Compositor> {
558        self.this.upgrade().expect("the compositor is alive while it is used")
559    }
560
561    fn trigger_commit_requested(&self) {
562        if let Some(scheduler) = self.scheduler.upgrade() {
563            scheduler.commit_requested(&self.this());
564        }
565    }
566
567    /// Requests the pending changes to be committed and returns the batch
568    /// they will be part of.
569    pub fn request_composition_batch_commit_async(&self) -> Arc<CompositionBatch> {
570        self.dispatcher.verify_access();
571        if let Some(next_commit) = self.next_commit.borrow().clone() {
572            return next_commit;
573        }
574        let next_commit = CompositionBatch::new();
575        *self.next_commit.borrow_mut() = Some(next_commit.clone());
576        let pending = self.pending_batch.lock().unwrap_or_else(|e| e.into_inner()).clone();
577        match pending {
578            Some(pending) => {
579                // The previous batch has not been processed yet: ask for
580                // the commit once it has been.
581                let key = self.key;
582                let dispatcher = self.dispatcher.clone();
583                pending.processed().on_completed(move || {
584                    dispatcher.post(
585                        move || {
586                            if let Some(compositor) = find_compositor(key) {
587                                compositor.trigger_commit_requested();
588                            }
589                        },
590                        DispatcherPriority::SEND,
591                    );
592                });
593            }
594            None => self.trigger_commit_requested(),
595        }
596        next_commit
597    }
598
599    /// Requests the pending changes to be committed.
600    pub fn request_commit_async(&self) -> Arc<CompositionBatch> {
601        self.request_composition_batch_commit_async()
602    }
603
604    /// Serializes the pending changes into a batch and hands it to the
605    /// server compositor.
606    pub fn commit(&self) -> Arc<CompositionBatch> {
607        let result = self.commit_core();
608        if !self.invoke_before_commit_write.borrow().is_empty() {
609            self.request_commit_async();
610        }
611        if !self.after_commit.is_empty() {
612            for (_, handler) in self.after_commit.snapshot().iter() {
613                handler();
614            }
615        }
616        result
617    }
618
619    fn commit_core(&self) -> Arc<CompositionBatch> {
620        self.dispatcher.verify_access();
621
622        let commit = self.next_commit.borrow_mut().get_or_insert_with(CompositionBatch::new).clone();
623
624        std::mem::swap(
625            &mut *self.invoke_before_commit_read.borrow_mut(),
626            &mut *self.invoke_before_commit_write.borrow_mut(),
627        );
628        loop {
629            let Some(action) = self.invoke_before_commit_read.borrow_mut().pop_front() else { break };
630            action();
631        }
632
633        let mut changes = self.batches.rent_data();
634        let disposed: Vec<PendingDisposal>;
635        {
636            let mut writer = BatchStreamWriter::new(&mut changes);
637
638            // Objects serialize references to server objects, so the objects
639            // created since the last commit come first.
640            let mut objects_written = false;
641            // The set of queued objects knows an object by its address, so
642            // an object stays alive until the set is cleared: an object
643            // created while another one is serialized can then never get the
644            // address of one that was serialized (and released) before it,
645            // and be taken for queued.
646            let mut serialized: Vec<Rc<dyn ICompositorSerializable>> = Vec::new();
647            loop {
648                self.write_pending_creations(&mut writer);
649                let Some(object) = self.object_serialization_queue.borrow_mut().pop_front() else { break };
650                if let Some(server_object) = object.try_get_server(self) {
651                    // Server objects the object creates for its changes
652                    // must exist before the changes are read.
653                    object.prepare_serialization(self);
654                    self.write_pending_creations(&mut writer);
655                    writer.write_server_object(Some(server_object));
656                    object.serialize_changes(self, &mut writer);
657                    if cfg!(debug_assertions) {
658                        writer.write(super::server::OBJECT_END_MAGIC);
659                        writer.write_marker(BatchMarker::ObjectEnd);
660                    }
661                    objects_written = true;
662                }
663                serialized.push(object);
664            }
665            let _ = objects_written;
666            self.object_serialization_hash_set.borrow_mut().clear();
667            drop(serialized);
668
669            disposed = std::mem::take(&mut *self.dispose_on_next_batch.borrow_mut());
670            if !disposed.is_empty() {
671                writer.write_marker(BatchMarker::RenderThreadDisposeStart);
672                writer.write(disposed.len() as i32);
673                for entry in &disposed {
674                    writer.write_server_object(Some(entry.id));
675                    writer.write(entry.dispose);
676                    writer.write(entry.release);
677                }
678            }
679
680            Self::serialize_server_jobs(
681                &mut writer,
682                &self.pending_server_compositor_jobs,
683                BatchMarker::RenderThreadJobsStart,
684                BatchMarker::RenderThreadJobsEnd,
685            );
686            Self::serialize_server_jobs(
687                &mut writer,
688                &self.pending_server_compositor_post_target_jobs,
689                BatchMarker::RenderThreadPostTargetJobsStart,
690                BatchMarker::RenderThreadPostTargetJobsEnd,
691            );
692        }
693        // Ids released by this batch can be reused by objects created for a
694        // later one.
695        {
696            let mut in_use = self.server_object_id_in_use.borrow_mut();
697            let mut free = self.free_server_object_ids.borrow_mut();
698            for entry in disposed.iter().filter(|entry| entry.release) {
699                if in_use.get_mut(entry.id.index()).is_some_and(|in_use| std::mem::replace(in_use, false)) {
700                    free.push(entry.id.0);
701                }
702            }
703        }
704
705        self.enqueue_batch(CommittedBatch { batch: commit.clone(), changes, committed_at: self.clock_elapsed() });
706
707        *self.pending_batch.lock().unwrap_or_else(|e| e.into_inner()) = Some(commit.clone());
708        let pending_batch = self.pending_batch.clone();
709        let sequence_id = commit.sequence_id();
710        commit.processed().on_completed(move || {
711            let mut pending = pending_batch.lock().unwrap_or_else(|e| e.into_inner());
712            if pending.as_ref().is_some_and(|p| p.sequence_id() == sequence_id) {
713                *pending = None;
714            }
715        });
716        *self.next_commit.borrow_mut() = None;
717        // The animations started in the frame are on their way: the commit
718        // batch of the frame is ended.
719        self.animation_batches.on_commit();
720
721        commit
722    }
723
724    fn write_pending_creations(&self, writer: &mut BatchStreamWriter<'_>) {
725        let creations = std::mem::take(&mut *self.pending_creations.borrow_mut());
726        if creations.is_empty() {
727            return;
728        }
729        writer.write_marker(BatchMarker::CreateStart);
730        writer.write(creations.len() as i32);
731        for (id, factory) in creations {
732            writer.write_server_object(Some(id));
733            writer.write_object(BatchObject::Create(factory));
734        }
735    }
736
737    fn serialize_server_jobs(
738        writer: &mut BatchStreamWriter<'_>,
739        list: &RefCell<Vec<PendingServerJob>>,
740        start_marker: BatchMarker,
741        end_marker: BatchMarker,
742    ) {
743        let jobs = std::mem::take(&mut *list.borrow_mut());
744        if !jobs.is_empty() {
745            writer.write_marker(start_marker);
746            for job in jobs {
747                match job {
748                    PendingServerJob::Job(job) => writer.write_object(BatchObject::Job(job)),
749                    PendingServerJob::ObjectJob(target, job) => {
750                        writer.write_server_object(Some(target));
751                        writer.write_object(BatchObject::ObjectJob(job));
752                    }
753                }
754            }
755            writer.write_marker(end_marker);
756        }
757    }
758
759    fn enqueue_batch(&self, batch: CommittedBatch) {
760        self.batches.enqueue(batch);
761        self.render_loop.wakeup();
762    }
763
764    /// Disposes a server object in a batch of its own, outside of the
765    /// regular commit.
766    ///
767    /// The object is disposed, not released: it stays in the table of the
768    /// server under its id until the UI-thread object that names it is
769    /// dropped, as after an ordinary disposal. An object whose creation has
770    /// not been committed yet is created by the same batch, before it is
771    /// disposed: upstream the server object exists from the moment the
772    /// UI-thread object does, so the disposal always finds it. Left to the
773    /// next commit, the creation would follow the disposal and the object
774    /// would never be disposed.
775    pub(crate) fn oob_dispose(&self, server: ServerObjectId) -> Arc<CompositionBatch> {
776        // A disposal that was queued for the next batch is done here.
777        self.dispose_on_next_batch.borrow_mut().retain_mut(|entry| {
778            if entry.id == server {
779                entry.dispose = false;
780                entry.release
781            } else {
782                true
783            }
784        });
785        let pending_creation = {
786            let mut creations = self.pending_creations.borrow_mut();
787            creations.iter().position(|(id, _)| *id == server).map(|index| creations.remove(index))
788        };
789        let batch = CompositionBatch::new();
790        let mut changes = self.batches.rent_data();
791        {
792            let mut writer = BatchStreamWriter::new(&mut changes);
793            if let Some((id, factory)) = pending_creation {
794                writer.write_marker(BatchMarker::CreateStart);
795                writer.write(1i32);
796                writer.write_server_object(Some(id));
797                writer.write_object(BatchObject::Create(factory));
798            }
799            writer.write_marker(BatchMarker::RenderThreadDisposeStart);
800            writer.write(1i32);
801            writer.write_server_object(Some(server));
802            writer.write(true);
803            writer.write(false);
804        }
805        self.enqueue_batch(CommittedBatch { batch: batch.clone(), changes, committed_at: self.clock_elapsed() });
806        batch
807    }
808
809    /// Allocates the id of a new server object and queues its creation for
810    /// the next batch. `factory` runs on the render thread.
811    pub fn create_server_object(
812        &self,
813        factory: impl FnOnce(&Rc<ServerCompositor>, ServerObjectId) -> Rc<dyn IServerObject> + Send + 'static,
814    ) -> ServerObjectId {
815        self.dispatcher.verify_access();
816        let id = match self.free_server_object_ids.borrow_mut().pop() {
817            Some(id) => ServerObjectId(id),
818            None => {
819                let id = self.next_server_object_id.get();
820                self.next_server_object_id.set(id + 1);
821                self.server_object_id_in_use.borrow_mut().push(false);
822                ServerObjectId(id)
823            }
824        };
825        self.server_object_id_in_use.borrow_mut()[id.index()] = true;
826        self.pending_creations.borrow_mut().push((id, Box::new(factory)));
827        self.request_commit_async();
828        id
829    }
830
831    /// Queues an object whose changes must be written to the next batch.
832    pub fn register_for_serialization(&self, composition_object: Rc<dyn ICompositorSerializable>) {
833        self.dispatcher.verify_access();
834        let key = composition_object.serialization_key();
835        if self.object_serialization_hash_set.borrow_mut().insert(key) {
836            self.object_serialization_queue.borrow_mut().push_back(composition_object);
837        }
838        self.request_commit_async();
839    }
840
841    /// Disposes a server object with the next batch, and releases it: the
842    /// server compositor drops it and its id can be reused.
843    pub fn dispose_on_next_batch(&self, obj: ServerObjectId) {
844        if self.queue_disposal(obj, true, true) {
845            self.request_commit_async();
846        }
847    }
848
849    /// Disposes a server object with the next batch but keeps it, under its
850    /// id, until [`release_with_a_later_batch`](Self::release_with_a_later_batch):
851    /// the disposal of a composition object, whose server object stays
852    /// reachable by the jobs and animations that refer to it while the
853    /// UI-thread object is alive, as upstream the UI-thread object holds its
854    /// server object.
855    pub(crate) fn dispose_and_keep_on_next_batch(&self, obj: ServerObjectId) {
856        if self.queue_disposal(obj, true, false) {
857            self.request_commit_async();
858        }
859    }
860
861    /// Disposes a server object with a later batch, without asking for a
862    /// commit: used when a UI-thread object is dropped, which must not
863    /// schedule work by itself.
864    pub(crate) fn dispose_with_a_later_batch(&self, obj: ServerObjectId) {
865        self.queue_disposal(obj, true, true);
866    }
867
868    /// Releases a server object that was already disposed with a later
869    /// batch, without asking for a commit: used when a disposed UI-thread
870    /// object is dropped.
871    pub(crate) fn release_with_a_later_batch(&self, obj: ServerObjectId) {
872        self.queue_disposal(obj, false, true);
873    }
874
875    /// Adds `obj` to the disposal list of the next batch, or merges the
876    /// request into its entry. Returns whether the list changed.
877    fn queue_disposal(&self, obj: ServerObjectId, dispose: bool, release: bool) -> bool {
878        // An id that was released already names no object: asking again
879        // for its release must not return it to the free list twice.
880        if !self.server_object_id_in_use.borrow().get(obj.index()).copied().unwrap_or(false) {
881            return false;
882        }
883        let Ok(mut list) = self.dispose_on_next_batch.try_borrow_mut() else { return false };
884        match list.iter_mut().find(|entry| entry.id == obj) {
885            Some(entry) => {
886                let changed = (dispose && !entry.dispose) || (release && !entry.release);
887                entry.dispose |= dispose;
888                entry.release |= release;
889                changed
890            }
891            None => {
892                list.push(PendingDisposal { id: obj, dispose, release });
893                true
894            }
895        }
896    }
897
898    /// Enqueues a callback to be run before the next commit.
899    pub fn request_composition_update(&self, action: impl FnOnce() + 'static) {
900        self.dispatcher.verify_access();
901        self.invoke_before_commit_write.borrow_mut().push_back(Box::new(action));
902        self.request_commit_async();
903    }
904
905    /// Posts a job that runs on the render thread with the next batch:
906    /// before the composition targets are rendered, or after them when
907    /// `post_target` is set.
908    pub fn post_server_job(&self, job: impl FnOnce(&ServerCompositor) + Send + 'static, post_target: bool) {
909        self.dispatcher.verify_access();
910        let list = if post_target {
911            &self.pending_server_compositor_post_target_jobs
912        } else {
913            &self.pending_server_compositor_jobs
914        };
915        list.borrow_mut().push(PendingServerJob::Job(Box::new(job)));
916        self.request_commit_async();
917    }
918
919    /// Posts a job that runs on the render thread with the next batch and
920    /// receives the server object `target` (`None` when it does not
921    /// exist). Upstream a job holds the server object itself; this job
922    /// names it, and the object is resolved when the batch is read, so the
923    /// job reaches an object that the same batch disposes.
924    pub fn post_server_object_job(
925        &self,
926        target: ServerObjectId,
927        job: impl FnOnce(&ServerCompositor, Option<Rc<dyn IServerObject>>) + Send + 'static,
928        post_target: bool,
929    ) {
930        self.dispatcher.verify_access();
931        let list = if post_target {
932            &self.pending_server_compositor_post_target_jobs
933        } else {
934            &self.pending_server_compositor_jobs
935        };
936        list.borrow_mut().push(PendingServerJob::ObjectJob(target, Box::new(job)));
937        self.request_commit_async();
938    }
939
940    /// [`invoke_server_job_async`](Self::invoke_server_job_async) with a job
941    /// that receives the server object `target`; see
942    /// [`post_server_object_job`](Self::post_server_object_job).
943    pub fn invoke_server_object_job_async<T: Send + 'static>(
944        &self,
945        target: ServerObjectId,
946        job: impl FnOnce(&ServerCompositor, Option<Rc<dyn IServerObject>>) -> Result<T, ServerJobError>
947            + Send
948            + 'static,
949        post_target: bool,
950    ) -> ServerJobTask<T> {
951        let task = ServerJobTask::new();
952        let completion = task.clone();
953        self.post_server_object_job(
954            target,
955            move |compositor, object| match job(compositor, object) {
956                Ok(result) => completion.set_result(result),
957                Err(error) => completion.set_exception(error),
958            },
959            post_target,
960        );
961        task
962    }
963
964    /// Runs `job` on the render thread with the next batch and returns a
965    /// task that completes with its result (`InvokeServerJobAsync`). A job
966    /// that fails faults the task with its error.
967    pub fn invoke_server_job_async<T: Send + 'static>(
968        &self,
969        job: impl FnOnce(&ServerCompositor) -> Result<T, ServerJobError> + Send + 'static,
970        post_target: bool,
971    ) -> ServerJobTask<T> {
972        let task = ServerJobTask::new();
973        let completion = task.clone();
974        self.post_server_job(
975            move |compositor| match job(compositor) {
976                Ok(result) => completion.set_result(result),
977                Err(error) => completion.set_exception(error),
978            },
979            post_target,
980        );
981        task
982    }
983
984    /// Renders a visual and its children into a new bitmap, on the render
985    /// thread with the next batch, after the composition targets are
986    /// rendered (`CreateCompositionVisualSnapshot`).
987    ///
988    /// The task is faulted with a [`CompositionVisualSnapshotError`] when
989    /// the visual belongs to another compositor or is not attached to a
990    /// composition target (`InvalidOperationException` upstream, which the
991    /// asynchronous method also reports through its task).
992    pub fn create_composition_visual_snapshot(&self, visual: &CompositionVisual, scaling: f64) -> ServerJobTask<Bitmap> {
993        if !Rc::ptr_eq(visual.compositor(), &self.this()) {
994            return ServerJobTask::from_exception(Arc::new(CompositionVisualSnapshotError(
995                "the visual belongs to another compositor",
996            )));
997        }
998        if visual.root().is_none() {
999            return ServerJobTask::from_exception(Arc::new(CompositionVisualSnapshotError(
1000                "the visual is not attached to a composition target",
1001            )));
1002        }
1003        // The job names the server visual by id, as the other jobs of an object do: the id
1004        // stays bound to the server visual while the visual is alive.
1005        let server = visual.server();
1006        self.invoke_server_job_async(
1007            move |compositor| match compositor.get::<ServerCompositionVisual>(server) {
1008                Some(visual) => Ok(Bitmap::from_impl(compositor.create_composition_visual_snapshot(&visual, scaling, true))),
1009                None => {
1010                    let error: ServerJobError =
1011                        Arc::new(CompositionVisualSnapshotError("the server visual no longer exists"));
1012                    Err(error)
1013                }
1014            },
1015            true,
1016        )
1017    }
1018
1019    /// `GetRenderInterfacePublicFeatures`: the public features of the
1020    /// render interface, handed to `f` inside the compositor lock; what `f`
1021    /// takes from them is the outcome.
1022    ///
1023    /// As upstream: the features that are cached are the answer at once; a
1024    /// compositor whose loop does not run in the background asks the server
1025    /// on this thread (which creates the backend context when needed); a
1026    /// compositor confined to its render thread without a cache asks with a
1027    /// job of the render thread (`InvokeServerJobAsync(Server.RT_GetRenderInterfaceFeatures)`)
1028    /// and the outcome comes with the job, so the caller awaits it. `f` then
1029    /// runs in the job, on the render thread.
1030    ///
1031    /// Upstream hands the dictionary itself to the caller, and the caller
1032    /// takes the feature from it; here a feature is an `Rc` of the server
1033    /// side, so `f` takes what it needs inside the lock, where the features
1034    /// are, and only what it returns leaves. Its third argument tells
1035    /// whether the calling thread may create the backend context: `false`
1036    /// only on the thread of a confined compositor, which answers from the
1037    /// cache. Without a backend context a compositor that is not confined
1038    /// and whose render interface is not ready has no features (`None`).
1039    fn get_render_interface_public_features<R: Send + 'static>(
1040        &self,
1041        f: impl FnOnce(&ServerCompositor, &HashMap<std::any::TypeId, Rc<dyn std::any::Any>>, bool) -> R + Send + 'static,
1042    ) -> LocalFuture<Option<R>> {
1043        self.dispatcher.verify_access();
1044        if self.is_confined_to_render_thread() {
1045            // The features are objects of the server side: they are read
1046            // under the lock, and `f` runs there; it is only taken out of
1047            // its cell when there is a cache to give it.
1048            let f = Cell::new(Some(f));
1049            let cached = self.server.with(|server| {
1050                let features = server.at_try_get_cached_render_interface_features()?;
1051                let f = f.take().expect("taken once");
1052                Some(f(server, &features, false))
1053            });
1054            if let Some(result) = cached {
1055                return Box::pin(std::future::ready(Some(result)));
1056            }
1057            let f = f.take().expect("taken once");
1058            let task = self.invoke_server_job_async(
1059                move |server| {
1060                    let features = server.rt_get_render_interface_features();
1061                    Ok(f(server, &features, true))
1062                },
1063                false,
1064            );
1065            return Box::pin(ServerJobResult::new(task));
1066        }
1067        let result = self.server.with(|server| {
1068            if let Some(features) = server.at_try_get_cached_render_interface_features() {
1069                return Some(f(server, &features, true));
1070            }
1071            if !server.render_interface().is_ready() {
1072                return None;
1073            }
1074            let features = server.rt_get_render_interface_features();
1075            Some(f(server, &features, true))
1076        });
1077        Box::pin(std::future::ready(result))
1078    }
1079
1080    /// Attempts to query for a feature from the platform render interface
1081    /// (`TryGetRenderInterfaceFeature`): a future that completes with the
1082    /// feature, or `None` when the render interface has none. Awaited on the
1083    /// thread of the compositor; see
1084    /// [`get_render_interface_public_features`](Self::get_render_interface_public_features)
1085    /// for when it completes.
1086    ///
1087    /// Upstream hands out the feature object itself, which the two threads
1088    /// then share. Here the feature is an `Rc` of the server side, so what
1089    /// is handed out is a [`RenderInterfaceFeature`](super::RenderInterfaceFeature):
1090    /// a handle to the same feature that stays inside the compositor lock
1091    /// and lends the feature there. It keeps the feature alive as the
1092    /// object upstream does.
1093    pub fn try_get_render_interface_feature(
1094        &self,
1095        feature_type: std::any::TypeId,
1096    ) -> LocalFuture<Option<super::RenderInterfaceFeature>> {
1097        // The handle is cloned from the map and bound to the lock inside
1098        // the lock: no count of the server side is touched outside it.
1099        let locked = self.server.clone();
1100        let features = self.get_render_interface_public_features(move |server, features, _| {
1101            let feature = features.get(&feature_type)?;
1102            Some(super::RenderInterfaceFeature::new(super::server::LockBound::new(&locked, server, feature.clone())))
1103        });
1104        Box::pin(async move { features.await.flatten() })
1105    }
1106
1107    /// Attempts to query for the GPU interop feature from the platform
1108    /// render interface (`TryGetCompositionGpuInterop`): a future that
1109    /// completes with the interop with GPU objects created outside of the
1110    /// framework, or `None` when the render interface does not support it.
1111    /// Awaited on the thread of the compositor.
1112    ///
1113    /// Upstream awaits the feature (`TryGetRenderInterfaceFeature`) and
1114    /// constructs the interop where the await continues: on the render
1115    /// thread, when a job of the render thread gave the feature. Here the
1116    /// feature is an object of the server side: the server part of the
1117    /// interop is made from it inside the lock, where the feature is read
1118    /// (in that job, for a confined compositor without a cache), and this
1119    /// thread makes the interop around it.
1120    pub fn try_get_composition_gpu_interop(&self) -> LocalFuture<Option<Rc<dyn super::ICompositionGpuInterop>>> {
1121        let feature_type = std::any::TypeId::of::<dyn crate::platform::IExternalObjectsRenderInterfaceContextFeature>();
1122        let locked = self.server.clone();
1123        let parts = self.get_render_interface_public_features(move |server, features, may_create_context| {
1124            let external_objects = features
1125                .get(&feature_type)?
1126                .downcast_ref::<Rc<dyn crate::platform::IExternalObjectsRenderInterfaceContextFeature>>()?
1127                .clone();
1128            // `None` where this thread may not create the context of the
1129            // render interface and the render thread has none at present.
1130            super::CompositionInterop::server_part(&locked, server, external_objects, may_create_context)
1131        });
1132        let this = self.this();
1133        Box::pin(async move {
1134            let part = parts.await.flatten()?;
1135            let interop: Rc<dyn super::ICompositionGpuInterop> = super::CompositionInterop::from_server_part(&this, part);
1136            Some(interop)
1137        })
1138    }
1139
1140    /// Whether an object is queued for serialization (for unit tests).
1141    pub fn unit_test_is_registered_for_serialization(&self, serializable: &dyn ICompositorSerializable) -> bool {
1142        self.object_serialization_hash_set.borrow().contains(&serializable.serialization_key())
1143    }
1144
1145    /// The compositor registered in the service locator, if any.
1146    pub fn try_get_default_compositor() -> Option<Rc<Compositor>> {
1147        FerroLocator::current().get_service::<Compositor>()
1148    }
1149}
1150
1151impl Drop for Compositor {
1152    fn drop(&mut self) {
1153        match &self.confined_loop_task {
1154            Some(task) => {
1155                // The graph holds what only the render thread may drop (the
1156                // render targets, the graphics context), so its release is
1157                // the last job of that thread: the task stays in the loop
1158                // for one more tick, releases the graph there and then has
1159                // itself removed. Asked for before the lock is entered: no
1160                // frame starts after this thread has been inside.
1161                task.request_release();
1162                // What the graph shares with this thread is dropped here,
1163                // for the same reason the other modes release all of it
1164                // here: the count of the handle of the render interface
1165                // belongs to this thread. (The platform graphics and their
1166                // ready state are shared by their contracts and go with it.)
1167                self.server.try_with(|server| server.render_interface().release_platform_handles());
1168                // A loop that does not tick again keeps the task, and the
1169                // graph with it: whichever thread drops the loop then leaks
1170                // the graph instead of dropping it (`LockedServerCompositor`).
1171                self.render_loop.wakeup();
1172            }
1173            None => {
1174                self.render_loop.remove(&self.loop_task);
1175                // The server compositor is released here, on the thread of
1176                // the compositor: a tick in progress on the render thread
1177                // may still hold the task, and with it the lock object, but
1178                // not the graph.
1179                self.server.release();
1180            }
1181        }
1182        // No frame of this compositor applies a batch from here on. The
1183        // batches that are still in the queue are completed as they are:
1184        // the media context commits no compositor while one of the batches
1185        // it committed is not processed.
1186        self.batches.complete_all();
1187        let key = self.key;
1188        // The registry may already be gone when the thread is exiting.
1189        let _ = COMPOSITORS.try_with(|compositors| {
1190            if let Ok(mut compositors) = compositors.try_borrow_mut() {
1191                compositors.remove(&key);
1192            }
1193        });
1194    }
1195}
1196
1197/// A server object listed for disposal by the next batch: the server
1198/// object is disposed when `dispose` is set, and dropped by the server
1199/// compositor, its id becoming free, when `release` is set.
1200struct PendingDisposal {
1201    id: ServerObjectId,
1202    dispose: bool,
1203    release: bool,
1204}
1205
1206/// The error of a snapshot that cannot be taken (`InvalidOperationException`):
1207/// see [`Compositor::create_composition_visual_snapshot`].
1208#[derive(Clone, Debug)]
1209pub struct CompositionVisualSnapshotError(&'static str);
1210
1211impl std::fmt::Display for CompositionVisualSnapshotError {
1212    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1213        f.write_str(self.0)
1214    }
1215}
1216
1217impl std::error::Error for CompositionVisualSnapshotError {}
1218
1219/// A job waiting for the next batch.
1220enum PendingServerJob {
1221    Job(ServerJob),
1222    ObjectJob(ServerObjectId, ServerObjectJob),
1223}
1224
1225/// The error of a job run on the render thread: it crosses to the thread
1226/// that asked for the job.
1227pub type ServerJobError = Arc<dyn std::error::Error + Send + Sync>;
1228
1229/// The outcome of a job run on the render thread: what
1230/// [`Compositor::invoke_server_job_async`] returns (a task upstream).
1231///
1232/// The handle is shared by its clones; the job completes it once, from the
1233/// render thread. Continuations belong to the thread that created the task
1234/// and run there: right away when the job completes on that thread,
1235/// otherwise from its dispatcher.
1236pub struct ServerJobTask<T> {
1237    shared: Arc<ServerJobTaskShared<T>>,
1238}
1239
1240struct ServerJobTaskShared<T> {
1241    outcome: std::sync::Mutex<ServerJobOutcome<T>>,
1242    continuations: crate::utilities::ThreadBound<RefCell<Vec<Box<dyn FnOnce()>>>>,
1243    thread: std::thread::ThreadId,
1244}
1245
1246enum ServerJobOutcome<T> {
1247    Running,
1248    RanToCompletion(Option<T>),
1249    Faulted(ServerJobError),
1250}
1251
1252impl<T> Clone for ServerJobTask<T> {
1253    fn clone(&self) -> Self {
1254        Self { shared: self.shared.clone() }
1255    }
1256}
1257
1258impl<T: Send + 'static> ServerJobTask<T> {
1259    fn with_outcome(outcome: ServerJobOutcome<T>) -> Self {
1260        Self {
1261            shared: Arc::new(ServerJobTaskShared {
1262                outcome: std::sync::Mutex::new(outcome),
1263                continuations: crate::utilities::ThreadBound::new(RefCell::new(Vec::new())),
1264                thread: std::thread::current().id(),
1265            }),
1266        }
1267    }
1268
1269    pub(crate) fn new() -> Self {
1270        Self::with_outcome(ServerJobOutcome::Running)
1271    }
1272
1273    /// A task that has completed with `result`.
1274    pub fn from_result(result: T) -> Self {
1275        Self::with_outcome(ServerJobOutcome::RanToCompletion(Some(result)))
1276    }
1277
1278    /// A task that has failed with `error`.
1279    pub fn from_exception(error: ServerJobError) -> Self {
1280        Self::with_outcome(ServerJobOutcome::Faulted(error))
1281    }
1282
1283    fn run_continuations(shared: &ServerJobTaskShared<T>) {
1284        let continuations = std::mem::take(&mut *shared.continuations.get().borrow_mut());
1285        for continuation in continuations {
1286            continuation();
1287        }
1288    }
1289
1290    fn complete(&self, outcome: ServerJobOutcome<T>) {
1291        {
1292            let mut current = self.shared.outcome.lock().unwrap();
1293            if !matches!(*current, ServerJobOutcome::Running) {
1294                return;
1295            }
1296            *current = outcome;
1297        }
1298
1299        if self.shared.continuations.is_on_thread() {
1300            Self::run_continuations(&self.shared);
1301        } else if let Some(dispatcher) = Dispatcher::from_thread(self.shared.thread) {
1302            // Completed by the render thread: the continuations run on the
1303            // thread of the task.
1304            let shared = self.shared.clone();
1305            dispatcher.post(move || Self::run_continuations(&shared), DispatcherPriority::SEND);
1306        }
1307    }
1308
1309    pub(crate) fn set_result(&self, result: T) {
1310        self.complete(ServerJobOutcome::RanToCompletion(Some(result)))
1311    }
1312
1313    pub(crate) fn set_exception(&self, error: ServerJobError) {
1314        self.complete(ServerJobOutcome::Faulted(error))
1315    }
1316
1317    /// Whether the job has run.
1318    pub fn is_completed(&self) -> bool {
1319        !matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::Running)
1320    }
1321
1322    /// Whether the job has run without an error.
1323    pub fn is_completed_successfully(&self) -> bool {
1324        matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::RanToCompletion(_))
1325    }
1326
1327    /// Whether the job failed.
1328    pub fn is_faulted(&self) -> bool {
1329        matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::Faulted(_))
1330    }
1331
1332    /// The error of a failed job.
1333    pub fn exception(&self) -> Option<ServerJobError> {
1334        match &*self.shared.outcome.lock().unwrap() {
1335            ServerJobOutcome::Faulted(error) => Some(error.clone()),
1336            _ => None,
1337        }
1338    }
1339
1340    /// Takes the result of a job that ran to completion: `None` while it
1341    /// runs, and once taken; the error of a failed job.
1342    pub fn take_result(&self) -> Option<Result<T, ServerJobError>> {
1343        match &mut *self.shared.outcome.lock().unwrap() {
1344            ServerJobOutcome::Running => None,
1345            ServerJobOutcome::RanToCompletion(result) => result.take().map(Ok),
1346            ServerJobOutcome::Faulted(error) => Some(Err(error.clone())),
1347        }
1348    }
1349
1350    /// Runs `continuation` once the job has run; right away if it has.
1351    /// Called on the thread that created the task.
1352    pub fn on_completed(&self, continuation: impl FnOnce() + 'static) {
1353        {
1354            // The outcome is read under its lock, so a job completing on the
1355            // render thread either sees the continuation or has completed.
1356            let outcome = self.shared.outcome.lock().unwrap();
1357            if matches!(*outcome, ServerJobOutcome::Running) {
1358                self.shared.continuations.get().borrow_mut().push(Box::new(continuation));
1359                return;
1360            }
1361        }
1362        continuation();
1363    }
1364}
1365
1366/// A future of the thread of the compositor: what its asynchronous members
1367/// return (a `ValueTask` upstream), awaited on that thread.
1368pub type LocalFuture<T> = std::pin::Pin<Box<dyn std::future::Future<Output = T>>>;
1369
1370/// `await` of the task of a job of the render thread: the result of the
1371/// job. A job that fails ends the awaiting method with its error, as the
1372/// exception that `await` rethrows upstream.
1373struct ServerJobResult<T> {
1374    task: ServerJobTask<T>,
1375    registered: bool,
1376}
1377
1378impl<T: Send + 'static> ServerJobResult<T> {
1379    fn new(task: ServerJobTask<T>) -> Self {
1380        Self { task, registered: false }
1381    }
1382}
1383
1384impl<T: Send + 'static> std::future::Future for ServerJobResult<T> {
1385    type Output = Option<T>;
1386
1387    fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
1388        match self.task.take_result() {
1389            Some(Ok(result)) => return std::task::Poll::Ready(Some(result)),
1390            Some(Err(error)) => panic!("{error}"),
1391            None => {}
1392        }
1393        if !self.registered {
1394            self.registered = true;
1395            let waker = cx.waker().clone();
1396            self.task.on_completed(move || waker.wake());
1397        }
1398        std::task::Poll::Pending
1399    }
1400}