Merge pull request #9 from jtgeibel/remove-over-capacity-rejection

Remove rejection of requests when over capacity

Author: jtgeibel

examples/server.rs

@@ -7,16 +7,14 @@ use conduit_router::RouteBuilder;
 use std::io;
 use std::thread::sleep;
 
-const MAX_THREADS: usize = 1;
-
 #[tokio::main]
 async fn main() {
     env_logger::init();
 
     let app = build_conduit_handler();
     let addr = ([127, 0, 0, 1], 12345).into();
 
-    Server::serve(&addr, app, MAX_THREADS).await;
+    Server::serve(&addr, app).await;
 }
 
 fn build_conduit_handler() -> impl Handler {

src/lib.rs

@@ -17,13 +17,11 @@
 //! use conduit_hyper::Server;
 //! use tokio::runtime::Runtime;
 //!
-//! const MAX_THREADS: usize = 10;
-//!
 //! #[tokio::main]
 //! async fn main() {
 //!     let app = build_conduit_handler();
 //!     let addr = ([127, 0, 0, 1], 12345).into();
-//!     let server = Server::serve(&addr, app, MAX_THREADS);
+//!     let server = Server::serve(&addr, app);
 //!
 //!     server.await;
 //! }

src/server.rs

@@ -19,12 +19,8 @@ impl Server {
     /// `tokio::Runtime` it is not possible to furter configure the `hyper::Server`.  If more
     /// control, such as configuring a graceful shutdown is necessary, then call
     /// `Service::from_blocking` instead.
-    pub fn serve<H: conduit::Handler>(
-        addr: &SocketAddr,
-        handler: H,
-        max_threads: usize,
-    ) -> impl Future {
-        let handler = Arc::new(BlockingHandler::new(handler, max_threads));
+    pub fn serve<H: conduit::Handler>(addr: &SocketAddr, handler: H) -> impl Future {
+        let handler = Arc::new(BlockingHandler::new(handler));
         let make_service = make_service_fn(move |socket: &AddrStream| {
             let handler = handler.clone();
             let remote_addr = socket.remote_addr();

src/service.rs

@@ -32,8 +32,7 @@ impl Service {
     /// #     }
     /// # }
     /// # let app = Endpoint();
-    /// const MAX_THREADS: usize = 10;
-    /// let handler = Arc::new(BlockingHandler::new(app, MAX_THREADS));
+    /// let handler = Arc::new(BlockingHandler::new(app));
     /// let make_service =
     ///     hyper::service::make_service_fn(move |socket: &hyper::server::conn::AddrStream| {
     ///         let addr = socket.remote_addr();

src/service/blocking_handler.rs

@@ -4,27 +4,20 @@ use crate::service::ServiceError;
 use crate::{ConduitResponse, HyperResponse};
 
 use std::net::SocketAddr;
-use std::sync::{
-    atomic::{AtomicUsize, Ordering},
-    Arc,
-};
+use std::sync::Arc;
 
-use conduit::{header, Handler, StatusCode};
+use conduit::{Handler, StatusCode};
 use hyper::{Body, Request, Response};
 use tracing::error;
 
 #[derive(Debug)]
 pub struct BlockingHandler<H: Handler> {
-    thread_count: AtomicUsize,
-    max_thread_count: usize,
     handler: Arc<H>,
 }
 
 impl<H: Handler> BlockingHandler<H> {
-    pub fn new(handler: H, max_thread_count: usize) -> Self {
+    pub fn new(handler: H) -> Self {
         Self {
-            thread_count: AtomicUsize::new(0),
-            max_thread_count,
             handler: Arc::new(handler),
         }
     }
@@ -40,14 +33,6 @@ impl<H: Handler> BlockingHandler<H> {
         let full_body = hyper::body::to_bytes(body).await?;
         let mut request_info = RequestInfo::new(parts, full_body);
 
-        // The _drop_on_return ensures the counter is decreased for all exit paths
-        let (_drop_on_return, prev_count) = ThreadCounter::begin_with(&self.thread_count);
-
-        // Comparison is against the "previous value" from an atomic fetch_add, so using `>=`
-        if prev_count >= self.max_thread_count {
-            return Ok(over_capacity_error_response());
-        }
-
         let handler = self.handler.clone();
         tokio::task::spawn_blocking(move || {
             let mut request = ConduitRequest::new(&mut request_info, remote_addr);
@@ -83,36 +68,3 @@ fn server_error_response(message: &str) -> HyperResponse {
         .body(body)
         .expect("Unexpected invalid header")
 }
-
-/// Logs an error message and returns a 503 status saying the service is over capacity
-fn over_capacity_error_response() -> HyperResponse {
-    const RETRY_AFTER: u32 = 2;
-    error!("Server Capacity Exceeded");
-    let body = hyper::Body::from(format!(
-        "Service Unavailable: Please retry after {} seconds.",
-        RETRY_AFTER
-    ));
-    Response::builder()
-        .status(StatusCode::SERVICE_UNAVAILABLE)
-        .header(header::RETRY_AFTER, RETRY_AFTER)
-        .body(body)
-        .expect("Unexpected invalid header")
-}
-
-/// A struct that stores a reference to an atomic counter so it can be decremented when dropped
-struct ThreadCounter<'a> {
-    counter: &'a AtomicUsize,
-}
-
-impl<'a> ThreadCounter<'a> {
-    fn begin_with(counter: &'a AtomicUsize) -> (Self, usize) {
-        let previous = counter.fetch_add(1, Ordering::SeqCst);
-        (Self { counter }, previous)
-    }
-}
-
-impl<'a> Drop for ThreadCounter<'a> {
-    fn drop(&mut self) {
-        self.counter.fetch_sub(1, Ordering::SeqCst);
-    }
-}

src/tests.rs

@@ -1,5 +1,5 @@
 use conduit::{box_error, Body, Handler, HandlerResult, RequestExt, Response, StatusCode};
-use futures_util::future::{Future, FutureExt};
+use futures_util::future::Future;
 use hyper::{body::to_bytes, service::Service};
 
 use super::service::{BlockingHandler, ServiceError};
@@ -71,7 +71,6 @@ impl Handler for Sleep {
 
 fn make_service<H: Handler>(
     handler: H,
-    max_thread_count: usize,
 ) -> impl Service<
     hyper::Request<hyper::Body>,
     Response = HyperResponse,
@@ -80,7 +79,7 @@ fn make_service<H: Handler>(
 > {
     use hyper::service::service_fn;
 
-    let handler = std::sync::Arc::new(BlockingHandler::new(handler, max_thread_count));
+    let handler = std::sync::Arc::new(BlockingHandler::new(handler));
 
     service_fn(move |request: hyper::Request<hyper::Body>| {
         let remote_addr = ([0, 0, 0, 0], 0).into();
@@ -89,7 +88,7 @@ fn make_service<H: Handler>(
 }
 
 async fn simulate_request<H: Handler>(handler: H) -> HyperResponse {
-    let mut service = make_service(handler, 1);
+    let mut service = make_service(handler);
     service.call(hyper::Request::default()).await.unwrap()
 }
 
@@ -128,7 +127,7 @@ async fn recover_from_panic() {
 
 #[tokio::test]
 async fn normalize_path() {
-    let mut service = make_service(AssertPathNormalized, 1);
+    let mut service = make_service(AssertPathNormalized);
     let req = hyper::Request::put("//removed/.././.././normalized")
         .body(hyper::Body::default())
         .unwrap();
@@ -144,26 +143,9 @@ async fn normalize_path() {
     assert_eq!(resp.headers().len(), 1);
 }
 
-#[tokio::test]
-async fn limits_thread_count() {
-    let mut service = make_service(Sleep, 1);
-    let first = service.call(hyper::Request::default());
-    let second = service.call(hyper::Request::default());
-
-    let first_completed = futures_util::select! {
-        // The first thead is spawned and sleeps for 100ms
-        sleep = first.fuse() => sleep,
-        // The second request is rejected immediately
-        over_capacity = second.fuse() => over_capacity,
-    }
-    .unwrap();
-
-    assert_eq!(first_completed.status(), StatusCode::SERVICE_UNAVAILABLE)
-}
-
 #[tokio::test]
 async fn sleeping_doesnt_block_another_request() {
-    let mut service = make_service(Sleep, 2);
+    let mut service = make_service(Sleep);
 
     let first = service.call(hyper::Request::default());
     let second = service.call(hyper::Request::default());