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
23pub 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 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
41pub 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 server: Arc<super::server::LockedServerCompositor>,
51 render_thread: bool,
55 confined_loop_task: Option<Arc<RenderThreadLoopTask>>,
60 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 free_server_object_ids: RefCell<Vec<u32>>,
79 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 animation_batches: super::composition_batch::CompositorBatches,
92}
93
94struct 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
129pub type SharedCompositorClock = Arc<dyn Fn() -> Duration + Send + Sync>;
132
133struct RenderThreadLoopTask {
136 this: std::sync::Weak<RenderThreadLoopTask>,
137 server: Arc<super::server::LockedServerCompositor>,
138 confined: bool,
141 release_requested: AtomicBool,
144 removal_posted: AtomicBool,
145 render_loop: std::sync::Weak<dyn IRenderLoop>,
146 dispatcher: Arc<Dispatcher>,
147}
148
149impl RenderThreadLoopTask {
150 fn request_release(&self) {
153 self.release_requested.store(true, Ordering::SeqCst);
154 }
155
156 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 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 self.server.try_with(|server| server.release_gpu_resources());
198 self.server.release();
199 self.remove_from_loop();
200 false
201 }
202 None => false,
204 }
205 }
206}
207
208impl Compositor {
209 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 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 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 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 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 pub(crate) fn default_easing(&self) -> Rc<dyn IEasing> {
434 self.default_easing.clone()
435 }
436
437 pub(crate) fn animation_batches(&self) -> &super::composition_batch::CompositorBatches {
439 &self.animation_batches
440 }
441
442 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 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 pub fn with_server<R>(&self, f: impl FnOnce(&Rc<ServerCompositor>) -> R) -> R {
466 self.server.with(f)
467 }
468
469 pub fn locked_server(&self) -> Arc<super::server::LockedServerCompositor> {
480 self.server.clone()
481 }
482
483 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 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 pub fn renders_on_render_thread(&self) -> bool {
509 self.render_thread
510 }
511
512 pub fn is_confined_to_render_thread(&self) -> bool {
524 self.confined_loop_task.is_some()
525 }
526
527 pub fn readback(&self) -> &Arc<super::server::ReadbackIndices> {
529 &self.readback
530 }
531
532 pub fn dispatcher(&self) -> &Arc<Dispatcher> {
534 &self.dispatcher
535 }
536
537 pub fn clock_elapsed(&self) -> Duration {
539 (self.clock)()
540 }
541
542 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 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 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 pub fn request_commit_async(&self) -> Arc<CompositionBatch> {
601 self.request_composition_batch_commit_async()
602 }
603
604 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 let mut objects_written = false;
641 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 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 {
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 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 pub(crate) fn oob_dispose(&self, server: ServerObjectId) -> Arc<CompositionBatch> {
776 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 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 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 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 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 pub(crate) fn dispose_with_a_later_batch(&self, obj: ServerObjectId) {
865 self.queue_disposal(obj, true, true);
866 }
867
868 pub(crate) fn release_with_a_later_batch(&self, obj: ServerObjectId) {
872 self.queue_disposal(obj, false, true);
873 }
874
875 fn queue_disposal(&self, obj: ServerObjectId, dispose: bool, release: bool) -> bool {
878 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 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 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 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 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 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 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 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 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 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 pub fn try_get_render_interface_feature(
1094 &self,
1095 feature_type: std::any::TypeId,
1096 ) -> LocalFuture<Option<super::RenderInterfaceFeature>> {
1097 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 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 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 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 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 task.request_release();
1162 self.server.try_with(|server| server.render_interface().release_platform_handles());
1168 self.render_loop.wakeup();
1172 }
1173 None => {
1174 self.render_loop.remove(&self.loop_task);
1175 self.server.release();
1180 }
1181 }
1182 self.batches.complete_all();
1187 let key = self.key;
1188 let _ = COMPOSITORS.try_with(|compositors| {
1190 if let Ok(mut compositors) = compositors.try_borrow_mut() {
1191 compositors.remove(&key);
1192 }
1193 });
1194 }
1195}
1196
1197struct PendingDisposal {
1201 id: ServerObjectId,
1202 dispose: bool,
1203 release: bool,
1204}
1205
1206#[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
1219enum PendingServerJob {
1221 Job(ServerJob),
1222 ObjectJob(ServerObjectId, ServerObjectJob),
1223}
1224
1225pub type ServerJobError = Arc<dyn std::error::Error + Send + Sync>;
1228
1229pub 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 pub fn from_result(result: T) -> Self {
1275 Self::with_outcome(ServerJobOutcome::RanToCompletion(Some(result)))
1276 }
1277
1278 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 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 pub fn is_completed(&self) -> bool {
1319 !matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::Running)
1320 }
1321
1322 pub fn is_completed_successfully(&self) -> bool {
1324 matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::RanToCompletion(_))
1325 }
1326
1327 pub fn is_faulted(&self) -> bool {
1329 matches!(*self.shared.outcome.lock().unwrap(), ServerJobOutcome::Faulted(_))
1330 }
1331
1332 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 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 pub fn on_completed(&self, continuation: impl FnOnce() + 'static) {
1353 {
1354 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
1366pub type LocalFuture<T> = std::pin::Pin<Box<dyn std::future::Future<Output = T>>>;
1369
1370struct 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}