From 410198b099aa0138ef9976e32d167a178770b82b Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Tue, 20 Aug 2024 10:20:26 +0200 Subject: [PATCH] realising, that an extra msg_thread will be nececary --- src/core/instance/connection.rs | 2 ++ src/core/instance/mod.rs | 28 ++++++++++++++++++++++--- src/core/instance/updater.rs | 25 +++++++++++++++++++++-- src/core/item.rs | 36 ++++++++++++++++++++++++++++----- src/core/proto/mod.rs | 23 +++++++++++++++++++++ src/platform/os/cli/set.rs | 18 ++++++----------- src/platform/os/fsstore.rs | 7 ++++++- src/platform/os/main.rs | 1 + src/platform/os/unix_socket.rs | 3 ++- 9 files changed, 119 insertions(+), 24 deletions(-) diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index 6bfda05..64c9ec9 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -55,3 +55,5 @@ pub fn value_raw_from_peer(item: &mut Item) -> MizeResult { return Ok(data); } + + diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 4e8fe16..7ca1911 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -5,7 +5,7 @@ use std::thread::JoinHandle; use std::{thread, vec}; use crossbeam::channel::{self, bounded, Receiver, Sender}; use std::sync::{Arc, Mutex}; -use tracing::{trace, debug, info, warn, error}; +use tracing::{debug, error, info, trace, warn, Instrument}; use uuid::Uuid; use std::fs::File; use colored::Colorize; @@ -56,11 +56,12 @@ pub struct Instance { // the namespace of the instance itself // TODO: set to a random uuid pub self_namespace: Arc>, - op_tx: channel::Sender, + pub op_tx: channel::Sender, context: Vec, threads: Vec, give_msg_wait: Arc>>>>, create_msg_wait: Arc>>>, + update_thread_busy: Arc>, #[cfg(feature = "async")] pub runtime: Arc>, } @@ -95,15 +96,21 @@ impl Instance { threads: vec![], next_con_id: Arc::new(Mutex::new(1)), give_msg_wait, create_msg_wait, + update_thread_busy: Arc::new(Mutex::new(false)), #[cfg(feature = "async")] runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)), }; let instance_clone = instance.clone(); - let closure = move || updater_thread(op_rx, &instance_clone); + let op_rx_clone = op_rx.clone(); + let closure = move || updater_thread(op_rx_clone, &instance_clone); instance.spawn("updater_thread", closure)?; + let instance_clone_two = instance.clone(); + let closure_two = move || updater_thread(op_rx, &instance_clone_two); + instance.spawn("updater_thread", closure_two)?; + // will probably move the msg stuff into it's own thread //let msg_instance_clone = instance.clone(); //let msg_closure = move || msg_thread(op_rx, instance_clone); @@ -259,6 +266,10 @@ impl Instance { return Ok(self_namespace_inner.clone()); } + pub fn we_are_namespace(&self) -> MizeResult { + Ok(self.get_namespace()? == self.get_self_namespace()?) + } + pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { let mut instance_clone = self.clone(); self.spawn("some_listener", move || listener.listen(instance_clone)); @@ -387,6 +398,17 @@ impl Instance { thread::sleep_ms(10000000) } } + + pub fn wait_for_updaater_thread(&self) -> MizeResult<()> { + thread::sleep_ms(10000000); + return Ok(()); + while !self.op_tx.is_empty() { + thread::sleep_ms(50); + } + self.update_thread_busy.lock()?; + trace!("wait_for_updaater_thread Thread Idle"); + Ok(()) + } } impl fmt::Debug for Instance { diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 86d92b7..f342098 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -17,14 +17,27 @@ pub enum Operation { } pub fn updater_thread(operation_rx : Receiver, instance: &Instance) -> MizeResult<()> { - for mut operation in operation_rx { - trace!("OPERATION"); + let mut count = 0; + + loop { + println!("updater_thread loop beginning: {}", operation_rx.len()); + + let mut operation = operation_rx.recv()?; + trace!("OPERATION {}", count); + + //let mut busy = instance.update_thread_busy.lock()?; + let result = handle_operation(&mut operation, instance); + trace!("OPERATION DONE {}", count); + count += 1; + if let Err(err) = result { error!("OPERATION FAILED: {:?}", operation); err.log(); } + + //drop(busy); } Ok(()) } @@ -59,6 +72,14 @@ fn handle_msg(msg: &mut MizeMessage, instance: &Instance) -> MizeResult<()> { instance.set(id, data); }, + // this should check, if the update is valid + // but for now, we do just always accept it + MessageCmd::UpdateRequest => { + let data = msg.data()?; + let id = msg.id(instance)?; + instance.set(id, data); + }, + MessageCmd::Give => { let id = msg.id(instance)?; let data = msg.data()?; diff --git a/src/core/item.rs b/src/core/item.rs index 99bc90e..376b23e 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -11,7 +11,7 @@ use std::io; use tracing::{span, Level}; use crate::error::{MizeError, MizeResult, IntoMizeResult}; -use crate::instance::Instance; +use crate::instance::{connection, Instance}; use crate::instance::store::Store; use crate::id::MizeId; use crate::proto::MizeMessage; @@ -106,14 +106,26 @@ impl Item<'_> { } pub fn merge>(&mut self, mut value: V) -> MizeResult<()> { - let store_inner = self.instance.store.lock()?; - let mut old_data = store_inner.get_value_data_full(self.id())?; + + println!("before as_data_full: {}", self.instance.op_tx.len()); + let mut data = self.as_data_full()?; + println!("data: {:?}", data); let new_data = value.into(); - old_data.merge(new_data); + data.merge(new_data); + println!("new_data: {:?}", data); + + if self.instance.we_are_namespace()? { + let store_inner = self.instance.store.lock()?; + store_inner.set(self.id(), data); + } else { + let namespace = self.id().namespace(); + let mut connection = self.instance.get_connection_by_ns(namespace)?; + let msg = MizeMessage::new_update_request(self.id(), data, connection.id); + connection.send(msg)?; + } - store_inner.set(self.id(), old_data)?; Ok(()) } } @@ -131,6 +143,10 @@ impl ItemData { ItemData(CborValue::Null) } + pub fn empty() -> ItemData { + ItemData(CborValue::Null) + } + pub fn from_string>(into_string: S) -> ItemData { let string = into_string.into(); let data = CborValue::Text(string); @@ -419,6 +435,16 @@ impl IntoItemData for &str { ItemData::parse(self) } } +impl IntoItemData for String { + fn into_item_data(self) -> ItemData { + ItemData(self.into()) + } +} +impl IntoItemData for &String { + fn into_item_data(self) -> ItemData { + ItemData(self.to_owned().into()) + } +} impl IntoItemData for CborValue { fn into_item_data(self) -> ItemData { ItemData ( self ) diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index a402ac4..1d4e14a 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -1,5 +1,6 @@ use core::fmt; use std::io; +use nix::libc::CTRL_CMD_UNSPEC; use serde::Serialize; use ciborium::{value::Integer, Value as CborValue}; @@ -81,6 +82,28 @@ impl MizeMessage { MizeMessage::new(value, conn_id) } + pub fn new_update_request(id: MizeId, data: ItemData, conn_id: u64) -> MizeMessage { + let id_path = id.path().into_iter().map(|string| CborValue::Text(string.to_owned())).collect(); + + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_UPDATE_REQUEST.into())); + let id = (CborValue::Integer(MSG_ID.into()), CborValue::Array(id_path)); + let data = (CborValue::Integer(MSG_DATA.into()), data.cbor().to_owned()); + let value = CborValue::Map(vec![cmd, id, data]); + + MizeMessage::new(value, conn_id) + } + + pub fn new_update(id: MizeId, data: ItemData, conn_id: u64) -> MizeMessage { + let id_path = id.path().into_iter().map(|string| CborValue::Text(string.to_owned())).collect(); + + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_UPDATE.into())); + let id = (CborValue::Integer(MSG_ID.into()), CborValue::Array(id_path)); + let data = (CborValue::Integer(MSG_DATA.into()), data.cbor().to_owned()); + let value = CborValue::Map(vec![cmd, id, data]); + + MizeMessage::new(value, conn_id) + } + pub fn value(self) -> CborValue { self.value } diff --git a/src/platform/os/cli/set.rs b/src/platform/os/cli/set.rs index e1131ba..36918f4 100644 --- a/src/platform/os/cli/set.rs +++ b/src/platform/os/cli/set.rs @@ -5,6 +5,7 @@ use home::home_dir; use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait}; use mize::instance::Instance; +use mize::item::IntoItemData; use mize::platform::os::config_from_cli_args; @@ -16,19 +17,12 @@ pub fn set(sub_matches: &ArgMatches) -> MizeResult<()> { let id = sub_matches.get_one::("id") .ok_or(MizeError::new().msg("No id Argument specified"))?; - let item = instance.get(id)?; + let value = sub_matches.get_one::("value") + .ok_or(MizeError::new().msg("No value Argument specified"))?; - let recurse_count = sub_matches.get_count("recurse"); + instance.set(id, value.into_item_data())?; - println!("{}", item.as_data_full()?); - return Ok(()); + instance.wait_for_updaater_thread()?; - if recurse_count > 0 { - println!("{}", item.id()); - println!("{}", item.as_data_full()?); - } else { - println!("{}", item); - }; - - return Ok(()); + Ok(()) } diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 3005804..73de353 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -42,7 +42,9 @@ impl FileStore { } // init the store - fs::write(path.join("next_id"), "1"); + if !path.join("next_id").exists() { + fs::write(path.join("next_id"), "1"); + } Ok(FileStore { path: Path::new(&path).to_owned() }) } @@ -110,6 +112,9 @@ impl Store for FileStore { fn get_value_data_full(&self, id: MizeId) -> MizeResult { let file_path = self.path.join("store").join(id.namespace_str()).join(id.store_part()); println!("file_path: {:?}", file_path.display()); + if !file_path.exists() { + return Ok(ItemData::empty()); + } let file = OpenOptions::new().read(true).open(file_path)?; let cbor_value: CborValue = ciborium::from_reader(file) diff --git a/src/platform/os/main.rs b/src/platform/os/main.rs index 0a8e106..240a20c 100644 --- a/src/platform/os/main.rs +++ b/src/platform/os/main.rs @@ -164,6 +164,7 @@ fn cli_matches() -> clap::ArgMatches { Command::new("set") .aliases(["s"]) .arg(Arg::new("id").help("The id or path to set")) + .arg(Arg::new("value").help("The value to set the path to")) ) .subcommand( Command::new("show") diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index 9cd8eb3..85871c4 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -10,7 +10,7 @@ use tracing::{info, warn}; use crate::instance::connection::{ConnListener, Connection}; use crate::instance::{self, Instance}; use crate::error::{MizeError, MizeResult, IntoMizeResult}; -use crate::proto::MizeMessage; +use crate::proto::{self, MizeMessage}; pub struct UnixListener { sock_path: PathBuf, @@ -140,6 +140,7 @@ fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id: } let msg = MizeMessage::new(value, conn_id); println!("unix incoming got msg: {}", msg); + println!("op channel len: {}", instance.op_tx.len()); instance.got_msg(msg); } Ok(())