diff --git a/.idea/misc.xml b/.idea/misc.xml index c6c4293..f1a0458 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -43,6 +43,7 @@ + diff --git a/.idea/rust.iml b/.idea/rust.iml index e27f61d..2ed216a 100644 --- a/.idea/rust.iml +++ b/.idea/rust.iml @@ -173,6 +173,10 @@ + + + + @@ -191,6 +195,7 @@ + diff --git a/.idea/workspace.xml b/.idea/workspace.xml index 7307ac7..7ce4a57 100644 --- a/.idea/workspace.xml +++ b/.idea/workspace.xml @@ -14,51 +14,24 @@ - - - - - - - - - - - - - - - - - - - - - - - + - + - + - - - - - - - - - - - + + + + + + @@ -92,12 +65,6 @@ @@ -199,19 +172,12 @@ - + - - - - - - - - + @@ -227,7 +193,7 @@ - + @@ -243,6 +209,12 @@ + + + + + + - + - - + + + - @@ -377,7 +349,7 @@ - + @@ -394,7 +366,7 @@ - + @@ -408,59 +380,6 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - @@ -824,21 +743,21 @@ - + - + - + @@ -854,9 +773,48 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + diff --git a/hello/404.html b/hello/404.html new file mode 100644 index 0000000..0d6b5b0 --- /dev/null +++ b/hello/404.html @@ -0,0 +1,11 @@ + + + + + Hello! + + +

Oops!

+

Sorry, I don't know what you're asking for.

+ + diff --git a/hello/Cargo.toml b/hello/Cargo.toml new file mode 100644 index 0000000..cae316d --- /dev/null +++ b/hello/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "hello" +version = "0.1.0" +authors = ["Timothy Warren "] +edition = "2018" + +[dependencies] diff --git a/hello/hello.html b/hello/hello.html new file mode 100644 index 0000000..4c6d8d3 --- /dev/null +++ b/hello/hello.html @@ -0,0 +1,11 @@ + + + + + Hello! + + +

Hello!

+

Hi from Rust

+ + diff --git a/hello/src/bin/main.rs b/hello/src/bin/main.rs new file mode 100644 index 0000000..33b79f0 --- /dev/null +++ b/hello/src/bin/main.rs @@ -0,0 +1,48 @@ +use std::fs; +use std::io::prelude::*; +use std::net::TcpStream; +use std::net::TcpListener; +use std::thread; +use std::time::Duration; + +use hello::ThreadPool; + +fn main() { + let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); + let pool = ThreadPool::new(4); + + for stream in listener.incoming().take(2) { + let stream = stream.unwrap(); + + pool.execute(|| { + handle_connection(stream); + }); + } + + println!("Shutting down."); +} + +fn handle_connection(mut stream: TcpStream) { + let mut buffer = [0; 512]; + + stream.read(&mut buffer).unwrap(); + + let get = b"GET / HTTP/1.1\r\n"; + let sleep = b"GET /sleep HTTP/1.1\r\n"; + + let (status_line, filename) = if buffer.starts_with(get) { + ("HTTP/1.1 200 OK\r\n\r\n", "hello.html") + } else if buffer.starts_with(sleep) { + thread::sleep(Duration::from_secs(5)); + ("HTTP/1.1 200 OK\r\n\r\n", "hello.html") + } else { + ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html") + }; + + let contents = fs::read_to_string(filename).unwrap(); + + let response = format!("{}{}", status_line, contents); + + stream.write(response.as_bytes()).unwrap(); + stream.flush().unwrap(); +} diff --git a/hello/src/lib.rs b/hello/src/lib.rs new file mode 100644 index 0000000..a03ae88 --- /dev/null +++ b/hello/src/lib.rs @@ -0,0 +1,115 @@ +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::mpsc; +use std::thread; + +enum Message { + NewJob(Job), + Terminate, +} + +trait FnBox { + fn call_box(self: Box); +} + +impl FnBox for F { + fn call_box(self: Box) { + (*self)() + } +} + +pub struct ThreadPool { + workers: Vec, + sender: mpsc::Sender, +} + +type Job = Box; + +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 (sender, receiver) = mpsc::channel(); + + let receiver = Arc::new(Mutex::new(receiver)); + + let mut workers = Vec::with_capacity(size); + + for id in 0..size { + workers.push(Worker::new(id, Arc::clone(&receiver))); + } + + ThreadPool { + workers, + sender, + } + } + + pub fn execute(&self, f: F) + where + F: FnOnce() + Send + 'static + { + let job = Box::new(f); + + self.sender.send(Message::NewJob(job)).unwrap(); + } +} + +impl Drop for ThreadPool { + fn drop(&mut self) { + println!("Sending terminate message to all workers."); + + for _ in &mut self.workers { + self.sender.send(Message::Terminate).unwrap(); + } + + println!("Shutting down all workers"); + + 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: Arc>>) -> Worker { + let thread = thread::spawn(move|| { + loop { + let message = receiver.lock().unwrap().recv().unwrap(); + + match message { + Message::NewJob(job) => { + println!("Worker {} got a job; executing.", id); + job.call_box(); + }, + Message::Terminate => { + println!("Worker {} was told to terminate", id); + + break; + }, + } + } + }); + + Worker { + id, + thread: Some(thread), + } + } +}