From e16aae911cff9386473e199aaf1336c654848ac4 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Fri, 6 Dec 2024 15:18:40 +0100 Subject: [PATCH] switched from crossbeam to flume, for async support + fixed Instance::report_err() and Instance::new_connection_join_namespace() --- Cargo.lock | 75 ++++++++++++++++++------------- Cargo.toml | 2 +- src/core/instance/connection.rs | 2 +- src/core/instance/mod.rs | 14 +++--- src/core/instance/subscription.rs | 2 +- src/core/instance/updater.rs | 2 +- src/platform/os/cli.rs | 2 +- src/platform/os/unix_socket.rs | 2 +- 8 files changed, 56 insertions(+), 45 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2480871..731dc2c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -475,28 +475,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "crossbeam" -version = "0.8.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8" -dependencies = [ - "crossbeam-channel", - "crossbeam-deque", - "crossbeam-epoch", - "crossbeam-queue", - "crossbeam-utils", -] - -[[package]] -name = "crossbeam-channel" -version = "0.5.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33480d6946193aa8033910124896ca395333cae7e2d1113d1fef6c3272217df2" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "crossbeam-deque" version = "0.8.5" @@ -516,15 +494,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "crossbeam-queue" -version = "0.3.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df0346b5d5e76ac2fe4e327c5fd1118d6be7c51dfb18f9b7922923f287471e35" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "crossbeam-utils" version = "0.8.20" @@ -678,6 +647,18 @@ dependencies = [ "miniz_oxide 0.8.0", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "nanorand", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -766,8 +747,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4567c8db10ae91089c99af84c68c38da3ec2f087c3f82960bcdbf3656b6f4d7" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1068,6 +1051,16 @@ version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78b3ae25bc7c8c38cec158d1f2757ee79e9b3740fbc7ccf0e59e4b08d793fa89" +[[package]] +name = "lock_api" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07af8b9cdd281b7915f413fa73f29ebd5d55d0d3f0155584dade1ff18cea1b17" +dependencies = [ + "autocfg", + "scopeguard", +] + [[package]] name = "log" version = "0.4.21" @@ -1152,10 +1145,10 @@ dependencies = [ "clap", "colored", "console_error_panic_hook", - "crossbeam", "daemonize", "dyn-clone", "flate2", + "flume", "home", "http_req", "interner", @@ -1197,6 +1190,15 @@ dependencies = [ "twoway", ] +[[package]] +name = "nanorand" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3" +dependencies = [ + "getrandom", +] + [[package]] name = "native-tls" version = "0.2.12" @@ -1646,6 +1648,12 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "security-framework" version = "2.11.0" @@ -1821,6 +1829,9 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] [[package]] name = "stable_deref_trait" diff --git a/Cargo.toml b/Cargo.toml index bff5d41..7d6b1e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,10 +36,10 @@ serde = "1.0.203" interner = { git = "https://github.com/c2vi/interner" } tracing = "0.1.40" serde_json = { version = "1.0.117", optional = false } -crossbeam = "0.8.4" toml = { version = "0.8.13", features = ["parse"], optional = false } dyn-clone = "1.0.17" sha2 = "0.10.8" +flume = "0.11.1" # buildin modules #mize_module_blob = { path = "./modules/modules/Blob" } diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index 3a1ec81..ee49ae9 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -1,4 +1,4 @@ -use crossbeam::channel::{Sender, Receiver}; +use flume::{Sender, Receiver}; use tracing::{trace, debug, info, warn, error}; use crate::id::Namespace; diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index abdcc55..27a6b23 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -2,7 +2,7 @@ use core::fmt; use std::collections::HashMap; use std::fs::create_dir; use std::{thread, vec}; -use crossbeam::channel::{self, bounded, Receiver, Sender}; +use flume::{bounded, Receiver, Sender, unbounded}; use std::sync::{Arc, Mutex}; use tracing::{debug, error, info, trace, warn, Instrument}; use uuid::Uuid; @@ -69,7 +69,7 @@ pub struct Instance { // the namespace of the instance itself // TODO: set to a random uuid pub self_namespace: Arc>, - pub op_tx: channel::Sender, + pub op_tx: Sender, threads: Arc>>, next_thread_id: Arc>, give_msg_wait: Arc>>>>, @@ -96,7 +96,7 @@ impl Instance { let namespace_pool_raw = StringPool::default(); let connections = Arc::new(Mutex::new(Vec::new())); let subs = Arc::new(Mutex::new(HashMap::new())); - let (op_tx, op_rx) = channel::unbounded(); + let (op_tx, op_rx) = unbounded(); let give_msg_wait = Arc::new(Mutex::new(HashMap::new())); let create_msg_wait = Arc::new(Mutex::new(None)); let namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); @@ -368,13 +368,15 @@ impl Instance { } pub fn new_connection_join_namespace(&self, tx: Sender) -> MizeResult { - let conn_id = self.new_connection(&self, tx)?; + let conn_id = self.new_connection(tx)?; let ns_of_peer_str = self.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?; let ns_of_peer = self.namespace_from_string(ns_of_peer_str)?; self.connection_set_namespace(conn_id, ns_of_peer.clone()); self.set_namespace(ns_of_peer); + + Ok(conn_id) } pub fn connection_set_namespace(&self, conn_id: u64, namespace: Namespace) -> MizeResult<()> { @@ -428,8 +430,6 @@ impl Instance { pub fn spawn(&mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static) -> MizeResult<()> { - console_log!("in instance::spawn with wasm target"); - let mut threads_inner = self.threads.lock()?; let mut next_thread_id = self.next_thread_id.lock()?; @@ -552,7 +552,7 @@ impl Instance { } pub fn report_error(err: MizeError) { - err.log() + err.log(); } } diff --git a/src/core/instance/subscription.rs b/src/core/instance/subscription.rs index 679a675..26d2ee3 100644 --- a/src/core/instance/subscription.rs +++ b/src/core/instance/subscription.rs @@ -1,6 +1,6 @@ use core::fmt; use std::rc::Rc; -use crossbeam::channel::Sender; +use flume::Sender; use std::sync::Arc; use tracing::trace; diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 7547774..c59b985 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -1,4 +1,4 @@ -use crossbeam::channel::Receiver; +use flume::Receiver; use tracing::{error, trace, warn}; use std::borrow::BorrowMut; use std::sync::Arc; diff --git a/src/platform/os/cli.rs b/src/platform/os/cli.rs index 59f2ec0..f0eb4b6 100644 --- a/src/platform/os/cli.rs +++ b/src/platform/os/cli.rs @@ -1,7 +1,7 @@ use std::path::PathBuf; use clap::ArgMatches; use home::home_dir; -use crossbeam::channel::bounded; +use flume::bounded; use std::io::Read; use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait}; diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index b7ebd5d..0f86c30 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -3,7 +3,7 @@ use std::path::{Path, PathBuf}; use tokio::net::{UnixListener as TokioUnixListener, UnixStream}; use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; use tokio::io::{Interest, AsyncReadExt, AsyncWriteExt}; -use crossbeam::channel::{Receiver, Sender, unbounded}; +use flume::{Receiver, Sender, unbounded}; use ciborium::Value as CborValue; use tracing::{info, warn, debug};