switched from crossbeam to flume, for async support + fixed Instance::report_err() and Instance::new_connection_join_namespace()

2024-12-06 15:18:40 +01:00
parent 59c42b01ed
commit e16aae911c
8 changed files with 56 additions and 45 deletions
--- 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"
--- 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" }
--- 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;
--- 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<Mutex<Namespace>>,
-    pub op_tx: channel::Sender<Operation>,
+    pub op_tx: Sender<Operation>,
    threads: Arc<Mutex<Vec<(u32, String)>>>,
    next_thread_id: Arc<Mutex<u32>>,
    give_msg_wait: Arc<Mutex<HashMap<MizeId, Vec<Sender<ItemData>>>>>,
@@ -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<MizeMessage>) -> MizeResult<u64> {
-        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();
    }
 }

--- 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;

--- 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;
--- 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};
--- 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};