Mirrored_Repos/veloren
1
0
Fork 0
mirror of https://gitlab.com/veloren/veloren.git synced 2024-08-30 18:12:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fixing the DEADLOCK in handshake -> channel creation

Browse Source 
- this bug was initially called imbris bug, as it happened on his runners and i couldn't reproduce it locally at fist :)
- When in a Handshake a seperate mpsc::Channel was created for (Cid, Frame) transport
  however the protocol could already catch non handshake data any more and push in into this
  mpsc::Channel.
  Then this channel got dropped and a fresh one was created for the network::Channel.
  These droped Frames are ofc a BUG!
  I tried multiple things to solve this:
   - dont create a new mpsc::Channel, but instead bind it to the Protocol itself and always use 1.
     This would work theoretically, but in bParticipant side we are using 1 mpsc::Channel<(Cid, Frame)>
     to handle ALL the network::channel.
     If now ever Protocol would have it's own, and with that every network::Channel had it's own it would no longer work out
     Bad Idea...
   - using the first method but creating the mpsc::Channel inside the scheduler instead protocol neither works, as the
     scheduler doesnt know the remote_pid yet
   - i dont want a hack to say the protocol only listen to 2 messages and then stop no matter what
  So i switched over to the simply method now:
   - Do everything like before with 2 mpsc::Channels
   - after the handshake. close the receiver and listen for all remaining (cid, frame) combinations
   - when starting the channel, reapply them to the new sender/listener combination
   - added tracing
- switched Protocol RwLock to Mutex, as it's only ever 1
- Additionally changed the layout and introduces the c2w_frame_s and w2s_cid_frame_s name schema
- Fixed a bug in scheduler which WOULD cause a DEADLOCK if handshake would fail
- fixd a but in api_send_send_main, i need to store the stream_p otherwise it's immeadiatly closed and a stream_a.send() isn't guaranteed
- add extra test to verify that a send message is received even if the Stream is already closed
- changed OutGoing to Outgoing
- fixed a bug that `metrics.tick()` was never called
- removed 2 unused nightly features and added `deny_code`
This commit is contained in:
Marcel Märtens 2020-06-03 09:13:00 +02:00
parent 2a7c5807ff
commit 3324c08640
12 changed files with 320 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
network/examples/async_recv/src/main.rs
View File

@ -21,7 +21,7 @@ enum Msg {
Pong(u64),
}
/// This utility checks if async functionatily of veloren-network works
/// This utility checks if async functionality of veloren-network works
/// correctly and outputs it at the end
fn main() {
let matches = App::new("Veloren Async Prove Utility")

61
network/src/api.rs
View File

@ -3,7 +3,7 @@
//!
//! (cd network/examples/async_recv && RUST_BACKTRACE=1 cargo run)
use crate::{
message::{self, InCommingMessage, MessageBuffer, OutGoingMessage},
message::{self, IncomingMessage, MessageBuffer, OutgoingMessage},
scheduler::Scheduler,
types::{Mid, Pid, Prio, Promises, Sid},
};
@ -76,8 +76,8 @@ pub struct Stream {
mid: Mid,
prio: Prio,
promises: Promises,
a2b_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutGoingMessage)>,
b2a_msg_recv_r: mpsc::UnboundedReceiver<InCommingMessage>,
a2b_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutgoingMessage)>,
b2a_msg_recv_r: mpsc::UnboundedReceiver<IncomingMessage>,
closed: Arc<AtomicBool>,
a2b_close_stream_s: Option<mpsc::UnboundedSender<Sid>>,
}
@ -586,8 +586,8 @@ impl Stream {
sid: Sid,
prio: Prio,
promises: Promises,
a2b_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutGoingMessage)>,
b2a_msg_recv_r: mpsc::UnboundedReceiver<InCommingMessage>,
a2b_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutgoingMessage)>,
b2a_msg_recv_r: mpsc::UnboundedReceiver<IncomingMessage>,
closed: Arc<AtomicBool>,
a2b_close_stream_s: mpsc::UnboundedSender<Sid>,
) -> Self {
@ -629,56 +629,27 @@ impl Stream {
/// are also dropped.
///
/// # Example
/// ```rust
/// ```
/// use veloren_network::{Network, Address, Pid};
/// # use veloren_network::{PROMISES_ORDERED, PROMISES_CONSISTENCY};
/// use uvth::ThreadPoolBuilder;
/// use futures::executor::block_on;
/// use tracing::*;
/// use tracing_subscriber::EnvFilter;
///
/// # fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
///
/// std::thread::spawn(|| {
/// let filter = EnvFilter::from_default_env()
/// .add_directive("trace".parse().unwrap())
/// .add_directive("async_std::task::block_on=warn".parse().unwrap())
/// .add_directive("veloren_network::tests=trace".parse().unwrap())
/// .add_directive("veloren_network::controller=trace".parse().unwrap())
/// .add_directive("veloren_network::channel=trace".parse().unwrap())
/// .add_directive("veloren_network::message=trace".parse().unwrap())
/// .add_directive("veloren_network::metrics=trace".parse().unwrap())
/// .add_directive("veloren_network::types=trace".parse().unwrap());
/// let _sub = tracing_subscriber::FmtSubscriber::builder()
/// // all spans/events with a level higher than TRACE (e.g, info, warn, etc.)
/// // will be written to stdout.
/// .with_max_level(Level::TRACE)
/// .with_env_filter(filter)
/// // sets this to be the default, global subscriber for this application.
/// .try_init();
///
/// // Create a Network, listen on Port `2200` and wait for a Stream to be opened, then answer `Hello World`
/// let network = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
/// # let remote = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
/// block_on(async {
/// network.listen(Address::Tcp("127.0.0.1:2200".parse().unwrap())).await.unwrap();
/// # let remote_p = remote.connect(Address::Tcp("127.0.0.1:2200".parse().unwrap())).await.unwrap();
/// # remote_p.open(16, PROMISES_ORDERED | PROMISES_CONSISTENCY).await.unwrap();
/// let participant_a = network.connected().await.unwrap();
/// let mut stream_a = participant_a.opened().await.unwrap();
/// network.listen(Address::Tcp("127.0.0.1:2200".parse().unwrap())).await?;
/// # let remote_p = remote.connect(Address::Tcp("127.0.0.1:2200".parse().unwrap())).await?;
/// # // keep it alive
/// # let _stream_p = remote_p.open(16, PROMISES_ORDERED | PROMISES_CONSISTENCY).await?;
/// let participant_a = network.connected().await?;
/// let mut stream_a = participant_a.opened().await?;
/// //Send Message
/// stream_a.send("Hello World").unwrap();
/// stream_a.send("Hello World")?;
/// # Ok(())
/// })
/// });
///
/// std::thread::sleep(std::time::Duration::from_secs(70));
/// println!("Sleep another 10s");
/// std::thread::sleep(std::time::Duration::from_secs(10));
/// println!("TRACING THE DEADLOCK");
/// assert!(false);
///
/// std::thread::sleep(std::time::Duration::from_secs(150));
/// Ok(())
/// # }
/// ```
///
@ -740,7 +711,7 @@ impl Stream {
return Err(StreamError::StreamClosed);
}
//debug!(?messagebuffer, "sending a message");
self.a2b_msg_s.send((self.prio, self.sid, OutGoingMessage {
self.a2b_msg_s.send((self.prio, self.sid, OutgoingMessage {
buffer: messagebuffer,
cursor: 0,
mid: self.mid,
@ -760,7 +731,7 @@ impl Stream {
/// `Stream` got closed already.
///
/// # Example
/// ```rust
/// ```
/// use veloren_network::{Network, Address, Pid};
/// # use veloren_network::{PROMISES_ORDERED, PROMISES_CONSISTENCY};
/// use uvth::ThreadPoolBuilder;

110
network/src/channel.rs
View File

@ -18,26 +18,21 @@ use tracing::*;
pub(crate) struct Channel {
cid: Cid,
remote_pid: Pid,
to_wire_receiver: Option<mpsc::UnboundedReceiver<Frame>>,
c2w_frame_r: Option<mpsc::UnboundedReceiver<Frame>>,
read_stop_receiver: Option<oneshot::Receiver<()>>,
}
impl Channel {
pub fn new(
cid: u64,
remote_pid: Pid,
) -> (Self, mpsc::UnboundedSender<Frame>, oneshot::Sender<()>) {
let (to_wire_sender, to_wire_receiver) = mpsc::unbounded::<Frame>();
pub fn new(cid: u64) -> (Self, mpsc::UnboundedSender<Frame>, oneshot::Sender<()>) {
let (c2w_frame_s, c2w_frame_r) = mpsc::unbounded::<Frame>();
let (read_stop_sender, read_stop_receiver) = oneshot::channel();
(
Self {
cid,
remote_pid,
to_wire_receiver: Some(to_wire_receiver),
c2w_frame_r: Some(c2w_frame_r),
read_stop_receiver: Some(read_stop_receiver),
},
to_wire_sender,
c2w_frame_s,
read_stop_sender,
)
}
@ -45,28 +40,37 @@ impl Channel {
pub async fn run(
mut self,
protocol: Protocols,
from_wire_sender: mpsc::UnboundedSender<(Cid, Frame)>,
mut w2c_cid_frame_s: mpsc::UnboundedSender<(Cid, Frame)>,
mut leftover_cid_frame: Vec<(Cid, Frame)>,
) {
let to_wire_receiver = self.to_wire_receiver.take().unwrap();
let c2w_frame_r = self.c2w_frame_r.take().unwrap();
let read_stop_receiver = self.read_stop_receiver.take().unwrap();
trace!(?self.remote_pid, "start up channel");
//reapply leftovers from handshake
let cnt = leftover_cid_frame.len();
trace!(?self.cid, ?cnt, "reapplying leftovers");
for cid_frame in leftover_cid_frame.drain(..) {
w2c_cid_frame_s.send(cid_frame).await.unwrap();
}
trace!(?self.cid, ?cnt, "all leftovers reapplied");
trace!(?self.cid, "start up channel");
match protocol {
Protocols::Tcp(tcp) => {
futures::join!(
tcp.read(self.cid, from_wire_sender, read_stop_receiver),
tcp.write(self.cid, to_wire_receiver),
tcp.read_from_wire(self.cid, &mut w2c_cid_frame_s, read_stop_receiver),
tcp.write_to_wire(self.cid, c2w_frame_r),
);
},
Protocols::Udp(udp) => {
futures::join!(
udp.read(self.cid, from_wire_sender, read_stop_receiver),
udp.write(self.cid, to_wire_receiver),
udp.read_from_wire(self.cid, &mut w2c_cid_frame_s, read_stop_receiver),
udp.write_to_wire(self.cid, c2w_frame_r),
);
},
}
trace!(?self.remote_pid, "shut down channel");
trace!(?self.cid, "shut down channel");
}
}
@ -106,37 +110,55 @@ impl Handshake {
}
}
pub async fn setup(self, protocol: &Protocols) -> Result<(Pid, Sid, u128), ()> {
let (to_wire_sender, to_wire_receiver) = mpsc::unbounded::<Frame>();
let (from_wire_sender, from_wire_receiver) = mpsc::unbounded::<(Cid, Frame)>();
let (read_stop_sender, read_stop_receiver) = oneshot::channel();
pub async fn setup(
self,
protocol: &Protocols,
) -> Result<(Pid, Sid, u128, Vec<(Cid, Frame)>), ()> {
let (c2w_frame_s, c2w_frame_r) = mpsc::unbounded::<Frame>();
let (mut w2c_cid_frame_s, mut w2c_cid_frame_r) = mpsc::unbounded::<(Cid, Frame)>();
let (read_stop_sender, read_stop_receiver) = oneshot::channel();
let handler_future =
self.frame_handler(from_wire_receiver, to_wire_sender, read_stop_sender);
match protocol {
self.frame_handler(&mut w2c_cid_frame_r, c2w_frame_s, read_stop_sender);
let res = match protocol {
Protocols::Tcp(tcp) => {
(join! {
tcp.read(self.cid, from_wire_sender, read_stop_receiver),
tcp.write(self.cid, to_wire_receiver).fuse(),
tcp.read_from_wire(self.cid, &mut w2c_cid_frame_s, read_stop_receiver),
tcp.write_to_wire(self.cid, c2w_frame_r).fuse(),
handler_future,
})
.2
},
Protocols::Udp(udp) => {
(join! {
udp.read(self.cid, from_wire_sender, read_stop_receiver),
udp.write(self.cid, to_wire_receiver),
udp.read_from_wire(self.cid, &mut w2c_cid_frame_s, read_stop_receiver),
udp.write_to_wire(self.cid, c2w_frame_r),
handler_future,
})
.2
},
};
match res {
Ok(res) => {
let mut leftover_frames = vec![];
while let Ok(Some(cid_frame)) = w2c_cid_frame_r.try_next() {
leftover_frames.push(cid_frame);
}
let cnt = leftover_frames.len();
if cnt > 0 {
debug!(?self.cid, ?cnt, "Some additional frames got already transfered, piping them to the bparticipant as leftover_frames");
}
Ok((res.0, res.1, res.2, leftover_frames))
},
Err(e) => Err(e),
}
}
async fn frame_handler(
&self,
mut from_wire_receiver: mpsc::UnboundedReceiver<(Cid, Frame)>,
mut to_wire_sender: mpsc::UnboundedSender<Frame>,
w2c_cid_frame_r: &mut mpsc::UnboundedReceiver<(Cid, Frame)>,
mut c2w_frame_s: mpsc::UnboundedSender<Frame>,
_read_stop_sender: oneshot::Sender<()>,
) -> Result<(Pid, Sid, u128), ()> {
const ERR_S: &str = "Got A Raw Message, these are usually Debug Messages indicating that \
@ -145,10 +167,10 @@ impl Handshake {
let cid_string = self.cid.to_string();
if self.init_handshake {
self.send_handshake(&mut to_wire_sender).await;
self.send_handshake(&mut c2w_frame_s).await;
}
match from_wire_receiver.next().await {
match w2c_cid_frame_r.next().await {
Some((
_,
Frame::Handshake {
@ -170,11 +192,11 @@ impl Handshake {
.with_label_values(&["", &cid_string, "Raw"])
.inc();
debug!("sending client instructions before killing");
to_wire_sender
c2w_frame_s
.send(Frame::Raw(Self::WRONG_NUMBER.to_vec()))
.await
.unwrap();
to_wire_sender.send(Frame::Shutdown).await.unwrap();
c2w_frame_s.send(Frame::Shutdown).await.unwrap();
}
return Err(());
}
@ -187,7 +209,7 @@ impl Handshake {
.frames_out_total
.with_label_values(&["", &cid_string, "Raw"])
.inc();
to_wire_sender
c2w_frame_s
.send(Frame::Raw(
format!(
"{} Our Version: {:?}\nYour Version: {:?}\nClosing the \
@ -201,15 +223,15 @@ impl Handshake {
))
.await
.unwrap();
to_wire_sender.send(Frame::Shutdown {}).await.unwrap();
c2w_frame_s.send(Frame::Shutdown {}).await.unwrap();
}
return Err(());
}
debug!("handshake completed");
if self.init_handshake {
self.send_init(&mut to_wire_sender, &pid_string).await;
self.send_init(&mut c2w_frame_s, &pid_string).await;
} else {
self.send_handshake(&mut to_wire_sender).await;
self.send_handshake(&mut c2w_frame_s).await;
}
},
Some((_, Frame::Shutdown)) => {
@ -241,7 +263,7 @@ impl Handshake {
None => return Err(()),
};
match from_wire_receiver.next().await {
match w2c_cid_frame_r.next().await {
Some((_, Frame::Init { pid, secret })) => {
debug!(?pid, "Participant send their ID");
pid_string = pid.to_string();
@ -252,7 +274,7 @@ impl Handshake {
let stream_id_offset = if self.init_handshake {
STREAM_ID_OFFSET1
} else {
self.send_init(&mut to_wire_sender, &pid_string).await;
self.send_init(&mut c2w_frame_s, &pid_string).await;
STREAM_ID_OFFSET2
};
info!(?pid, "this Handshake is now configured!");
@ -288,12 +310,12 @@ impl Handshake {
};
}
async fn send_handshake(&self, to_wire_sender: &mut mpsc::UnboundedSender<Frame>) {
async fn send_handshake(&self, c2w_frame_s: &mut mpsc::UnboundedSender<Frame>) {
self.metrics
.frames_out_total
.with_label_values(&["", &self.cid.to_string(), "Handshake"])
.inc();
to_wire_sender
c2w_frame_s
.send(Frame::Handshake {
magic_number: VELOREN_MAGIC_NUMBER,
version: VELOREN_NETWORK_VERSION,
@ -302,12 +324,12 @@ impl Handshake {
.unwrap();
}
async fn send_init(&self, to_wire_sender: &mut mpsc::UnboundedSender<Frame>, pid_string: &str) {
async fn send_init(&self, c2w_frame_s: &mut mpsc::UnboundedSender<Frame>, pid_string: &str) {
self.metrics
.frames_out_total
.with_label_values(&[pid_string, &self.cid.to_string(), "ParticipantId"])
.inc();
to_wire_sender
c2w_frame_s
.send(Frame::Init {
pid: self.local_pid,
secret: self.secret,

3
network/src/lib.rs
View File

@ -1,4 +1,5 @@
#![feature(trait_alias, try_trait, async_closure, const_if_match)]
#![deny(unsafe_code)]
#![feature(try_trait, const_if_match)]
//! Crate to handle high level networking of messages with different
//! requirements and priorities over a number of protocols

13
network/src/message.rs
View File

@ -19,7 +19,7 @@ pub struct MessageBuffer {
}
#[derive(Debug)]
pub(crate) struct OutGoingMessage {
pub(crate) struct OutgoingMessage {
pub buffer: Arc<MessageBuffer>,
pub cursor: u64,
pub mid: Mid,
@ -27,7 +27,7 @@ pub(crate) struct OutGoingMessage {
}
#[derive(Debug)]
pub(crate) struct InCommingMessage {
pub(crate) struct IncomingMessage {
pub buffer: MessageBuffer,
pub length: u64,
pub mid: Mid,
@ -35,13 +35,20 @@ pub(crate) struct InCommingMessage {
}
pub(crate) fn serialize<M: Serialize>(message: &M) -> MessageBuffer {
//this will never fail: https://docs.rs/bincode/0.8.0/bincode/fn.serialize.html
let writer = bincode::serialize(message).unwrap();
MessageBuffer { data: writer }
}
pub(crate) fn deserialize<M: DeserializeOwned>(buffer: MessageBuffer) -> M {
let span = buffer.data;
let decoded: M = bincode::deserialize(span.as_slice()).unwrap();
//this might fail if you choose the wrong type for M. in that case probably X
// got transfered while you assume Y. probably this means your application
// logic is wrong. E.g. You expect a String, but just get a u8.
let decoded: M = bincode::deserialize(span.as_slice()).expect(
"deserialisation failed, this is probably due to a programming error on YOUR side, \
probably the type send by remote isn't what you are expecting. change the type of `M`",
);
decoded
}

92
network/src/participant.rs
View File

@ -1,7 +1,7 @@
use crate::{
api::Stream,
channel::Channel,
message::{InCommingMessage, MessageBuffer, OutGoingMessage},
message::{IncomingMessage, MessageBuffer, OutgoingMessage},
metrics::{NetworkMetrics, PidCidFrameCache},
prios::PrioManager,
protocols::Protocols,
@ -37,7 +37,7 @@ struct ChannelInfo {
struct StreamInfo {
prio: Prio,
promises: Promises,
b2a_msg_recv_s: mpsc::UnboundedSender<InCommingMessage>,
b2a_msg_recv_s: mpsc::UnboundedSender<IncomingMessage>,
closed: Arc<AtomicBool>,
}
@ -45,7 +45,8 @@ struct StreamInfo {
struct ControlChannels {
a2b_steam_open_r: mpsc::UnboundedReceiver<(Prio, Promises, oneshot::Sender<Stream>)>,
b2a_stream_opened_s: mpsc::UnboundedSender<Stream>,
s2b_create_channel_r: mpsc::UnboundedReceiver<(Cid, Sid, Protocols, oneshot::Sender<()>)>,
s2b_create_channel_r:
mpsc::UnboundedReceiver<(Cid, Sid, Protocols, Vec<(Cid, Frame)>, oneshot::Sender<()>)>,
a2b_close_stream_r: mpsc::UnboundedReceiver<Sid>,
a2b_close_stream_s: mpsc::UnboundedSender<Sid>,
s2b_shutdown_bparticipant_r: oneshot::Receiver<oneshot::Sender<async_std::io::Result<()>>>, /* own */
@ -72,7 +73,7 @@ impl BParticipant {
Self,
mpsc::UnboundedSender<(Prio, Promises, oneshot::Sender<Stream>)>,
mpsc::UnboundedReceiver<Stream>,
mpsc::UnboundedSender<(Cid, Sid, Protocols, oneshot::Sender<()>)>,
mpsc::UnboundedSender<(Cid, Sid, Protocols, Vec<(Cid, Frame)>, oneshot::Sender<()>)>,
oneshot::Sender<oneshot::Sender<async_std::io::Result<()>>>,
) {
let (a2b_steam_open_s, a2b_steam_open_r) =
@ -80,8 +81,7 @@ impl BParticipant {
let (b2a_stream_opened_s, b2a_stream_opened_r) = mpsc::unbounded::<Stream>();
let (a2b_close_stream_s, a2b_close_stream_r) = mpsc::unbounded();
let (s2b_shutdown_bparticipant_s, s2b_shutdown_bparticipant_r) = oneshot::channel();
let (s2b_create_channel_s, s2b_create_channel_r) =
mpsc::unbounded::<(Cid, Sid, Protocols, oneshot::Sender<()>)>();
let (s2b_create_channel_s, s2b_create_channel_r) = mpsc::unbounded();
let run_channels = Some(ControlChannels {
a2b_steam_open_r,
@ -136,7 +136,7 @@ impl BParticipant {
run_channels.a2b_close_stream_s,
a2p_msg_s.clone(),
),
self.create_channel_mgr(run_channels.s2b_create_channel_r, w2b_frames_s,),
self.create_channel_mgr(run_channels.s2b_create_channel_r, w2b_frames_s),
self.send_mgr(
prios,
shutdown_send_mgr_receiver,
@ -243,7 +243,7 @@ impl BParticipant {
mut w2b_frames_r: mpsc::UnboundedReceiver<(Cid, Frame)>,
mut b2a_stream_opened_s: mpsc::UnboundedSender<Stream>,
a2b_close_stream_s: mpsc::UnboundedSender<Sid>,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutGoingMessage)>,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutgoingMessage)>,
) {
self.running_mgr.fetch_add(1, Ordering::Relaxed);
trace!("start handle_frames_mgr");
@ -300,7 +300,7 @@ impl BParticipant {
}
},
Frame::DataHeader { mid, sid, length } => {
let imsg = InCommingMessage {
let imsg = IncomingMessage {
buffer: MessageBuffer { data: Vec::new() },
length,
mid,
@ -367,40 +367,50 @@ impl BParticipant {
async fn create_channel_mgr(
&self,
s2b_create_channel_r: mpsc::UnboundedReceiver<(Cid, Sid, Protocols, oneshot::Sender<()>)>,
s2b_create_channel_r: mpsc::UnboundedReceiver<(
Cid,
Sid,
Protocols,
Vec<(Cid, Frame)>,
oneshot::Sender<()>,
)>,
w2b_frames_s: mpsc::UnboundedSender<(Cid, Frame)>,
) {
self.running_mgr.fetch_add(1, Ordering::Relaxed);
trace!("start create_channel_mgr");
s2b_create_channel_r
.for_each_concurrent(None, |(cid, _, protocol, b2s_create_channel_done_s)| {
// This channel is now configured, and we are running it in scope of the
// participant.
let w2b_frames_s = w2b_frames_s.clone();
let channels = self.channels.clone();
async move {
let (channel, b2w_frame_s, b2r_read_shutdown) =
Channel::new(cid, self.remote_pid);
channels.write().await.push(ChannelInfo {
cid,
cid_string: cid.to_string(),
b2w_frame_s,
b2r_read_shutdown,
});
b2s_create_channel_done_s.send(()).unwrap();
self.metrics
.channels_connected_total
.with_label_values(&[&self.remote_pid_string])
.inc();
trace!(?cid, "running channel in participant");
channel.run(protocol, w2b_frames_s).await;
self.metrics
.channels_disconnected_total
.with_label_values(&[&self.remote_pid_string])
.inc();
trace!(?cid, "channel got closed");
}
})
.for_each_concurrent(
None,
|(cid, _, protocol, leftover_cid_frame, b2s_create_channel_done_s)| {
// This channel is now configured, and we are running it in scope of the
// participant.
let w2b_frames_s = w2b_frames_s.clone();
let channels = self.channels.clone();
async move {
let (channel, b2w_frame_s, b2r_read_shutdown) = Channel::new(cid);
channels.write().await.push(ChannelInfo {
cid,
cid_string: cid.to_string(),
b2w_frame_s,
b2r_read_shutdown,
});
b2s_create_channel_done_s.send(()).unwrap();
self.metrics
.channels_connected_total
.with_label_values(&[&self.remote_pid_string])
.inc();
trace!(?cid, "running channel in participant");
channel
.run(protocol, w2b_frames_s, leftover_cid_frame)
.await;
self.metrics
.channels_disconnected_total
.with_label_values(&[&self.remote_pid_string])
.inc();
trace!(?cid, "channel got closed");
}
},
)
.await;
trace!("stop create_channel_mgr");
self.running_mgr.fetch_sub(1, Ordering::Relaxed);
@ -410,7 +420,7 @@ impl BParticipant {
&self,
mut a2b_steam_open_r: mpsc::UnboundedReceiver<(Prio, Promises, oneshot::Sender<Stream>)>,
a2b_close_stream_s: mpsc::UnboundedSender<Sid>,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutGoingMessage)>,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutgoingMessage)>,
shutdown_open_mgr_receiver: oneshot::Receiver<()>,
) {
self.running_mgr.fetch_add(1, Ordering::Relaxed);
@ -562,10 +572,10 @@ impl BParticipant {
sid: Sid,
prio: Prio,
promises: Promises,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutGoingMessage)>,
a2p_msg_s: std::sync::mpsc::Sender<(Prio, Sid, OutgoingMessage)>,
a2b_close_stream_s: &mpsc::UnboundedSender<Sid>,
) -> Stream {
let (b2a_msg_recv_s, b2a_msg_recv_r) = mpsc::unbounded::<InCommingMessage>();
let (b2a_msg_recv_s, b2a_msg_recv_r) = mpsc::unbounded::<IncomingMessage>();
let closed = Arc::new(AtomicBool::new(false));
self.streams.write().await.insert(sid, StreamInfo {
prio,

22
network/src/prios.rs
View File

@ -6,7 +6,7 @@
//! immeadiatly when found!
use crate::{
message::OutGoingMessage,
message::OutgoingMessage,
metrics::NetworkMetrics,
types::{Frame, Prio, Sid},
};
@ -30,8 +30,8 @@ struct PidSidInfo {
pub(crate) struct PrioManager {
points: [u32; PRIO_MAX],
messages: [VecDeque<(Sid, OutGoingMessage)>; PRIO_MAX],
messages_rx: Receiver<(Prio, Sid, OutGoingMessage)>,
messages: [VecDeque<(Sid, OutgoingMessage)>; PRIO_MAX],
messages_rx: Receiver<(Prio, Sid, OutgoingMessage)>,
sid_owned: HashMap<Sid, PidSidInfo>,
//you can register to be notified if a pid_sid combination is flushed completly here
sid_flushed_rx: Receiver<(Sid, oneshot::Sender<()>)>,
@ -55,7 +55,7 @@ impl PrioManager {
pid: String,
) -> (
Self,
Sender<(Prio, Sid, OutGoingMessage)>,
Sender<(Prio, Sid, OutgoingMessage)>,
Sender<(Sid, oneshot::Sender<()>)>,
) {
// (a2p_msg_s, a2p_msg_r)
@ -205,7 +205,7 @@ impl PrioManager {
/// returns if msg is empty
fn tick_msg<E: Extend<(Sid, Frame)>>(
msg: &mut OutGoingMessage,
msg: &mut OutgoingMessage,
msg_sid: Sid,
frames: &mut E,
) -> bool {
@ -311,7 +311,7 @@ impl std::fmt::Debug for PrioManager {
#[cfg(test)]
mod tests {
use crate::{
message::{MessageBuffer, OutGoingMessage},
message::{MessageBuffer, OutgoingMessage},
metrics::NetworkMetrics,
prios::*,
types::{Frame, Pid, Prio, Sid},
@ -327,7 +327,7 @@ mod tests {
fn mock_new() -> (
PrioManager,
Sender<(Prio, Sid, OutGoingMessage)>,
Sender<(Prio, Sid, OutgoingMessage)>,
Sender<(Sid, oneshot::Sender<()>)>,
) {
let pid = Pid::fake(1);
@ -337,9 +337,9 @@ mod tests {
)
}
fn mock_out(prio: Prio, sid: u64) -> (Prio, Sid, OutGoingMessage) {
fn mock_out(prio: Prio, sid: u64) -> (Prio, Sid, OutgoingMessage) {
let sid = Sid::new(sid);
(prio, sid, OutGoingMessage {
(prio, sid, OutgoingMessage {
buffer: Arc::new(MessageBuffer {
data: vec![48, 49, 50],
}),
@ -349,12 +349,12 @@ mod tests {
})
}
fn mock_out_large(prio: Prio, sid: u64) -> (Prio, Sid, OutGoingMessage) {
fn mock_out_large(prio: Prio, sid: u64) -> (Prio, Sid, OutgoingMessage) {
let sid = Sid::new(sid);
let mut data = vec![48; USIZE];
data.append(&mut vec![49; USIZE]);
data.append(&mut vec![50; 20]);
(prio, sid, OutGoingMessage {
(prio, sid, OutgoingMessage {
buffer: Arc::new(MessageBuffer { data }),
cursor: 0,
mid: 1,

194
network/src/protocols.rs
View File

@ -5,11 +5,11 @@ use crate::{
use async_std::{
net::{TcpStream, UdpSocket},
prelude::*,
sync::RwLock,
};
use futures::{
channel::{mpsc, oneshot},
future::FutureExt,
lock::Mutex,
select,
sink::SinkExt,
stream::StreamExt,
@ -49,7 +49,7 @@ pub(crate) struct UdpProtocol {
socket: Arc<UdpSocket>,
remote_addr: SocketAddr,
metrics: Arc<NetworkMetrics>,
data_in: RwLock<mpsc::UnboundedReceiver<Vec<u8>>>,
data_in: Mutex<mpsc::UnboundedReceiver<Vec<u8>>>,
}
//TODO: PERFORMACE: Use BufWriter and BufReader from std::io!
@ -63,25 +63,22 @@ impl TcpProtocol {
cid: Cid,
mut stream: &TcpStream,
mut bytes: &mut [u8],
from_wire_sender: &mut mpsc::UnboundedSender<(Cid, Frame)>,
w2c_cid_frame_s: &mut mpsc::UnboundedSender<(Cid, Frame)>,
) {
match stream.read_exact(&mut bytes).await {
Err(e) => {
warn!(
?e,
"closing tcp protocol due to read error, sending close frame to gracefully \
shutdown"
);
from_wire_sender.send((cid, Frame::Shutdown)).await.unwrap();
},
_ => (),
if let Err(e) = stream.read_exact(&mut bytes).await {
warn!(
?e,
"closing tcp protocol due to read error, sending close frame to gracefully \
shutdown"
);
w2c_cid_frame_s.send((cid, Frame::Shutdown)).await.unwrap();
}
}
pub async fn read(
pub async fn read_from_wire(
&self,
cid: Cid,
mut from_wire_sender: mpsc::UnboundedSender<(Cid, Frame)>,
w2c_cid_frame_s: &mut mpsc::UnboundedSender<(Cid, Frame)>,
end_receiver: oneshot::Receiver<()>,
) {
trace!("starting up tcp read()");
@ -107,8 +104,7 @@ impl TcpProtocol {
let frame = match frame_no {
FRAME_HANDSHAKE => {
let mut bytes = [0u8; 19];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let magic_number = [
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
];
@ -123,8 +119,7 @@ impl TcpProtocol {
},
FRAME_INIT => {
let mut bytes = [0u8; 16];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let pid = Pid::from_le_bytes(bytes);
stream.read_exact(&mut bytes).await.unwrap();
let secret = u128::from_le_bytes(bytes);
@ -133,8 +128,7 @@ impl TcpProtocol {
FRAME_SHUTDOWN => Frame::Shutdown,
FRAME_OPEN_STREAM => {
let mut bytes = [0u8; 10];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let sid = Sid::from_le_bytes([
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
bytes[7],
@ -149,8 +143,7 @@ impl TcpProtocol {
},
FRAME_CLOSE_STREAM => {
let mut bytes = [0u8; 8];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let sid = Sid::from_le_bytes([
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
bytes[7],
@ -159,8 +152,7 @@ impl TcpProtocol {
},
FRAME_DATA_HEADER => {
let mut bytes = [0u8; 24];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let mid = Mid::from_le_bytes([
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
bytes[7],
@ -177,8 +169,7 @@ impl TcpProtocol {
},
FRAME_DATA => {
let mut bytes = [0u8; 18];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let mid = Mid::from_le_bytes([
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6],
bytes[7],
@ -190,31 +181,27 @@ impl TcpProtocol {
let length = u16::from_le_bytes([bytes[16], bytes[17]]);
let mut data = vec![0; length as usize];
throughput_cache.inc_by(length as i64);
Self::read_except_or_close(cid, &mut stream, &mut data, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut data, w2c_cid_frame_s).await;
Frame::Data { mid, start, data }
},
FRAME_RAW => {
let mut bytes = [0u8; 2];
Self::read_except_or_close(cid, &mut stream, &mut bytes, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut bytes, w2c_cid_frame_s).await;
let length = u16::from_le_bytes([bytes[0], bytes[1]]);
let mut data = vec![0; length as usize];
Self::read_except_or_close(cid, &mut stream, &mut data, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut data, w2c_cid_frame_s).await;
Frame::Raw(data)
},
_ => {
// report a RAW frame, but cannot rely on the next 2 bytes to be a size.
// guessing 256 bytes, which might help to sort down issues
let mut data = vec![0; 256];
Self::read_except_or_close(cid, &mut stream, &mut data, &mut from_wire_sender)
.await;
Self::read_except_or_close(cid, &mut stream, &mut data, w2c_cid_frame_s).await;
Frame::Raw(data)
},
};
metrics_cache.with_label_values(&frame).inc();
from_wire_sender.send((cid, frame)).await.unwrap();
w2c_cid_frame_s.send((cid, frame)).await.unwrap();
}
trace!("shutting down tcp read()");
}
@ -241,7 +228,7 @@ impl TcpProtocol {
//dezerialize here as this is executed in a seperate thread PER channel.
// Limites Throughput per single Receiver but stays in same thread (maybe as its
// in a threadpool) for TCP, UDP and MPSC
pub async fn write(&self, cid: Cid, mut to_wire_receiver: mpsc::UnboundedReceiver<Frame>) {
pub async fn write_to_wire(&self, cid: Cid, mut c2w_frame_r: mpsc::UnboundedReceiver<Frame>) {
trace!("starting up tcp write()");
let mut stream = self.stream.clone();
let mut metrics_cache = CidFrameCache::new(self.metrics.frames_wire_out_total.clone(), cid);
@ -249,7 +236,7 @@ impl TcpProtocol {
.metrics
.wire_out_throughput
.with_label_values(&[&cid.to_string()]);
while let Some(frame) = to_wire_receiver.next().await {
while let Some(frame) = c2w_frame_r.next().await {
metrics_cache.with_label_values(&frame).inc();
if match frame {
Frame::Handshake {
@ -259,47 +246,38 @@ impl TcpProtocol {
Self::write_or_close(
&mut stream,
&FRAME_HANDSHAKE.to_be_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(&mut stream, &magic_number, &mut to_wire_receiver)
.await
|| Self::write_or_close(&mut stream, &magic_number, &mut c2w_frame_r).await
|| Self::write_or_close(
&mut stream,
&version[0].to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(
&mut stream,
&version[1].to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(
&mut stream,
&version[2].to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
},
Frame::Init { pid, secret } => {
Self::write_or_close(
&mut stream,
&FRAME_INIT.to_be_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(
&mut stream,
&pid.to_le_bytes(),
&mut to_wire_receiver,
)
Self::write_or_close(&mut stream, &FRAME_INIT.to_be_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(&mut stream, &pid.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(
&mut stream,
&secret.to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
},
@ -307,7 +285,7 @@ impl TcpProtocol {
Self::write_or_close(
&mut stream,
&FRAME_SHUTDOWN.to_be_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
},
@ -319,25 +297,17 @@ impl TcpProtocol {
Self::write_or_close(
&mut stream,
&FRAME_OPEN_STREAM.to_be_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(
&mut stream,
&sid.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(
&mut stream,
&prio.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(&mut stream, &sid.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(&mut stream, &prio.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(
&mut stream,
&promises.to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
},
@ -345,84 +315,56 @@ impl TcpProtocol {
Self::write_or_close(
&mut stream,
&FRAME_CLOSE_STREAM.to_be_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(
&mut stream,
&sid.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(&mut stream, &sid.to_le_bytes(), &mut c2w_frame_r)
.await
},
Frame::DataHeader { mid, sid, length } => {
Self::write_or_close(
&mut stream,
&FRAME_DATA_HEADER.to_be_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(
&mut stream,
&mid.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(
&mut stream,
&sid.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(&mut stream, &mid.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(&mut stream, &sid.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(
&mut stream,
&length.to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
},
Frame::Data { mid, start, data } => {
throughput_cache.inc_by(data.len() as i64);
Self::write_or_close(
&mut stream,
&FRAME_DATA.to_be_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(
&mut stream,
&mid.to_le_bytes(),
&mut to_wire_receiver,
)
.await
|| Self::write_or_close(
&mut stream,
&start.to_le_bytes(),
&mut to_wire_receiver,
)
Self::write_or_close(&mut stream, &FRAME_DATA.to_be_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(&mut stream, &mid.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(&mut stream, &start.to_le_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(
&mut stream,
&(data.len() as u16).to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(&mut stream, &data, &mut to_wire_receiver).await
|| Self::write_or_close(&mut stream, &data, &mut c2w_frame_r).await
},
Frame::Raw(data) => {
Self::write_or_close(
&mut stream,
&FRAME_RAW.to_be_bytes(),
&mut to_wire_receiver,
)
.await
Self::write_or_close(&mut stream, &FRAME_RAW.to_be_bytes(), &mut c2w_frame_r)
.await
|| Self::write_or_close(
&mut stream,
&(data.len() as u16).to_le_bytes(),
&mut to_wire_receiver,
&mut c2w_frame_r,
)
.await
|| Self::write_or_close(&mut stream, &data, &mut to_wire_receiver).await
|| Self::write_or_close(&mut stream, &data, &mut c2w_frame_r).await
},
} {
//failure
@ -444,14 +386,14 @@ impl UdpProtocol {
socket,
remote_addr,
metrics,
data_in: RwLock::new(data_in),
data_in: Mutex::new(data_in),
}
}
pub async fn read(
pub async fn read_from_wire(
&self,
cid: Cid,
mut from_wire_sender: mpsc::UnboundedSender<(Cid, Frame)>,
w2c_cid_frame_s: &mut mpsc::UnboundedSender<(Cid, Frame)>,
end_receiver: oneshot::Receiver<()>,
) {
trace!("starting up udp read()");
@ -460,7 +402,7 @@ impl UdpProtocol {
.metrics
.wire_in_throughput
.with_label_values(&[&cid.to_string()]);
let mut data_in = self.data_in.write().await;
let mut data_in = self.data_in.lock().await;
let mut end_receiver = end_receiver.fuse();
while let Some(bytes) = select! {
r = data_in.next().fuse() => r,
@ -559,12 +501,12 @@ impl UdpProtocol {
_ => Frame::Raw(bytes),
};
metrics_cache.with_label_values(&frame).inc();
from_wire_sender.send((cid, frame)).await.unwrap();
w2c_cid_frame_s.send((cid, frame)).await.unwrap();
}
trace!("shutting down udp read()");
}
pub async fn write(&self, cid: Cid, mut to_wire_receiver: mpsc::UnboundedReceiver<Frame>) {
pub async fn write_to_wire(&self, cid: Cid, mut c2w_frame_r: mpsc::UnboundedReceiver<Frame>) {
trace!("starting up udp write()");
let mut buffer = [0u8; 2000];
let mut metrics_cache = CidFrameCache::new(self.metrics.frames_wire_out_total.clone(), cid);
@ -572,7 +514,7 @@ impl UdpProtocol {
.metrics
.wire_out_throughput
.with_label_values(&[&cid.to_string()]);
while let Some(frame) = to_wire_receiver.next().await {
while let Some(frame) = c2w_frame_r.next().await {
metrics_cache.with_label_values(&frame).inc();
let len = match frame {
Frame::Handshake {

61
network/src/scheduler.rs
View File

@ -4,7 +4,7 @@ use crate::{
metrics::NetworkMetrics,
participant::BParticipant,
protocols::{Protocols, TcpProtocol, UdpProtocol},
types::{Cid, Pid, Sid},
types::{Cid, Frame, Pid, Sid},
};
use async_std::{
io, net,
@ -33,7 +33,8 @@ use tracing_futures::Instrument;
#[derive(Debug)]
struct ParticipantInfo {
secret: u128,
s2b_create_channel_s: mpsc::UnboundedSender<(Cid, Sid, Protocols, oneshot::Sender<()>)>,
s2b_create_channel_s:
mpsc::UnboundedSender<(Cid, Sid, Protocols, Vec<(Cid, Frame)>, oneshot::Sender<()>)>,
s2b_shutdown_bparticipant_s:
Option<oneshot::Sender<oneshot::Sender<async_std::io::Result<()>>>>,
}
@ -45,6 +46,7 @@ struct ParticipantInfo {
/// - p: prios
/// - r: protocol
/// - w: wire
/// - c: channel/handshake
#[derive(Debug)]
struct ControlChannels {
a2s_listen_r: mpsc::UnboundedReceiver<(Address, oneshot::Sender<io::Result<()>>)>,
@ -205,8 +207,10 @@ impl Scheduler {
},
};
info!("Connecting Tcp to: {}", stream.peer_addr().unwrap());
let protocol = Protocols::Tcp(TcpProtocol::new(stream, self.metrics.clone()));
(protocol, false)
(
Protocols::Tcp(TcpProtocol::new(stream, self.metrics.clone())),
false,
)
},
Address::Udp(addr) => {
self.metrics
@ -226,17 +230,17 @@ impl Scheduler {
};
info!("Connecting Udp to: {}", addr);
let (udp_data_sender, udp_data_receiver) = mpsc::unbounded::<Vec<u8>>();
let protocol = Protocols::Udp(UdpProtocol::new(
let protocol = UdpProtocol::new(
socket.clone(),
addr,
self.metrics.clone(),
udp_data_receiver,
));
);
self.pool.spawn_ok(
Self::udp_single_channel_connect(socket.clone(), udp_data_sender)
.instrument(tracing::info_span!("udp", ?addr)),
);
(protocol, true)
(Protocols::Udp(protocol), true)
},
_ => unimplemented!(),
};
@ -360,12 +364,9 @@ impl Scheduler {
} {
let stream = stream.unwrap();
info!("Accepting Tcp from: {}", stream.peer_addr().unwrap());
self.init_protocol(
Protocols::Tcp(TcpProtocol::new(stream, self.metrics.clone())),
None,
true,
)
.await;
let protocol = TcpProtocol::new(stream, self.metrics.clone());
self.init_protocol(Protocols::Tcp(protocol), None, true)
.await;
}
},
Address::Udp(addr) => {
@ -400,13 +401,14 @@ impl Scheduler {
info!("Accepting Udp from: {}", &remote_addr);
let (udp_data_sender, udp_data_receiver) = mpsc::unbounded::<Vec<u8>>();
listeners.insert(remote_addr.clone(), udp_data_sender);
let protocol = Protocols::Udp(UdpProtocol::new(
let protocol = UdpProtocol::new(
socket.clone(),
remote_addr.clone(),
self.metrics.clone(),
udp_data_receiver,
));
self.init_protocol(protocol, None, false).await;
);
self.init_protocol(Protocols::Udp(protocol), None, false)
.await;
}
let udp_data_sender = listeners.get_mut(&remote_addr).unwrap();
udp_data_sender.send(datavec).await.unwrap();
@ -476,7 +478,7 @@ impl Scheduler {
send_handshake,
);
match handshake.setup(&protocol).await {
Ok((pid, sid, secret)) => {
Ok((pid, sid, secret, leftover_cid_frame)) => {
trace!(
?cid,
?pid,
@ -515,13 +517,20 @@ impl Scheduler {
//create a new channel within BParticipant and wait for it to run
let (b2s_create_channel_done_s, b2s_create_channel_done_r) =
oneshot::channel();
//From now on wire connects directly with bparticipant!
s2b_create_channel_s
.send((cid, sid, protocol, b2s_create_channel_done_s))
.send((
cid,
sid,
protocol,
leftover_cid_frame,
b2s_create_channel_done_s,
))
.await
.unwrap();
b2s_create_channel_done_r.await.unwrap();
if let Some(pid_oneshot) = s2a_return_pid_s {
// someone is waiting with connect, so give them their PID
// someone is waiting with `connect`, so give them their PID
pid_oneshot.send(Ok(participant)).unwrap();
} else {
// noone is waiting on this Participant, return in to Network
@ -543,7 +552,7 @@ impl Scheduler {
error!("just dropping here, TODO handle this correctly!");
//TODO
if let Some(pid_oneshot) = s2a_return_pid_s {
// someone is waiting with connect, so give them their Error
// someone is waiting with `connect`, so give them their Error
pid_oneshot
.send(Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
@ -561,7 +570,17 @@ impl Scheduler {
//From now on this CHANNEL can receiver other frames!
// move directly to participant!
},
Err(()) => {},
Err(()) => {
if let Some(pid_oneshot) = s2a_return_pid_s {
// someone is waiting with `connect`, so give them their Error
pid_oneshot
.send(Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"handshake failed, denying connection",
)))
.unwrap();
}
},
}
}
.instrument(tracing::trace_span!("")),

30
network/tests/closing.rs
View File

@ -46,6 +46,23 @@ fn close_stream() {
);
}
///THIS is actually a bug which currently luckily doesn't trigger, but with new
/// async-std WE must make sure, if a stream is `drop`ed inside a `block_on`,
/// that no panic is thrown.
#[test]
fn close_streams_in_block_on() {
let (_, _) = helper::setup(false, 0);
let (_n_a, _p_a, s1_a, _n_b, _p_b, s1_b) = block_on(network_participant_stream(tcp()));
block_on(async {
//make it locally so that they are dropped later
let mut s1_a = s1_a;
let mut s1_b = s1_b;
s1_a.send("ping").unwrap();
assert_eq!(s1_b.recv().await, Ok("ping".to_string()));
drop(s1_a);
});
}
#[test]
fn stream_simple_3msg_then_close() {
let (_, _) = helper::setup(false, 0);
@ -79,6 +96,19 @@ fn stream_send_first_then_receive() {
assert_eq!(s1_b.send("Hello World"), Err(StreamError::StreamClosed));
}
#[test]
fn stream_send_1_then_close_stream() {
let (_, _) = helper::setup(false, 0);
let (_n_a, _, mut s1_a, _n_b, _, mut s1_b) = block_on(network_participant_stream(tcp()));
s1_a.send("this message must be received, even if stream is closed already!")
.unwrap();
drop(s1_a);
std::thread::sleep(std::time::Duration::from_millis(500));
let exp = Ok("this message must be received, even if stream is closed already!".to_string());
assert_eq!(block_on(s1_b.recv()), exp);
println!("all received and done");
}
#[test]
fn stream_send_100000_then_close_stream() {
let (_, _) = helper::setup(false, 0);

33
network/tests/integration.rs
View File

@ -62,7 +62,7 @@ fn stream_simple_udp_3msg() {
#[test]
#[ignore]
fn tcp_and_udp_2_connections() -> std::result::Result<(), Box<dyn std::error::Error>> {
let (_, _) = helper::setup(true, 0);
let (_, _) = helper::setup(false, 0);
let network = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
let remote = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
block_on(async {
@ -110,23 +110,24 @@ fn failed_listen_on_used_ports() -> std::result::Result<(), Box<dyn std::error::
/// There is a bug an impris-desktop-1 which fails the DOC tests,
/// it fails exactly `api_stream_send_main` and `api_stream_recv_main` by
/// deadlocking at different times!
/// So i rather put the same test into a unit test, as my gues is that it's
/// compiler related
/// So i rather put the same test into a unit test, these are now duplicate to
/// the api, but are left here, just to be save!
#[test]
fn api_stream_send_main() -> std::result::Result<(), Box<dyn std::error::Error>> {
let (_, _) = helper::setup(false, 0);
// Create a Network, listen on Port `2200` and wait for a Stream to be opened,
// Create a Network, listen on Port `1200` and wait for a Stream to be opened,
// then answer `Hello World`
let network = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
let remote = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
block_on(async {
network
.listen(Address::Tcp("127.0.0.1:2200".parse().unwrap()))
.listen(Address::Tcp("127.0.0.1:1200".parse().unwrap()))
.await?;
let remote_p = remote
.connect(Address::Tcp("127.0.0.1:2200".parse().unwrap()))
.connect(Address::Tcp("127.0.0.1:1200".parse().unwrap()))
.await?;
remote_p
// keep it alive
let _stream_p = remote_p
.open(16, PROMISES_ORDERED | PROMISES_CONSISTENCY)
.await?;
let participant_a = network.connected().await?;
@ -140,16 +141,16 @@ fn api_stream_send_main() -> std::result::Result<(), Box<dyn std::error::Error>>
#[test]
fn api_stream_recv_main() -> std::result::Result<(), Box<dyn std::error::Error>> {
let (_, _) = helper::setup(false, 0);
// Create a Network, listen on Port `2220` and wait for a Stream to be opened,
// Create a Network, listen on Port `1220` and wait for a Stream to be opened,
// then listen on it
let network = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
let remote = Network::new(Pid::new(), &ThreadPoolBuilder::new().build(), None);
block_on(async {
network
.listen(Address::Tcp("127.0.0.1:2220".parse().unwrap()))
.listen(Address::Tcp("127.0.0.1:1220".parse().unwrap()))
.await?;
let remote_p = remote
.connect(Address::Tcp("127.0.0.1:2220".parse().unwrap()))
.connect(Address::Tcp("127.0.0.1:1220".parse().unwrap()))
.await?;
let mut stream_p = remote_p
.open(16, PROMISES_ORDERED | PROMISES_CONSISTENCY)
@ -158,7 +159,17 @@ fn api_stream_recv_main() -> std::result::Result<(), Box<dyn std::error::Error>>
let participant_a = network.connected().await?;
let mut stream_a = participant_a.opened().await?;
//Send Message
println!("{}", stream_a.recv::<String>().await?);
assert_eq!("Hello World".to_string(), stream_a.recv::<String>().await?);
Ok(())
})
}
#[test]
#[should_panic]
fn wrong_parse() {
let (_, _) = helper::setup(false, 0);
let (_n_a, _, mut s1_a, _n_b, _, mut s1_b) = block_on(network_participant_stream(tcp()));
s1_a.send(1337).unwrap();
assert_eq!(block_on(s1_b.recv()), Ok("Hello World".to_string()));
}

5
server/src/lib.rs
View File

@ -79,7 +79,7 @@ pub struct Server {
thread_pool: ThreadPool,
server_info: ServerInfo,
_metrics: ServerMetrics,
metrics: ServerMetrics,
tick_metrics: TickMetrics,
server_settings: ServerSettings,
@ -242,7 +242,7 @@ impl Server {
git_date: common::util::GIT_DATE.to_string(),
auth_provider: settings.auth_server_address.clone(),
},
_metrics: metrics,
metrics,
tick_metrics,
server_settings: settings.clone(),
};
@ -494,6 +494,7 @@ impl Server {
.tick_time
.with_label_values(&["metrics"])
.set(end_of_server_tick.elapsed().as_nanos() as i64);
self.metrics.tick();
// 8) Finish the tick, pass control back to the frontend.