backend/sys/client: Use unbounded buffering, with a test case

wayland-backend/CHANGELOG.md

@@ -8,6 +8,9 @@
 
 - backend/server/rs: Send protocol error if method requires newer protocol version
 
+#### Changes
+- backend/client/rs: Unbounded buffering, matching libwayland behavior
+
 ### Additions
 
 - backend: Added a `destroy_object` method

wayland-backend/src/rs/client_impl/mod.rs

@@ -150,7 +150,7 @@ impl InnerBackend {
     }
 
     pub fn connect(stream: UnixStream) -> Result<Self, NoWaylandLib> {
-        let socket = BufferedSocket::new(Socket::from(stream));
+        let socket = BufferedSocket::new(Socket::from(stream), true);
         let mut map = ObjectMap::new();
         map.insert_at(
             1,

wayland-backend/src/rs/server_impl/client.rs

@@ -69,7 +69,7 @@ impl<D> Client<D> {
         debug: bool,
         data: Arc<dyn ClientData>,
     ) -> Self {
-        let socket = BufferedSocket::new(Socket::from(stream));
+        let socket = BufferedSocket::new(Socket::from(stream), false);
         let mut map = ObjectMap::new();
         map.insert_at(
             1,

wayland-backend/src/rs/socket.rs

@@ -134,17 +134,19 @@ pub struct BufferedSocket {
     in_fds: VecDeque<OwnedFd>,
     out_data: Buffer<u8>,
     out_fds: Vec<OwnedFd>,
+    unbounded: bool,
 }
 
 impl BufferedSocket {
     /// Wrap a Socket into a Buffered Socket
-    pub fn new(socket: Socket) -> Self {
+    pub fn new(socket: Socket, unbounded: bool) -> Self {
         Self {
             socket,
             in_data: Buffer::new(2 * MAX_BYTES_OUT), // Incoming buffers are twice as big in order to be
             in_fds: VecDeque::new(),                 // able to store leftover data if needed
             out_data: Buffer::new(MAX_BYTES_OUT),
             out_fds: Vec::new(),
+            unbounded,
         }
     }
 
@@ -206,13 +208,24 @@ impl BufferedSocket {
     // if false is returned, it means there is not enough space
     // in the buffer
     fn attempt_write_message(&mut self, msg: &Message<u32, RawFd>) -> IoResult<bool> {
-        match write_to_buffers(msg, self.out_data.get_writable_storage(), &mut self.out_fds) {
-            Ok(bytes_out) => {
-                self.out_data.advance(bytes_out);
-                Ok(true)
+        let fds_len = self.out_fds.len();
+        loop {
+            match write_to_buffers(msg, self.out_data.get_writable_storage(), &mut self.out_fds) {
+                Ok(bytes_out) => {
+                    self.out_data.advance(bytes_out);
+                    return Ok(true);
+                }
+                Err(MessageWriteError::BufferTooSmall) => {
+                    self.out_fds.truncate(fds_len);
+                    if !self.unbounded {
+                        return Ok(false);
+                    }
+                    self.out_data.increase_capacity();
+                }
+                Err(MessageWriteError::DupFdFailed(e)) => {
+                    return Err(e);
+                }
             }
-            Err(MessageWriteError::BufferTooSmall) => Ok(false),
-            Err(MessageWriteError::DupFdFailed(e)) => Err(e),
         }
     }
 
@@ -333,6 +346,11 @@ impl<T: Copy + Default> Buffer<T> {
         self.offset += bytes;
     }
 
+    fn increase_capacity(&mut self) {
+        let new_len = self.storage.len() * 2;
+        self.storage.resize_with(new_len, Default::default);
+    }
+
     /// Clears the contents of the buffer
     ///
     /// This only sets the counter of occupied space back to zero,
@@ -419,8 +437,8 @@ mod tests {
         };
 
         let (client, server) = ::std::os::unix::net::UnixStream::pair().unwrap();
-        let mut client = BufferedSocket::new(Socket::from(client));
-        let mut server = BufferedSocket::new(Socket::from(server));
+        let mut client = BufferedSocket::new(Socket::from(client), false);
+        let mut server = BufferedSocket::new(Socket::from(server), false);
 
         client.write_message(&msg).unwrap();
         client.flush().unwrap();
@@ -463,8 +481,8 @@ mod tests {
         };
 
         let (client, server) = ::std::os::unix::net::UnixStream::pair().unwrap();
-        let mut client = BufferedSocket::new(Socket::from(client));
-        let mut server = BufferedSocket::new(Socket::from(server));
+        let mut client = BufferedSocket::new(Socket::from(client), false);
+        let mut server = BufferedSocket::new(Socket::from(server), false);
 
         client.write_message(&msg).unwrap();
         client.flush().unwrap();
@@ -522,8 +540,8 @@ mod tests {
         ];
 
         let (client, server) = ::std::os::unix::net::UnixStream::pair().unwrap();
-        let mut client = BufferedSocket::new(Socket::from(client));
-        let mut server = BufferedSocket::new(Socket::from(server));
+        let mut client = BufferedSocket::new(Socket::from(client), false);
+        let mut server = BufferedSocket::new(Socket::from(server), false);
 
         for msg in &messages {
             client.write_message(msg).unwrap();
@@ -561,8 +579,8 @@ mod tests {
         };
 
         let (client, server) = ::std::os::unix::net::UnixStream::pair().unwrap();
-        let mut client = BufferedSocket::new(Socket::from(client));
-        let mut server = BufferedSocket::new(Socket::from(server));
+        let mut client = BufferedSocket::new(Socket::from(client), false);
+        let mut server = BufferedSocket::new(Socket::from(server), false);
 
         client.write_message(&msg).unwrap();
         client.flush().unwrap();

wayland-backend/src/test/protocol_error.rs

@@ -114,7 +114,7 @@ expand_test!(client_wrong_id, {
     let mut server = server_backend::Backend::<()>::new().unwrap();
     let _client_id = server.handle().insert_client(rx, Arc::new(())).unwrap();
 
-    let mut socket = BufferedSocket::new(Socket::from(tx));
+    let mut socket = BufferedSocket::new(Socket::from(tx), false);
 
     socket
         .write_message(&Message {
@@ -140,7 +140,7 @@ expand_test!(client_wrong_opcode, {
     let mut server = server_backend::Backend::<()>::new().unwrap();
     let _client_id = server.handle().insert_client(rx, Arc::new(())).unwrap();
 
-    let mut socket = BufferedSocket::new(Socket::from(tx));
+    let mut socket = BufferedSocket::new(Socket::from(tx), false);
 
     socket
         .write_message(&Message {
@@ -164,7 +164,7 @@ expand_test!(client_wrong_sender, {
     let mut server = server_backend::Backend::<()>::new().unwrap();
     let _client_id = server.handle().insert_client(rx, Arc::new(())).unwrap();
 
-    let mut socket = BufferedSocket::new(Socket::from(tx));
+    let mut socket = BufferedSocket::new(Socket::from(tx), false);
 
     socket
         .write_message(&Message {
@@ -332,7 +332,7 @@ expand_test!(protocol_version_check, {
         Arc::new(ServerData(object_id.clone())),
     );
 
-    let mut socket = BufferedSocket::new(Socket::from(tx));
+    let mut socket = BufferedSocket::new(Socket::from(tx), false);
 
     socket
         .write_message(&Message {

wayland-tests/tests/buffer_size.rs

@@ -53,6 +53,55 @@ fn buffer_size() {
     }
 }
 
+// TODO: Enable if libwayland >= 1.23.0
+#[cfg(not(feature = "client_system"))]
+#[test]
+fn buffer_size_client() {
+    let mut server = TestServer::new();
+    server
+        .display
+        .handle()
+        .create_global::<ServerHandler, ways::protocol::wl_seat::WlSeat, _>(1, ());
+
+    let (_, mut client) = server.add_client();
+    let mut client_ddata = ClientHandler { globals: globals::GlobalList::new() };
+
+    let registry = client.display.get_registry(&client.event_queue.handle(), ());
+
+    let mut server_handler = ServerHandler::default();
+
+    roundtrip(&mut client, &mut server, &mut client_ddata, &mut server_handler).unwrap();
+
+    let seat = client_ddata
+        .globals
+        .bind::<wayc::protocol::wl_seat::WlSeat, _, _>(
+            &client.event_queue.handle(),
+            &registry,
+            1..2,
+            (),
+        )
+        .unwrap();
+
+    let pointer = seat.get_pointer(&client.event_queue.handle(), ());
+
+    // Send too many Wayland events to buffer for default server buffer
+    for _ in 0..10_000 {
+        pointer.set_cursor(0, None, 0, 0);
+    }
+
+    while let Err(wayc::backend::WaylandError::Io(err)) = client.conn.flush() {
+        if err.kind() == ErrorKind::WouldBlock {
+            server.answer(&mut server_handler);
+        } else {
+            break;
+        }
+    }
+    server.answer(&mut server_handler);
+    client.conn.flush().unwrap();
+
+    // roundtrip(&mut client, &mut server, &mut client_ddata, &mut server_handler).unwrap();
+}
+
 /*
  * Client handler
  */