mirror of
https://gitlab.com/veloren/veloren.git
synced 2024-08-30 18:12:32 +00:00
9028578bc8
Instead of keeping Runtime and manually spawn a task on `drop` this task is spawned at start and will wait to be triggered. The `drop` methods then wait for completion, UNLESS they are in a async context, then they MUST NOT BLOCK (deadlock potential), so they defer it to the Runtime and HOPE for the runtime to exist long enough. This get rid of the weird `block_in_place` which is only accessable with `rt-multi-threaded` and has some disadvantages. We also wont requiere the runtime to be active all the time. Though its needed for a clean shutdown
450 lines
16 KiB
Rust
450 lines
16 KiB
Rust
//! How to read those tests:
|
|
//! - in the first line we call the helper, this is only debug code. in case
|
|
//! you want to have tracing for a special test you set set the bool = true
|
|
//! and the sleep to 10000 and your test will start 10 sec delayed with
|
|
//! tracing. You need a delay as otherwise the other tests pollute your trace
|
|
//! - the second line is to simulate a client and a server
|
|
//! `network_participant_stream` will return
|
|
//! - 2 networks
|
|
//! - 2 participants
|
|
//! - 2 streams
|
|
//! each one `linked` to their counterpart.
|
|
//! You see a cryptic use of rust `_` this is because we are testing the
|
|
//! `drop` behavior here.
|
|
//! - A `_` means this is directly dropped after the line executes, thus
|
|
//! immediately executing its `Drop` impl.
|
|
//! - A `_p1_a` e.g. means we don't use that Participant yet, but we must
|
|
//! not `drop` it yet as we might want to use the Streams.
|
|
//! - You sometimes see sleep(1000ms) this is used when we rely on the
|
|
//! underlying TCP functionality, as this simulates client and server
|
|
|
|
use std::sync::Arc;
|
|
use tokio::runtime::Runtime;
|
|
use veloren_network::{Network, ParticipantError, Pid, Promises, StreamError};
|
|
mod helper;
|
|
use helper::{network_participant_stream, tcp};
|
|
|
|
#[test]
|
|
fn close_network() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _, _p1_a, mut s1_a, _, _p1_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
|
|
assert_eq!(s1_a.send("Hello World"), Err(StreamError::StreamClosed));
|
|
let msg1: Result<String, _> = r.block_on(s1_b.recv());
|
|
assert_eq!(msg1, Err(StreamError::StreamClosed));
|
|
}
|
|
|
|
#[test]
|
|
fn close_participant() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p1_a, mut s1_a, _n_b, p1_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
r.block_on(p1_a.disconnect()).unwrap();
|
|
r.block_on(p1_b.disconnect()).unwrap();
|
|
|
|
assert_eq!(s1_a.send("Hello World"), Err(StreamError::StreamClosed));
|
|
assert_eq!(
|
|
r.block_on(s1_b.recv::<String>()),
|
|
Err(StreamError::StreamClosed)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn close_stream() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _, mut s1_a, _n_b, _, _) = network_participant_stream(tcp());
|
|
|
|
// s1_b is dropped directly while s1_a isn't
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
|
|
assert_eq!(s1_a.send("Hello World"), Err(StreamError::StreamClosed));
|
|
assert_eq!(
|
|
r.block_on(s1_a.recv::<String>()),
|
|
Err(StreamError::StreamClosed)
|
|
);
|
|
}
|
|
|
|
///WE must NOT create runtimes inside a Runtime, this check needs to verify
|
|
/// that we dont panic there
|
|
#[test]
|
|
fn close_streams_in_block_on() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, s1_a, _n_b, _p_b, s1_b) = network_participant_stream(tcp());
|
|
r.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);
|
|
});
|
|
drop((_n_a, _p_a, _n_b, _p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn stream_simple_3msg_then_close() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
s1_a.send(1u8).unwrap();
|
|
s1_a.send(42).unwrap();
|
|
s1_a.send("3rdMessage").unwrap();
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok(1u8));
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok(42));
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok("3rdMessage".to_string()));
|
|
drop(s1_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
assert_eq!(s1_b.send("Hello World"), Err(StreamError::StreamClosed));
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_first_then_receive() {
|
|
// recv should still be possible even if stream got closed if they are in queue
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
s1_a.send(1u8).unwrap();
|
|
s1_a.send(42).unwrap();
|
|
s1_a.send("3rdMessage").unwrap();
|
|
drop(s1_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok(1u8));
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok(42));
|
|
assert_eq!(r.block_on(s1_b.recv()), Ok("3rdMessage".to_string()));
|
|
assert_eq!(s1_b.send("Hello World"), Err(StreamError::StreamClosed));
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_1_then_close_stream() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = 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(1000));
|
|
let exp = Ok("this message must be received, even if stream is closed already!".to_string());
|
|
assert_eq!(r.block_on(s1_b.recv()), exp);
|
|
println!("all received and done");
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_100000_then_close_stream() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(s1_a);
|
|
let exp = Ok("woop_PARTY_HARD_woop".to_string());
|
|
println!("start receiving");
|
|
r.block_on(async {
|
|
for _ in 0..100000 {
|
|
assert_eq!(s1_b.recv().await, exp);
|
|
}
|
|
});
|
|
println!("all received and done");
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_100000_then_close_stream_remote() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(s1_a);
|
|
drop(_s1_b);
|
|
//no receiving
|
|
drop((_n_a, _p_a, _n_b, _p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_100000_then_close_stream_remote2() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(_s1_b);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(s1_a);
|
|
//no receiving
|
|
drop((_n_a, _p_a, _n_b, _p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn stream_send_100000_then_close_stream_remote3() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(_s1_b);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(s1_a);
|
|
//no receiving
|
|
drop((_n_a, _p_a, _n_b, _p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn close_part_then_network() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, n_a, p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..1000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(n_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
}
|
|
|
|
#[test]
|
|
fn close_network_then_part() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, n_a, p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..1000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(n_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
}
|
|
|
|
#[test]
|
|
fn close_network_then_disconnect_part() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, n_a, p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..1000 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(n_a);
|
|
assert!(r.block_on(p_a.disconnect()).is_err());
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop((_n_b, _p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn close_runtime_then_network() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(r);
|
|
drop(_n_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(_p_b);
|
|
}
|
|
|
|
#[test]
|
|
fn close_runtime_then_part() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
drop(r);
|
|
drop(_p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
drop(_p_b);
|
|
drop(_n_a);
|
|
}
|
|
|
|
#[test]
|
|
fn close_network_from_async() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, _p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
r.block_on(async move {
|
|
drop(_n_a);
|
|
});
|
|
drop(_p_b);
|
|
}
|
|
|
|
#[test]
|
|
fn close_part_from_async() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p_a, mut s1_a, _n_b, _p_b, _s1_b) = network_participant_stream(tcp());
|
|
for _ in 0..100 {
|
|
s1_a.send("woop_PARTY_HARD_woop").unwrap();
|
|
}
|
|
r.block_on(async move {
|
|
p_a.disconnect().await.unwrap();
|
|
drop(_p_b);
|
|
});
|
|
drop(_n_a);
|
|
}
|
|
|
|
#[test]
|
|
fn opened_stream_before_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p_a, _, _n_b, p_b, _) = network_participant_stream(tcp());
|
|
let mut s2_a = r.block_on(p_a.open(4, Promises::empty(), 0)).unwrap();
|
|
s2_a.send("HelloWorld").unwrap();
|
|
let mut s2_b = r.block_on(p_b.opened()).unwrap();
|
|
drop(p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
assert_eq!(r.block_on(s2_b.recv()), Ok("HelloWorld".to_string()));
|
|
drop((_n_a, _n_b, p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn opened_stream_after_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p_a, _, _n_b, p_b, _) = network_participant_stream(tcp());
|
|
let mut s2_a = r.block_on(p_a.open(3, Promises::empty(), 0)).unwrap();
|
|
s2_a.send("HelloWorld").unwrap();
|
|
drop(p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
let mut s2_b = r.block_on(p_b.opened()).unwrap();
|
|
assert_eq!(r.block_on(s2_b.recv()), Ok("HelloWorld".to_string()));
|
|
assert_eq!(
|
|
r.block_on(p_b.opened()).unwrap_err(),
|
|
ParticipantError::ParticipantDisconnected
|
|
);
|
|
drop((_n_a, _n_b, p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn open_stream_after_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p_a, _, _n_b, p_b, _) = network_participant_stream(tcp());
|
|
let mut s2_a = r.block_on(p_a.open(4, Promises::empty(), 0)).unwrap();
|
|
s2_a.send("HelloWorld").unwrap();
|
|
drop(p_a);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
let mut s2_b = r.block_on(p_b.opened()).unwrap();
|
|
assert_eq!(r.block_on(s2_b.recv()), Ok("HelloWorld".to_string()));
|
|
assert_eq!(
|
|
r.block_on(p_b.open(5, Promises::empty(), 0)).unwrap_err(),
|
|
ParticipantError::ParticipantDisconnected
|
|
);
|
|
drop((_n_a, _n_b, p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn failed_stream_open_after_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (r, _n_a, p_a, _, _n_b, p_b, _) = network_participant_stream(tcp());
|
|
drop(p_a);
|
|
assert_eq!(
|
|
r.block_on(p_b.opened()).unwrap_err(),
|
|
ParticipantError::ParticipantDisconnected
|
|
);
|
|
drop((_n_a, _n_b, p_b)); //clean teardown
|
|
}
|
|
|
|
#[test]
|
|
fn open_participant_before_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let r = Arc::new(Runtime::new().unwrap());
|
|
let n_a = Network::new(Pid::fake(0), &r);
|
|
let n_b = Network::new(Pid::fake(1), &r);
|
|
let addr = tcp();
|
|
r.block_on(n_a.listen(addr.clone())).unwrap();
|
|
let p_b = r.block_on(n_b.connect(addr)).unwrap();
|
|
let mut s1_b = r.block_on(p_b.open(4, Promises::empty(), 0)).unwrap();
|
|
s1_b.send("HelloWorld").unwrap();
|
|
let p_a = r.block_on(n_a.connected()).unwrap();
|
|
drop(s1_b);
|
|
drop(p_b);
|
|
drop(n_b);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
let mut s1_a = r.block_on(p_a.opened()).unwrap();
|
|
assert_eq!(r.block_on(s1_a.recv()), Ok("HelloWorld".to_string()));
|
|
}
|
|
|
|
#[test]
|
|
fn open_participant_after_remote_part_is_closed() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let r = Arc::new(Runtime::new().unwrap());
|
|
let n_a = Network::new(Pid::fake(0), &r);
|
|
let n_b = Network::new(Pid::fake(1), &r);
|
|
let addr = tcp();
|
|
r.block_on(n_a.listen(addr.clone())).unwrap();
|
|
let p_b = r.block_on(n_b.connect(addr)).unwrap();
|
|
let mut s1_b = r.block_on(p_b.open(4, Promises::empty(), 0)).unwrap();
|
|
s1_b.send("HelloWorld").unwrap();
|
|
drop(s1_b);
|
|
drop(p_b);
|
|
drop(n_b);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
let p_a = r.block_on(n_a.connected()).unwrap();
|
|
let mut s1_a = r.block_on(p_a.opened()).unwrap();
|
|
assert_eq!(r.block_on(s1_a.recv()), Ok("HelloWorld".to_string()));
|
|
}
|
|
|
|
#[test]
|
|
fn close_network_scheduler_completely() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let r = Arc::new(Runtime::new().unwrap());
|
|
let n_a = Network::new(Pid::fake(0), &r);
|
|
let n_b = Network::new(Pid::fake(1), &r);
|
|
let addr = tcp();
|
|
r.block_on(n_a.listen(addr.clone())).unwrap();
|
|
let p_b = r.block_on(n_b.connect(addr)).unwrap();
|
|
let mut s1_b = r.block_on(p_b.open(4, Promises::empty(), 0)).unwrap();
|
|
s1_b.send("HelloWorld").unwrap();
|
|
|
|
let p_a = r.block_on(n_a.connected()).unwrap();
|
|
let mut s1_a = r.block_on(p_a.opened()).unwrap();
|
|
assert_eq!(r.block_on(s1_a.recv()), Ok("HelloWorld".to_string()));
|
|
drop(n_a);
|
|
drop(n_b);
|
|
std::thread::sleep(std::time::Duration::from_millis(1000));
|
|
|
|
drop(p_b);
|
|
drop(p_a);
|
|
let runtime = Arc::try_unwrap(r).expect("runtime is not alone, there still exist a reference");
|
|
runtime.shutdown_timeout(std::time::Duration::from_secs(300));
|
|
}
|
|
|
|
#[test]
|
|
fn dont_panic_on_multiply_recv_after_close() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
s1_a.send(11u32).unwrap();
|
|
drop(s1_a);
|
|
std::thread::sleep(std::time::Duration::from_secs(1));
|
|
assert_eq!(s1_b.try_recv::<u32>(), Ok(Some(11u32)));
|
|
assert_eq!(s1_b.try_recv::<String>(), Err(StreamError::StreamClosed));
|
|
// There was a "Feature" in futures::channels that they panic when you call recv
|
|
// a second time after it showed end of stream
|
|
assert_eq!(s1_b.try_recv::<String>(), Err(StreamError::StreamClosed));
|
|
}
|
|
|
|
#[test]
|
|
fn dont_panic_on_recv_send_after_close() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
s1_a.send(11u32).unwrap();
|
|
drop(s1_a);
|
|
std::thread::sleep(std::time::Duration::from_secs(1));
|
|
assert_eq!(s1_b.try_recv::<u32>(), Ok(Some(11u32)));
|
|
assert_eq!(s1_b.try_recv::<String>(), Err(StreamError::StreamClosed));
|
|
assert_eq!(s1_b.send("foobar"), Err(StreamError::StreamClosed));
|
|
}
|
|
|
|
#[test]
|
|
fn dont_panic_on_multiple_send_after_close() {
|
|
let (_, _) = helper::setup(false, 0);
|
|
let (_r, _n_a, _p_a, mut s1_a, _n_b, _p_b, mut s1_b) = network_participant_stream(tcp());
|
|
|
|
s1_a.send(11u32).unwrap();
|
|
drop(s1_a);
|
|
drop(_p_a);
|
|
std::thread::sleep(std::time::Duration::from_secs(1));
|
|
assert_eq!(s1_b.try_recv::<u32>(), Ok(Some(11u32)));
|
|
assert_eq!(s1_b.try_recv::<String>(), Err(StreamError::StreamClosed));
|
|
assert_eq!(s1_b.send("foobar"), Err(StreamError::StreamClosed));
|
|
assert_eq!(s1_b.send("foobar"), Err(StreamError::StreamClosed));
|
|
}
|