WIP Update wayland-rs and calloop

Cargo.toml

@@ -23,20 +23,21 @@ log = "0.4"
 memmap2 = "0.7.0"
 nix = { version = "0.26.1", default-features = false, features = ["fs", "mman"] }
 thiserror = "1.0.30"
-wayland-backend = "0.1.0"
-wayland-client = "0.30.2"
-wayland-cursor = "0.30.0"
-wayland-protocols = { version = "0.30.1", features = ["client", "staging", "unstable"] }
-wayland-protocols-wlr = { version = "0.1.0", features = ["client"] }
-wayland-scanner = "0.30.0"
-wayland-csd-frame = { version = "0.2.2", default-features = false, features = ["wayland-backend_0_1"] }
+wayland-backend = "0.3.0"
+wayland-client = "0.31.0"
+wayland-cursor = "0.31.0"
+wayland-protocols = { version = "0.31.0", features = ["client", "staging", "unstable"] }
+wayland-protocols-wlr = { version = "0.2.0", features = ["client"] }
+wayland-scanner = "0.31.0"
+# wayland-csd-frame = { version = "0.2.2", default-features = false, features = ["wayland-backend_0_3"] }
+wayland-csd-frame = { git = "https://github.com/ids1024/wayland-csd-frame", branch = "wayland_backend_0_3", default-features = false, features = ["wayland-backend_0_3"] }
 
 xkbcommon = { version = "0.5", optional = true, features = ["wayland"] }
-calloop = { version = "0.10.5", optional = true }
+calloop = { version = "0.11.0", optional = true }
 
 [features]
 default = ["calloop", "xkbcommon"]
-calloop = ["dep:calloop", "wayland-client/calloop"]
+calloop = ["dep:calloop"]
 xkbcommon = ["dep:xkbcommon", "pkg-config"]
 
 [build-dependencies]
@@ -55,3 +56,7 @@ pollster = "0.2.5"
 [[example]]
 name = "wgpu"
 required-features = ["wayland-backend/client_system"]
+
+[patch.crates-io]
+wayland-backend = { git = "https://github.com/ids1024/wayland-rs", branch = "client-as-fd" }
+wayland-client = { git = "https://github.com/ids1024/wayland-rs", branch = "client-as-fd" }

examples/simple_window.rs

@@ -2,6 +2,7 @@ use std::{convert::TryInto, time::Duration};
 
 use calloop::{EventLoop, LoopHandle};
 use smithay_client_toolkit::{
+    calloop::WaylandSource,
     compositor::{CompositorHandler, CompositorState},
     delegate_compositor, delegate_keyboard, delegate_output, delegate_pointer, delegate_registry,
     delegate_seat, delegate_shm, delegate_xdg_shell, delegate_xdg_window,
@@ -28,7 +29,7 @@ use smithay_client_toolkit::{
 use wayland_client::{
     globals::registry_queue_init,
     protocol::{wl_keyboard, wl_output, wl_pointer, wl_seat, wl_shm, wl_surface},
-    Connection, QueueHandle, WaylandSource,
+    Connection, QueueHandle,
 };
 
 fn main() {

src/calloop.rs

@@ -0,0 +1,200 @@
+//! Utilities for using an [`EventQueue`] from wayland-client with an event loop that performs polling with
+//! [`calloop`](https://crates.io/crates/calloop).
+
+use std::io;
+
+use calloop::{
+    generic::Generic, EventSource, InsertError, Interest, LoopHandle, Mode, Poll, PostAction,
+    Readiness, RegistrationToken, Token, TokenFactory,
+};
+use nix::errno::Errno;
+use wayland_backend::client::{ReadEventsGuard, WaylandError};
+use wayland_client::{DispatchError, EventQueue};
+
+/// An adapter to insert an [`EventQueue`] into a calloop [`EventLoop`](calloop::EventLoop).
+///
+/// This type implements [`EventSource`] which generates an event whenever events on the event queue need to be
+/// dispatched. The event queue available in the callback calloop registers may be used to dispatch pending
+/// events using [`EventQueue::dispatch_pending`].
+///
+/// [`WaylandSource::insert`] can be used to insert this source into an event loop and automatically dispatch
+/// pending events on the event queue.
+#[derive(Debug)]
+pub struct WaylandSource<D> {
+    queue: Generic<EventQueue<D>>,
+    read_guard: Option<ReadEventsGuard>,
+}
+
+impl<D> WaylandSource<D> {
+    /// Wrap an [`EventQueue`] as a [`WaylandSource`].
+    pub fn new(queue: EventQueue<D>) -> Result<WaylandSource<D>, WaylandError> {
+        let queue = Generic::new(queue, Interest::READ, Mode::Level);
+
+        Ok(WaylandSource { queue, read_guard: None })
+    }
+
+    /// Access the underlying event queue
+    ///
+    /// Note that you should be careful when interacting with it if you invoke methods that
+    /// interact with the wayland socket (such as `dispatch()` or `prepare_read()`). These may
+    /// interfere with the proper waking up of this event source in the event loop.
+    pub fn queue(&mut self) -> &mut EventQueue<D> {
+        &mut self.queue.file
+    }
+
+    /// Insert this source into the given event loop.
+    ///
+    /// This adapter will pass the event loop's shared data as the `D` type for the event loop.
+    pub fn insert(self, handle: LoopHandle<D>) -> Result<RegistrationToken, InsertError<Self>>
+    where
+        D: 'static,
+    {
+        handle.insert_source(self, |_, queue, data| queue.dispatch_pending(data))
+    }
+}
+
+impl<D> EventSource for WaylandSource<D> {
+    type Event = ();
+
+    /// The underlying event queue.
+    ///
+    /// You should call [`EventQueue::dispatch_pending`] inside your callback using this queue.
+    type Metadata = EventQueue<D>;
+    type Ret = Result<usize, DispatchError>;
+    type Error = calloop::Error;
+
+    fn process_events<F>(
+        &mut self,
+        readiness: Readiness,
+        token: Token,
+        mut callback: F,
+    ) -> Result<PostAction, Self::Error>
+    where
+        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
+    {
+        let read_guard = &mut self.read_guard;
+
+        let action = self.queue.process_events(readiness, token, |_, queue| {
+            // 1. read events from the socket if any are available
+            if let Some(guard) = read_guard.take() {
+                // might be None if some other thread read events before us, concurrently
+                if let Err(WaylandError::Io(err)) = guard.read() {
+                    if err.kind() != io::ErrorKind::WouldBlock {
+                        return Err(err);
+                    }
+                }
+            }
+
+            // 2. dispatch any pending events in the queue
+            // This is done to ensure we are not waiting for messages that are already in the buffer.
+            Self::loop_callback_pending(queue, &mut callback)?;
+            *read_guard = queue.prepare_read();
+
+            // 3. Once dispatching is finished, flush the responses to the compositor
+            if let Err(WaylandError::Io(e)) = queue.flush() {
+                if e.kind() != io::ErrorKind::WouldBlock {
+                    // in case of error, forward it and fast-exit
+                    return Err(e);
+                }
+                // WouldBlock error means the compositor could not process all our messages
+                // quickly. Either it is slowed down or we are a spammer.
+                // Should not really happen, if it does we do nothing and will flush again later
+            }
+
+            Ok(PostAction::Continue)
+        })?;
+
+        Ok(action)
+    }
+
+    fn register(
+        &mut self,
+        poll: &mut Poll,
+        token_factory: &mut TokenFactory,
+    ) -> calloop::Result<()> {
+        self.queue.register(poll, token_factory)
+    }
+
+    fn reregister(
+        &mut self,
+        poll: &mut Poll,
+        token_factory: &mut TokenFactory,
+    ) -> calloop::Result<()> {
+        self.queue.reregister(poll, token_factory)
+    }
+
+    fn unregister(&mut self, poll: &mut Poll) -> calloop::Result<()> {
+        self.queue.unregister(poll)
+    }
+
+    fn pre_run<F>(&mut self, mut callback: F) -> calloop::Result<()>
+    where
+        F: FnMut((), &mut Self::Metadata) -> Self::Ret,
+    {
+        debug_assert!(self.read_guard.is_none());
+
+        // flush the display before starting to poll
+        if let Err(WaylandError::Io(err)) = self.queue().flush() {
+            if err.kind() != io::ErrorKind::WouldBlock {
+                // in case of error, don't prepare a read, if the error is persistent, it'll trigger in other
+                // wayland methods anyway
+                log::error!("Error trying to flush the wayland display: {}", err);
+                return Err(err.into());
+            }
+        }
+
+        // ensure we are not waiting for messages that are already in the buffer.
+        Self::loop_callback_pending(&mut self.queue(), &mut callback)?;
+        self.read_guard = self.queue().prepare_read();
+
+        Ok(())
+    }
+
+    fn post_run<F>(&mut self, _: F) -> calloop::Result<()>
+    where
+        F: FnMut((), &mut Self::Metadata) -> Self::Ret,
+    {
+        // Drop implementation of ReadEventsGuard will do cleanup
+        self.read_guard.take();
+        Ok(())
+    }
+}
+
+impl<D> WaylandSource<D> {
+    /// Loop over the callback until all pending messages have been dispatched.
+    fn loop_callback_pending<F>(queue: &mut EventQueue<D>, callback: &mut F) -> io::Result<()>
+    where
+        F: FnMut((), &mut EventQueue<D>) -> Result<usize, DispatchError>,
+    {
+        // Loop on the callback until no pending events are left.
+        loop {
+            match callback((), queue) {
+                // No more pending events.
+                Ok(0) => break Ok(()),
+
+                Ok(_) => continue,
+
+                Err(DispatchError::Backend(WaylandError::Io(err))) => {
+                    return Err(err);
+                }
+
+                Err(DispatchError::Backend(WaylandError::Protocol(err))) => {
+                    log::error!("Protocol error received on display: {}", err);
+
+                    break Err(Errno::EPROTO.into());
+                }
+
+                Err(DispatchError::BadMessage { interface, sender_id, opcode }) => {
+                    log::error!(
+                        "Bad message on interface \"{}\": (sender_id: {}, opcode: {})",
+                        interface,
+                        sender_id,
+                        opcode,
+                    );
+
+                    break Err(Errno::EPROTO.into());
+                }
+            }
+        }
+    }
+}

src/compositor.rs

@@ -423,7 +423,7 @@ impl wayland_client::backend::ObjectData for RegionData {
         _: &wayland_client::backend::Backend,
         _: wayland_client::backend::protocol::Message<
             wayland_client::backend::ObjectId,
-            wayland_backend::io_lifetimes::OwnedFd,
+            std::os::unix::io::OwnedFd,
         >,
     ) -> Option<Arc<(dyn wayland_client::backend::ObjectData + 'static)>> {
         unreachable!("wl_region has no events");

src/data_device_manager/data_offer.rs

@@ -1,6 +1,6 @@
 use std::{
     ops::{Deref, DerefMut},
-    os::unix::prelude::{FromRawFd, RawFd},
+    os::unix::prelude::{BorrowedFd, FromRawFd, RawFd},
     sync::{Arc, Mutex},
 };
 
@@ -407,7 +407,7 @@ pub fn receive(offer: &WlDataOffer, mime_type: String) -> std::io::Result<ReadPi
     // create a pipe
     let (readfd, writefd) = pipe2(OFlag::O_CLOEXEC)?;
 
-    offer.receive(mime_type, writefd);
+    offer.receive(mime_type, unsafe { BorrowedFd::borrow_raw(writefd) });
 
     if let Err(err) = close(writefd) {
         log::warn!("Failed to close write pipe: {}", err);
@@ -434,7 +434,7 @@ pub fn receive(offer: &WlDataOffer, mime_type: String) -> std::io::Result<ReadPi
 pub unsafe fn receive_to_fd(offer: &WlDataOffer, mime_type: String, writefd: RawFd) {
     use nix::unistd::close;
 
-    offer.receive(mime_type, writefd);
+    offer.receive(mime_type, unsafe { BorrowedFd::borrow_raw(writefd) });
 
     if let Err(err) = close(writefd) {
         log::warn!("Failed to close write pipe: {}", err);

src/data_device_manager/read_pipe.rs

@@ -1,8 +1,7 @@
 use std::{
     fs, io,
-    os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd},
+    os::unix::io::{AsFd, AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd},
 };
-use wayland_backend::io_lifetimes::{AsFd, OwnedFd};
 
 /// If the `calloop` cargo feature is enabled, this can be used
 /// as an `EventSource` in a calloop event loop.
@@ -78,7 +77,7 @@ impl AsRawFd for ReadPipe {
 
 #[cfg(feature = "calloop")]
 impl AsFd for ReadPipe {
-    fn as_fd(&self) -> wayland_backend::io_lifetimes::BorrowedFd<'_> {
+    fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
         self.file.file.as_fd()
     }
 }
@@ -92,7 +91,7 @@ impl AsRawFd for ReadPipe {
 #[cfg(not(feature = "calloop"))]
 
 impl AsFd for ReadPipe {
-    fn as_fd(&self) -> wayland_backend::io_lifetimes::BorrowedFd<'_> {
+    fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
         self.file.as_fd()
     }
 }

src/data_device_manager/write_pipe.rs

@@ -1,8 +1,7 @@
 use std::{
     fs, io,
-    os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd},
+    os::unix::io::{AsFd, AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd},
 };
-use wayland_backend::io_lifetimes::{AsFd, OwnedFd};
 
 /// If the `calloop` cargo feature is enabled, this can be used
 /// as an `EventSource` in a calloop event loop.
@@ -86,7 +85,7 @@ impl AsRawFd for WritePipe {
 
 #[cfg(feature = "calloop")]
 impl AsFd for WritePipe {
-    fn as_fd(&self) -> wayland_backend::io_lifetimes::BorrowedFd<'_> {
+    fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
         self.file.file.as_fd()
     }
 }
@@ -100,7 +99,7 @@ impl AsRawFd for WritePipe {
 #[cfg(not(feature = "calloop"))]
 
 impl AsFd for WritePipe {
-    fn as_fd(&self) -> wayland_backend::io_lifetimes::BorrowedFd<'_> {
+    fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
         self.file.as_fd()
     }
 }

src/lib.rs

@@ -16,6 +16,8 @@ pub mod reexports {
     pub use wayland_protocols_wlr as protocols_wlr;
 }
 
+#[cfg(feature = "calloop")]
+pub mod calloop;
 pub mod compositor;
 pub mod data_device_manager;
 pub mod error;

src/primary_selection/offer.rs

@@ -1,5 +1,5 @@
 use std::{
-    os::unix::io::{FromRawFd, RawFd},
+    os::unix::io::{BorrowedFd, FromRawFd, RawFd},
     sync::Mutex,
 };
 
@@ -41,7 +41,7 @@ impl PrimarySelectionOffer {
         // create a pipe
         let (readfd, writefd) = pipe2(OFlag::O_CLOEXEC)?;
 
-        self.offer.receive(mime_type, writefd);
+        self.offer.receive(mime_type, unsafe { BorrowedFd::borrow_raw(writefd) });
 
         if let Err(err) = close(writefd) {
             log::warn!("Failed to close write pipe: {}", err);
@@ -57,7 +57,7 @@ impl PrimarySelectionOffer {
     pub unsafe fn receive_to_fd(&self, mime_type: String, writefd: RawFd) {
         use nix::unistd::close;
 
-        self.offer.receive(mime_type, writefd);
+        self.offer.receive(mime_type, unsafe { BorrowedFd::borrow_raw(writefd) });
 
         if let Err(err) = close(writefd) {
             log::warn!("Failed to close write pipe: {}", err);

src/shell/xdg/mod.rs

@@ -205,7 +205,7 @@ impl wayland_client::backend::ObjectData for PositionerData {
         _: &wayland_client::backend::Backend,
         _: wayland_client::backend::protocol::Message<
             wayland_client::backend::ObjectId,
-            wayland_backend::io_lifetimes::OwnedFd,
+            std::os::unix::io::OwnedFd,
         >,
     ) -> Option<Arc<(dyn wayland_client::backend::ObjectData + 'static)>> {
         unreachable!("xdg_positioner has no events");

src/shm/multi.rs

@@ -406,7 +406,7 @@ impl wayland_client::backend::ObjectData for BufferObjectData {
         _backend: &wayland_backend::client::Backend,
         msg: wayland_backend::protocol::Message<
             wayland_backend::client::ObjectId,
-            wayland_backend::io_lifetimes::OwnedFd,
+            std::os::unix::io::OwnedFd,
         >,
     ) -> Option<Arc<dyn wayland_backend::client::ObjectData>> {
         debug_assert!(wayland_client::backend::protocol::same_interface(