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: fix connection buffer issue, optimize

Browse source
This commit is contained in:
Joakim Frostegård 2020-07-20 15:20:33 +02:00
parent ff6eddfc85
commit 75ebe3208d
1 changed files with 26 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

46
aquatic_http_load_test/src/network.rs
View file

@ -13,9 +13,9 @@ use crate::utils::create_random_request;
pub struct Connection { pub struct Connection {
stream: TcpStream, stream: TcpStream,
read_buffer: [u8; 2048], read_buffer: [u8; 4096],
bytes_read: usize, bytes_read: usize,
can_send: bool, can_send_initial: bool,
} }
@ -34,9 +34,9 @@ impl Connection {
let connection = Connection { let connection = Connection {
stream, stream,
read_buffer: [0; 2048], read_buffer: [0; 4096],
bytes_read: 0, bytes_read: 0,
can_send: true, can_send_initial: true,
}; };
connections.insert(*token_counter, connection); connections.insert(*token_counter, connection);
@ -51,27 +51,27 @@ impl Connection {
config: &Config, config: &Config,
state: &LoadTestState, state: &LoadTestState,
rng: &mut impl Rng, rng: &mut impl Rng,
) -> bool { // true = response received ){
loop { loop {
match self.stream.read(&mut self.read_buffer){ match self.stream.read(&mut self.read_buffer[self.bytes_read..]){
Ok(bytes_read) => { Ok(bytes_read) => {
self.bytes_read = bytes_read; self.bytes_read = bytes_read;
break; break;
}, },
Err(err) if err.kind() == ErrorKind::WouldBlock => { Err(err) if err.kind() == ErrorKind::WouldBlock => {
self.can_send = false; self.can_send_initial = false;
eprintln!("handle_read_event error would block: {}", err); eprintln!("handle_read_event error would block: {}", err);
return false; return;
}, },
Err(err) => { Err(err) => {
self.bytes_read = 0; self.bytes_read = 0;
eprintln!("handle_read_event error: {}", err); eprintln!("handle_read_event error: {}", err);
return false; return;
} }
} }
}; };
@ -98,7 +98,7 @@ impl Connection {
Err(err) => { Err(err) => {
eprintln!("response from bytes error: {}", err); eprintln!("response from bytes error: {}", err);
return false; return;
} }
} }
@ -107,8 +107,6 @@ impl Connection {
state, state,
rng rng
); );
true
} }
pub fn send_request( pub fn send_request(
@ -132,7 +130,7 @@ impl Connection {
} }
} }
self.can_send = false; self.can_send_initial = false;
} }
fn send_request_inner(&mut self, request: &[u8]) -> ::std::io::Result<()> { fn send_request_inner(&mut self, request: &[u8]) -> ::std::io::Result<()> {
@ -171,9 +169,9 @@ pub fn run_socket_thread(
).unwrap(); ).unwrap();
} }
loop { let mut initial_sent = false;
let mut responses_received = 0usize;
loop {
poll.poll(&mut events, Some(timeout)) poll.poll(&mut events, Some(timeout))
.expect("failed polling"); .expect("failed polling");
@ -182,22 +180,28 @@ 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){
if connection.read_response_and_send_request(config, &state, &mut rng){ connection.read_response_and_send_request(
responses_received += 1; config,
} &state,
&mut rng
);
} else { } else {
eprintln!("connection not found: {:?}", token); eprintln!("connection not found: {:?}", token);
} }
} }
} }
for (_, connection) in connections.iter_mut(){ if !initial_sent {
if connection.can_send { for connection in connections.values_mut(){
if connection.can_send_initial {
connection.send_request(config, &state, &mut rng); connection.send_request(config, &state, &mut rng);
initial_sent = true;
}
} }
} }
if token_counter < 1 && responses_received > 0 { if token_counter < 1 { // Only create one connection at the moment
Connection::create_and_register( Connection::create_and_register(
config, config,
&mut connections, &mut connections,