vnc_worker: improve VNC server compatibility and performance

Cargo.lock

@@ -10108,6 +10108,7 @@ dependencies = [
 name = "vnc"
 version = "0.0.0"
 dependencies = [
+ "flate2",
  "futures",
  "pal_async",
  "socket2",

workers/vnc_worker/src/lib.rs

@@ -217,9 +217,64 @@ impl<T: Listener> Server<T> {
                         abort: abort_send,
                     };
                 }
-                State::Connected { task, .. } => {
-                    let (view, input) = task.await;
-                    self.state = State::Listening { view, input };
+                State::Connected { .. } => {
+                    // Take ownership of the connected state so we can
+                    // move abort_send into the new-connection branch.
+                    let (mut task, abort, _old_addr) = if let State::Connected {
+                        task,
+                        abort,
+                        remote_addr,
+                    } =
+                        std::mem::replace(&mut self.state, State::Invalid)
+                    {
+                        (task, abort, remote_addr)
+                    } else {
+                        unreachable!()
+                    };
+
+                    // Race: current connection vs new incoming connection.
+                    futures::select! {
+                        result = task.as_mut().fuse() => {
+                            let (view, input) = result;
+                            self.state = State::Listening { view, input };
+                        }
+                        accept = self.listener.accept().fuse() => {
+                            let (new_socket, remote_addr) = accept?;
+                            tracing::info!(address = ?remote_addr, "New VNC client, disconnecting previous");
+                            // Abort the current connection and wait for resources.
+                            abort.send(());
+                            let (view, input) = task.await;
+                            // Now set up the new connection.
+                            let socket = PolledSocket::new(driver, new_socket.into())?;
+                            let mut vncserver = vnc::Server::new("OpenVMM VM".into(), socket, view, input);
+                            let mut timer = PolledTimer::new(driver);
+                            let (abort_send, abort_recv) = mesh::oneshot();
+                            let connection = Box::pin(async move {
+                                let updater = vncserver.updater();
+                                let update_task = async {
+                                    loop {
+                                        timer.sleep(Duration::from_millis(30)).await;
+                                        updater.update();
+                                    }
+                                };
+                                let r = futures::select! {
+                                    r = vncserver.run().fuse() => r.context("VNC error"),
+                                    _ = abort_recv.fuse() => Err(anyhow!("VNC connection aborted")),
+                                    _ = update_task.fuse() => unreachable!(),
+                                };
+                                match r {
+                                    Ok(_) => tracing::info!("VNC client disconnected"),
+                                    Err(err) => tracing::error!(error = err.as_error(), "VNC client error"),
+                                }
+                                vncserver.done()
+                            });
+                            self.state = State::Connected {
+                                remote_addr,
+                                task: connection,
+                                abort: abort_send,
+                            };
+                        }
+                    }
                 }
                 State::Invalid => unreachable!(),
             }

workers/vnc_worker/vnc/Cargo.toml

@@ -9,6 +9,7 @@ rust-version.workspace = true
 [dependencies]
 pal_async.workspace = true
 
+flate2.workspace = true
 futures.workspace = true
 thiserror.workspace = true
 zerocopy.workspace = true