diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index f429958..4ccb6cf 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -12,7 +12,6 @@ use crate::item::{get_raw_from_cbor, Item}; // only created with Instance::new_connection() #[derive(Clone)] pub struct Connection { - pub rx: Receiver, pub tx: Sender, pub id: u64, pub ns: Option, @@ -46,13 +45,15 @@ 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, connection.id); + let msg = MizeMessage::new_get(new_id.clone(), connection.id); connection.tx.send(msg)?; - let data = item.instance.give_msg_wait(item.id())?; + 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()); diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index b42fb76..99bedbf 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use std::fs::create_dir; use std::thread::JoinHandle; use std::{thread, vec}; -use crossbeam::channel::{self, Receiver, Sender}; +use crossbeam::channel::{self, bounded, Receiver, Sender}; use std::sync::{Arc, Mutex}; use tracing::{trace, debug, info, warn, error}; use uuid::Uuid; @@ -23,6 +23,7 @@ use crate::memstore::MemStore; use crate::instance::updater::Operation; use crate::mize_err; use crate::proto::MizeMessage; +use tokio::runtime::Handle; use self::connection::{ConnListener, Connection}; @@ -56,7 +57,7 @@ pub struct Instance { threads: Vec, #[cfg(feature = "async")] pub runtime: Arc>, - give_msg_wait: Arc, Receiver)>>>, + give_msg_wait: Arc>>>>, } pub struct InstanceRef { @@ -213,14 +214,15 @@ impl Instance { Ok(()) } - pub fn new_connection(&mut self, rx: Receiver, tx: Sender) -> MizeResult { + pub fn new_connection(&mut self, tx: Sender) -> MizeResult { let mut conn_inner = self.connections.lock()?; let mut next_con_id = self.next_con_id.lock()?; + let old_next_con_id = *next_con_id; - let connection = Connection { id: next_con_id.to_owned(), rx, tx, ns: None}; + let connection = Connection { id: next_con_id.to_owned(), tx, ns: None}; conn_inner.push(connection); *next_con_id += 1; - Ok(next_con_id.to_owned()) + Ok(old_next_con_id) } pub fn connection_set_namespace(&mut self, conn_id: u64, namespace: Namespace) -> MizeResult<()> { @@ -230,6 +232,10 @@ impl Instance { Ok(()) } + pub(crate) fn got_msg(&mut self, msg: MizeMessage) -> MizeResult<()> { + Ok(self.op_tx.send(Operation::Msg(msg))?) + } + fn set_connection(&mut self, conn_id: u64, new_connection: Connection) -> MizeResult<()> { let mut conn_inner = self.connections.lock()?; @@ -248,6 +254,7 @@ 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()); } @@ -266,18 +273,20 @@ impl Instance { let mut give_msg_wait_inner = self.give_msg_wait.lock()?; - let rx = match give_msg_wait_inner.get(&id) { - // if this id is already in the process of being requested, just use the rx already there - Some((tx, rx)) => rx.to_owned(), + let (tx, rx) = bounded::(1); + let vec = match give_msg_wait_inner.get_mut(&id) { + Some(vec) => vec, None => { - let (tx, rx) = crossbeam::channel::bounded::(1); - give_msg_wait_inner.insert(id, (tx, rx.clone())); - rx - }, + give_msg_wait_inner.insert(id.clone(), Vec::new()); + give_msg_wait_inner.get_mut(&id).unwrap() + } }; + vec.push(tx); + // so that another thread can also give_msg_wait(), while we wait in the recv() of rx + drop(vec); drop(give_msg_wait_inner); let data = rx.recv()?; @@ -291,15 +300,27 @@ impl Instance { let runtime_inner = self.runtime.lock().unwrap(); runtime_inner.spawn(func); } + #[cfg(feature = "async")] + pub fn async_get_handle(&mut self) -> Handle { + let runtime_inner = self.runtime.lock().unwrap(); + let handle = runtime_inner.handle().to_owned(); + handle + } #[cfg(feature = "async")] - pub fn spawn_async_blocking + Send + Sync + 'static>(&mut self, name: &str, func: F) { + pub fn spawn_async_blocking + Send + Sync + 'static>(&mut self, name: &str, func: F) -> F::Output { + use std::process::Output; + self.threads.push(name.to_owned()); let runtime_inner = self.runtime.lock().unwrap(); - runtime_inner.block_on(func); + let handle = runtime_inner.handle().to_owned(); + drop(runtime_inner); + let result = handle.block_on(func); + return result; } pub fn wait(&self) { + info!("Instance main thread waiting"); loop { thread::sleep_ms(10000000) } diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 6c81eeb..8437e14 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -1,5 +1,5 @@ use crossbeam::channel::Receiver; -use tracing::{trace, error}; +use tracing::{error, trace, warn}; use crate::mize_err; use crate::{instance::Instance, item::ItemData}; @@ -23,6 +23,7 @@ pub fn updater_thread(operation_rx : Receiver, mut instance: Instance if let Err(err) = result { error!("OPERATION FAILED: {:?}", operation); + err.log(); } } Ok(()) @@ -57,6 +58,18 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> let id = msg.id(instance)?; instance.set(id, data); }, + MessageCmd::Give => { + let id = msg.id(instance)?; + let data = msg.data()?; + let give_msg_wait_inner = instance.give_msg_wait.lock()?; + if let Some(vec) = give_msg_wait_inner.get(&id) { + for tx in vec { + tx.send(data.clone()); + } + } else { + warn!("got give msg for id '{}', that has no one waiting for it", id); + } + }, _ => { return Err(mize_err!("got a message, that is not handeled")); } diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index 8bb29d2..babe30f 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -81,6 +81,7 @@ impl MizeMessage { let cmd = match val_of_c { 1 => MessageCmd::Get, 2 => MessageCmd::Set, + 3 => MessageCmd::Give, _ => { return Err(MizeError::new().msg("error cmd of msg was not a valid command")); }, @@ -104,7 +105,7 @@ impl MizeMessage { for (key, val) in msg_as_map { if let CborValue::Integer(key_int) = key { // number 2 indicates the id field - let two: Integer = 1.into(); + let two: Integer = 2.into(); if key_int == &two { data = val; } @@ -170,6 +171,7 @@ impl MizeMessage { pub enum MessageCmd { Get, Set, + Give, } diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 392169c..8b12d8c 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -41,12 +41,13 @@ impl FileStore { } // init the store - fs::write(path.join("next_id"), "0"); + fs::write(path.join("next_id"), "1"); Ok(FileStore { path: Path::new(&path).to_owned() }) } pub fn store_is_opened(store_path: String) -> MizeResult { + println!("valid_pid_file: {}", valid_pid_file(Path::new(&store_path))?.is_some()); Ok(valid_pid_file(Path::new(&store_path))?.is_some()) } } diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index 1317822..9968f1f 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -25,22 +25,8 @@ impl UnixListener { pub fn connect(instance: &mut Instance, store_path: PathBuf) -> MizeResult<()> { - instance.spawn_async_blocking("unix_connection", connect_async(instance.clone(), store_path) ); + let conn_id = instance.spawn_async_blocking("unix_connection", connect_async(instance.clone(), store_path) )?; - Ok(()) -} - - -async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResult<()> { - println!("we are never hereeeeeeeeeeeeee"); - let mut stream = UnixStream::connect(store_path.join("sock")).await?; - let (read_half, write_half) = stream.split(); - println!("we are never hoooooooooooooooooooooooooooooo"); - - let (recieve_tx, recieve_rx) = unbounded::(); - let (send_tx, send_rx) = unbounded::(); - - let conn_id = instance.new_connection(recieve_rx, send_tx)?; let ns_of_peer = instance.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?; info!("connect_async ... ns_of_peer: {}", ns_of_peer); @@ -51,6 +37,39 @@ async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResul } +async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResult { + let mut stream = UnixStream::connect(store_path.join("sock")).await?; + let (unix_read, unix_write) = stream.into_split(); + + let (send_tx, send_rx) = unbounded::(); + + let conn_id = instance.new_connection(send_tx)?; + + + let cloned_instance = instance.clone(); + instance.spawn("incomming", move || { + 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 because of EOF"); + } + Ok(()) + }); + + let outgoing_cloned_instance = instance.clone(); + instance.spawn("outgoing", move || { + 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 because of EOF"); + } + Ok(()) + }); + + return Ok(conn_id); +} + + impl ConnListener for UnixListener { fn listen(self, mut instance: Instance) -> MizeResult<()> { instance.spawn_async("unix listen async", unix_listen(self, instance.clone())); @@ -71,18 +90,17 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu .mize_result_msg("Error while accepting Unix sock connection")?; info!("new connection"); - let (recieve_tx, recieve_rx) = unbounded::(); let (send_tx, send_rx) = unbounded::(); let (unix_read, unix_write) = unix_sock.into_split(); - let conn_id = instance.new_connection(recieve_rx, send_tx)?; + let conn_id = instance.new_connection(send_tx)?; let cloned_instance = instance.clone(); instance.spawn("incomming", move || { - let result = unix_incomming(unix_read, recieve_tx, cloned_instance, conn_id); + 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") + warn!("Connection closing, {:?}", err); } Ok(()) }); @@ -92,7 +110,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") + warn!("Connection closing, {:?}", err); } Ok(()) }); @@ -102,6 +120,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver, mut instance: Instance, conn_id: u64) -> MizeResult<()> { for msg in send_rx { + println!("unix outgoing got msg: {:?}", msg); let adapter = MyCiboriumWriter { inner: &mut unix_write, instance: &mut instance }; let value = msg.value(); ciborium::into_writer(&value, adapter)? @@ -109,15 +128,17 @@ fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver, Ok(()) } -fn unix_incomming(mut unix_read: OwnedReadHalf, recieve_tx: Sender, mut instance: Instance, conn_id: u64) -> MizeResult<()> { +fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id: u64) -> MizeResult<()> { 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 { break; } let msg = MizeMessage::new(value, conn_id); - recieve_tx.send(msg)?; + println!("unix incoming got msg: {:?}", msg); + instance.got_msg(msg); } Ok(()) } @@ -131,11 +152,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"); let runtime = self.instance.runtime.lock()?; + 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"); + let handle = self.instance.async_get_handle(); + let result = handle.block_on(self.inner.flush()); + println!("flush result: {:?}", result); Ok(()) } } @@ -145,17 +172,37 @@ struct MyCiboriumReader<'a> { instance: &'a mut Instance, } +async fn test(test: &mut OwnedReadHalf, buf: &mut [u8]) { + println!("hello from async..."); + println!("len: {}", buf.len()); + let num_read = test.read_u8().await; + println!("num_read: {:?}", num_read); + //test.read_exact(buf).await; + println!("hello from async... twoooooooooooooo"); +} + impl<'a> ciborium_io::Read for MyCiboriumReader<'a> { type Error = MizeError; fn read_exact(&mut self, data: &mut [u8]) -> Result<(), Self::Error> { - let runtime = self.instance.runtime.lock()?; - let num_read = runtime.block_on(self.inner.read_exact(data))?; + //println!("data before: {:?}", data); + //let runtime = self.instance.runtime.lock()?; + //println!("are we here after runtime lock()"); + //runtime.block_on(test(self.inner, data)); + //println!("are we hereeeeeeeeeeeeeeeeeeee"); + + //return Ok(()); + + 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!("data after: {:?}", data); // if we read 0 bytes, that means the reader stream closed and we should terminate the // connection - if num_read == 0 { - return Err(MizeError::new().msg("ReceiverStream closed")); - } + //if num_read == 0 { + //return Err(MizeError::new().msg("ReceiverStream closed")); + //} Ok(()) } }