Allow to set io tag for web server (#426)

ntex-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-server"
-version = "2.3.0"
+version = "2.4.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "Server for ntex framework"
 keywords = ["network", "framework", "async", "futures"]

ntex-server/src/manager.rs

@@ -3,7 +3,8 @@ use std::{cell::Cell, cell::RefCell, collections::VecDeque, rc::Rc, sync::Arc};
 
 use async_channel::{unbounded, Receiver, Sender};
 use ntex_rt::System;
-use ntex_util::{future::join_all, time::sleep, time::Millis};
+use ntex_util::future::join_all;
+use ntex_util::time::{sleep, timeout, Millis};
 
 use crate::server::ServerShared;
 use crate::signals::Signal;
@@ -238,16 +239,13 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
 
         // stop workers
         if !self.workers.is_empty() {
-            let timeout = self.mgr.0.cfg.shutdown_timeout;
+            let to = self.mgr.0.cfg.shutdown_timeout;
 
-            if graceful && !timeout.is_zero() {
-                let futs: Vec<_> = self
-                    .workers
-                    .iter()
-                    .map(|worker| worker.stop(timeout))
-                    .collect();
+            if graceful && !to.is_zero() {
+                let futs: Vec<_> =
+                    self.workers.iter().map(|worker| worker.stop(to)).collect();
 
-                let _ = join_all(futs).await;
+                let _ = timeout(to, join_all(futs)).await;
             } else {
                 self.workers.iter().for_each(|worker| {
                     let _ = worker.stop(Millis::ZERO);

ntex-server/src/net/builder.rs

@@ -288,7 +288,7 @@ impl ServerBuilder {
         Ok(self)
     }
 
-    /// Add new service to the server.
+    /// Set io tag for named service.
     pub fn set_tag<N: AsRef<str>>(mut self, name: N, tag: &'static str) -> Self {
         let mut token = None;
         for sock in &self.sockets {

ntex-server/src/net/config.rs

@@ -16,12 +16,14 @@ pub struct Config(Rc<InnerServiceConfig>);
 #[derive(Debug)]
 pub(super) struct InnerServiceConfig {
     pub(super) pool: Cell<PoolId>,
+    pub(super) tag: Cell<Option<&'static str>>,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self(Rc::new(InnerServiceConfig {
             pool: Cell::new(PoolId::DEFAULT),
+            tag: Cell::new(None),
         }))
     }
 }
@@ -35,9 +37,19 @@ impl Config {
         self
     }
 
+    /// Set io tag for the service.
+    pub fn tag(&self, tag: &'static str) -> &Self {
+        self.0.tag.set(Some(tag));
+        self
+    }
+
     pub(super) fn get_pool_id(&self) -> PoolId {
         self.0.pool.get()
     }
+
+    pub(super) fn get_tag(&self) -> Option<&'static str> {
+        self.0.tag.get()
+    }
 }
 
 #[derive(Clone, Debug)]

ntex-server/src/net/factory.rs

@@ -91,20 +91,24 @@ where
 
     fn create(&self) -> BoxFuture<'static, Result<Vec<NetService>, ()>> {
         let cfg = Config::default();
-        let pool = cfg.get_pool_id();
         let name = self.name.clone();
-        let tokens = self.tokens.clone();
-        let factory_fut = (self.factory)(cfg);
+        let mut tokens = self.tokens.clone();
+        let factory_fut = (self.factory)(cfg.clone());
 
         Box::pin(async move {
             let factory = factory_fut.await.map_err(|_| {
                 log::error!("Cannot create {:?} service", name);
             })?;
+            if let Some(tag) = cfg.get_tag() {
+                for item in &mut tokens {
+                    item.1 = tag;
+                }
+            }
 
             Ok(vec![NetService {
                 tokens,
                 factory,
-                pool,
+                pool: cfg.get_pool_id(),
             }])
         })
     }

ntex-server/src/wrk.rs

@@ -12,7 +12,7 @@ use ntex_util::time::{sleep, timeout_checked, Millis};
 
 use crate::{ServerConfiguration, WorkerId};
 
-const STOP_TIMEOUT: Millis = Millis(5000);
+const STOP_TIMEOUT: Millis = Millis(3000);
 
 #[derive(Debug)]
 /// Shutdown worker
@@ -284,6 +284,7 @@ where
             }
         }
 
+        // re-create service
         loop {
             match select(wrk.factory.create(()), stream_recv(&mut wrk.stop)).await {
                 Either::Left(Ok(service)) => {

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [2.5.0] - 2024-09-24
+
+* Allow to set io tag for web server
+
 ## [2.4.0] - 2024-09-05
 
 * Add experimental `compio` runtime support

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "2.4.1"
+version = "2.5.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -68,7 +68,7 @@ ntex-service = "3.0"
 ntex-macros = "0.1.3"
 ntex-util = "2"
 ntex-bytes = "0.1.27"
-ntex-server = "2.3"
+ntex-server = "2.4"
 ntex-h2 = "1.1"
 ntex-rt = "0.4.17"
 ntex-io = "2.5"

ntex/examples/basic.rs

@@ -40,6 +40,7 @@ async fn main() -> std::io::Result<()> {
     .bind("0.0.0.0:8081")?
     .workers(4)
     .keep_alive(http::KeepAlive::Disabled)
+    .tag("MY-SERVER")
     .run()
     .await
 }

ntex/src/web/server.rs

@@ -21,6 +21,7 @@ struct Config {
     ssl_handshake_timeout: Seconds,
     headers_read_rate: Option<ReadRate>,
     payload_read_rate: Option<ReadRate>,
+    tag: &'static str,
     pool: PoolId,
 }
 
@@ -107,6 +108,7 @@ where
                     max_timeout: Seconds(13),
                 }),
                 payload_read_rate: None,
+                tag: "WEB",
                 pool: PoolId::P0,
             })),
             backlog: 1024,
@@ -308,6 +310,12 @@ where
         self
     }
 
+    /// Set io tag for web server
+    pub fn tag(self, tag: &'static str) -> Self {
+        self.config.lock().unwrap().tag = tag;
+        self
+    }
+
     /// Set memory pool.
     ///
     /// Use specified memory pool for memory allocations.
@@ -334,6 +342,7 @@ where
                         addr,
                         c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                     );
+                    r.tag(c.tag);
                     r.memory_pool(c.pool);
 
                     HttpService::build_with_config(c.into_cfg())
@@ -373,6 +382,7 @@ where
                         addr,
                         c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                     );
+                    r.tag(c.tag);
                     r.memory_pool(c.pool);
 
                     HttpService::build_with_config(c.into_cfg())
@@ -414,6 +424,7 @@ where
                     addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                 );
+                r.tag(c.tag);
                 r.memory_pool(c.pool);
 
                 HttpService::build_with_config(c.into_cfg())
@@ -522,6 +533,7 @@ where
                 socket_addr,
                 c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
             );
+            r.tag(c.tag);
             r.memory_pool(c.pool);
 
             HttpService::build_with_config(c.into_cfg())
@@ -553,6 +565,7 @@ where
                     socket_addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
                 );
+                r.tag(c.tag);
                 r.memory_pool(c.pool);
 
                 HttpService::build_with_config(c.into_cfg())