From 2e7a528857dfdbc2661e0e6fe3e8c436ab01c2ee Mon Sep 17 00:00:00 2001 From: galister <22305755+galister@users.noreply.github.com> Date: Sun, 12 Jul 2026 15:06:24 +0900 Subject: [PATCH] swich tokio to smol --- Cargo.lock | 47 ++------------------- Cargo.toml | 3 +- wayvr-ipc/Cargo.toml | 8 ++-- wayvr-ipc/src/client.rs | 76 +++++++++++++++++++--------------- wayvr-ipc/src/util/notifier.rs | 24 ++++++----- wayvrctl/Cargo.toml | 2 +- wayvrctl/src/main.rs | 51 ++++++++++++----------- 7 files changed, 93 insertions(+), 118 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e1f665b3..557029a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2662,10 +2662,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "069323743400cb7ab06a8fe5c1ed911d36b6919ec531661d034c89083629595b" dependencies = [ "doctest-file", - "futures-core", "libc", "recvmsg", - "tokio", "widestring", "windows-sys 0.61.2", ] @@ -3175,17 +3173,6 @@ version = "0.5.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e53debba6bda7a793e5f99b8dacf19e626084f525f7829104ba9898f367d85ff" -[[package]] -name = "mio" -version = "1.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" -dependencies = [ - "libc", - "wasi 0.11.1+wasi-snapshot-preview1", - "windows-sys 0.61.2", -] - [[package]] name = "moveit" version = "0.6.0" @@ -5414,16 +5401,6 @@ version = "0.3.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4aaa7368fcf4852a4c2dd92df0cace6a71f2091ca0a23391ce7f3a31833f1523" -[[package]] -name = "socket2" -version = "0.6.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" -dependencies = [ - "libc", - "windows-sys 0.61.2", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -5823,23 +5800,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" dependencies = [ "bytes", - "libc", - "mio", "pin-project-lite", - "socket2", - "tokio-macros", - "windows-sys 0.61.2", -] - -[[package]] -name = "tokio-macros" -version = "2.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.118", ] [[package]] @@ -6550,14 +6511,14 @@ name = "wayvr-ipc" version = "0.1.0" dependencies = [ "anyhow", + "async-channel", + "async-io", "bytes", - "interprocess", "log", "serde", "serde_json", "smallvec", - "tokio", - "tokio-util", + "smol", ] [[package]] @@ -6571,7 +6532,7 @@ dependencies = [ "serde", "serde_json", "shell-words", - "tokio", + "smol", "wayvr-ipc", ] diff --git a/Cargo.toml b/Cargo.toml index 4214b9c0..b57a0502 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,8 @@ members = [ [workspace.dependencies] anyhow = "1.0.100" +async-channel = "2.3" +async-io = "2.1" bytes = { version = "1.11.1" } chrono = "0.4.42" clap = { version = "4.5.53", features = ["derive"] } @@ -34,7 +36,6 @@ slotmap = "1.1.1" smallvec = "1.15.2" smol = "2.0.2" strum = { version = "0.27.2", features = ["derive"] } -tokio = "1.48.0" # TODO: switch to smol for ipc tracing-subscriber = { version = "0.3.22", features = ["env-filter"] } uuid = { version = "1.19.0", features = ["fast-rng", "serde", "v4"] } vulkano = { diff --git a/wayvr-ipc/Cargo.toml b/wayvr-ipc/Cargo.toml index 04a0338f..47520dd3 100644 --- a/wayvr-ipc/Cargo.toml +++ b/wayvr-ipc/Cargo.toml @@ -9,17 +9,17 @@ repository = "https://github.com/wlx-team/wayvr" [dependencies] anyhow.workspace = true +async-channel.workspace = true +async-io.workspace = true bytes.workspace = true log.workspace = true serde.workspace = true smallvec.workspace = true # client-only deps -interprocess = { workspace = true, features = ["tokio"], optional = true } serde_json.workspace = true -tokio = { workspace = true, features = ["macros"], optional = true } -tokio-util = { version = "0.7.18", optional = true } +smol = { workspace = true, optional = true } [features] default = ["client"] -client = ["dep:interprocess", "dep:tokio", "dep:tokio-util"] +client = ["dep:smol"] diff --git a/wayvr-ipc/src/client.rs b/wayvr-ipc/src/client.rs index a1a6cb0c..e16c9ada 100644 --- a/wayvr-ipc/src/client.rs +++ b/wayvr-ipc/src/client.rs @@ -1,16 +1,11 @@ use bytes::BufMut; -use interprocess::local_socket::{ - self, GenericNamespaced, - tokio::{Stream, prelude::*}, -}; use serde::Serialize; use smallvec::SmallVec; +use smol::io::AsyncReadExt; +use smol::io::AsyncWriteExt; +use smol::lock::Mutex; +use smol::net::unix::UnixStream; use std::sync::{Arc, Weak}; -use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, - sync::Mutex, -}; -use tokio_util::sync::CancellationToken; use crate::{ gen_id, @@ -43,10 +38,10 @@ type SignalFunc = Box bool + Send>; pub struct WayVRClient { receiver: ReceiverMutex, sender: SenderMutex, - cancel_token: CancellationToken, + cancel_tx: async_channel::Sender<()>, exiting: bool, queued_packets: QueuedPacketVec, - pub auth: Option, // Some if authenticated + pub auth: Option, pub on_signal: Option, } @@ -67,8 +62,8 @@ pub async fn send_packet(sender: &SenderMutex, data: &[u8]) -> anyhow::Result<() pub type WayVRClientMutex = Arc>; pub type WayVRClientWeak = Weak>; -type ReceiverMutex = Arc>; -type SenderMutex = Arc>; +type ReceiverMutex = Arc>; +type SenderMutex = Arc>; async fn client_runner(client: WayVRClientMutex) -> anyhow::Result<()> { loop { @@ -79,7 +74,7 @@ async fn client_runner(client: WayVRClientMutex) -> anyhow::Result<()> { type Payload = SmallVec<[u8; 64]>; async fn read_payload( - conn: &mut local_socket::tokio::RecvHalf, + conn: &mut smol::net::unix::UnixStream, size: u32, ) -> anyhow::Result { let mut payload = Payload::new(); @@ -122,32 +117,29 @@ impl WayVRClient { pub async fn new(client_name: &str) -> anyhow::Result { let printname = "/tmp/wayvr_ipc.sock"; - let name = printname.to_ns_name::()?; - let stream = match Stream::connect(name).await { + let stream = match UnixStream::connect(printname).await { Ok(c) => c, Err(e) => { anyhow::bail!("Failed to connect to the WayVR IPC: {}", e) } }; - let (receiver, sender) = stream.split(); + let receiver = Arc::new(Mutex::new(stream.clone())); + let sender = Arc::new(Mutex::new(stream)); - let receiver = Arc::new(Mutex::new(receiver)); - let sender = Arc::new(Mutex::new(sender)); - - let cancel_token = CancellationToken::new(); + let (cancel_tx, cancel_rx) = async_channel::bounded(1); let client = Arc::new(Mutex::new(Self { receiver, sender: sender.clone(), + cancel_tx, exiting: false, - cancel_token: cancel_token.clone(), queued_packets: QueuedPacketVec::new(), auth: None, on_signal: None, })); - WayVRClient::start_runner(client.clone(), cancel_token); + WayVRClient::start_runner(client.clone(), cancel_rx); // Send handshake to the server send_packet( @@ -163,17 +155,31 @@ impl WayVRClient { Ok(client) } - fn start_runner(client: WayVRClientMutex, cancel_token: CancellationToken) { - tokio::spawn(async move { - tokio::select! { - _ = cancel_token.cancelled() => { - log::info!("Exiting IPC runner gracefully"); - } - e = client_runner(client.clone()) => { - log::info!("IPC Runner failed: {:?}", e); + fn start_runner(client: WayVRClientMutex, cancel_rx: async_channel::Receiver<()>) { + smol::spawn(async move { + let cancel_fut = async { + cancel_rx + .recv() + .await + .map_err(|_| anyhow::anyhow!("cancelled")) + }; + + let result = smol::future::race(client_runner(client.clone()), cancel_fut).await; + + match result { + Ok(()) => { + log::info!("IPC Runner completed"); + } + Err(e) => { + if e.to_string() == "cancelled" { + log::info!("Exiting IPC runner gracefully"); + } else { + log::info!("IPC Runner failed: {:?}", e); } + } } - }); + }) + .detach(); } async fn tick(client_mtx: WayVRClientMutex) -> anyhow::Result<()> { @@ -185,7 +191,9 @@ impl WayVRClient { // read packet let packet = { let mut receiver = receiver.lock().await; - let packet_size = receiver.read_u32().await?; + let mut size_buf = [0u8; 4]; + receiver.read_exact(&mut size_buf).await?; + let packet_size = u32::from_be_bytes(size_buf); log::trace!("packet size {}", packet_size); if packet_size > 128 * 1024 { anyhow::bail!("packet size too large (> 128 KiB)"); @@ -442,6 +450,6 @@ impl WayVRClient { impl Drop for WayVRClient { fn drop(&mut self) { self.exiting = true; - self.cancel_token.cancel(); + let _ = self.cancel_tx.try_send(()); } } diff --git a/wayvr-ipc/src/util/notifier.rs b/wayvr-ipc/src/util/notifier.rs index b985cc5c..38acefc7 100644 --- a/wayvr-ipc/src/util/notifier.rs +++ b/wayvr-ipc/src/util/notifier.rs @@ -1,25 +1,29 @@ -use std::sync::Arc; - -use tokio::sync::Notify; +use async_channel::{Receiver, Sender}; // Copyable wrapped Notify struct for easier usage -#[derive(Default, Clone)] +#[derive(Clone)] pub struct Notifier { - notifier: Arc, + sender: Sender<()>, + receiver: Receiver<()>, } impl Notifier { pub fn new() -> Self { - Self { - notifier: Arc::new(Notify::new()), - } + let (sender, receiver) = async_channel::bounded(1); + Self { sender, receiver } } pub fn notify(&self) { - self.notifier.notify_waiters(); + let _ = self.sender.try_send(()); } pub async fn wait(&self) { - self.notifier.notified().await; + let _ = self.receiver.recv().await; + } +} + +impl Default for Notifier { + fn default() -> Self { + Self::new() } } diff --git a/wayvrctl/Cargo.toml b/wayvrctl/Cargo.toml index fbd982f6..659ef4b3 100644 --- a/wayvrctl/Cargo.toml +++ b/wayvrctl/Cargo.toml @@ -15,7 +15,7 @@ clap.workspace = true log.workspace = true serde.workspace = true serde_json.workspace = true -tokio.workspace = true +smol.workspace = true env_logger = "0.11.11" shell-words = "1.1.1" diff --git a/wayvrctl/src/main.rs b/wayvrctl/src/main.rs index 7b08c0b2..ed3e7994 100644 --- a/wayvrctl/src/main.rs +++ b/wayvrctl/src/main.rs @@ -22,37 +22,38 @@ use crate::helper::{ mod helper; mod types; -#[tokio::main(flavor = "current_thread")] -async fn main() -> ExitCode { +fn main() -> ExitCode { env_logger::init_from_env(Env::default().default_filter_or("info")); let args = Args::parse(); - let mut state = WayVRClientState { - wayvr_client: WayVRClient::new(&format!("wayvrctl-{}", process::id())) - .await - .inspect_err(|e| { - log::error!("Failed to initialize WayVR connection: {e:?}"); - process::exit(1); - }) - .unwrap(), - serial_generator: ipc::SerialGenerator::new(), - pretty_print: args.pretty, - }; + smol::block_on(async move { + let mut state = WayVRClientState { + wayvr_client: WayVRClient::new(&format!("wayvrctl-{}", process::id())) + .await + .inspect_err(|e| { + log::error!("Failed to initialize WayVR connection: {e:?}"); + process::exit(1); + }) + .unwrap(), + serial_generator: ipc::SerialGenerator::new(), + pretty_print: args.pretty, + }; - let maybe_err = if let Subcommands::Batch { fail_fast } = args.command { - run_batch(&mut state, fail_fast).await - } else { - run_once(&mut state, args).await - }; + let maybe_err = if let Subcommands::Batch { fail_fast } = args.command { + run_batch(&mut state, fail_fast).await + } else { + run_once(&mut state, args).await + }; - if let Err(e) = maybe_err { - log::error!("{e:?}"); - return ExitCode::FAILURE; - } else { - std::thread::sleep(Duration::from_millis(20)); - } + if let Err(e) = maybe_err { + log::error!("{e:?}"); + return ExitCode::FAILURE; + } else { + std::thread::sleep(Duration::from_millis(20)); + } - ExitCode::SUCCESS + ExitCode::SUCCESS + }) } async fn run_batch(state: &mut WayVRClientState, fail_fast: bool) -> anyhow::Result<()> {