Remove Rc<S> where S: Service as it brakes readiness check validity (#198)

* Add SharedService, a service that can be checked for readiness by multiple tasks

ntex-service/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.0.2] - 2023-04-14
+
+* Remove Rc<S> where S: Service as it brakes readiness check validity
+
 ## [1.0.1] - 2023-01-24
 
 * Add `FnShutdown` service to provide on_shutdown callback

ntex-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-service"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex service"
 keywords = ["network", "framework", "async", "futures"]

ntex-service/src/lib.rs

@@ -279,28 +279,6 @@ where
     }
 }
 
-impl<S, Req> Service<Req> for Rc<S>
-where
-    S: Service<Req> + ?Sized,
-{
-    type Response = S::Response;
-    type Error = S::Error;
-    type Future<'f> = S::Future<'f> where S: 'f, Req: 'f;
-
-    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        (**self).poll_ready(cx)
-    }
-
-    #[inline]
-    fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
-        (**self).poll_shutdown(cx)
-    }
-
-    fn call(&self, request: Req) -> S::Future<'_> {
-        (**self).call(request)
-    }
-}
-
 impl<S, Req, Cfg> ServiceFactory<Req, Cfg> for Rc<S>
 where
     S: ServiceFactory<Req, Cfg>,

ntex-util/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.2.1] - 2023-04-14
+
+* Add SharedService, a service that can be checked for readiness by multiple tasks
+
 ## [0.2.0] - 2023-01-04
 
 * Release

ntex-util/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-util"
-version = "0.2.0"
+version = "0.2.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for ntex framework"
 keywords = ["network", "framework", "async", "futures"]

ntex-util/src/services/mod.rs

@@ -3,6 +3,7 @@ pub mod counter;
 mod extensions;
 pub mod inflight;
 pub mod keepalive;
+pub mod shared;
 pub mod timeout;
 pub mod variant;
 

ntex-util/src/services/shared.rs

@@ -0,0 +1,169 @@
+/// A service that can be checked for readiness by multiple tasks
+use std::{
+    cell::Cell, cell::RefCell, marker::PhantomData, rc::Rc, task::Context, task::Poll,
+};
+
+use ntex_service::{Middleware, Service};
+
+use crate::channel::{condition, oneshot};
+use crate::future::{poll_fn, select, Either};
+
+/// A middleware that construct sharable service
+pub struct Shared<R>(PhantomData<R>);
+
+impl<R> Shared<R> {
+    pub fn new() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<R> Default for Shared<R> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<S: Service<R>, R> Middleware<S> for Shared<R> {
+    type Service = SharedService<S, R>;
+
+    fn create(&self, service: S) -> Self::Service {
+        SharedService::new(service)
+    }
+}
+
+/// A service that can be checked for readiness by multiple tasks
+pub struct SharedService<S: Service<R>, R> {
+    inner: Rc<Inner<S, R>>,
+    readiness: condition::Waiter,
+}
+
+struct Inner<S: Service<R>, R> {
+    service: S,
+    ready: condition::Condition,
+    driver_stop: Cell<Option<oneshot::Sender<()>>>,
+    driver_running: Cell<bool>,
+    error: RefCell<Option<S::Error>>,
+}
+
+impl<S: Service<R>, R> SharedService<S, R> {
+    pub fn new(service: S) -> Self {
+        let condition = condition::Condition::default();
+        Self {
+            readiness: condition.wait(),
+            inner: Rc::new(Inner {
+                service,
+                ready: condition,
+                driver_stop: Cell::default(),
+                driver_running: Cell::default(),
+                error: RefCell::default(),
+            }),
+        }
+    }
+}
+
+impl<S: Service<R>, R> Clone for SharedService<S, R> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+            readiness: self.readiness.clone(),
+        }
+    }
+}
+
+impl<S: Service<R>, R> Drop for SharedService<S, R> {
+    fn drop(&mut self) {
+        if self.inner.driver_running.get() {
+            // the only live references to inner are in this SharedService instance and the driver task
+            if Rc::strong_count(&self.inner) == 2 {
+                if let Some(stop) = self.inner.driver_stop.take() {
+                    let _ = stop.send(());
+                }
+            }
+        }
+    }
+}
+
+impl<S, R> Service<R> for SharedService<S, R>
+where
+    S: Service<R> + 'static,
+    S::Error: Clone,
+    R: 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future<'f> = S::Future<'f> where Self: 'f, R: 'f;
+
+    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        // if there is an error, it should be returned to all tasks checking readiness
+        if let Some(error) = self.inner.error.borrow().as_ref() {
+            return Poll::Ready(Err(error.clone()));
+        }
+
+        // if the service is being driven to readiness we must register our waker and wait
+        if self.inner.driver_running.get() {
+            log::trace!("polled SharedService driver, driver is already running");
+            // register waker to be notified, regardless of any previous notification
+            let _ = self.readiness.poll_ready(cx);
+            return Poll::Pending;
+        }
+
+        // driver is not running, check the inner service is ready
+        let result = self.inner.service.poll_ready(cx);
+        log::trace!(
+            "polled SharedService, ready: {}, errored: {}",
+            result.is_ready(),
+            matches!(result, Poll::Ready(Err(_)))
+        );
+
+        match result {
+            // pass through service is ready, allow call
+            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
+            // capture error, all future readiness checks will fail
+            Poll::Ready(Err(e)) => {
+                *self.inner.error.borrow_mut() = Some(e.clone());
+                Poll::Ready(Err(e))
+            }
+            // start driver and elide all poll_ready calls until it is complete
+            Poll::Pending => {
+                let inner = self.inner.clone();
+                let (tx, rx) = oneshot::channel();
+                inner.driver_running.set(true);
+                inner.driver_stop.set(Some(tx));
+
+                ntex_rt::spawn(async move {
+                    log::trace!("SharedService driver has started");
+                    let service_ready = poll_fn(|cx| inner.service.poll_ready(cx));
+                    let clients_gone = rx;
+                    let result = select(service_ready, clients_gone).await;
+                    if let Either::Left(result) = result {
+                        log::trace!(
+                            "SharedService driver completed, errored: {}",
+                            result.is_err()
+                        );
+                        if let Err(e) = result {
+                            inner.error.borrow_mut().replace(e);
+                        }
+                        inner.driver_running.set(false);
+                        inner.driver_stop.set(None);
+                        inner.ready.notify();
+                    } else {
+                        log::trace!("SharedService driver task stopped because all clients are gone");
+                    }
+                });
+
+                // register waker to be notified, regardless of any previous notification
+                let _ = self.readiness.poll_ready(cx);
+
+                Poll::Pending
+            }
+        }
+    }
+
+    fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
+        self.inner.service.poll_shutdown(cx)
+    }
+
+    fn call(&self, req: R) -> Self::Future<'_> {
+        self.inner.service.call(req)
+    }
+}

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.6.7] - 2023-04-14
+
+* Remove Rc<Service> usage, update service and util deps
+
 ## [0.6.6] - 2023-04-11
 
 * http: Better http2 config handling

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "0.6.6"
+version = "0.6.7"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -52,11 +52,11 @@ ntex-codec = "0.6.2"
 ntex-connect = "0.2.1"
 ntex-http = "0.1.9"
 ntex-router = "0.5.1"
-ntex-service = "1.0.0"
+ntex-service = "1.0.2"
 ntex-macros = "0.1.3"
-ntex-util = "0.2.0"
+ntex-util = "0.2.1"
 ntex-bytes = "0.1.19"
-ntex-h2 = "0.2.1"
+ntex-h2 = "0.2.3"
 ntex-rt = "0.4.9"
 ntex-io = "0.2.9"
 ntex-tls = "0.2.4"

ntex/src/http/client/connector.rs

@@ -1,11 +1,13 @@
-use std::{rc::Rc, task::Context, task::Poll, time::Duration};
+use std::{task::Context, task::Poll, time::Duration};
 
 use ntex_h2::{self as h2};
 
 use crate::connect::{Connect as TcpConnect, Connector as TcpConnector};
 use crate::service::{apply_fn, boxed, Service};
 use crate::time::{Millis, Seconds};
-use crate::util::{timeout::TimeoutError, timeout::TimeoutService, Either, Ready};
+use crate::util::{
+    shared::SharedService, timeout::TimeoutError, timeout::TimeoutService, Either, Ready,
+};
 use crate::{http::Uri, io::IoBoxed};
 
 use super::{connection::Connection, error::ConnectError, pool::ConnectionPool, Connect};
@@ -231,7 +233,7 @@ impl Connector {
             None
         };
 
-        Rc::new(InnerConnector {
+        SharedService::new(InnerConnector {
             tcp_pool: ConnectionPool::new(
                 tcp_service,
                 self.conn_lifetime,

ntex/src/http/client/error.rs

@@ -1,5 +1,5 @@
 //! Http client errors
-use std::{error::Error, io};
+use std::{error::Error, io, rc::Rc};
 
 use serde_json::error::Error as JsonError;
 use thiserror::Error;
@@ -34,7 +34,7 @@ pub enum ConnectError {
     /// SSL error
     #[cfg(feature = "openssl")]
     #[error("{0}")]
-    SslError(#[from] SslError),
+    SslError(Rc<SslError>),
 
     /// SSL Handshake error
     #[cfg(feature = "openssl")]
@@ -62,6 +62,42 @@ pub enum ConnectError {
     Unresolved,
 }
 
+impl Clone for ConnectError {
+    fn clone(&self) -> Self {
+        match self {
+            ConnectError::SslIsNotSupported => ConnectError::SslIsNotSupported,
+            #[cfg(feature = "openssl")]
+            ConnectError::SslError(e) => ConnectError::SslError(e.clone()),
+            #[cfg(feature = "openssl")]
+            ConnectError::SslHandshakeError(e) => {
+                ConnectError::SslHandshakeError(e.clone())
+            }
+            ConnectError::Resolver(e) => {
+                ConnectError::Resolver(io::Error::new(e.kind(), format!("{}", e)))
+            }
+            ConnectError::NoRecords => ConnectError::NoRecords,
+            ConnectError::Timeout => ConnectError::Timeout,
+            ConnectError::Disconnected(e) => {
+                if let Some(e) = e {
+                    ConnectError::Disconnected(Some(io::Error::new(
+                        e.kind(),
+                        format!("{}", e),
+                    )))
+                } else {
+                    ConnectError::Disconnected(None)
+                }
+            }
+            ConnectError::Unresolved => ConnectError::Unresolved,
+        }
+    }
+}
+
+impl From<SslError> for ConnectError {
+    fn from(err: SslError) -> Self {
+        ConnectError::SslError(Rc::new(err))
+    }
+}
+
 impl From<crate::connect::ConnectError> for ConnectError {
     fn from(err: crate::connect::ConnectError) -> ConnectError {
         match err {