Merge pull request #67 from w-henderson/connection-timeout

Implemented a connection timeout (fixes #66)

docs/src/server/configuration.md

@@ -9,6 +9,7 @@ server {
   address   "0.0.0.0"        # Address to host the server on
   port      443              # Port to host the server on
   threads   32               # Number of threads to use for the server
+  timeout   5                # Timeout for requests, highly recommended to avoid deadlocking the thread pool
 
   plugins { # Plugin configuration (only supported with the `plugins` feature)
     include "php.conf"       # Include PHP configuration (see next page)

humphrey-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "humphrey_server"
-version = "0.4.2"
+version = "0.4.3"
 edition = "2018"
 license = "MIT"
 homepage = "https://github.com/w-henderson/Humphrey"
@@ -11,7 +11,7 @@ keywords = ["http", "server", "http-server"]
 categories = ["web-programming::http-server", "network-programming", "command-line-utilities"]
 
 [dependencies]
-humphrey = { version = "^0.4", path = "../humphrey" }
+humphrey = { version = "^0.4.4", path = "../humphrey" }
 libloading = { version = "0.7", optional = true }
 
 [features]

humphrey-server/src/config/config.rs

@@ -12,6 +12,7 @@ use std::env::args;
 use std::fs::File;
 use std::io::Read;
 use std::net::IpAddr;
+use std::time::Duration;
 
 /// Represents the parsed and validated configuration.
 #[derive(Debug, PartialEq)]
@@ -40,6 +41,8 @@ pub struct Config {
     pub cache: CacheConfig,
     /// Blacklist configuration
     pub blacklist: BlacklistConfig,
+    /// The amount of time to wait between requests
+    pub connection_timeout: Option<Duration>,
 }
 
 /// Represents the configuration for a specific host.
@@ -179,6 +182,13 @@ impl Config {
         let threads: usize =
             hashmap.get_optional_parsed("server.threads", 32, "Invalid number of threads")?;
         let default_websocket_proxy = hashmap.get_owned("server.websocket");
+        let connection_timeout_seconds: u64 =
+            hashmap.get_optional_parsed("server.timeout", 0, "Invalid connection timeout")?;
+        let connection_timeout = if connection_timeout_seconds > 0 {
+            Some(Duration::from_secs(connection_timeout_seconds))
+        } else {
+            None
+        };
 
         // Get and validate the blacklist file and mode
         let blacklist = {
@@ -314,6 +324,7 @@ impl Config {
             logging,
             cache,
             blacklist,
+            connection_timeout,
         })
     }
 

humphrey-server/src/config/default.rs

@@ -21,6 +21,7 @@ impl Default for Config {
             logging: Default::default(),
             cache: Default::default(),
             blacklist: Default::default(),
+            connection_timeout: Default::default(),
         }
     }
 }

humphrey-server/src/server/server.rs

@@ -8,7 +8,6 @@ use humphrey::{App, SubApp};
 use crate::plugins::manager::PluginManager;
 #[cfg(feature = "plugins")]
 use crate::plugins::plugin::PluginLoadResult;
-use std::error::Error;
 #[cfg(feature = "plugins")]
 use std::process::exit;
 
@@ -18,6 +17,7 @@ use crate::logger::Logger;
 use crate::proxy::proxy_handler;
 use crate::r#static::{directory_handler, file_handler, redirect_handler};
 
+use std::error::Error;
 use std::io::{Read, Write};
 use std::net::TcpStream;
 use std::sync::{Arc, RwLock};
@@ -52,8 +52,11 @@ impl From<Config> for AppState {
 
 /// Main function for the static server.
 pub fn main(config: Config) {
+    let connection_timeout = config.connection_timeout;
+
     let mut app: App<AppState> = App::new_with_config(config.threads, AppState::from(config))
-        .with_connection_condition(verify_connection);
+        .with_connection_condition(verify_connection)
+        .with_connection_timeout(connection_timeout);
 
     let state = app.get_state();
 

humphrey-server/src/tests/config.rs

@@ -14,6 +14,7 @@ use humphrey_server::proxy::{EqMutex, LoadBalancer};
 use humphrey_server::rand::Lcg;
 
 use std::collections::HashMap;
+use std::time::Duration;
 
 #[test]
 fn test_parse_config() {
@@ -80,6 +81,7 @@ fn test_parse_config() {
             list: Vec::new(),
             mode: BlacklistMode::Block,
         },
+        connection_timeout: Some(Duration::from_secs(5)),
     };
 
     assert_eq!(conf, expected_conf);
@@ -144,6 +146,7 @@ fn test_host_config() {
             list: Vec::new(),
             mode: BlacklistMode::Block,
         },
+        connection_timeout: None,
     };
 
     assert_eq!(conf, expected_conf);
@@ -196,6 +199,7 @@ fn comma_separated_routes() {
             list: Vec::new(),
             mode: BlacklistMode::Block,
         },
+        connection_timeout: None,
     };
 
     assert_eq!(conf, expected_conf);

humphrey-server/src/tests/include.rs

@@ -52,6 +52,7 @@ fn include_route() {
             list: Vec::new(),
             mode: BlacklistMode::Block,
         },
+        connection_timeout: None,
     });
 
     assert_eq!(config, expected_conf);
@@ -104,6 +105,7 @@ fn nested_include() {
             list: Vec::new(),
             mode: BlacklistMode::Block,
         },
+        connection_timeout: None,
     });
 
     assert_eq!(config, expected_conf);

humphrey-server/src/tests/testcases/valid.conf

@@ -6,6 +6,7 @@ server {
     port       80
     threads    32
     websocket  "localhost:1234"
+    timeout    5
 
     plugins { # this is a comment on a section header
         php {

humphrey-server/src/tests/tree.rs

@@ -14,6 +14,7 @@ fn test_build_tree() {
         ConfigNode::Number("port".into(), "80".into()),
         ConfigNode::Number("threads".into(), "32".into()),
         ConfigNode::String("websocket".into(), "localhost:1234".into()),
+        ConfigNode::Number("timeout".into(), "5".into()),
         ConfigNode::Section("plugins".into(), vec![
             ConfigNode::Section("php".into(), vec![
                 ConfigNode::String("library".into(), "plugins/php/target/release/php.dll".into()),
@@ -58,6 +59,7 @@ fn test_flatten_config() {
     expected_hashmap.insert("server.port".into(), ConfigNode::Number("port".into(), "80".into()));
     expected_hashmap.insert("server.threads".into(), ConfigNode::Number("threads".into(), "32".into()));
     expected_hashmap.insert("server.websocket".into(), ConfigNode::String("websocket".into(), "localhost:1234".into()));
+    expected_hashmap.insert("server.timeout".into(), ConfigNode::Number("timeout".into(), "5".into()));
     expected_hashmap.insert("server.blacklist.mode".into(), ConfigNode::String("mode".into(), "block".into()));
     expected_hashmap.insert("server.log.level".into(), ConfigNode::String("level".into(), "info".into()));
     expected_hashmap.insert("server.log.console".into(), ConfigNode::Boolean("console".into(), "true".into()));

humphrey/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "humphrey"
-version = "0.4.3"
+version = "0.4.4"
 edition = "2018"
 license = "MIT"
 homepage = "https://github.com/w-henderson/Humphrey"

humphrey/src/app.rs

@@ -13,6 +13,7 @@ use crate::thread::pool::ThreadPool;
 
 use std::net::{TcpListener, TcpStream, ToSocketAddrs};
 use std::sync::Arc;
+use std::time::Duration;
 
 #[cfg(feature = "tls")]
 use rustls::ServerConfig;
@@ -33,6 +34,7 @@ where
     state: Arc<State>,
     connection_handler: ConnectionHandler<State>,
     connection_condition: ConnectionCondition<State>,
+    connection_timeout: Option<Duration>,
     #[cfg(feature = "tls")]
     tls_config: Option<Arc<ServerConfig>>,
     #[cfg(feature = "tls")]
@@ -41,8 +43,14 @@ where
 
 /// Represents a function able to handle a connection.
 /// In most cases, the default connection handler should be used.
-pub type ConnectionHandler<State> =
-    fn(Stream, Arc<Vec<SubApp<State>>>, Arc<SubApp<State>>, Arc<ErrorHandler>, Arc<State>);
+pub type ConnectionHandler<State> = fn(
+    Stream,
+    Arc<Vec<SubApp<State>>>,
+    Arc<SubApp<State>>,
+    Arc<ErrorHandler>,
+    Arc<State>,
+    Option<Duration>,
+);
 
 /// Represents a function able to calculate whether a connection will be accepted.
 pub type ConnectionCondition<State> = fn(&mut TcpStream, Arc<State>) -> bool;
@@ -141,6 +149,7 @@ where
             state: Arc::new(State::default()),
             connection_handler: client_handler,
             connection_condition: |_, _| true,
+            connection_timeout: None,
             #[cfg(feature = "tls")]
             tls_config: None,
             #[cfg(feature = "tls")]
@@ -158,6 +167,7 @@ where
             state: Arc::new(state),
             connection_handler: client_handler,
             connection_condition: |_, _| true,
+            connection_timeout: None,
             #[cfg(feature = "tls")]
             tls_config: None,
             #[cfg(feature = "tls")]
@@ -186,6 +196,7 @@ where
                 let cloned_default_subapp = default_subapp.clone();
                 let cloned_error_handler = error_handler.clone();
                 let cloned_handler = self.connection_handler;
+                let cloned_timeout = self.connection_timeout;
 
                 // Spawn a new thread to handle the connection
                 self.thread_pool.execute(move || {
@@ -195,6 +206,7 @@ where
                         cloned_default_subapp,
                         cloned_error_handler,
                         cloned_state,
+                        cloned_timeout,
                     )
                 });
             }
@@ -232,6 +244,7 @@ where
                 let cloned_default_subapp = default_subapp.clone();
                 let cloned_error_handler = error_handler.clone();
                 let cloned_handler = self.connection_handler;
+                let cloned_timeout = self.connection_timeout;
                 let cloned_config = self
                     .tls_config
                     .as_ref()
@@ -250,6 +263,7 @@ where
                         cloned_default_subapp,
                         cloned_error_handler,
                         cloned_state,
+                        cloned_timeout,
                     )
                 });
             }
@@ -341,6 +355,12 @@ where
         self
     }
 
+    /// Sets the connection timeout, the amount of time to wait between keep-alive requests.
+    pub fn with_connection_timeout(mut self, timeout: Option<Duration>) -> Self {
+        self.connection_timeout = timeout;
+        self
+    }
+
     /// Sets whether HTTPS should be forced on all connections. Defaults to false.
     ///
     /// If this is set to true, a background thread will be spawned when `run_tls` is called to send
@@ -427,6 +447,7 @@ fn client_handler<State>(
     default_subapp: Arc<SubApp<State>>,
     error_handler: Arc<ErrorHandler>,
     state: Arc<State>,
+    timeout: Option<Duration>,
 ) {
     use std::collections::btree_map::Entry;
     use std::io::Write;
@@ -443,7 +464,11 @@ fn client_handler<State>(
 
     loop {
         // Parses the request from the stream
-        let request = Request::from_stream(&mut stream, addr);
+        let request = match timeout {
+            Some(timeout) => Request::from_stream_with_timeout(&mut stream, addr, timeout),
+            None => Request::from_stream(&mut stream, addr),
+        };
+
         let cloned_state = state.clone();
 
         // If the request is valid an is a WebSocket request, call the corresponding handler
@@ -454,14 +479,6 @@ fn client_handler<State>(
             }
         }
 
-        // If the request could not be parsed due to a stream error, close the thread
-        if match &request {
-            Ok(_) => false,
-            Err(e) => e == &RequestError::Stream,
-        } {
-            break;
-        }
-
         // Get the keep alive information from the request before it is consumed by the handler
         let keep_alive = if let Ok(request) = &request {
             if let Some(connection) = request.headers.get(&RequestHeader::Connection) {
@@ -516,7 +533,11 @@ fn client_handler<State>(
 
                 response
             }
-            Err(_) => error_handler(StatusCode::BadRequest),
+            Err(e) => match e {
+                RequestError::Request => error_handler(StatusCode::BadRequest),
+                RequestError::Timeout => error_handler(StatusCode::RequestTimeout),
+                RequestError::Stream => return,
+            },
         };
 
         // Write the response to the stream