From b6b51375cfd96e330d6ee22096dacf831a992aa7 Mon Sep 17 00:00:00 2001 From: Héctor Ramón Jiménez Date: Tue, 16 Apr 2024 21:50:28 +0200 Subject: Implement backpressure mechanism in `iced_winit::Proxy` --- winit/src/application.rs | 79 +++++++++++++-------------------- winit/src/multi_window.rs | 77 +++++++++++++-------------------- winit/src/proxy.rs | 108 +++++++++++++++++++++++++++++++++++++++------- 3 files changed, 152 insertions(+), 112 deletions(-) diff --git a/winit/src/application.rs b/winit/src/application.rs index 1ca80609..a447c9da 100644 --- a/winit/src/application.rs +++ b/winit/src/application.rs @@ -149,13 +149,14 @@ where let event_loop = EventLoopBuilder::with_user_event() .build() .expect("Create event loop"); - let proxy = event_loop.create_proxy(); + + let (proxy, worker) = Proxy::new(event_loop.create_proxy()); let runtime = { - let proxy = Proxy::new(event_loop.create_proxy()); let executor = E::new().map_err(Error::ExecutorCreationFailed)?; + executor.spawn(worker); - Runtime::new(executor, proxy) + Runtime::new(executor, proxy.clone()) }; let (application, init_command) = { @@ -305,7 +306,7 @@ async fn run_instance( mut compositor: C, mut renderer: A::Renderer, mut runtime: Runtime, A::Message>, - mut proxy: winit::event_loop::EventLoopProxy, + mut proxy: Proxy, mut debug: Debug, mut event_receiver: mpsc::UnboundedReceiver< winit::event::Event, @@ -370,6 +371,7 @@ async fn run_instance( let mut mouse_interaction = mouse::Interaction::default(); let mut events = Vec::new(); let mut messages = Vec::new(); + let mut user_events = 0; let mut redraw_pending = false; debug.startup_finished(); @@ -396,6 +398,7 @@ async fn run_instance( } event::Event::UserEvent(message) => { messages.push(message); + user_events += 1; } event::Event::WindowEvent { event: event::WindowEvent::RedrawRequested { .. }, @@ -593,6 +596,11 @@ async fn run_instance( if should_exit { break; } + + if user_events > 0 { + proxy.free_slots(user_events); + user_events = 0; + } } if !redraw_pending { @@ -667,7 +675,7 @@ pub fn update( runtime: &mut Runtime, A::Message>, clipboard: &mut Clipboard, should_exit: &mut bool, - proxy: &mut winit::event_loop::EventLoopProxy, + proxy: &mut Proxy, debug: &mut Debug, messages: &mut Vec, window: &winit::window::Window, @@ -717,7 +725,7 @@ pub fn run_command( runtime: &mut Runtime, A::Message>, clipboard: &mut Clipboard, should_exit: &mut bool, - proxy: &mut winit::event_loop::EventLoopProxy, + proxy: &mut Proxy, debug: &mut Debug, window: &winit::window::Window, ) where @@ -742,9 +750,7 @@ pub fn run_command( clipboard::Action::Read(tag, kind) => { let message = tag(clipboard.read(kind)); - proxy - .send_event(message) - .expect("Send message to event loop"); + proxy.send(message); } clipboard::Action::Write(contents, kind) => { clipboard.write(kind, contents); @@ -774,25 +780,16 @@ pub fn run_command( let size = window.inner_size().to_logical(window.scale_factor()); - proxy - .send_event(callback(Size::new( - size.width, - size.height, - ))) - .expect("Send message to event loop"); + proxy.send(callback(Size::new(size.width, size.height))); } window::Action::FetchMaximized(_id, callback) => { - proxy - .send_event(callback(window.is_maximized())) - .expect("Send message to event loop"); + proxy.send(callback(window.is_maximized())); } window::Action::Maximize(_id, maximized) => { window.set_maximized(maximized); } window::Action::FetchMinimized(_id, callback) => { - proxy - .send_event(callback(window.is_minimized())) - .expect("Send message to event loop"); + proxy.send(callback(window.is_minimized())); } window::Action::Minimize(_id, minimized) => { window.set_minimized(minimized); @@ -808,9 +805,7 @@ pub fn run_command( }) .ok(); - proxy - .send_event(callback(position)) - .expect("Send message to event loop"); + proxy.send(callback(position)); } window::Action::Move(_id, position) => { window.set_outer_position(winit::dpi::LogicalPosition { @@ -835,9 +830,7 @@ pub fn run_command( core::window::Mode::Hidden }; - proxy - .send_event(tag(mode)) - .expect("Send message to event loop"); + proxy.send(tag(mode)); } window::Action::ToggleMaximize(_id) => { window.set_maximized(!window.is_maximized()); @@ -865,17 +858,13 @@ pub fn run_command( } } window::Action::FetchId(_id, tag) => { - proxy - .send_event(tag(window.id().into())) - .expect("Send message to event loop"); + proxy.send(tag(window.id().into())); } window::Action::RunWithHandle(_id, tag) => { use window::raw_window_handle::HasWindowHandle; if let Ok(handle) = window.window_handle() { - proxy - .send_event(tag(&handle)) - .expect("Send message to event loop"); + proxy.send(tag(&handle)); } } @@ -888,12 +877,10 @@ pub fn run_command( &debug.overlay(), ); - proxy - .send_event(tag(window::Screenshot::new( - bytes, - state.physical_size(), - ))) - .expect("Send message to event loop."); + proxy.send(tag(window::Screenshot::new( + bytes, + state.physical_size(), + ))); } }, command::Action::System(action) => match action { @@ -901,7 +888,7 @@ pub fn run_command( #[cfg(feature = "system")] { let graphics_info = compositor.fetch_information(); - let proxy = proxy.clone(); + let mut proxy = proxy.clone(); let _ = std::thread::spawn(move || { let information = @@ -909,9 +896,7 @@ pub fn run_command( let message = _tag(information); - proxy - .send_event(message) - .expect("Send message to event loop"); + proxy.send(message); }); } } @@ -934,9 +919,7 @@ pub fn run_command( match operation.finish() { operation::Outcome::None => {} operation::Outcome::Some(message) => { - proxy - .send_event(message) - .expect("Send message to event loop"); + proxy.send(message); } operation::Outcome::Chain(next) => { current_operation = Some(next); @@ -951,9 +934,7 @@ pub fn run_command( // TODO: Error handling (?) compositor.load_font(bytes); - proxy - .send_event(tagger(Ok(()))) - .expect("Send message to event loop"); + proxy.send(tagger(Ok(()))); } command::Action::Custom(_) => { log::warn!("Unsupported custom action in `iced_winit` shell"); diff --git a/winit/src/multi_window.rs b/winit/src/multi_window.rs index 3537ac18..f832eb81 100644 --- a/winit/src/multi_window.rs +++ b/winit/src/multi_window.rs @@ -123,13 +123,13 @@ where .build() .expect("Create event loop"); - let proxy = event_loop.create_proxy(); + let (proxy, worker) = Proxy::new(event_loop.create_proxy()); let runtime = { - let proxy = Proxy::new(event_loop.create_proxy()); let executor = E::new().map_err(Error::ExecutorCreationFailed)?; + executor.spawn(worker); - Runtime::new(executor, proxy) + Runtime::new(executor, proxy.clone()) }; let (application, init_command) = { @@ -343,7 +343,7 @@ async fn run_instance( mut application: A, mut compositor: C, mut runtime: Runtime, A::Message>, - mut proxy: winit::event_loop::EventLoopProxy, + mut proxy: Proxy, mut debug: Debug, mut event_receiver: mpsc::UnboundedReceiver>, mut control_sender: mpsc::UnboundedSender, @@ -408,6 +408,7 @@ async fn run_instance( runtime.track(application.subscription().into_recipes()); let mut messages = Vec::new(); + let mut user_events = 0; debug.startup_finished(); @@ -482,6 +483,7 @@ async fn run_instance( } event::Event::UserEvent(message) => { messages.push(message); + user_events += 1; } event::Event::WindowEvent { window_id: id, @@ -803,6 +805,11 @@ async fn run_instance( &mut window_manager, cached_interfaces, )); + + if user_events > 0 { + proxy.free_slots(user_events); + user_events = 0; + } } } _ => {} @@ -845,7 +852,7 @@ fn update( runtime: &mut Runtime, A::Message>, clipboard: &mut Clipboard, control_sender: &mut mpsc::UnboundedSender, - proxy: &mut winit::event_loop::EventLoopProxy, + proxy: &mut Proxy, debug: &mut Debug, messages: &mut Vec, window_manager: &mut WindowManager, @@ -887,7 +894,7 @@ fn run_command( runtime: &mut Runtime, A::Message>, clipboard: &mut Clipboard, control_sender: &mut mpsc::UnboundedSender, - proxy: &mut winit::event_loop::EventLoopProxy, + proxy: &mut Proxy, debug: &mut Debug, window_manager: &mut WindowManager, ui_caches: &mut FxHashMap, @@ -913,9 +920,7 @@ fn run_command( clipboard::Action::Read(tag, kind) => { let message = tag(clipboard.read(kind)); - proxy - .send_event(message) - .expect("Send message to event loop"); + proxy.send(message); } clipboard::Action::Write(contents, kind) => { clipboard.write(kind, contents); @@ -967,18 +972,12 @@ fn run_command( .to_logical(window.raw.scale_factor()); proxy - .send_event(callback(Size::new( - size.width, - size.height, - ))) - .expect("Send message to event loop"); + .send(callback(Size::new(size.width, size.height))); } } window::Action::FetchMaximized(id, callback) => { if let Some(window) = window_manager.get_mut(id) { - proxy - .send_event(callback(window.raw.is_maximized())) - .expect("Send message to event loop"); + proxy.send(callback(window.raw.is_maximized())); } } window::Action::Maximize(id, maximized) => { @@ -988,9 +987,7 @@ fn run_command( } window::Action::FetchMinimized(id, callback) => { if let Some(window) = window_manager.get_mut(id) { - proxy - .send_event(callback(window.raw.is_minimized())) - .expect("Send message to event loop"); + proxy.send(callback(window.raw.is_minimized())); } } window::Action::Minimize(id, minimized) => { @@ -1012,9 +1009,7 @@ fn run_command( }) .ok(); - proxy - .send_event(callback(position)) - .expect("Send message to event loop"); + proxy.send(callback(position)); } } window::Action::Move(id, position) => { @@ -1049,9 +1044,7 @@ fn run_command( core::window::Mode::Hidden }; - proxy - .send_event(tag(mode)) - .expect("Send message to event loop"); + proxy.send(tag(mode)); } } window::Action::ToggleMaximize(id) => { @@ -1099,9 +1092,7 @@ fn run_command( } window::Action::FetchId(id, tag) => { if let Some(window) = window_manager.get_mut(id) { - proxy - .send_event(tag(window.raw.id().into())) - .expect("Send message to event loop"); + proxy.send(tag(window.raw.id().into())); } } window::Action::RunWithHandle(id, tag) => { @@ -1111,9 +1102,7 @@ fn run_command( .get_mut(id) .and_then(|window| window.raw.window_handle().ok()) { - proxy - .send_event(tag(&handle)) - .expect("Send message to event loop"); + proxy.send(tag(&handle)); } } window::Action::Screenshot(id, tag) => { @@ -1126,12 +1115,10 @@ fn run_command( &debug.overlay(), ); - proxy - .send_event(tag(window::Screenshot::new( - bytes, - window.state.physical_size(), - ))) - .expect("Event loop doesn't exist."); + proxy.send(tag(window::Screenshot::new( + bytes, + window.state.physical_size(), + ))); } } }, @@ -1140,7 +1127,7 @@ fn run_command( #[cfg(feature = "system")] { let graphics_info = compositor.fetch_information(); - let proxy = proxy.clone(); + let mut proxy = proxy.clone(); let _ = std::thread::spawn(move || { let information = @@ -1148,9 +1135,7 @@ fn run_command( let message = _tag(information); - proxy - .send_event(message) - .expect("Event loop doesn't exist."); + proxy.send(message); }); } } @@ -1175,9 +1160,7 @@ fn run_command( match operation.finish() { operation::Outcome::None => {} operation::Outcome::Some(message) => { - proxy - .send_event(message) - .expect("Event loop doesn't exist."); + proxy.send(message); // operation completed, don't need to try to operate on rest of UIs break 'operate; @@ -1197,9 +1180,7 @@ fn run_command( // TODO: Error handling (?) compositor.load_font(bytes.clone()); - proxy - .send_event(tagger(Ok(()))) - .expect("Send message to event loop"); + proxy.send(tagger(Ok(()))); } command::Action::Custom(_) => { log::warn!("Unsupported custom action in `iced_winit` shell"); diff --git a/winit/src/proxy.rs b/winit/src/proxy.rs index 1d6c48bb..a35e8a27 100644 --- a/winit/src/proxy.rs +++ b/winit/src/proxy.rs @@ -1,28 +1,101 @@ use crate::futures::futures::{ channel::mpsc, + stream, task::{Context, Poll}, - Sink, + Future, Sink, StreamExt, }; use std::pin::Pin; -/// An event loop proxy that implements `Sink`. +/// An event loop proxy with backpressure that implements `Sink`. #[derive(Debug)] pub struct Proxy { raw: winit::event_loop::EventLoopProxy, + sender: mpsc::Sender, + notifier: mpsc::Sender, } impl Clone for Proxy { fn clone(&self) -> Self { Self { raw: self.raw.clone(), + sender: self.sender.clone(), + notifier: self.notifier.clone(), } } } impl Proxy { + const MAX_SIZE: usize = 100; + /// Creates a new [`Proxy`] from an `EventLoopProxy`. - pub fn new(raw: winit::event_loop::EventLoopProxy) -> Self { - Self { raw } + pub fn new( + raw: winit::event_loop::EventLoopProxy, + ) -> (Self, impl Future) { + let (notifier, processed) = mpsc::channel(Self::MAX_SIZE); + let (sender, receiver) = mpsc::channel(Self::MAX_SIZE); + let proxy = raw.clone(); + + let worker = async move { + enum Item { + MessageProduced(T), + BatchProcessed(usize), + } + + let mut receiver = receiver.map(Item::MessageProduced); + let mut processed = processed.map(Item::BatchProcessed); + + let mut count = 0; + + loop { + if count < Self::MAX_SIZE { + let mut stream = + stream::select(receiver.by_ref(), processed.by_ref()); + + match stream.select_next_some().await { + Item::MessageProduced(message) => { + let _ = proxy.send_event(message); + + count += 1; + } + Item::BatchProcessed(amount) => { + count = count.saturating_sub(amount); + } + } + } else if let Item::BatchProcessed(amount) = + processed.select_next_some().await + { + count = count.saturating_sub(amount); + } + } + }; + + ( + Self { + raw, + sender, + notifier, + }, + worker, + ) + } + + /// Sends a `Message` to the event loop. + /// + /// Note: This skips the backpressure mechanism with an unbounded + /// channel. Use sparingly! + pub fn send(&mut self, message: Message) + where + Message: std::fmt::Debug, + { + self.raw + .send_event(message) + .expect("Send message to event loop"); + } + + /// Frees an amount of slots for additional messages to be queued in + /// this [`Proxy`]. + pub fn free_slots(&mut self, amount: usize) { + let _ = self.notifier.start_send(amount); } } @@ -30,32 +103,37 @@ impl Sink for Proxy { type Error = mpsc::SendError; fn poll_ready( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, ) -> Poll> { - Poll::Ready(Ok(())) + self.sender.poll_ready(cx) } fn start_send( - self: Pin<&mut Self>, + mut self: Pin<&mut Self>, message: Message, ) -> Result<(), Self::Error> { - let _ = self.raw.send_event(message); - - Ok(()) + self.sender.start_send(message) } fn poll_flush( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, ) -> Poll> { - Poll::Ready(Ok(())) + match self.sender.poll_ready(cx) { + Poll::Ready(Err(ref e)) if e.is_disconnected() => { + // If the receiver disconnected, we consider the sink to be flushed. + Poll::Ready(Ok(())) + } + x => x, + } } fn poll_close( - self: Pin<&mut Self>, + mut self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll> { + self.sender.disconnect(); Poll::Ready(Ok(())) } } -- cgit