From b60c9c2f698782d2885a40bdb706f4021531e721 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Thu, 15 Aug 2024 09:35:34 +0200 Subject: [PATCH] most of getting stuff with two instances works --- src/core/id.rs | 9 +++- src/core/instance/connection.rs | 12 ++---- src/core/instance/mod.rs | 65 +++++++++++++++++++++------- src/core/item.rs | 76 ++++++++++++++++++++++++++------- src/platform/os/unix_socket.rs | 34 ++++++++------- 5 files changed, 140 insertions(+), 56 deletions(-) diff --git a/src/core/id.rs b/src/core/id.rs index 6da91df..f6e9196 100644 --- a/src/core/id.rs +++ b/src/core/id.rs @@ -17,8 +17,13 @@ pub struct MizeId { pub struct Namespace ( pub SharedString ); impl Namespace { - fn as_string(self) -> SharedString { - self.0 + pub fn as_string(&self) -> SharedString { + self.0.clone() + } + pub fn as_real_string(&self) -> String { + let a = self.as_string(); + let b = a.to_string(); + return b; } } diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index 4ccb6cf..6bfda05 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -7,7 +7,7 @@ use crate::proto::MizeMessage; use crate::error::{MizeError, MizeResult, IntoMizeResult}; use super::Instance; -use crate::item::{get_raw_from_cbor, Item}; +use crate::item::{get_raw_from_cbor, Item, ItemData}; // only created with Instance::new_connection() #[derive(Clone)] @@ -27,7 +27,7 @@ impl Connection { } } -pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult> { +pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult { if item.id().nth_part(3)? == "peer" { return value_raw_from_peer(item); } else { @@ -35,7 +35,7 @@ pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult> { } } -pub fn value_raw_from_peer(item: &mut Item) -> MizeResult> { +pub fn value_raw_from_peer(item: &mut Item) -> MizeResult { let id = item.id(); let conn_id_str = id.nth_part(2)?; @@ -45,9 +45,7 @@ pub fn value_raw_from_peer(item: &mut Item) -> MizeResult> { let skipped = new_id_str.skip(4); let new_id = item.instance.new_id(skipped.collect::>())?; - println!("value_raw_from_peer hereeeeeeeeeee"); let connection = item.instance.get_connection(conn_id)?; - println!("value_raw_from_peer afteeeeerrrrrrrrrrrrrrrr"); let msg = MizeMessage::new_get(new_id.clone(), connection.id); @@ -55,7 +53,5 @@ pub fn value_raw_from_peer(item: &mut Item) -> MizeResult> { let data = item.instance.give_msg_wait(new_id)?; - let raw_data = get_raw_from_cbor(data.cbor(), vec![])?; - return Ok(raw_data.to_owned()); - + return Ok(data); } diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 99bedbf..327b7d2 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -42,22 +42,26 @@ static MSG_CHANNEL_SIZE: usize = 200; /// The Instance type is the heart of the mize system #[derive(Clone)] pub struct Instance { - // question vec of stores??? - // would mean replication logic and such things is handeled by the instance - // i think it would be better thogh to then implement a ReplicatedStore + // a bit a lot of Mutexes isn-it??? pub store: Box, connections: Arc>>, next_con_id: Arc>, subs: Arc>>, pub id_pool: Arc>, pub namespace_pool: Arc>, - pub namespace: Namespace, + + // the namespace the instance operates in + pub namespace: Arc>, + + // the namespace of the instance itself + // TODO: set to a random uuid + pub self_namespace: Arc>, op_tx: channel::Sender, context: Vec, threads: Vec, + give_msg_wait: Arc>>>>, #[cfg(feature = "async")] pub runtime: Arc>, - give_msg_wait: Arc>>>>, } pub struct InstanceRef { @@ -76,14 +80,15 @@ 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 namespace = Namespace ( namespace_pool_raw.get("mize.default.namespace") ); let (op_tx, op_rx) = channel::unbounded(); let give_msg_wait = Arc::new(Mutex::new(HashMap::new())); + let namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); + let self_namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); let mut instance = Instance { store: Box::new(store), connections, subs, id_pool, - namespace, op_tx, + namespace, self_namespace, op_tx, namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)), context: vec![], threads: vec![], @@ -189,16 +194,28 @@ impl Instance { pub fn id_from_string(&self, string: String) -> MizeResult { let vec_string: Vec = string.split("/").map(|v| v.to_owned()).collect(); - let id_pool_inner = self.id_pool.lock()?; - - let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() }; - - Ok(id) + self.id_from_vec_string(vec_string) } - pub fn id_from_vec_string(&self, vec_string: Vec) -> MizeResult { + pub fn id_from_vec_string(&self, mut vec_string: Vec) -> MizeResult { let id_pool_inner = self.id_pool.lock()?; - let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() }; + let namespace_inner = self.namespace.lock()?; + let first_el = vec_string.first_mut().ok_or(mize_err!("MizeId was empty"))?; + + let id = if first_el.contains(":") { // first el is a namespace + store_part + let new_first_el = first_el.clone(); + let vec: Vec<&str> = new_first_el.split(":").collect(); + let ns_part = vec.iter().nth(0).ok_or(mize_err!("should really not happen"))?; + let store_part = vec.iter().nth(1).ok_or(mize_err!("mizeid was like 'namespace:/hi', why are you doing that"))?; + *first_el = store_part.to_owned().to_owned(); + + MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace_from_string(ns_part.to_owned().to_owned())? } + } else { + + MizeId { path: id_pool_inner.get(vec_string), namespace: namespace_inner.clone() } + }; + trace!("new MizeId made: {:?}", id); + Ok(id) } @@ -208,6 +225,13 @@ impl Instance { Ok(namespace) } + pub fn set_namespace(&mut self, ns: Namespace) -> MizeResult<()> { + let mut namespace_inner = self.namespace.lock()?; + *namespace_inner = ns; + + Ok(()) + } + pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { let mut instance_clone = self.clone(); self.spawn("some_listener", move || listener.listen(instance_clone)); @@ -254,7 +278,6 @@ impl Instance { let mut conn_inner = self.connections.lock()?; for connection in conn_inner.iter() { - println!("get_connection iterrrrrrrrrrrrrr"); if connection.id == conn_id { return Ok(connection.clone()); } @@ -263,6 +286,18 @@ impl Instance { return Err(mize_err!("Connection with id {} not known to instance", conn_id)); } + pub fn get_connection_by_ns(&self, ns: Namespace) -> MizeResult { + let mut conn_inner = self.connections.lock()?; + + for connection in conn_inner.iter() { + if connection.ns == Some(ns.clone()) { + return Ok(connection.clone()); + } + } + + return Err(mize_err!("Connection with namespace {} not known to instance", ns.as_string())); + } + pub fn spawn(&mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static) -> MizeResult<()> { self.threads.push(name.to_owned()); thread::spawn(move || func()); diff --git a/src/core/item.rs b/src/core/item.rs index 7da9935..36b4456 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -14,6 +14,7 @@ use crate::error::{MizeError, MizeResult, IntoMizeResult}; use crate::instance::Instance; use crate::instance::store::Store; use crate::id::MizeId; +use crate::proto::MizeMessage; use ciborium::Value as CborValue; use crate::instance::connection::value_raw_con_by_id; @@ -40,20 +41,9 @@ impl Item<'_> { } pub fn value_raw(&self) -> MizeResult> { - let id = self.id(); - - // some paths like inst/con_by_id/x/ are not stored in the store, but gotten from the instance - // itself (the data for that lifes in memory in the instance struct) - // so here we set handlers for certain paths - if id.store_part() == "inst" { - match id.nth_part(1)? { - "con_by_id" => { return value_raw_con_by_id(&mut self.clone()); }, - _ => {}, - } - } - - // this will call from the instance which gets the value from the store - self.instance.store.get_value_raw(self.id()) + let data = self.as_data_full()?; + let raw = get_raw_from_cbor(data.cbor(), vec![])?; + return Ok(raw.to_owned()); } pub fn value_string(&self) -> MizeResult { @@ -63,7 +53,57 @@ impl Item<'_> { } pub fn as_data_full(&self) -> MizeResult { - self.instance.store.get_value_data_full(self.id()) + let id = self.id(); + + // some paths like inst/con_by_id/x/ are not stored in the store, but gotten from the instance + // itself (the data for that lifes in memory in the instance struct) + // so here we set handlers for certain paths + if id.store_part() == "inst" { + match id.nth_part(1)? { + "con_by_id" => { return value_raw_con_by_id(&mut self.clone()); }, + "namespace" => { + let namespace_inner = self.instance.namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + "self_namespace" => { + let namespace_inner = self.instance.self_namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + _ => {}, + } + } + + // the instance we are + if id.store_part() == "self" { + match id.nth_part(1)? { + "con_by_id" => { return value_raw_con_by_id(&mut self.clone()); }, + "namespace" => { + let namespace_inner = self.instance.namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + "self_namespace" => { + let namespace_inner = self.instance.self_namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + _ => {}, + } + } + + let self_namespace_inner = self.instance.self_namespace.lock()?; + + if self.id().namespace() == self_namespace_inner.to_owned() { + debug!("getting item from store"); + return self.instance.store.get_value_data_full(self.id()) + + } else { + + let mut connection = self.instance.get_connection_by_ns(self.id().namespace())?; + let msg = MizeMessage::new_get(self.id(), connection.id); + connection.send(msg); + let data = self.instance.give_msg_wait(self.id())?; + return Ok(data); + + } } pub fn merge>(&mut self, mut value: V) -> MizeResult<()> { @@ -91,6 +131,12 @@ impl ItemData { ItemData(CborValue::Null) } + pub fn from_string>(into_string: S) -> ItemData { + let string = into_string.into(); + let data = CborValue::Text(string); + return ItemData (data); + } + pub fn merge(&mut self, other: ItemData) { item_data_merge(&mut self.0, &other.0); } diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index 9968f1f..ccfef9d 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -27,11 +27,13 @@ impl UnixListener { pub fn connect(instance: &mut Instance, store_path: PathBuf) -> MizeResult<()> { let conn_id = instance.spawn_async_blocking("unix_connection", connect_async(instance.clone(), store_path) )?; - let ns_of_peer = instance.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?; + let ns_of_peer_str = instance.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?; + let ns_of_peer = instance.namespace_from_string(ns_of_peer_str)?; - info!("connect_async ... ns_of_peer: {}", ns_of_peer); + info!("connect_async ... ns_of_peer: {}", ns_of_peer.clone().as_string()); - //instance.connection_set_namespace(conn_id, namespace) + instance.connection_set_namespace(conn_id, ns_of_peer.clone()); + instance.set_namespace(ns_of_peer); Ok(()) } @@ -100,7 +102,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu let result = unix_incomming(unix_read, cloned_instance, conn_id); // if unix incomming fails, close the connection if let Err(err) = result { - warn!("Connection closing, {:?}", err); + warn!("Connection closing because of EOF"); } Ok(()) }); @@ -110,7 +112,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu let result = unix_outgoing(unix_write, send_rx, outgoing_cloned_instance, conn_id); // if writing fails, close this connection if let Err(err) = result { - warn!("Connection closing, {:?}", err); + warn!("Connection closing because of EOF"); } Ok(()) }); @@ -130,7 +132,7 @@ fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver, fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id: u64) -> MizeResult<()> { loop { - println!("unix incoming loop..."); + //println!("unix incoming loop..."); let adapter = MyCiboriumReader { inner: &mut unix_read, instance: &mut instance }; let value: CborValue = ciborium::from_reader(adapter)?; if let CborValue::Integer(_) = value { @@ -152,17 +154,17 @@ struct MyCiboriumWriter<'a> { impl<'a> ciborium_io::Write for MyCiboriumWriter<'a> { type Error = MizeError; fn write_all(&mut self, data: &[u8]) -> Result<(), Self::Error> { - println!("CALL: write_all"); + //println!("CALL: write_all"); let runtime = self.instance.runtime.lock()?; - println!("after runtime"); + //println!("after runtime"); let num_read = runtime.block_on(self.inner.write_all(data))?; Ok(()) } fn flush(&mut self) -> Result<(), Self::Error> { - println!("CALL flush"); + //println!("CALL flush"); let handle = self.instance.async_get_handle(); let result = handle.block_on(self.inner.flush()); - println!("flush result: {:?}", result); + //println!("flush result: {:?}", result); Ok(()) } } @@ -173,12 +175,12 @@ struct MyCiboriumReader<'a> { } async fn test(test: &mut OwnedReadHalf, buf: &mut [u8]) { - println!("hello from async..."); - println!("len: {}", buf.len()); + //println!("hello from async..."); + //println!("len: {}", buf.len()); let num_read = test.read_u8().await; - println!("num_read: {:?}", num_read); + //println!("num_read: {:?}", num_read); //test.read_exact(buf).await; - println!("hello from async... twoooooooooooooo"); + //println!("hello from async... twoooooooooooooo"); } impl<'a> ciborium_io::Read for MyCiboriumReader<'a> { @@ -192,10 +194,10 @@ impl<'a> ciborium_io::Read for MyCiboriumReader<'a> { //return Ok(()); - println!("incoming before block_on"); + //println!("incoming before block_on"); let handle = self.instance.async_get_handle(); let num_read = handle.block_on(self.inner.read_exact(data))?; - println!("incoming num_read: {:?}", num_read); + //println!("incoming num_read: {:?}", num_read); //println!("data after: {:?}", data); // if we read 0 bytes, that means the reader stream closed and we should terminate the