YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-04-02 10:45:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

aquatic_http_load_test: refactor timer code; panic on an error case

Browse source
This commit is contained in:
Joakim Frostegård 2021-10-28 01:21:47 +02:00
parent e458cc54db
commit 80447d9d1b
1 changed files with 25 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

30
aquatic_http_load_test/src/network.rs
View file

@ -2,7 +2,7 @@ use std::{cell::RefCell, convert::TryInto, io::{Cursor, ErrorKind, Read}, rc::Rc
use aquatic_http_protocol::response::Response; use aquatic_http_protocol::response::Response;
use futures_lite::{AsyncReadExt, AsyncWriteExt}; use futures_lite::{AsyncReadExt, AsyncWriteExt};
use glommio::{enclose, prelude::*, timer::TimerActionRepeat}; use glommio::{prelude::*, timer::TimerActionRepeat};
use glommio::net::TcpStream; use glommio::net::TcpStream;
use rand::{SeedableRng, prelude::SmallRng}; use rand::{SeedableRng, prelude::SmallRng};
use rustls::ClientConnection; use rustls::ClientConnection;
@ -17,8 +17,24 @@ pub async fn run_socket_thread(
let config = Rc::new(config); let config = Rc::new(config);
let num_active_connections = Rc::new(RefCell::new(0usize)); let num_active_connections = Rc::new(RefCell::new(0usize));
TimerActionRepeat::repeat(enclose!((config, tls_config, load_test_state, num_active_connections) move || { TimerActionRepeat::repeat(move || periodically_open_connections(
enclose!((config, tls_config, load_test_state, num_active_connections) move || async move { config.clone(),
tls_config.clone(),
load_test_state.clone(),
num_active_connections.clone())
);
futures_lite::future::pending::<bool>().await;
Ok(())
}
async fn periodically_open_connections(
config: Rc<Config>,
tls_config: Arc<rustls::ClientConfig>,
load_test_state: LoadTestState,
num_active_connections: Rc<RefCell<usize>>,
) -> Option<Duration> {
if *num_active_connections.borrow() < config.num_connections { if *num_active_connections.borrow() < config.num_connections {
spawn_local(async move { spawn_local(async move {
if let Err(err) = Connection::run(config, tls_config, load_test_state, num_active_connections).await { if let Err(err) = Connection::run(config, tls_config, load_test_state, num_active_connections).await {
@ -28,12 +44,6 @@ pub async fn run_socket_thread(
} }
Some(Duration::from_secs(1)) Some(Duration::from_secs(1))
})()
}));
futures_lite::future::pending::<bool>().await;
Ok(())
} }
struct Connection { struct Connection {
@ -148,7 +158,7 @@ impl Connection {
break; break;
} }
Err(err) => { Err(err) => {
break; panic!("tls.reader().read: {}", err);
} }
} }
} }