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

aquatic http load test: only send request when appropriate, other fixes

Browse source
This commit is contained in:
Joakim Frostegård 2020-08-02 10:24:27 +02:00
parent 49414e9006
commit 93a7ad0344
2 changed files with 24 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

2
aquatic_http_load_test/src/config.rs
View file

@ -48,7 +48,7 @@ impl Default for Config {
Self { Self {
server_address: "127.0.0.1:3000".parse().unwrap(), server_address: "127.0.0.1:3000".parse().unwrap(),
num_workers: 1, num_workers: 1,
num_connections: 128, num_connections: 8,
duration: 0, duration: 0,
network: NetworkConfig::default(), network: NetworkConfig::default(),
torrents: TorrentConfig::default(), torrents: TorrentConfig::default(),

36
aquatic_http_load_test/src/network.rs
View file

@ -15,6 +15,7 @@ pub struct Connection {
stream: TcpStream, stream: TcpStream,
read_buffer: [u8; 4096], read_buffer: [u8; 4096],
bytes_read: usize, bytes_read: usize,
can_send: bool,
} }
@ -35,6 +36,7 @@ impl Connection {
stream, stream,
read_buffer: [0; 4096], read_buffer: [0; 4096],
bytes_read: 0, bytes_read: 0,
can_send: true,
}; };
connections.insert(*token_counter, connection); connections.insert(*token_counter, connection);
@ -73,8 +75,6 @@ impl Connection {
state.statistics.bytes_received state.statistics.bytes_received
.fetch_add(self.bytes_read, Ordering::SeqCst); .fetch_add(self.bytes_read, Ordering::SeqCst);
self.bytes_read = 0;
match response { match response {
Response::Announce(_) => { Response::Announce(_) => {
state.statistics.responses_announce state.statistics.responses_announce
@ -90,6 +90,9 @@ impl Connection {
}, },
} }
self.bytes_read = 0;
self.can_send = true;
break false; break false;
}, },
Err(err) => { Err(err) => {
@ -124,6 +127,10 @@ impl Connection {
rng: &mut impl Rng, rng: &mut impl Rng,
request_buffer: &mut Cursor<&mut [u8]>, request_buffer: &mut Cursor<&mut [u8]>,
) -> bool { // bool = remove connection ) -> bool { // bool = remove connection
if !self.can_send {
return false;
}
let request = create_random_request( let request = create_random_request(
&config, &config,
&state, &state,
@ -138,6 +145,8 @@ impl Connection {
Ok(_) => { Ok(_) => {
state.statistics.requests.fetch_add(1, Ordering::SeqCst); state.statistics.requests.fetch_add(1, Ordering::SeqCst);
self.can_send = false;
false false
}, },
Err(err) => { Err(err) => {
@ -195,10 +204,11 @@ pub fn run_socket_thread(
).unwrap(); ).unwrap();
} }
let mut initial_sent = false;
let mut iter_counter = 0usize; let mut iter_counter = 0usize;
let mut num_to_create = 0usize; let mut num_to_create = 0usize;
let mut drop_connections = Vec::with_capacity(config.num_connections);
loop { loop {
poll.poll(&mut events, Some(timeout)) poll.poll(&mut events, Some(timeout))
.expect("failed polling"); .expect("failed polling");
@ -208,9 +218,9 @@ pub fn run_socket_thread(
let token = event.token(); let token = event.token();
if let Some(connection) = connections.get_mut(&token.0){ if let Some(connection) = connections.get_mut(&token.0){
let remove = connection.read_response(&state); let remove_connection = connection.read_response(&state);
if remove { if remove_connection {
connections.remove(&token.0); connections.remove(&token.0);
num_to_create += 1; num_to_create += 1;
} }
@ -221,8 +231,6 @@ pub fn run_socket_thread(
} }
} }
let mut drop_keys = Vec::new();
for (k, connection) in connections.iter_mut(){ for (k, connection) in connections.iter_mut(){
let remove_connection = connection.send_request( let remove_connection = connection.send_request(
config, config,
@ -232,17 +240,21 @@ pub fn run_socket_thread(
); );
if remove_connection { if remove_connection {
drop_keys.push(*k); drop_connections.push(*k);
} }
} }
for k in drop_keys { for k in drop_connections.drain(..) {
connections.remove(&k); connections.remove(&k);
num_to_create += 1; num_to_create += 1;
} }
// num_to_create += 1; let max_new = config.num_connections - connections.len();
let max_new = 8 - connections.len();
if max_new != 0 && iter_counter % create_conn_interval == 0 {
num_to_create += 1;
}
let num_new = num_to_create.min(max_new); let num_new = num_to_create.min(max_new);
for _ in 0..num_new { for _ in 0..num_new {
@ -256,8 +268,6 @@ pub fn run_socket_thread(
if !err { if !err {
num_to_create -= 1; num_to_create -= 1;
} }
initial_sent = false;
} }
iter_counter = iter_counter.wrapping_add(1); iter_counter = iter_counter.wrapping_add(1);