ferroui-base 0.2.0

use super::{
    CompositorPools, IServerObject, IServerRenderResource, ReadbackIndices, ServerCompositionTarget,
    ServerCompositionVisual, ServerCompositorAnimations, ServerObjectId,
};
use crate::platform::surfaces::IPlatformRenderSurface;
use crate::platform::{IBitmapImpl, IPlatformGraphics, IRenderTarget, LtrbRect};
use crate::rendering::composition::transport::{
    BatchMarker, BatchObject, BatchStreamData, BatchStreamReader, CommittedBatch, CompositionBatch, ServerJob,
};
use crate::rendering::composition::CompositionOptions;
use crate::rendering::PlatformRenderInterfaceContextManager;
use crate::reactive::IDisposable;
use crate::threading::Dispatcher;
use crate::{Matrix, PixelSize, Size, Vector};
use std::any::{Any, TypeId};
use std::cell::{Cell, RefCell};
use std::collections::{HashMap, HashSet, VecDeque};
use std::rc::{Rc, Weak};
use std::sync::{Arc, Mutex};
use std::time::Duration;

/// The source of the compositor clock: time elapsed since an arbitrary
/// origin. It is read from both sides of the transport; like the server
/// compositor it is confined to the compositor's thread for now, because
/// the dispatcher's clock is.
pub type CompositorClock = Rc<dyn Fn() -> Duration>;

/// The queue of committed batches: the one point where data passes from
/// the UI-thread compositor to the server compositor.
///
/// The UI thread enqueues and rents buffers, the render thread dequeues and
/// returns them: both under a lock.
#[derive(Default)]
pub struct BatchQueue {
    batches: Mutex<VecDeque<CommittedBatch>>,
    /// Emptied stream buffers, returned by the server for reuse.
    data_pool: Mutex<Vec<BatchStreamData>>,
}

impl BatchQueue {
    pub(crate) fn enqueue(&self, batch: CommittedBatch) {
        self.batches.lock().unwrap().push_back(batch);
    }

    fn dequeue(&self) -> Option<CommittedBatch> {
        self.batches.lock().unwrap().pop_front()
    }

    /// Takes a stream buffer for a new batch.
    pub(crate) fn rent_data(&self) -> BatchStreamData {
        self.data_pool.lock().unwrap().pop().unwrap_or_default()
    }

    /// Completes the batches nothing will apply any more, without applying
    /// them: the compositor is gone and its server compositor is released,
    /// or about to be. Whoever waits for such a batch, or holds back its
    /// next commit until the batch is processed (the media context does, for
    /// every compositor at once), is answered; what the batches carry (their
    /// jobs among it) is dropped.
    ///
    /// Not from upstream, where a compositor and its server live as long as
    /// the process.
    pub(crate) fn complete_all(&self) {
        loop {
            // Outside the lock of the queue: a continuation of the batch
            // runs here.
            let Some(batch) = self.dequeue() else { break };
            batch.batch.notify_processed();
            batch.batch.notify_rendered();
        }
    }

    fn return_data(&self, mut data: BatchStreamData) {
        data.reset();
        let mut pool = self.data_pool.lock().unwrap();
        if pool.len() < 16 {
            pool.push(data);
        }
    }
}

/// The queue is what the two threads share.
const _: fn() = || {
    fn shared<T: Send + Sync>() {}
    shared::<BatchQueue>();
};

const COMMIT_GRACE_TICKS: i32 = 10;

/// A job in the queues of the server compositor: a job of a batch, or one
/// bound to the server object it was sent for. It never leaves the render
/// thread, unlike the [`ServerJob`] of a batch.
type RenderThreadJob = Box<dyn FnOnce(&ServerCompositor)>;

/// Server-side counterpart of the compositor: owns the server objects,
/// applies the batches committed by the UI thread and renders.
///
/// Confined to the render thread. UI-thread code interacts with it only by
/// enqueueing batches (and by reading the clock and the readback indices,
/// which are thread-safe values).
pub struct ServerCompositor {
    this: Weak<ServerCompositor>,
    batches: Arc<BatchQueue>,
    objects: RefCell<Vec<Option<Rc<dyn IServerObject>>>>,
    received_job_queue: RefCell<VecDeque<RenderThreadJob>>,
    disposed_in_batch: RefCell<Vec<ServerObjectId>>,
    received_post_target_job_queue: RefCell<VecDeque<RenderThreadJob>>,
    last_batch_id: Cell<i64>,
    clock: CompositorClock,
    server_now: Cell<Duration>,
    ui_thread_is_inside_render: Cell<bool>,
    /// Set by [`stop_rendering`](Self::stop_rendering).
    rendering_stopped: Cell<bool>,
    render_interface: Rc<PlatformRenderInterfaceContextManager>,
    options: CompositionOptions,
    readback: Arc<ReadbackIndices>,
    ticks_since_last_commit: Cell<i32>,
    to_notify_processed: RefCell<Vec<Arc<CompositionBatch>>>,
    to_notify_rendered: RefCell<Vec<Arc<CompositionBatch>>>,
    render_resources_invalidation_queue: RefCell<VecDeque<Rc<dyn IServerRenderResource>>>,
    render_resources_invalidation_set: RefCell<HashSet<*const ()>>,
    // TODO: parallel processing maybe
    visual_own_properties_recompute_pass: RefCell<VecDeque<Rc<ServerCompositionVisual>>>,
    visual_readback_update_pass_queue: RefCell<VecDeque<Rc<ServerCompositionVisual>>>,
    adorner_update_queue: RefCell<VecDeque<Rc<ServerCompositionVisual>>>,
    active_targets: RefCell<Vec<Rc<ServerCompositionTarget>>>,
    pools: CompositorPools,
    animations: ServerCompositorAnimations,
    render_interface_feature_cache: RefCell<Option<Rc<HashMap<TypeId, Rc<dyn Any>>>>>,
    /// The text renderer of the debug overlays. It is made from the default
    /// typeface, which belongs to the thread of the font manager: see
    /// [`diagnostic_text_renderer`](Self::diagnostic_text_renderer).
    diagnostic_text_renderer: RefCell<Option<Rc<super::DiagnosticTextRenderer>>>,
    _context_subscriptions: RefCell<Vec<Rc<dyn IDisposable>>>,
}

impl ServerCompositor {
    /// Creates the server compositor on the thread that renders. The queue of
    /// batches and the readback are what it shares with its compositor, which
    /// creates them.
    pub(crate) fn new(
        platform_graphics: Option<Arc<dyn IPlatformGraphics>>,
        options: CompositionOptions,
        batches: Arc<BatchQueue>,
        readback: Arc<ReadbackIndices>,
        clock: CompositorClock,
    ) -> Rc<ServerCompositor> {
        let compositor = Rc::new_cyclic(|this| ServerCompositor {
            this: this.clone(),
            batches,
            objects: RefCell::new(Vec::new()),
            received_job_queue: RefCell::new(VecDeque::new()),
            disposed_in_batch: RefCell::new(Vec::new()),
            received_post_target_job_queue: RefCell::new(VecDeque::new()),
            last_batch_id: Cell::new(0),
            clock,
            server_now: Cell::new(Duration::ZERO),
            ui_thread_is_inside_render: Cell::new(false),
            rendering_stopped: Cell::new(false),
            render_interface: PlatformRenderInterfaceContextManager::new(platform_graphics),
            options,
            readback,
            ticks_since_last_commit: Cell::new(0),
            to_notify_processed: RefCell::new(Vec::new()),
            to_notify_rendered: RefCell::new(Vec::new()),
            render_resources_invalidation_queue: RefCell::new(VecDeque::new()),
            render_resources_invalidation_set: RefCell::new(HashSet::new()),
            visual_own_properties_recompute_pass: RefCell::new(VecDeque::new()),
            visual_readback_update_pass_queue: RefCell::new(VecDeque::new()),
            adorner_update_queue: RefCell::new(VecDeque::new()),
            active_targets: RefCell::new(Vec::new()),
            pools: CompositorPools::default(),
            animations: ServerCompositorAnimations::new(),
            render_interface_feature_cache: RefCell::new(None),
            diagnostic_text_renderer: RefCell::new(None),
            _context_subscriptions: RefCell::new(Vec::new()),
        });
        let weak = Rc::downgrade(&compositor);
        let disposed = compositor.render_interface.context_disposed(move || {
            if let Some(compositor) = weak.upgrade() {
                compositor.rt_on_context_disposed();
            }
        });
        let weak = Rc::downgrade(&compositor);
        let created = compositor.render_interface.context_created(move |context| {
            if let Some(compositor) = weak.upgrade() {
                compositor.rt_on_context_created(&**context);
            }
        });
        *compositor._context_subscriptions.borrow_mut() = vec![disposed, created];
        compositor
    }

    /// The object pools of the tree walks.
    pub fn pools(&self) -> &CompositorPools {
        &self.pools
    }

    /// The clock items and the animated objects of the compositor.
    pub fn animations(&self) -> &ServerCompositorAnimations {
        &self.animations
    }

    /// The sequence id of the batch applied last.
    pub fn last_batch_id(&self) -> i64 {
        self.last_batch_id.get()
    }

    /// The time elapsed on the compositor clock.
    pub fn clock_elapsed(&self) -> Duration {
        (self.clock)()
    }

    /// The compositor clock itself.
    pub fn clock(&self) -> &CompositorClock {
        &self.clock
    }

    /// The compositor time of the frame being rendered.
    pub fn server_now(&self) -> Duration {
        self.server_now.get()
    }

    pub fn render_interface(&self) -> &Rc<PlatformRenderInterfaceContextManager> {
        &self.render_interface
    }

    /// The text renderer of the debug overlays, shared by the targets.
    ///
    /// It can only be created on a thread that has the font manager (the UI
    /// thread; upstream's services are global and the first overlay creates
    /// it wherever it is drawn). The compositor asks for it under the lock
    /// when debug overlays are switched on; a frame of the render thread
    /// uses what is there and draws no overlay text until then.
    pub fn diagnostic_text_renderer(&self) -> Option<Rc<super::DiagnosticTextRenderer>> {
        if self.diagnostic_text_renderer.borrow().is_none() {
            // Absent in a unit test without a platform, and on a thread
            // without services.
            {
                use crate::LocatorExtensions as _;
                crate::FerroLocator::current().get_service::<dyn crate::platform::IFontManagerImpl>()?;
            }
            let renderer = Rc::new(super::DiagnosticTextRenderer::create_default());
            // The labels of the FPS counter, shaped here on the UI thread: the counter is
            // created by the first frame of the render thread, which cannot create glyph runs
            // where it has no services (the render thread of the browser).
            for label in super::FpsCounter::LABELS {
                renderer.shape_label(label);
            }
            *self.diagnostic_text_renderer.borrow_mut() = Some(renderer);
        }
        self.diagnostic_text_renderer.borrow().clone()
    }

    pub fn options(&self) -> &CompositionOptions {
        &self.options
    }

    /// The indices of the values the UI thread reads back from the server.
    pub fn readback(&self) -> &Arc<ReadbackIndices> {
        &self.readback
    }

    pub(crate) fn update_server_time(&self) {
        self.server_now.set(self.clock_elapsed());
    }

    // --- object table -------------------------------------------------------

    /// The server object with the given id, if it is alive.
    pub fn get_object(&self, id: ServerObjectId) -> Option<Rc<dyn IServerObject>> {
        self.objects.borrow().get(id.index()).and_then(Clone::clone)
    }

    /// The server object with the given id as its concrete type.
    pub fn get<T: IServerObject>(&self, id: ServerObjectId) -> Option<Rc<T>> {
        self.get_object(id).and_then(|object| object.into_any_rc().downcast::<T>().ok())
    }

    /// The server object with the given id as an animatable object.
    pub fn get_animated_object(&self, id: ServerObjectId) -> Option<Rc<dyn super::IAnimatedServerObject>> {
        self.get_object(id).and_then(|object| object.as_animated())
    }

    /// The number of live server objects.
    pub fn object_count(&self) -> usize {
        self.objects.borrow().iter().filter(|o| o.is_some()).count()
    }

    fn insert_object(&self, id: ServerObjectId, object: Rc<dyn IServerObject>) {
        let mut objects = self.objects.borrow_mut();
        if objects.len() <= id.index() {
            objects.resize_with(id.index() + 1, || None);
        }
        debug_assert!(objects[id.index()].is_none(), "a server object id was reused while still alive");
        objects[id.index()] = Some(object);
    }

    fn remove_object(&self, id: ServerObjectId) -> Option<Rc<dyn IServerObject>> {
        self.objects.borrow_mut().get_mut(id.index()).and_then(Option::take)
    }

    // --- batches ------------------------------------------------------------

    fn apply_pending_batches(&self) {
        let mut had_batches = false;
        while let Some(mut batch) = self.batches.dequeue() {
            {
                let mut stream = BatchStreamReader::new(&mut batch.changes);
                while !stream.is_object_eof() {
                    match stream.read_object() {
                        BatchObject::Marker(BatchMarker::CreateStart) => self.read_create_jobs(&mut stream),
                        BatchObject::Marker(BatchMarker::RenderThreadJobsStart) => self.read_server_jobs(
                            &mut stream,
                            &self.received_job_queue,
                            BatchMarker::RenderThreadJobsEnd,
                        ),
                        BatchObject::Marker(BatchMarker::RenderThreadPostTargetJobsStart) => self.read_server_jobs(
                            &mut stream,
                            &self.received_post_target_job_queue,
                            BatchMarker::RenderThreadPostTargetJobsEnd,
                        ),
                        BatchObject::Marker(BatchMarker::RenderThreadDisposeStart) => {
                            self.read_dispose_jobs(&mut stream)
                        }
                        BatchObject::ServerObject(id) => {
                            let Some(target) = self.get_object(id) else {
                                panic!("a batch refers to server object {id:?}, which does not exist");
                            };
                            target.deserialize_changes(&mut stream, batch.committed_at);
                            if cfg!(debug_assertions) {
                                // Each object's changes are followed by a
                                // marker in debug builds, which catches a
                                // serializer and deserializer that disagree.
                                match stream.read_object() {
                                    BatchObject::Marker(BatchMarker::ObjectEnd) => {}
                                    _ => panic!("server object {id:?} failed to deserialize properly on the object stream"),
                                }
                                if stream.read::<u64>() != OBJECT_END_MAGIC {
                                    panic!("server object {id:?} failed to deserialize properly on the data stream");
                                }
                            }
                        }
                        _ => panic!("unexpected item at the top level of a batch object stream"),
                    }
                }
            }
            // The objects the batch released leave the table once the whole
            // batch has been read: its jobs, which follow the dispose list,
            // resolve them.
            for id in std::mem::take(&mut *self.disposed_in_batch.borrow_mut()) {
                self.remove_object(id);
            }
            self.batches.return_data(std::mem::take(&mut batch.changes));
            self.last_batch_id.set(batch.batch.sequence_id());
            self.to_notify_processed.borrow_mut().push(batch.batch);
            had_batches = true;
        }

        if had_batches {
            self.ticks_since_last_commit.set(0);
        } else if self.ticks_since_last_commit.get() < i32::MAX {
            self.ticks_since_last_commit.set(self.ticks_since_last_commit.get() + 1);
        }
    }

    fn read_create_jobs(&self, reader: &mut BatchStreamReader<'_>) {
        let Some(this) = self.this.upgrade() else { return };
        let mut count = reader.read::<i32>();
        while count > 0 {
            let Some(id) = reader.read_server_object() else {
                panic!("a batch creates a server object without an id");
            };
            let BatchObject::Create(factory) = reader.read_object() else {
                panic!("a batch creates a server object without a factory");
            };
            self.insert_object(id, factory(&this, id));
            count -= 1;
        }
    }

    fn read_server_jobs(&self, reader: &mut BatchStreamReader<'_>, queue: &RefCell<VecDeque<RenderThreadJob>>, end: BatchMarker) {
        loop {
            match reader.read_object() {
                BatchObject::Marker(marker) if marker == end => break,
                BatchObject::Job(job) => queue.borrow_mut().push_back(job),
                BatchObject::ServerObject(id) => {
                    let target = self.get_object(id);
                    match reader.read_object() {
                        BatchObject::ObjectJob(job) => {
                            queue.borrow_mut().push_back(Box::new(move |compositor| job(compositor, target)))
                        }
                        _ => panic!("a server object in the job list of a batch is not followed by its job"),
                    }
                }
                _ => panic!("unexpected item in the job list of a batch"),
            }
        }
    }

    fn read_dispose_jobs(&self, reader: &mut BatchStreamReader<'_>) {
        let mut count = reader.read::<i32>();
        while count > 0 {
            let id = reader.read_server_object();
            let dispose = reader.read::<bool>();
            let release = reader.read::<bool>();
            if let Some(id) = id {
                if let Some(object) = self.get_object(id) {
                    if dispose {
                        object.dispose();
                    }
                    if release {
                        self.disposed_in_batch.borrow_mut().push(id);
                    }
                }
            }
            count -= 1;
        }
    }

    fn execute_server_jobs(&self, queue: &RefCell<VecDeque<RenderThreadJob>>) {
        loop {
            let Some(job) = queue.borrow_mut().pop_front() else { break };
            job(self);
        }
    }

    fn notify_batches_processed(&self) {
        let processed = std::mem::take(&mut *self.to_notify_processed.borrow_mut());
        for batch in &processed {
            batch.notify_processed();
        }
        self.to_notify_rendered.borrow_mut().extend(processed);
    }

    fn notify_batches_rendered(&self) {
        let rendered = std::mem::take(&mut *self.to_notify_rendered.borrow_mut());
        for batch in &rendered {
            batch.notify_rendered();
        }
    }

    // --- passes -------------------------------------------------------------

    fn apply_enqueued_render_resource_changes_pass(&self) {
        loop {
            let Some(resource) = self.render_resources_invalidation_queue.borrow_mut().pop_front() else { break };
            // Removed before the call so a resource invalidated again while
            // observers react is queued again, like the dequeue upstream.
            resource.queued_invalidate();
        }
        self.render_resources_invalidation_set.borrow_mut().clear();
    }

    /// Queues a render resource whose observers must be told that it
    /// changed, once per frame.
    pub fn enqueue_render_resource_for_invalidation(&self, resource: Rc<dyn IServerRenderResource>) {
        let key = Rc::as_ptr(&resource) as *const ();
        if self.render_resources_invalidation_set.borrow_mut().insert(key) {
            self.render_resources_invalidation_queue.borrow_mut().push_back(resource);
        }
    }

    fn visual_own_properties_update_pass(&self) {
        loop {
            let Some(visual) = self.visual_own_properties_recompute_pass.borrow_mut().pop_front() else { break };
            visual.recompute_own_properties();
        }
    }

    pub fn enqueue_visual_for_own_properties_update_pass(&self, visual: Rc<ServerCompositionVisual>) {
        self.visual_own_properties_recompute_pass.borrow_mut().push_back(visual);
    }

    fn visual_readback_update_pass(&self) {
        if self.visual_readback_update_pass_queue.borrow().is_empty() {
            return;
        }

        // visual.HitTest is waiting for this lock to be released, so we need to be quick
        // this is why we have a queue in the first place
        let scope = self.readback.begin_write();
        let read = self.readback.read_revision();
        let write = self.readback.write_revision();
        loop {
            let Some(visual) = self.visual_readback_update_pass_queue.borrow_mut().pop_front() else { break };
            visual.update_readback(write, read);
        }
        drop(scope);
    }

    pub fn enqueue_visual_for_readback_update_pass(&self, visual: Rc<ServerCompositionVisual>) {
        self.visual_readback_update_pass_queue.borrow_mut().push_back(visual);
    }

    pub fn enqueue_adorner_update(&self, visual: Rc<ServerCompositionVisual>) {
        self.adorner_update_queue.borrow_mut().push_back(visual);
    }

    fn adorner_update_pass(&self) {
        loop {
            let Some(adorner) = self.adorner_update_queue.borrow_mut().pop_front() else { break };
            adorner.update_adorner();
        }
    }

    // --- targets ------------------------------------------------------------

    pub fn add_composition_target(&self, target: &Rc<ServerCompositionTarget>) {
        self.active_targets.borrow_mut().push(target.clone());
    }

    pub fn remove_composition_target(&self, target: &Rc<ServerCompositionTarget>) {
        let mut targets = self.active_targets.borrow_mut();
        if let Some(index) = targets.iter().position(|t| Rc::ptr_eq(t, target)) {
            targets.remove(index);
        }
    }

    pub fn create_render_target(&self, surfaces: &[std::sync::Arc<dyn IPlatformRenderSurface>]) -> Rc<dyn IRenderTarget> {
        let current = self.render_interface.ensure_current();
        let target = self.render_interface.create_render_target(surfaces);
        current.dispose();
        target
    }

    pub fn is_ready_to_create_render_target(&self, surfaces: &[std::sync::Arc<dyn IPlatformRenderSurface>]) -> bool {
        self.render_interface.is_ready_to_create_render_target(surfaces)
    }

    // --- user APIs ----------------------------------------------------------

    fn rt_on_context_created(&self, context: &dyn crate::platform::IPlatformRenderInterfaceContext) {
        *self.render_interface_feature_cache.borrow_mut() = Some(Rc::new(context.public_features()));
    }

    fn rt_on_context_disposed(&self) {
        *self.render_interface_feature_cache.borrow_mut() = None;
    }

    /// The public features of the render interface, if a backend context
    /// exists (callable from any place that can reach the compositor).
    pub fn at_try_get_cached_render_interface_features(&self) -> Option<Rc<HashMap<TypeId, Rc<dyn Any>>>> {
        self.render_interface_feature_cache.borrow().clone()
    }

    /// The public features of the render interface; creates the backend
    /// context when needed.
    pub fn rt_get_render_interface_features(&self) -> Rc<HashMap<TypeId, Rc<dyn Any>>> {
        if let Some(features) = self.render_interface_feature_cache.borrow().clone() {
            return features;
        }
        let features = Rc::new(self.render_interface.value().public_features());
        *self.render_interface_feature_cache.borrow_mut() = Some(features.clone());
        features
    }

    /// Renders a visual (and, with `render_children`, its subtree) into a
    /// new bitmap.
    pub fn create_composition_visual_snapshot(
        &self,
        visual: &Rc<ServerCompositionVisual>,
        scaling: f64,
        render_children: bool,
    ) -> std::sync::Arc<crate::platform::SharedBitmapImpl> {
        let current = self.render_interface.ensure_current();
        let size = visual.size();
        let pixel_size = PixelSize::from_size(Size::new(size.x, size.y), scaling);
        let scale_transform = Matrix::create_scale(scaling, scaling);

        let target = self.render_interface.value().create_offscreen_render_target(
            pixel_size,
            Vector::new(scaling, scaling),
            true,
        );
        {
            let mut canvas = target.create_drawing_context();
            canvas.set_transform(scale_transform);
            visual.render(&mut *canvas, LtrbRect::INFINITE, None, render_children, false, false);
            canvas.dispose();
        }

        let result: std::sync::Arc<crate::platform::SharedBitmapImpl> = match target.as_layer_with_render_context_affinity() {
            Some(affined) if affined.has_render_context_affinity() => {
                let snapshot = affined.create_non_affined_snapshot();
                target.dispose();
                snapshot
            }
            // The original returns the target itself; a layer does not
            // leave the render thread here, so its contents do.
            _ => {
                let snapshot = target.create_shared_snapshot();
                target.dispose();
                snapshot
            }
        };
        current.dispose();
        result
    }

    pub fn reset_all_gpu_resources(&self) {
        let targets = self.active_targets.borrow().clone();
        for target in targets {
            target.reset_render_target();
        }
        self.render_interface.reset();
    }

    /// Releases what the compositor draws with, while its graphics context
    /// is current: the layer and the render target of every active target,
    /// then the backend context. The context that was current before is
    /// restored. Does nothing when no backend context was ever created, and
    /// nothing for a graphics context that is lost (nothing can be released
    /// in it).
    ///
    /// For the two places where these objects go for good: before the graph
    /// of a compositor is dropped, and when the surface the compositor draws
    /// to is closed. An object of a graphics interface with a current
    /// context (OpenGL) is deleted in whatever context is current when it is
    /// dropped: with none current the call fails, and with the context of
    /// another compositor current it deletes an object of that one. Once the
    /// backend context is released here, what is dropped later touches no
    /// graphics context.
    ///
    /// Unlike [`reset_all_gpu_resources`](Self::reset_all_gpu_resources),
    /// nothing is asked for that may have to be created: it also works for
    /// graphics that are not ready, and after the compositor has taken its
    /// platform handles back.
    ///
    /// Not from upstream, where these objects are finalized by the garbage
    /// collector, on the thread of the context.
    pub fn release_gpu_resources(&self) {
        if self.render_interface.existing_backend_context().is_none() {
            return;
        }
        let gpu_context = self.render_interface.gpu_context();
        if gpu_context.as_ref().is_some_and(|context| context.is_lost()) {
            return;
        }
        let current = gpu_context.map(|context| context.ensure_current());
        let targets = self.active_targets.borrow().clone();
        for target in targets {
            target.release_render_target();
        }
        self.render_interface.reset();
        if let Some(current) = current {
            current.dispose();
        }
    }

    /// Ends the rendering of the compositor for good: from here on a frame
    /// applies the batches that arrive (so that whoever waits for one is
    /// answered), drops the jobs they carry, renders nothing and does not
    /// ask for another tick.
    ///
    /// For a compositor whose surface is closed while the compositor lives
    /// on: its graphics would be "not ready" for ever, and a compositor that
    /// is not ready asks for the next tick each time. Upstream's browser
    /// backend takes the server compositor out of the render loop at that
    /// point; here the compositor stays in the loop, because the thread of
    /// the compositor still waits for the batch that disposes its target.
    pub fn stop_rendering(&self) {
        self.rendering_stopped.set(true);
    }

    /// Whether [`stop_rendering`](Self::stop_rendering) was called.
    pub fn is_rendering_stopped(&self) -> bool {
        self.rendering_stopped.get()
    }

    pub fn invalidate_all_composition_targets(&self) {
        let targets = self.active_targets.borrow().clone();
        for target in targets {
            target.request_full_redraw();
        }
    }

    // --- rendering ----------------------------------------------------------

    /// Whether the caller may access the objects of the compositor: whether
    /// the compositor is rendering a frame. Upstream the frame is rendered
    /// under the compositor lock by the thread that holds it.
    pub fn check_access(&self) -> bool {
        self.ui_thread_is_inside_render.get()
    }

    /// Panics when the caller may not access the objects of the compositor
    /// (see [`check_access`](Self::check_access)).
    pub fn verify_access(&self) {
        if !self.check_access() {
            panic!("This object can be only accessed under compositor lock");
        }
    }

    /// [`verify_access`](Self::verify_access) of a compositor that may be
    /// gone: no frame of a compositor that is gone is being rendered.
    /// Returns the compositor.
    pub fn verify_access_of(compositor: Option<&Rc<ServerCompositor>>) -> Rc<ServerCompositor> {
        match compositor {
            Some(compositor) => {
                compositor.verify_access();
                compositor.clone()
            }
            None => panic!("This object can be only accessed under compositor lock"),
        }
    }

    /// Runs one frame: applies the pending batches, runs the global passes
    /// and the queued jobs. Returns whether another tick is needed even if
    /// nothing else wakes the render loop up.
    ///
    /// Panics on reentrancy.
    pub fn render(&self) -> bool {
        if self.ui_thread_is_inside_render.replace(true) {
            panic!("Reentrancy is not supported");
        }
        let _reset = ResetFlag(&self.ui_thread_is_inside_render);

        // While the UI thread renders, nothing may pump its dispatcher from
        // inside the frame. A frame of the render thread leaves the
        // dispatcher alone.
        let dispatcher = Dispatcher::ui_thread();
        let processing_disabled = dispatcher.check_access().then(|| dispatcher.disable_processing());
        let _notify = NotifyRendered(self);
        // Declared after `_notify`, so processing is enabled again before
        // the rendered notification runs.
        let _enable = EnableProcessing(processing_disabled);
        self.render_core()
    }

    fn execute_global_passes(&self) -> Duration {
        let compositor_global_passes_started = self.clock_elapsed();
        self.apply_pending_batches();
        self.notify_batches_processed();
        self.animations.process();
        self.apply_enqueued_render_resource_changes_pass();
        self.visual_own_properties_update_pass();
        // Adorners need to be updated after own properties recompute pass,
        // because they may depend on ancestor's transform chain to be consistent
        self.adorner_update_pass();
        self.clock_elapsed().saturating_sub(compositor_global_passes_started)
    }

    fn render_core(&self) -> bool {
        self.update_server_time();
        let compositor_global_passes_elapsed = self.execute_global_passes();

        if self.rendering_stopped.get() {
            // The jobs are dropped outside the borrow of their queue.
            let jobs = std::mem::take(&mut *self.received_job_queue.borrow_mut());
            let post_target_jobs = std::mem::take(&mut *self.received_post_target_job_queue.borrow_mut());
            drop((jobs, post_target_jobs));
            return false;
        }

        if !self.render_interface.is_ready() {
            return true;
        }
        self.render_interface.ensure_valid_backend_context();
        self.execute_server_jobs(&self.received_job_queue);

        let targets = self.active_targets.borrow().clone();
        for target in &targets {
            target.update(compositor_global_passes_elapsed);
            target.render();
        }

        self.visual_readback_update_pass();

        self.execute_server_jobs(&self.received_post_target_job_queue);

        // Request a tick if we have active animations or if there are recent batches
        if self.animations.need_next_tick() || self.ticks_since_last_commit.get() < COMMIT_GRACE_TICKS {
            return true;
        }

        // Request a tick if we had unready targets in the last tick, to check if they are ready next time
        // But skip targets that are waiting for a render loop wakeup from the platform
        let targets = self.active_targets.borrow().clone();
        for target in &targets {
            if target.is_waiting_for_ready_render_target() && !target.is_waiting_for_render_loop_wakeup() {
                return true;
            }
        }

        // Otherwise there is no need to waste CPU cycles, tell the timer to pause
        false
    }
}

/// The value written to the value stream after each object's changes in
/// debug builds.
pub(crate) const OBJECT_END_MAGIC: u64 = 0x4F42_4A45_4E44_2121;

struct ResetFlag<'a>(&'a Cell<bool>);

impl Drop for ResetFlag<'_> {
    fn drop(&mut self) {
        self.0.set(false);
    }
}

struct NotifyRendered<'a>(&'a ServerCompositor);

impl Drop for NotifyRendered<'_> {
    fn drop(&mut self) {
        self.0.notify_batches_rendered();
    }
}

/// Re-enables dispatcher processing when the frame ends, also when it
/// unwinds: the handle only re-enables on `dispose`.
struct EnableProcessing(Option<crate::threading::DispatcherProcessingDisabled>);

impl Drop for EnableProcessing {
    fn drop(&mut self) {
        if let Some(disabled) = &self.0 {
            crate::reactive::IDisposable::dispose(disabled);
        }
    }
}