Web: improve custom cursor handling and add animated cursors (#3384)

src/platform_impl/web/async/abortable.rs

@@ -0,0 +1,102 @@
+use std::error::Error;
+use std::fmt::{self, Display, Formatter};
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use pin_project::pin_project;
+
+use super::AtomicWaker;
+
+#[pin_project]
+pub struct Abortable<F: Future> {
+    #[pin]
+    future: F,
+    shared: Arc<Shared>,
+}
+
+impl<F: Future> Abortable<F> {
+    pub fn new(handle: AbortHandle, future: F) -> Self {
+        Self {
+            future,
+            shared: handle.0,
+        }
+    }
+}
+
+impl<F: Future> Future for Abortable<F> {
+    type Output = Result<F::Output, Aborted>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.shared.aborted.load(Ordering::Relaxed) {
+            return Poll::Ready(Err(Aborted));
+        }
+
+        if let Poll::Ready(value) = self.as_mut().project().future.poll(cx) {
+            return Poll::Ready(Ok(value));
+        }
+
+        self.shared.waker.register(cx.waker());
+
+        if self.shared.aborted.load(Ordering::Relaxed) {
+            return Poll::Ready(Err(Aborted));
+        }
+
+        Poll::Pending
+    }
+}
+
+#[derive(Debug)]
+struct Shared {
+    waker: AtomicWaker,
+    aborted: AtomicBool,
+}
+
+#[derive(Clone, Debug)]
+pub struct AbortHandle(Arc<Shared>);
+
+impl AbortHandle {
+    pub fn new() -> Self {
+        Self(Arc::new(Shared {
+            waker: AtomicWaker::new(),
+            aborted: AtomicBool::new(false),
+        }))
+    }
+
+    pub fn abort(&self) {
+        self.0.aborted.store(true, Ordering::Relaxed);
+        self.0.waker.wake()
+    }
+}
+
+#[derive(Debug)]
+pub struct DropAbortHandle(AbortHandle);
+
+impl DropAbortHandle {
+    pub fn new(handle: AbortHandle) -> Self {
+        Self(handle)
+    }
+
+    pub fn handle(&self) -> AbortHandle {
+        self.0.clone()
+    }
+}
+
+impl Drop for DropAbortHandle {
+    fn drop(&mut self) {
+        self.0.abort()
+    }
+}
+
+#[derive(Copy, Clone, Debug, Eq, Hash, PartialEq)]
+pub struct Aborted;
+
+impl Display for Aborted {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "`Abortable` future has been aborted")
+    }
+}
+
+impl Error for Aborted {}

src/platform_impl/web/async/atomic_waker.rs

@@ -0,0 +1,35 @@
+use std::cell::RefCell;
+use std::ops::Deref;
+use std::task::Waker;
+
+#[derive(Debug)]
+pub struct AtomicWaker(RefCell<Option<Waker>>);
+
+impl AtomicWaker {
+    pub const fn new() -> Self {
+        Self(RefCell::new(None))
+    }
+
+    pub fn register(&self, waker: &Waker) {
+        let mut this = self.0.borrow_mut();
+
+        if let Some(old_waker) = this.deref() {
+            if old_waker.will_wake(waker) {
+                return;
+            }
+        }
+
+        *this = Some(waker.clone());
+    }
+
+    pub fn wake(&self) {
+        if let Some(waker) = self.0.borrow_mut().take() {
+            waker.wake();
+        }
+    }
+}
+
+// SAFETY: Wasm without the `atomics` target feature is single-threaded.
+unsafe impl Send for AtomicWaker {}
+// SAFETY: Wasm without the `atomics` target feature is single-threaded.
+unsafe impl Sync for AtomicWaker {}

src/platform_impl/web/async/channel.rs

@@ -1,23 +1,24 @@
-use atomic_waker::AtomicWaker;
 use std::future;
 use std::rc::Rc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{self, Receiver, RecvError, SendError, Sender, TryRecvError};
+use std::sync::mpsc::{self, RecvError, SendError, TryRecvError};
 use std::sync::{Arc, Mutex};
 use std::task::Poll;
 
-pub fn channel<T>() -> (AsyncSender<T>, AsyncReceiver<T>) {
+use super::AtomicWaker;
+
+pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let (sender, receiver) = mpsc::channel();
     let shared = Arc::new(Shared {
         closed: AtomicBool::new(false),
         waker: AtomicWaker::new(),
     });
 
-    let sender = AsyncSender(Arc::new(SenderInner {
+    let sender = Sender(Arc::new(SenderInner {
         sender: Mutex::new(sender),
         shared: Arc::clone(&shared),
     }));
-    let receiver = AsyncReceiver {
+    let receiver = Receiver {
         receiver: Rc::new(receiver),
         shared,
     };
@@ -25,18 +26,18 @@ pub fn channel<T>() -> (AsyncSender<T>, AsyncReceiver<T>) {
     (sender, receiver)
 }
 
-pub struct AsyncSender<T>(Arc<SenderInner<T>>);
+pub struct Sender<T>(Arc<SenderInner<T>>);
 
 struct SenderInner<T> {
     // We need to wrap it into a `Mutex` to make it `Sync`. So the sender can't
     // be accessed on the main thread, as it could block. Additionally we need
     // to wrap `Sender` in an `Arc` to make it clonable on the main thread without
     // having to block.
-    sender: Mutex<Sender<T>>,
+    sender: Mutex<mpsc::Sender<T>>,
     shared: Arc<Shared>,
 }
 
-impl<T> AsyncSender<T> {
+impl<T> Sender<T> {
     pub fn send(&self, event: T) -> Result<(), SendError<T>> {
         self.0.sender.lock().unwrap().send(event)?;
         self.0.shared.waker.wake();
@@ -52,7 +53,7 @@ impl<T> SenderInner<T> {
     }
 }
 
-impl<T> Clone for AsyncSender<T> {
+impl<T> Clone for Sender<T> {
     fn clone(&self) -> Self {
         Self(Arc::clone(&self.0))
     }
@@ -64,12 +65,12 @@ impl<T> Drop for SenderInner<T> {
     }
 }
 
-pub struct AsyncReceiver<T> {
-    receiver: Rc<Receiver<T>>,
+pub struct Receiver<T> {
+    receiver: Rc<mpsc::Receiver<T>>,
     shared: Arc<Shared>,
 }
 
-impl<T> AsyncReceiver<T> {
+impl<T> Receiver<T> {
     pub async fn next(&self) -> Result<T, RecvError> {
         future::poll_fn(|cx| match self.receiver.try_recv() {
             Ok(event) => Poll::Ready(Ok(event)),
@@ -102,7 +103,7 @@ impl<T> AsyncReceiver<T> {
     }
 }
 
-impl<T> Clone for AsyncReceiver<T> {
+impl<T> Clone for Receiver<T> {
     fn clone(&self) -> Self {
         Self {
             receiver: Rc::clone(&self.receiver),
@@ -111,7 +112,7 @@ impl<T> Clone for AsyncReceiver<T> {
     }
 }
 
-impl<T> Drop for AsyncReceiver<T> {
+impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
         self.shared.closed.store(true, Ordering::Relaxed);
     }

src/platform_impl/web/async/concurrent_queue.rs

@@ -0,0 +1,55 @@
+use std::cell::{Cell, RefCell};
+
+#[derive(Debug)]
+pub struct ConcurrentQueue<T> {
+    queue: RefCell<Vec<T>>,
+    closed: Cell<bool>,
+}
+
+pub enum PushError<T> {
+    #[allow(dead_code)]
+    Full(T),
+    Closed(T),
+}
+
+pub enum PopError {
+    Empty,
+    Closed,
+}
+
+impl<T> ConcurrentQueue<T> {
+    pub fn unbounded() -> Self {
+        Self {
+            queue: RefCell::new(Vec::new()),
+            closed: Cell::new(false),
+        }
+    }
+
+    pub fn push(&self, value: T) -> Result<(), PushError<T>> {
+        if self.closed.get() {
+            return Err(PushError::Closed(value));
+        }
+
+        self.queue.borrow_mut().push(value);
+        Ok(())
+    }
+
+    pub fn pop(&self) -> Result<T, PopError> {
+        self.queue.borrow_mut().pop().ok_or_else(|| {
+            if self.closed.get() {
+                PopError::Closed
+            } else {
+                PopError::Empty
+            }
+        })
+    }
+
+    pub fn close(&self) -> bool {
+        !self.closed.replace(true)
+    }
+}
+
+// SAFETY: Wasm without the `atomics` target feature is single-threaded.
+unsafe impl<T> Send for ConcurrentQueue<T> {}
+// SAFETY: Wasm without the `atomics` target feature is single-threaded.
+unsafe impl<T> Sync for ConcurrentQueue<T> {}

src/platform_impl/web/async/dispatcher.rs

@@ -1,11 +1,11 @@
 use super::super::main_thread::MainThreadMarker;
-use super::{channel, AsyncReceiver, AsyncSender, Wrapper};
+use super::{channel, Receiver, Sender, Wrapper};
 use std::{
     cell::Ref,
     sync::{Arc, Condvar, Mutex},
 };
 
-pub struct Dispatcher<T: 'static>(Wrapper<true, T, AsyncSender<Closure<T>>, Closure<T>>);
+pub struct Dispatcher<T: 'static>(Wrapper<true, T, Sender<Closure<T>>, Closure<T>>);
 
 struct Closure<T>(Box<dyn FnOnce(&T) + Send>);
 
@@ -85,8 +85,8 @@ impl<T> Dispatcher<T> {
 }
 
 pub struct DispatchRunner<T: 'static> {
-    wrapper: Wrapper<true, T, AsyncSender<Closure<T>>, Closure<T>>,
-    receiver: AsyncReceiver<Closure<T>>,
+    wrapper: Wrapper<true, T, Sender<Closure<T>>, Closure<T>>,
+    receiver: Receiver<Closure<T>>,
 }
 
 impl<T> DispatchRunner<T> {

src/platform_impl/web/async/mod.rs

@@ -1,9 +1,19 @@
+mod abortable;
+#[cfg(not(target_feature = "atomics"))]
+mod atomic_waker;
 mod channel;
+#[cfg(not(target_feature = "atomics"))]
+mod concurrent_queue;
 mod dispatcher;
+mod notifier;
 mod waker;
 mod wrapper;
 
-pub use self::channel::{channel, AsyncReceiver, AsyncSender};
+pub use self::abortable::{AbortHandle, Abortable, DropAbortHandle};
+pub use self::channel::{channel, Receiver, Sender};
 pub use self::dispatcher::{DispatchRunner, Dispatcher};
+pub use self::notifier::{Notified, Notifier};
 pub use self::waker::{Waker, WakerSpawner};
 use self::wrapper::Wrapper;
+use atomic_waker::AtomicWaker;
+use concurrent_queue::{ConcurrentQueue, PushError};

src/platform_impl/web/async/notifier.rs

@@ -0,0 +1,78 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::OnceLock;
+use std::task::Context;
+use std::task::Poll;
+use std::task::Waker;
+
+use super::{ConcurrentQueue, PushError};
+
+#[derive(Debug)]
+pub struct Notifier<T: Clone>(Arc<Inner<T>>);
+
+impl<T: Clone> Notifier<T> {
+    pub fn new() -> Self {
+        Self(Arc::new(Inner {
+            queue: ConcurrentQueue::unbounded(),
+            value: OnceLock::new(),
+        }))
+    }
+
+    pub fn notify(self, value: T) {
+        if self.0.value.set(value).is_err() {
+            unreachable!("value set before")
+        }
+
+        self.0.queue.close();
+
+        while let Ok(waker) = self.0.queue.pop() {
+            waker.wake()
+        }
+    }
+
+    pub fn notified(&self) -> Notified<T> {
+        Notified(Some(Arc::clone(&self.0)))
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct Notified<T: Clone>(Option<Arc<Inner<T>>>);
+
+impl<T: Clone> Future for Notified<T> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.0.take().expect("`Receiver` polled after completion");
+
+        if this.value.get().is_none() {
+            match this.queue.push(cx.waker().clone()) {
+                Ok(()) => {
+                    if this.value.get().is_none() {
+                        self.0 = Some(this);
+                        return Poll::Pending;
+                    }
+                }
+                Err(PushError::Closed(_)) => (),
+                Err(PushError::Full(_)) => {
+                    unreachable!("found full queue despite using unbounded queue")
+                }
+            }
+        }
+
+        let (Ok(Some(value)) | Err(Some(value))) = Arc::try_unwrap(this)
+            .map(|mut inner| inner.value.take())
+            .map_err(|this| this.value.get().cloned())
+        else {
+            unreachable!("found no value despite being ready")
+        };
+
+        Poll::Ready(value)
+    }
+}
+
+#[derive(Debug)]
+struct Inner<T> {
+    queue: ConcurrentQueue<Waker>,
+    value: OnceLock<T>,
+}

src/platform_impl/web/async/waker.rs

@@ -1,11 +1,11 @@
-use super::super::main_thread::MainThreadMarker;
-use super::Wrapper;
-use atomic_waker::AtomicWaker;
 use std::future;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::task::Poll;
 
+use super::super::main_thread::MainThreadMarker;
+use super::{AtomicWaker, Wrapper};
+
 pub struct WakerSpawner<T: 'static>(Wrapper<false, Handler<T>, Sender, usize>);
 
 pub struct Waker<T: 'static>(Wrapper<false, Handler<T>, Sender, usize>);

src/platform_impl/web/event_loop/runner.rs

@@ -1,4 +1,3 @@
-use super::super::cursor::CustomCursorHandle;
 use super::super::main_thread::MainThreadMarker;
 use super::super::DeviceId;
 use super::{backend, state::State};
@@ -140,16 +139,6 @@ impl Runner {
                     )
                 }
             }
-            EventWrapper::CursorReady(result) => {
-                for (_, canvas, _) in runner.0.all_canvases.borrow().deref() {
-                    if let Some(canvas) = canvas.upgrade() {
-                        canvas
-                            .borrow_mut()
-                            .cursor
-                            .handle_cursor_ready(result.clone())
-                    }
-                }
-            }
         }
     }
 }
@@ -822,19 +811,6 @@ impl Shared {
     pub(crate) fn waker(&self) -> Waker<Weak<Execution>> {
         self.0.proxy_spawner.waker()
     }
-
-    pub(crate) fn weak(&self) -> WeakShared {
-        WeakShared(Rc::downgrade(&self.0))
-    }
-}
-
-#[derive(Clone, Debug)]
-pub(crate) struct WeakShared(Weak<Execution>);
-
-impl WeakShared {
-    pub(crate) fn upgrade(&self) -> Option<Shared> {
-        self.0.upgrade().map(Shared)
-    }
 }
 
 pub(crate) enum EventWrapper {
@@ -844,7 +820,6 @@ pub(crate) enum EventWrapper {
         size: PhysicalSize<u32>,
         scale: f64,
     },
-    CursorReady(Result<CustomCursorHandle, CustomCursorHandle>),
 }
 
 impl From<Event<()>> for EventWrapper {

src/platform_impl/web/main_thread.rs

@@ -6,7 +6,7 @@ use std::sync::OnceLock;
 use wasm_bindgen::prelude::wasm_bindgen;
 use wasm_bindgen::{JsCast, JsValue};
 
-use super::r#async::{self, AsyncSender};
+use super::r#async::{self, Sender};
 
 thread_local! {
     static MAIN_THREAD: bool = {
@@ -85,7 +85,7 @@ impl<T> Drop for MainThreadSafe<T> {
 unsafe impl<T> Send for MainThreadSafe<T> {}
 unsafe impl<T> Sync for MainThreadSafe<T> {}
 
-static DROP_HANDLER: OnceLock<AsyncSender<DropBox>> = OnceLock::new();
+static DROP_HANDLER: OnceLock<Sender<DropBox>> = OnceLock::new();
 
 struct DropBox(#[allow(dead_code)] Box<dyn Any>);
 

src/platform_impl/web/mod.rs

@@ -43,3 +43,4 @@ pub(crate) use crate::icon::NoIcon as PlatformIcon;
 pub(crate) use crate::platform_impl::Fullscreen;
 pub(crate) use cursor::CustomCursor as PlatformCustomCursor;
 pub(crate) use cursor::CustomCursorBuilder as PlatformCustomCursorBuilder;
+pub(crate) use cursor::CustomCursorFuture;

src/platform_impl/web/web_sys/canvas.rs

@@ -18,7 +18,6 @@ use crate::platform_impl::{OsError, PlatformSpecificWindowBuilderAttributes};
 use crate::window::{WindowAttributes, WindowId as RootWindowId};
 
 use super::super::cursor::CursorHandler;
-use super::super::event_loop::runner::WeakShared;
 use super::super::main_thread::MainThreadMarker;
 use super::super::WindowId;
 use super::animation_frame::AnimationFrameHandler;
@@ -71,7 +70,6 @@ pub struct Style {
 impl Canvas {
     pub(crate) fn create(
         main_thread: MainThreadMarker,
-        runner: WeakShared,
         id: WindowId,
         window: web_sys::Window,
         document: Document,
@@ -111,7 +109,7 @@ impl Canvas {
 
         let style = Style::new(&window, &canvas);
 
-        let cursor = CursorHandler::new(main_thread, runner, style.clone());
+        let cursor = CursorHandler::new(main_thread, canvas.clone(), style.clone());
 
         let common = Common {
             window: window.clone(),

src/platform_impl/web/window.rs

@@ -39,7 +39,6 @@ impl Window {
         let document = target.runner.document();
         let canvas = backend::Canvas::create(
             target.runner.main_thread(),
-            target.runner.weak(),
             id,
             window.clone(),
             document.clone(),