diff --git a/wrk/measure.sh b/wrk/measure.sh index b1c1694..5520f0a 100755 --- a/wrk/measure.sh +++ b/wrk/measure.sh @@ -54,6 +54,14 @@ if [ "$SUBJECT" = "rust-bythebook" ] ; then URL=http://127.0.0.1:7878 fi +if [ "$SUBJECT" = "rust-bythebook-improved" ] ; then + cd wrk/rust/bythebook-improved && cargo build --release + $TSK_SRV ./target/release/hello & + PID=$! + URL=http://127.0.0.1:7878 +fi + + if [ "$SUBJECT" = "rust-clean" ] ; then cd wrk/rust/clean && cargo build --release $TSK_SRV ./target/release/hello & diff --git a/wrk/measure_all.sh b/wrk/measure_all.sh index 0389740..3d774fa 100755 --- a/wrk/measure_all.sh +++ b/wrk/measure_all.sh @@ -9,7 +9,7 @@ fi SUBJECTS="$1" if [ -z "$SUBJECTS" ] ; then - SUBJECTS="zig-zap go python python-sanic rust-bythebook rust-clean rust-axum csharp cpp-beast" + SUBJECTS="zig-zap go python python-sanic rust-bythebook rust-bythebook-improved rust-clean rust-axum csharp cpp-beast" fi for S in $SUBJECTS; do diff --git a/wrk/rust/bythebook-improved/.gitignore b/wrk/rust/bythebook-improved/.gitignore new file mode 100644 index 0000000..6985cf1 --- /dev/null +++ b/wrk/rust/bythebook-improved/.gitignore @@ -0,0 +1,14 @@ +# Generated by Cargo +# will have compiled files and executables +debug/ +target/ + +# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries +# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html +Cargo.lock + +# These are backup files generated by rustfmt +**/*.rs.bk + +# MSVC Windows builds of rustc generate these, which store debugging information +*.pdb diff --git a/wrk/rust/bythebook-improved/Cargo.toml b/wrk/rust/bythebook-improved/Cargo.toml new file mode 100644 index 0000000..c9ba87b --- /dev/null +++ b/wrk/rust/bythebook-improved/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "hello" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +# crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] } diff --git a/wrk/rust/bythebook-improved/src/lib.rs b/wrk/rust/bythebook-improved/src/lib.rs new file mode 100644 index 0000000..6e5d88d --- /dev/null +++ b/wrk/rust/bythebook-improved/src/lib.rs @@ -0,0 +1,101 @@ +//Crossbeam should, but does not make this faster. +//use crossbeam::channel::bounded; +use std::{net::TcpStream, sync::mpsc, thread}; +type Job = (fn(TcpStream), TcpStream); + +type Sender = mpsc::Sender; +//type Sender = crossbeam::channel::Sender; + +type Receiver = mpsc::Receiver; +//type Receiver = crossbeam::channel::Receiver; + +pub struct ThreadPool { + workers: Vec, + senders: Vec, + + next_sender: usize, +} + +impl ThreadPool { + /// Create a new ThreadPool. + /// + /// The size is the number of threads in the pool. + /// + /// # Panics + /// + /// The `new` function will panic if the size is zero. + pub fn new(size: usize) -> ThreadPool { + assert!(size > 0); + + let mut workers = Vec::with_capacity(size); + let mut senders = Vec::with_capacity(size); + + for id in 0..size { + //let (sender, receiver) = bounded(2); + let (sender, receiver) = mpsc::channel(); + senders.push(sender); + workers.push(Worker::new(id, receiver)); + } + + ThreadPool { + workers, + senders, + next_sender: 0, + } + } + /// round robin over available workers to ensure we never have to buffer requests + pub fn execute(&mut self, handler: fn(TcpStream), stream: TcpStream) { + let job = (handler, stream); + self.senders[self.next_sender].send(job).unwrap(); + //self.senders[self.next_sender].try_send(job).unwrap(); + self.next_sender += 1; + if self.next_sender == self.senders.len() { + self.next_sender = 0; + } + } +} + +impl Drop for ThreadPool { + fn drop(&mut self) { + self.senders.clear(); + + for worker in &mut self.workers { + println!("Shutting down worker {}", worker.id); + if let Some(thread) = worker.thread.take() { + thread.join().unwrap(); + } + } + } +} + +struct Worker { + id: usize, + thread: Option>, +} + +impl Worker { + fn new(id: usize, receiver: Receiver) -> Worker { + let thread = thread::spawn(move || Self::work(receiver)); + + Worker { + id, + thread: Some(thread), + } + } + + fn work(receiver: Receiver) { + loop { + let message = receiver.recv(); + match message { + Ok((handler, stream)) => { + // println!("Worker got a job; executing."); + handler(stream); + } + Err(_) => { + // println!("Worker disconnected; shutting down."); + break; + } + } + } + } +} diff --git a/wrk/rust/bythebook-improved/src/main.rs b/wrk/rust/bythebook-improved/src/main.rs new file mode 100644 index 0000000..8b22471 --- /dev/null +++ b/wrk/rust/bythebook-improved/src/main.rs @@ -0,0 +1,34 @@ +use hello::ThreadPool; +use std::io::prelude::*; +use std::net::TcpListener; +use std::net::TcpStream; + +fn main() { + let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); + //Creating a massive amount of threads so we can always have one ready to go. + let mut pool = ThreadPool::new(128); + + for stream in listener.incoming() { + let stream = stream.unwrap(); + //handle_connection(stream); + pool.execute(handle_connection, stream); + } + + println!("Shutting down."); +} + +fn handle_connection(mut stream: TcpStream) { + stream.set_nodelay(true).expect("set_nodelay call failed"); + loop{ + let mut buffer = [0; 1024]; + match stream.read(&mut buffer){ + Err(_)=>return, + Ok(0)=>return, + Ok(_v)=>{}, + } + + let response_bytes = b"HTTP/1.1 200 OK\r\nContent-Length: 16\r\nConnection: keep-alive\r\n\r\nHELLO from RUST!"; + + stream.write_all(response_bytes).unwrap(); + } +} diff --git a/wrk/rust/clean/Cargo.toml b/wrk/rust/clean/Cargo.toml index c9ba87b..cf97cf6 100644 --- a/wrk/rust/clean/Cargo.toml +++ b/wrk/rust/clean/Cargo.toml @@ -3,7 +3,6 @@ name = "hello" version = "0.1.0" edition = "2021" -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -# crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] } + diff --git a/wrk/rust/clean/src/main.rs b/wrk/rust/clean/src/main.rs index d6138d9..cda767c 100644 --- a/wrk/rust/clean/src/main.rs +++ b/wrk/rust/clean/src/main.rs @@ -1,18 +1,14 @@ -use hello::ThreadPool; use std::io::prelude::*; use std::net::TcpListener; use std::net::TcpStream; fn main() { - let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); - //Creating a massive amount of threads so we can always have one ready to go. - let mut pool = ThreadPool::new(128); + let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); - // for stream in listener.incoming().take(2) { for stream in listener.incoming() { let stream = stream.unwrap(); //handle_connection(stream); - pool.execute(handle_connection, stream); + std::thread::spawn(||{handle_connection(stream)}); } println!("Shutting down."); @@ -20,22 +16,17 @@ fn main() { fn handle_connection(mut stream: TcpStream) { stream.set_nodelay(true).expect("set_nodelay call failed"); - let mut buffer = [0; 1024]; - let nbytes = stream.read(&mut buffer).unwrap(); - if nbytes == 0 { - return; + loop{ + let mut buffer = [0; 1024]; + match stream.read(&mut buffer){ + Err(_)=>return, + Ok(0)=>return, + Ok(_v)=>{}, + } + + let response_bytes = b"HTTP/1.1 200 OK\r\nContent-Length: 16\r\nConnection: keep-alive\r\n\r\nHELLO from RUST!"; + + stream.write_all(response_bytes).unwrap(); } - let status_line = "HTTP/1.1 200 OK"; - - let contents = "HELLO from RUST!"; - - let response = format!( - "{}\r\nContent-Length: {}\r\n\r\n{}", - status_line, - contents.len(), - contents - ); - - stream.write_all(response.as_bytes()).unwrap(); } diff --git a/wrk/samples/req_per_sec_graph.png b/wrk/samples/req_per_sec_graph.png index c0b031e..235a0ed 100644 Binary files a/wrk/samples/req_per_sec_graph.png and b/wrk/samples/req_per_sec_graph.png differ diff --git a/wrk/samples/xfer_per_sec_graph.png b/wrk/samples/xfer_per_sec_graph.png index c51402e..73fe3cd 100644 Binary files a/wrk/samples/xfer_per_sec_graph.png and b/wrk/samples/xfer_per_sec_graph.png differ