Prepare net release (#550)

* add public ServiceConfig::register constructor to support external configuration (#250)

* fix: doctest ServiceConfig::register() error (#250)

* add unit testing for ServiceConfig::register()

* replace pub(crate) to pub in ServiceConfig::new() (#250)

* replace pub to pub(crate) for ServiceConfig::new() and add pub for mod ntex::web::stack instead

* remove unsed DefaultError import in config.rs tests

---------

Co-authored-by: RuangyotN <ruangyotn@skyller.co>

.github/workflows/cov.yml

@@ -8,11 +8,6 @@ jobs:
     env:
       CARGO_TERM_COLOR: always
     steps:
-      - name: Free Disk Space
-        uses: jlumbroso/free-disk-space@main
-        with:
-          tool-cache: true
-
       - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update nightly

.github/workflows/linux.yml

@@ -16,11 +16,6 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: Free Disk Space
-        uses: jlumbroso/free-disk-space@main
-        with:
-          tool-cache: true
-
       - uses: actions/checkout@v4
 
       - name: Install ${{ matrix.version }}

Cargo.toml

@@ -46,7 +46,10 @@ ntex-compio = { path = "ntex-compio" }
 ntex-tokio = { path = "ntex-tokio" }
 
 [workspace.dependencies]
+async-channel = "2"
 async-task = "4.5.0"
+atomic-waker = "1.1"
+core_affinity = "0.8"
 bitflags = "2"
 cfg_aliases = "0.2.1"
 cfg-if = "1.0.0"
@@ -57,7 +60,8 @@ fxhash = "0.2"
 libc = "0.2.164"
 log = "0.4"
 io-uring = "0.7.4"
-polling = "3.3.0"
+oneshot = "0.1"
+polling = "3.7.4"
 nohash-hasher = "0.2.0"
 scoped-tls = "1.0.1"
 slab = "0.4.9"

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [2.11.1] - 2025-03-20
+
+* Add readiness check support
+
 ## [2.11.0] - 2025-03-10
 
 * Add single io context

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "2.11.0"
+version = "2.11.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -28,4 +28,3 @@ pin-project-lite = "0.2"
 [dev-dependencies]
 ntex = "2"
 rand = "0.8"
-env_logger = "0.11"

ntex-io/src/tasks.rs

@@ -537,7 +537,9 @@ impl IoContext {
                                 self.0.tag(),
                                 nbytes
                             );
-                            inner.dispatch_task.wake();
+                            if !inner.dispatch_task.wake_checked() {
+                                log::error!("Dispatcher waker is not registered");
+                            }
                         } else {
                             if nbytes >= hw {
                                 // read task is paused because of read back-pressure
@@ -722,28 +724,20 @@ impl IoContext {
     }
 
     /// Get read buffer
-    pub fn with_read_buf<F>(&self, f: F) -> Poll<()>
-    where
-        F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
-    {
-        let result = self.with_read_buf_inner(f);
-
+    pub fn is_read_ready(&self) -> bool {
         // check read readiness
-        if result.is_pending() {
-            if let Some(waker) = self.0 .0.read_task.take() {
-                let mut cx = Context::from_waker(&waker);
+        if let Some(waker) = self.0 .0.read_task.take() {
+            let mut cx = Context::from_waker(&waker);
 
-                if let Poll::Ready(ReadStatus::Ready) =
-                    self.0.filter().poll_read_ready(&mut cx)
-                {
-                    return Poll::Pending;
-                }
+            if let Poll::Ready(ReadStatus::Ready) = self.0.filter().poll_read_ready(&mut cx)
+            {
+                return true;
             }
         }
-        result
+        false
     }
 
-    fn with_read_buf_inner<F>(&self, f: F) -> Poll<()>
+    pub fn with_read_buf<F>(&self, f: F) -> Poll<()>
     where
         F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
     {
@@ -795,7 +789,9 @@ impl IoContext {
                                 self.0.tag(),
                                 nbytes
                             );
-                            inner.dispatch_task.wake();
+                            if !inner.dispatch_task.wake_checked() {
+                                log::error!("Dispatcher waker is not registered");
+                            }
                         } else {
                             if nbytes >= hw {
                                 // read task is paused because of read back-pressure
@@ -838,33 +834,8 @@ impl IoContext {
         }
     }
 
-    pub fn with_write_buf<F>(&self, f: F) -> Poll<()>
-    where
-        F: FnOnce(&BytesVec) -> Poll<io::Result<usize>>,
-    {
-        let result = self.with_write_buf_inner(f);
-
-        // check write readiness
-        if result.is_pending() {
-            let inner = &self.0 .0;
-            if let Some(waker) = inner.write_task.take() {
-                let ready = self
-                    .0
-                    .filter()
-                    .poll_write_ready(&mut Context::from_waker(&waker));
-                if !matches!(
-                    ready,
-                    Poll::Ready(WriteStatus::Ready | WriteStatus::Shutdown)
-                ) {
-                    return Poll::Ready(());
-                }
-            }
-        }
-        result
-    }
-
     /// Get write buffer
-    fn with_write_buf_inner<F>(&self, f: F) -> Poll<()>
+    pub fn with_write_buf<F>(&self, f: F) -> Poll<()>
     where
         F: FnOnce(&BytesVec) -> Poll<io::Result<usize>>,
     {

ntex-macros/Cargo.toml

@@ -18,4 +18,3 @@ proc-macro2 = "^1"
 [dev-dependencies]
 ntex = "2"
 futures = "0.3"
-env_logger = "0.11"

ntex-net/CHANGES.md

@@ -1,5 +1,25 @@
 # Changes
 
+## [2.5.10] - 2025-03-28
+
+* Better closed sockets handling
+
+## [2.5.9] - 2025-03-27
+
+* Handle closed sockets
+
+## [2.5.8] - 2025-03-25
+
+* Update neon runtime
+
+## [2.5.7] - 2025-03-21
+
+* Simplify neon poll impl
+
+## [2.5.6] - 2025-03-20
+
+* Redesign neon poll support
+
 ## [2.5.5] - 2025-03-17
 
 * Add check for required io-uring opcodes

ntex-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-net"
-version = "2.5.5"
+version = "2.5.10"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -27,27 +27,27 @@ compio = ["ntex-rt/compio", "ntex-compio"]
 # neon runtime
 neon = ["ntex-rt/neon", "ntex-neon", "slab", "socket2"]
 
-polling = ["ntex-neon/polling", "dep:polling"]
-io-uring = ["ntex-neon/io-uring", "dep:io-uring"]
+polling = ["ntex-neon/polling", "dep:polling", "socket2"]
+io-uring = ["ntex-neon/io-uring", "dep:io-uring", "socket2"]
 
 [dependencies]
 ntex-service = "3.3"
 ntex-bytes = "0.1"
 ntex-http = "0.1"
-ntex-io = "2.11"
+ntex-io = "2.11.1"
 ntex-rt = "0.4.25"
 ntex-util = "2.5"
 
 ntex-tokio = { version = "0.5.3", optional = true }
 ntex-compio = { version = "0.2.4", optional = true }
-ntex-neon = { version = "0.1.5", optional = true }
+ntex-neon = { version = "0.1.15", optional = true }
 
 bitflags = { workspace = true }
 cfg-if = { workspace = true }
 log = { workspace = true }
 libc = { workspace = true }
 slab = { workspace = true, optional = true }
-socket2 = { workspace = true, optional = true }
+socket2 = { workspace = true, optional = true, features = ["all"] }
 thiserror = { workspace = true }
 
 # Linux specific dependencies
@@ -57,4 +57,3 @@ polling = { workspace = true, optional = true }
 
 [dev-dependencies]
 ntex = "2"
-env_logger = "0.11"

ntex-net/src/rt_polling/connect.rs

@@ -1,7 +1,7 @@
 use std::os::fd::{AsRawFd, RawFd};
-use std::{cell::RefCell, collections::VecDeque, io, rc::Rc, task::Poll};
+use std::{cell::RefCell, io, rc::Rc, task::Poll};
 
-use ntex_neon::driver::{DriverApi, Handler, Interest};
+use ntex_neon::driver::{DriverApi, Event, Handler};
 use ntex_neon::{syscall, Runtime};
 use ntex_util::channel::oneshot::Sender;
 use slab::Slab;
@@ -12,13 +12,11 @@ pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
 
 #[derive(Debug)]
 enum Change {
-    Readable,
-    Writable,
+    Event(Event),
     Error(io::Error),
 }
 
 struct ConnectOpsBatcher {
-    feed: VecDeque<(usize, Change)>,
     inner: Rc<ConnectOpsInner>,
 }
 
@@ -42,10 +40,7 @@ impl ConnectOps {
                     connects: RefCell::new(Slab::new()),
                 });
                 inner = Some(ops.clone());
-                Box::new(ConnectOpsBatcher {
-                    inner: ops,
-                    feed: VecDeque::new(),
-                })
+                Box::new(ConnectOpsBatcher { inner: ops })
             });
 
             ConnectOps(inner.unwrap())
@@ -67,67 +62,50 @@ impl ConnectOps {
         let item = Item { fd, sender };
         let id = self.0.connects.borrow_mut().insert(item);
 
-        self.0.api.register(fd, id, Interest::Writable);
+        self.0.api.attach(fd, id as u32, Some(Event::writable(0)));
         Ok(id)
     }
 }
 
 impl Handler for ConnectOpsBatcher {
-    fn readable(&mut self, id: usize) {
+    fn event(&mut self, id: usize, event: Event) {
         log::debug!("connect-fd is readable {:?}", id);
-        self.feed.push_back((id, Change::Readable));
-    }
-
-    fn writable(&mut self, id: usize) {
-        log::debug!("connect-fd is writable {:?}", id);
-        self.feed.push_back((id, Change::Writable));
-    }
-
-    fn error(&mut self, id: usize, err: io::Error) {
-        self.feed.push_back((id, Change::Error(err)));
-    }
-
-    fn commit(&mut self) {
-        if self.feed.is_empty() {
-            return;
-        }
-        log::debug!("Commit connect driver changes, num: {:?}", self.feed.len());
 
         let mut connects = self.inner.connects.borrow_mut();
 
-        for (id, change) in self.feed.drain(..) {
-            if connects.contains(id) {
-                let item = connects.remove(id);
-                match change {
-                    Change::Readable => unreachable!(),
-                    Change::Writable => {
-                        let mut err: libc::c_int = 0;
-                        let mut err_len =
-                            std::mem::size_of::<libc::c_int>() as libc::socklen_t;
+        if connects.contains(id) {
+            let item = connects.remove(id);
+            if event.writable {
+                let mut err: libc::c_int = 0;
+                let mut err_len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
 
-                        let res = syscall!(libc::getsockopt(
-                            item.fd.as_raw_fd(),
-                            libc::SOL_SOCKET,
-                            libc::SO_ERROR,
-                            &mut err as *mut _ as *mut _,
-                            &mut err_len
-                        ));
+                let res = syscall!(libc::getsockopt(
+                    item.fd.as_raw_fd(),
+                    libc::SOL_SOCKET,
+                    libc::SO_ERROR,
+                    &mut err as *mut _ as *mut _,
+                    &mut err_len
+                ));
 
-                        let res = if err == 0 {
-                            res.map(|_| ())
-                        } else {
-                            Err(io::Error::from_raw_os_error(err))
-                        };
+                let res = if err == 0 {
+                    res.map(|_| ())
+                } else {
+                    Err(io::Error::from_raw_os_error(err))
+                };
 
-                        self.inner.api.unregister_all(item.fd);
-                        let _ = item.sender.send(res);
-                    }
-                    Change::Error(err) => {
-                        let _ = item.sender.send(Err(err));
-                        self.inner.api.unregister_all(item.fd);
-                    }
-                }
+                self.inner.api.detach(item.fd, id as u32);
+                let _ = item.sender.send(res);
             }
         }
     }
+
+    fn error(&mut self, id: usize, err: io::Error) {
+        let mut connects = self.inner.connects.borrow_mut();
+
+        if connects.contains(id) {
+            let item = connects.remove(id);
+            let _ = item.sender.send(Err(err));
+            self.inner.api.detach(item.fd, id as u32);
+        }
+    }
 }

ntex-net/src/rt_polling/driver.rs

@@ -1,7 +1,7 @@
 use std::os::fd::{AsRawFd, RawFd};
-use std::{cell::Cell, collections::VecDeque, future::Future, io, rc::Rc, task};
+use std::{cell::Cell, cell::RefCell, future::Future, io, mem, rc::Rc, task, task::Poll};
 
-use ntex_neon::driver::{DriverApi, Handler, Interest};
+use ntex_neon::driver::{DriverApi, Event, Handler};
 use ntex_neon::{syscall, Runtime};
 use slab::Slab;
 
@@ -9,37 +9,45 @@ use ntex_bytes::BufMut;
 use ntex_io::IoContext;
 
 pub(crate) struct StreamCtl<T> {
-    id: usize,
+    id: u32,
     inner: Rc<StreamOpsInner<T>>,
 }
 
+bitflags::bitflags! {
+    #[derive(Copy, Clone, Debug)]
+    struct Flags: u8 {
+        const RD     = 0b0000_0001;
+        const WR     = 0b0000_0010;
+    }
+}
+
 struct StreamItem<T> {
     io: Option<T>,
     fd: RawFd,
+    flags: Flags,
+    ref_count: u16,
     context: IoContext,
-    ref_count: usize,
 }
 
 pub(crate) struct StreamOps<T>(Rc<StreamOpsInner<T>>);
 
-#[derive(Debug)]
-enum Change {
-    Readable,
-    Writable,
-    Error(io::Error),
-}
-
 struct StreamOpsHandler<T> {
-    feed: VecDeque<(usize, Change)>,
     inner: Rc<StreamOpsInner<T>>,
 }
 
 struct StreamOpsInner<T> {
     api: DriverApi,
-    feed: Cell<Option<VecDeque<usize>>>,
+    delayd_drop: Cell<bool>,
+    feed: RefCell<Vec<u32>>,
     streams: Cell<Option<Box<Slab<StreamItem<T>>>>>,
 }
 
+impl<T> StreamItem<T> {
+    fn tag(&self) -> &'static str {
+        self.context.tag()
+    }
+}
+
 impl<T: AsRawFd + 'static> StreamOps<T> {
     pub(crate) fn current() -> Self {
         Runtime::value(|rt| {
@@ -47,14 +55,12 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
             rt.driver().register(|api| {
                 let ops = Rc::new(StreamOpsInner {
                     api,
-                    feed: Cell::new(Some(VecDeque::new())),
+                    feed: RefCell::new(Vec::new()),
+                    delayd_drop: Cell::new(false),
                     streams: Cell::new(Some(Box::new(Slab::new()))),
                 });
                 inner = Some(ops.clone());
-                Box::new(StreamOpsHandler {
-                    inner: ops,
-                    feed: VecDeque::new(),
-                })
+                Box::new(StreamOpsHandler { inner: ops })
             });
 
             StreamOps(inner.unwrap())
@@ -62,29 +68,27 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
     }
 
     pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl<T> {
-        let item = StreamItem {
-            context,
-            fd: io.as_raw_fd(),
-            io: Some(io),
-            ref_count: 1,
-        };
-        self.with(|streams| {
-            let id = streams.insert(item);
+        let fd = io.as_raw_fd();
+        let stream = self.0.with(move |streams| {
+            let item = StreamItem {
+                fd,
+                context,
+                io: Some(io),
+                ref_count: 1,
+                flags: Flags::empty(),
+            };
             StreamCtl {
-                id,
+                id: streams.insert(item) as u32,
                 inner: self.0.clone(),
             }
-        })
-    }
+        });
 
-    fn with<F, R>(&self, f: F) -> R
-    where
-        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
-    {
-        let mut inner = self.0.streams.take().unwrap();
-        let result = f(&mut inner);
-        self.0.streams.set(Some(inner));
-        result
+        self.0.api.attach(
+            fd,
+            stream.id,
+            Some(Event::new(0, false, false).with_interrupt()),
+        );
+        stream
     }
 }
 
@@ -95,131 +99,149 @@ impl<T> Clone for StreamOps<T> {
 }
 
 impl<T> Handler for StreamOpsHandler<T> {
-    fn readable(&mut self, id: usize) {
-        log::debug!("FD is readable {:?}", id);
-        self.feed.push_back((id, Change::Readable));
-    }
+    fn event(&mut self, id: usize, ev: Event) {
+        self.inner.with(|streams| {
+            if !streams.contains(id) {
+                return;
+            }
+            let item = &mut streams[id];
+            if item.io.is_none() {
+                return;
+            }
+            log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev);
 
-    fn writable(&mut self, id: usize) {
-        log::debug!("FD is writable {:?}", id);
-        self.feed.push_back((id, Change::Writable));
+            // handle HUP
+            if ev.is_interrupt() {
+                item.context.stopped(None);
+                close(id as u32, item, &self.inner.api, None, true);
+                return;
+            }
+
+            let mut renew_ev = Event::new(0, false, false).with_interrupt();
+
+            if ev.readable {
+                let res = item.context.with_read_buf(|buf| {
+                    let chunk = buf.chunk_mut();
+                    let result = task::ready!(syscall!(
+                        break libc::read(item.fd, chunk.as_mut_ptr() as _, chunk.len())
+                    ));
+                    if let Ok(size) = result {
+                        log::debug!("{}: data {:?}, s: {:?}", item.tag(), item.fd, size);
+                        unsafe { buf.advance_mut(size) };
+                    }
+                    Poll::Ready(result)
+                });
+
+                if res.is_pending() && item.context.is_read_ready() {
+                    renew_ev.readable = true;
+                    item.flags.insert(Flags::RD);
+                } else {
+                    item.flags.remove(Flags::RD);
+                }
+            } else if item.flags.contains(Flags::RD) {
+                renew_ev.readable = true;
+            }
+
+            if ev.writable {
+                let result = item.context.with_write_buf(|buf| {
+                    log::debug!("{}: write {:?} s: {:?}", item.tag(), item.fd, buf.len());
+                    syscall!(break libc::write(item.fd, buf[..].as_ptr() as _, buf.len()))
+                });
+                if result.is_pending() {
+                    renew_ev.writable = true;
+                    item.flags.insert(Flags::WR);
+                } else {
+                    item.flags.remove(Flags::WR);
+                }
+            } else if item.flags.contains(Flags::WR) {
+                renew_ev.writable = true;
+            }
+
+            self.inner.api.modify(item.fd, id as u32, renew_ev);
+
+            // delayed drops
+            if self.inner.delayd_drop.get() {
+                for id in self.inner.feed.borrow_mut().drain(..) {
+                    let item = &mut streams[id as usize];
+                    item.ref_count -= 1;
+                    if item.ref_count == 0 {
+                        let mut item = streams.remove(id as usize);
+                        log::debug!(
+                            "{}: Drop ({}), {:?}, has-io: {}",
+                            item.tag(),
+                            id,
+                            item.fd,
+                            item.io.is_some()
+                        );
+                        close(id, &mut item, &self.inner.api, None, true);
+                    }
+                }
+                self.inner.delayd_drop.set(false);
+            }
+        });
     }
 
     fn error(&mut self, id: usize, err: io::Error) {
-        log::debug!("FD is failed {:?}, err: {:?}", id, err);
-        self.feed.push_back((id, Change::Error(err)));
-    }
-
-    fn commit(&mut self) {
-        if self.feed.is_empty() {
-            return;
-        }
-        log::debug!("Commit changes, num: {:?}", self.feed.len());
-
-        let mut streams = self.inner.streams.take().unwrap();
-
-        for (id, change) in self.feed.drain(..) {
-            match change {
-                Change::Readable => {
-                    let item = &mut streams[id];
-                    let result = item.context.with_read_buf(|buf| {
-                        let chunk = buf.chunk_mut();
-                        let b = chunk.as_mut_ptr();
-                        task::Poll::Ready(
-                            task::ready!(syscall!(
-                                break libc::read(item.fd, b as _, chunk.len())
-                            ))
-                            .inspect(|size| {
-                                unsafe { buf.advance_mut(*size) };
-                                log::debug!(
-                                    "{}: {:?}, SIZE: {:?}",
-                                    item.context.tag(),
-                                    item.fd,
-                                    size
-                                );
-                            }),
-                        )
-                    });
-
-                    if item.io.is_some() && result.is_pending() {
-                        self.inner.api.register(item.fd, id, Interest::Readable);
-                    }
-                }
-                Change::Writable => {
-                    let item = &mut streams[id];
-                    let result = item.context.with_write_buf(|buf| {
-                        log::debug!(
-                            "{}: writing {:?} SIZE: {:?}",
-                            item.context.tag(),
-                            item.fd,
-                            buf.len()
-                        );
-                        let slice = &buf[..];
-                        syscall!(
-                            break libc::write(item.fd, slice.as_ptr() as _, slice.len())
-                        )
-                    });
-
-                    if item.io.is_some() && result.is_pending() {
-                        log::debug!("{}: want write {:?}", item.context.tag(), item.fd,);
-                        self.inner.api.register(item.fd, id, Interest::Writable);
-                    }
-                }
-                Change::Error(err) => {
-                    if let Some(item) = streams.get_mut(id) {
-                        item.context.stopped(Some(err));
-                        if let Some(_) = item.io.take() {
-                            close(id, item.fd, &self.inner.api);
-                        }
-                    }
-                }
-            }
-        }
-
-        // extra
-        let mut feed = self.inner.feed.take().unwrap();
-        for id in feed.drain(..) {
-            let item = &mut streams[id];
-            item.ref_count -= 1;
-            if item.ref_count == 0 {
-                let item = streams.remove(id);
+        self.inner.with(|streams| {
+            if let Some(item) = streams.get_mut(id) {
                 log::debug!(
-                    "{}: Drop io ({}), {:?}, has-io: {}",
-                    item.context.tag(),
+                    "{}: FD is failed ({}) {:?}, err: {:?}",
+                    item.tag(),
                     id,
                     item.fd,
-                    item.io.is_some()
+                    err
                 );
-                if item.io.is_some() {
-                    close(id, item.fd, &self.inner.api);
-                }
+                close(id as u32, item, &self.inner.api, Some(err), false);
             }
-        }
-
-        self.inner.feed.set(Some(feed));
-        self.inner.streams.set(Some(streams));
+        })
     }
 }
 
-fn close(id: usize, fd: RawFd, api: &DriverApi) -> ntex_rt::JoinHandle<io::Result<i32>> {
-    api.unregister_all(fd);
-    ntex_rt::spawn_blocking(move || {
-        syscall!(libc::shutdown(fd, libc::SHUT_RDWR))?;
-        syscall!(libc::close(fd))
-    })
+impl<T> StreamOpsInner<T> {
+    fn with<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
+    {
+        let mut streams = self.streams.take().unwrap();
+        let result = f(&mut streams);
+        self.streams.set(Some(streams));
+        result
+    }
+}
+
+fn close<T>(
+    id: u32,
+    item: &mut StreamItem<T>,
+    api: &DriverApi,
+    error: Option<io::Error>,
+    shutdown: bool,
+) -> Option<ntex_rt::JoinHandle<io::Result<i32>>> {
+    if let Some(io) = item.io.take() {
+        log::debug!("{}: Closing ({}), {:?}", item.tag(), id, item.fd);
+        mem::forget(io);
+        if let Some(err) = error {
+            item.context.stopped(Some(err));
+        }
+        let fd = item.fd;
+        api.detach(fd, id);
+        Some(ntex_rt::spawn_blocking(move || {
+            if shutdown {
+                let _ = syscall!(libc::shutdown(fd, libc::SHUT_RDWR));
+            }
+            syscall!(libc::close(fd))
+        }))
+    } else {
+        None
+    }
 }
 
 impl<T> StreamCtl<T> {
     pub(crate) fn close(self) -> impl Future<Output = io::Result<()>> {
-        let (io, fd) =
-            self.with(|streams| (streams[self.id].io.take(), streams[self.id].fd));
-        let fut = if let Some(io) = io {
-            log::debug!("Closing ({}), {:?}", self.id, fd);
-            std::mem::forget(io);
-            Some(close(self.id, fd, &self.inner.api))
-        } else {
-            None
-        };
+        let id = self.id as usize;
+        let fut = self.inner.with(|streams| {
+            let item = &mut streams[id];
+            close(self.id, item, &self.inner.api, None, false)
+        });
         async move {
             if let Some(fut) = fut {
                 fut.await
@@ -234,119 +256,85 @@ impl<T> StreamCtl<T> {
     where
         F: FnOnce(Option<&T>) -> R,
     {
-        self.with(|streams| f(streams[self.id].io.as_ref()))
+        self.inner
+            .with(|streams| f(streams[self.id as usize].io.as_ref()))
     }
 
-    pub(crate) fn pause_all(&self) {
-        self.with(|streams| {
-            let item = &mut streams[self.id];
+    pub(crate) fn modify(&self, rd: bool, wr: bool) {
+        self.inner.with(|streams| {
+            let item = &mut streams[self.id as usize];
 
             log::debug!(
-                "{}: Pause all io ({}), {:?}",
-                item.context.tag(),
+                "{}: Modify interest ({}), {:?} rd: {:?}, wr: {:?}",
+                item.tag(),
                 self.id,
-                item.fd
-            );
-            self.inner.api.unregister_all(item.fd);
-        })
-    }
-
-    pub(crate) fn pause_read(&self) {
-        self.with(|streams| {
-            let item = &mut streams[self.id];
-
-            log::debug!(
-                "{}: Pause io read ({}), {:?}",
-                item.context.tag(),
-                self.id,
-                item.fd
-            );
-            self.inner.api.unregister(item.fd, Interest::Readable);
-        })
-    }
-
-    pub(crate) fn resume_read(&self) {
-        self.with(|streams| {
-            let item = &mut streams[self.id];
-
-            log::debug!(
-                "{}: Resume io read ({}), {:?}",
-                item.context.tag(),
-                self.id,
-                item.fd
+                item.fd,
+                rd,
+                wr
             );
 
-            let result = item.context.with_read_buf(|buf| {
-                let chunk = buf.chunk_mut();
-                let b = chunk.as_mut_ptr();
-                task::Poll::Ready(
-                    task::ready!(syscall!(break libc::read(item.fd, b as _, chunk.len())))
-                        .inspect(|size| {
-                            unsafe { buf.advance_mut(*size) };
+            let mut event = Event::new(0, false, false).with_interrupt();
+
+            if rd {
+                if item.flags.contains(Flags::RD) {
+                    event.readable = true;
+                } else {
+                    let res = item.context.with_read_buf(|buf| {
+                        let chunk = buf.chunk_mut();
+                        let result = task::ready!(syscall!(
+                            break libc::read(item.fd, chunk.as_mut_ptr() as _, chunk.len())
+                        ));
+                        if let Ok(size) = result {
                             log::debug!(
-                                "{}: {:?}, SIZE: {:?}",
-                                item.context.tag(),
+                                "{}: read {:?}, s: {:?}",
+                                item.tag(),
                                 item.fd,
                                 size
                             );
-                        }),
-                )
-            });
+                            unsafe { buf.advance_mut(size) };
+                        }
+                        Poll::Ready(result)
+                    });
 
-            if item.io.is_some() && result.is_pending() {
-                self.inner
-                    .api
-                    .register(item.fd, self.id, Interest::Readable);
+                    if res.is_pending() && item.context.is_read_ready() {
+                        event.readable = true;
+                        item.flags.insert(Flags::RD);
+                    }
+                }
             }
-        })
-    }
 
-    pub(crate) fn resume_write(&self) {
-        self.with(|streams| {
-            let item = &mut streams[self.id];
+            if wr {
+                if item.flags.contains(Flags::WR) {
+                    event.writable = true;
+                } else {
+                    let result = item.context.with_write_buf(|buf| {
+                        log::debug!(
+                            "{}: Writing ({}), buf: {:?}",
+                            item.tag(),
+                            self.id,
+                            buf.len()
+                        );
+                        syscall!(
+                            break libc::write(item.fd, buf[..].as_ptr() as _, buf.len())
+                        )
+                    });
 
-            let result = item.context.with_write_buf(|buf| {
-                log::debug!(
-                    "{}: Writing io ({}), buf: {:?}",
-                    item.context.tag(),
-                    self.id,
-                    buf.len()
-                );
-
-                let slice = &buf[..];
-                syscall!(break libc::write(item.fd, slice.as_ptr() as _, slice.len()))
-            });
-
-            if item.io.is_some() && result.is_pending() {
-                log::debug!(
-                    "{}: Write is pending ({}), {:?}",
-                    item.context.tag(),
-                    self.id,
-                    item.context.flags()
-                );
-
-                self.inner
-                    .api
-                    .register(item.fd, self.id, Interest::Writable);
+                    if result.is_pending() {
+                        event.writable = true;
+                        item.flags.insert(Flags::WR);
+                    }
+                }
             }
-        })
-    }
 
-    fn with<F, R>(&self, f: F) -> R
-    where
-        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
-    {
-        let mut inner = self.inner.streams.take().unwrap();
-        let result = f(&mut inner);
-        self.inner.streams.set(Some(inner));
-        result
+            self.inner.api.modify(item.fd, self.id, event);
+        })
     }
 }
 
 impl<T> Clone for StreamCtl<T> {
     fn clone(&self) -> Self {
-        self.with(|streams| {
-            streams[self.id].ref_count += 1;
+        self.inner.with(|streams| {
+            streams[self.id as usize].ref_count += 1;
             Self {
                 id: self.id,
                 inner: self.inner.clone(),
@@ -358,25 +346,23 @@ impl<T> Clone for StreamCtl<T> {
 impl<T> Drop for StreamCtl<T> {
     fn drop(&mut self) {
         if let Some(mut streams) = self.inner.streams.take() {
-            streams[self.id].ref_count -= 1;
-            if streams[self.id].ref_count == 0 {
-                let item = streams.remove(self.id);
+            let id = self.id as usize;
+            streams[id].ref_count -= 1;
+            if streams[id].ref_count == 0 {
+                let mut item = streams.remove(id);
                 log::debug!(
                     "{}:  Drop io ({}), {:?}, has-io: {}",
-                    item.context.tag(),
+                    item.tag(),
                     self.id,
                     item.fd,
                     item.io.is_some()
                 );
-                if item.io.is_some() {
-                    close(self.id, item.fd, &self.inner.api);
-                }
+                close(self.id, &mut item, &self.inner.api, None, true);
             }
             self.inner.streams.set(Some(streams));
         } else {
-            let mut feed = self.inner.feed.take().unwrap();
-            feed.push_back(self.id);
-            self.inner.feed.set(Some(feed));
+            self.inner.delayd_drop.set(true);
+            self.inner.feed.borrow_mut().push(self.id);
         }
     }
 }

ntex-net/src/rt_polling/io.rs

@@ -54,21 +54,26 @@ enum Status {
 async fn run<T>(ctl: StreamCtl<T>, context: IoContext) {
     // Handle io read readiness
     let st = poll_fn(|cx| {
+        let mut modify = false;
+        let mut readable = false;
+        let mut writable = false;
         let read = match context.poll_read_ready(cx) {
             Poll::Ready(ReadStatus::Ready) => {
-                ctl.resume_read();
+                modify = true;
+                readable = true;
                 Poll::Pending
             }
             Poll::Ready(ReadStatus::Terminate) => Poll::Ready(()),
             Poll::Pending => {
-                ctl.pause_read();
+                modify = true;
                 Poll::Pending
             }
         };
 
         let write = match context.poll_write_ready(cx) {
             Poll::Ready(WriteStatus::Ready) => {
-                ctl.resume_write();
+                modify = true;
+                writable = true;
                 Poll::Pending
             }
             Poll::Ready(WriteStatus::Shutdown) => Poll::Ready(Status::Shutdown),
@@ -76,6 +81,10 @@ async fn run<T>(ctl: StreamCtl<T>, context: IoContext) {
             Poll::Pending => Poll::Pending,
         };
 
+        if modify {
+            ctl.modify(readable, writable);
+        }
+
         if read.is_pending() && write.is_pending() {
             Poll::Pending
         } else if write.is_ready() {
@@ -86,7 +95,7 @@ async fn run<T>(ctl: StreamCtl<T>, context: IoContext) {
     })
     .await;
 
-    ctl.resume_write();
+    ctl.modify(false, true);
     context.shutdown(st == Status::Shutdown).await;
     context.stopped(ctl.close().await.err());
 }

ntex-net/src/rt_polling/mod.rs

@@ -8,6 +8,9 @@ pub(crate) mod connect;
 mod driver;
 mod io;
 
+#[cfg(not(target_pointer_width = "64"))]
+compile_error!("Only 64bit platforms are supported");
+
 /// Tcp stream wrapper for neon TcpStream
 struct TcpStream(socket2::Socket);
 

ntex-net/src/rt_uring/driver.rs

@@ -33,6 +33,12 @@ struct StreamItem<T> {
     wr_op: Option<NonZeroU32>,
 }
 
+impl<T> StreamItem<T> {
+    fn tag(&self) -> &'static str {
+        self.context.tag()
+    }
+}
+
 enum Operation {
     Recv {
         id: usize,
@@ -249,7 +255,7 @@ impl<T> Handler for StreamOpsHandler<T> {
             if storage.streams[id].ref_count == 0 {
                 let mut item = storage.streams.remove(id);
 
-                log::debug!("{}: Drop io ({}), {:?}", item.context.tag(), id, item.fd);
+                log::debug!("{}: Drop io ({}), {:?}", item.tag(), id, item.fd);
 
                 if let Some(io) = item.io.take() {
                     mem::forget(io);
@@ -273,7 +279,7 @@ impl<T> StreamOpsStorage<T> {
             if let Poll::Ready(mut buf) = item.context.get_read_buf() {
                 log::debug!(
                     "{}: Recv resume ({}), {:?} rem: {:?}",
-                    item.context.tag(),
+                    item.tag(),
                     id,
                     item.fd,
                     buf.remaining_mut()
@@ -306,7 +312,7 @@ impl<T> StreamOpsStorage<T> {
             if let Poll::Ready(buf) = item.context.get_write_buf() {
                 log::debug!(
                     "{}: Send resume ({}), {:?} len: {:?}",
-                    item.context.tag(),
+                    item.tag(),
                     id,
                     item.fd,
                     buf.len()
@@ -396,12 +402,7 @@ impl<T> StreamCtl<T> {
 
         if let Some(rd_op) = item.rd_op {
             if !item.flags.contains(Flags::RD_CANCELING) {
-                log::debug!(
-                    "{}: Recv to pause ({}), {:?}",
-                    item.context.tag(),
-                    self.id,
-                    item.fd
-                );
+                log::debug!("{}: Recv to pause ({}), {:?}", item.tag(), self.id, item.fd);
                 item.flags.insert(Flags::RD_CANCELING);
                 self.inner.api.cancel(rd_op.get());
             }
@@ -426,12 +427,7 @@ impl<T> Drop for StreamCtl<T> {
             if storage.streams[self.id].ref_count == 0 {
                 let mut item = storage.streams.remove(self.id);
                 if let Some(io) = item.io.take() {
-                    log::debug!(
-                        "{}: Close io ({}), {:?}",
-                        item.context.tag(),
-                        self.id,
-                        item.fd
-                    );
+                    log::debug!("{}: Close io ({}), {:?}", item.tag(), self.id, item.fd);
                     mem::forget(io);
 
                     let id = storage.ops.insert(Operation::Close { tx: None });

ntex-rt/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.4.29] - 2025-03-26
+
+* Add Arbiter::get_value() helper method
+
 ## [0.4.27] - 2025-03-14
 
 * Add srbiters pings ttl

ntex-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-rt"
-version = "0.4.28"
+version = "0.4.29"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -32,8 +32,8 @@ neon = ["ntex-neon"]
 [dependencies]
 async-channel = "2"
 futures-timer = "3.0"
-log = "0.4"
 oneshot = "0.1"
+log = "0.4"
 
 compio-driver = { version = "0.6", optional = true }
 compio-runtime = { version = "0.6", optional = true }
@@ -42,7 +42,4 @@ tok-io = { version = "1", package = "tokio", default-features = false, features
     "net",
 ], optional = true }
 
-ntex-neon = { version = "0.1.1", optional = true }
-
-[dev-dependencies]
-env_logger = "0.11"
+ntex-neon = { version = "0.1.14", optional = true }

ntex-rt/src/arbiter.rs

@@ -286,6 +286,25 @@ impl Arbiter {
         })
     }
 
+    /// Get a type previously inserted to this runtime or create new one.
+    pub fn get_value<T, F>(f: F) -> T
+    where
+        T: Clone + 'static,
+        F: FnOnce() -> T,
+    {
+        STORAGE.with(move |cell| {
+            let mut st = cell.borrow_mut();
+            if let Some(boxed) = st.get(&TypeId::of::<T>()) {
+                if let Some(val) = (&**boxed as &(dyn Any + 'static)).downcast_ref::<T>() {
+                    return val.clone();
+                }
+            }
+            let val = f();
+            st.insert(TypeId::of::<T>(), Box::new(val.clone()));
+            val
+        })
+    }
+
     /// Wait for the event loop to stop by joining the underlying thread (if have Some).
     pub fn join(&mut self) -> thread::Result<()> {
         if let Some(thread_handle) = self.thread_handle.take() {
@@ -355,6 +374,7 @@ mod tests {
         assert!(Arbiter::get_item::<&'static str, _, _>(|s| *s == "test"));
         assert!(Arbiter::get_mut_item::<&'static str, _, _>(|s| *s == "test"));
         assert!(Arbiter::contains_item::<&'static str>());
+        assert!(Arbiter::get_value(|| 64u64) == 64);
         assert!(format!("{:?}", Arbiter::current()).contains("Arbiter"));
     }
 }

ntex-rt/src/lib.rs

@@ -112,6 +112,8 @@ mod tokio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> tok_io::task::JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -196,6 +198,8 @@ mod compio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -323,6 +327,8 @@ mod neon {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> Task<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -377,7 +383,7 @@ mod neon {
 
     impl<T> JoinHandle<T> {
         pub fn is_finished(&self) -> bool {
-            false
+            self.fut.is_none()
         }
     }
 

ntex-server/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [2.7.3] - 2025-03-28
+
+* Better worker availability handling
+
+## [2.7.2] - 2025-03-27
+
+* Handle paused state
+
 ## [2.7.1] - 2025-02-28
 
 * Fix set core affinity out of worker start #508

ntex-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-server"
-version = "2.7.1"
+version = "2.7.4"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Server for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -22,13 +22,13 @@ ntex-service = "3.4"
 ntex-rt = "0.4"
 ntex-util = "2.8"
 
-async-channel = "2"
-async-broadcast = "0.7"
-core_affinity = "0.8"
-polling = "3.3"
-log = "0.4"
-socket2 = "0.5"
-oneshot = { version = "0.1", default-features = false, features = ["async"] }
+async-channel = { workspace = true }
+atomic-waker = { workspace = true }
+core_affinity = { workspace = true }
+oneshot = { workspace = true }
+polling = { workspace = true }
+log = { workspace = true }
+socket2 = { workspace = true }
 
 [dev-dependencies]
 ntex = "2"

ntex-server/src/manager.rs

@@ -139,7 +139,6 @@ impl<F: ServerConfiguration> ServerManager<F> {
 fn start_worker<F: ServerConfiguration>(mgr: ServerManager<F>, cid: Option<CoreId>) {
     let _ = ntex_rt::spawn(async move {
         let id = mgr.next_id();
-
         let mut wrk = Worker::start(id, mgr.factory(), cid);
 
         loop {
@@ -181,7 +180,7 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
     fn process(&mut self, mut item: F::Item) {
         loop {
             if !self.workers.is_empty() {
-                if self.next > self.workers.len() {
+                if self.next >= self.workers.len() {
                     self.next = self.workers.len() - 1;
                 }
                 match self.workers[self.next].send(item) {
@@ -212,10 +211,9 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
         match upd {
             Update::Available(worker) => {
                 self.workers.push(worker);
+                self.workers.sort();
                 if self.workers.len() == 1 {
                     self.mgr.resume();
-                } else {
-                    self.workers.sort();
                 }
             }
             Update::Unavailable(worker) => {
@@ -234,6 +232,9 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
                 if let Err(item) = self.workers[0].send(item) {
                     self.backlog.push_back(item);
                     self.workers.remove(0);
+                    if self.workers.is_empty() {
+                        self.mgr.pause();
+                    }
                     break;
                 }
             }

ntex-server/src/net/accept.rs

@@ -92,12 +92,14 @@ impl AcceptLoop {
 
     /// Start accept loop
     pub fn start(mut self, socks: Vec<(Token, Listener)>, srv: Server) {
+        let (tx, rx_start) = oneshot::channel();
         let (rx, poll) = self
             .inner
             .take()
             .expect("AcceptLoop cannot be used multiple times");
 
         Accept::start(
+            tx,
             rx,
             poll,
             socks,
@@ -105,6 +107,8 @@ impl AcceptLoop {
             self.notify.clone(),
             self.status_handler.take(),
         );
+
+        let _ = rx_start.recv();
     }
 }
 
@@ -121,6 +125,7 @@ impl fmt::Debug for AcceptLoop {
 struct Accept {
     poller: Arc<Poller>,
     rx: mpsc::Receiver<AcceptorCommand>,
+    tx: Option<oneshot::Sender<()>>,
     sockets: Vec<ServerSocketInfo>,
     srv: Server,
     notify: AcceptNotify,
@@ -131,6 +136,7 @@ struct Accept {
 
 impl Accept {
     fn start(
+        tx: oneshot::Sender<()>,
         rx: mpsc::Receiver<AcceptorCommand>,
         poller: Arc<Poller>,
         socks: Vec<(Token, Listener)>,
@@ -145,11 +151,12 @@ impl Accept {
             .name("ntex-server accept loop".to_owned())
             .spawn(move || {
                 System::set_current(sys);
-                Accept::new(rx, poller, socks, srv, notify, status_handler).poll()
+                Accept::new(tx, rx, poller, socks, srv, notify, status_handler).poll()
             });
     }
 
     fn new(
+        tx: oneshot::Sender<()>,
         rx: mpsc::Receiver<AcceptorCommand>,
         poller: Arc<Poller>,
         socks: Vec<(Token, Listener)>,
@@ -175,6 +182,7 @@ impl Accept {
             notify,
             srv,
             status_handler,
+            tx: Some(tx),
             backpressure: true,
             backlog: VecDeque::new(),
         }
@@ -192,19 +200,23 @@ impl Accept {
         // Create storage for events
         let mut events = Events::with_capacity(NonZeroUsize::new(512).unwrap());
 
+        let mut timeout = Some(Duration::ZERO);
         loop {
-            if let Err(e) = self.poller.wait(&mut events, None) {
-                if e.kind() == io::ErrorKind::Interrupted {
-                    continue;
-                } else {
+            if let Err(e) = self.poller.wait(&mut events, timeout) {
+                if e.kind() != io::ErrorKind::Interrupted {
                     panic!("Cannot wait for events in poller: {}", e)
                 }
+            } else if timeout.is_some() {
+                timeout = None;
+                let _ = self.tx.take().unwrap().send(());
             }
 
-            for event in events.iter() {
-                let readd = self.accept(event.key);
-                if readd {
-                    self.add_source(event.key);
+            for idx in 0..self.sockets.len() {
+                if self.sockets[idx].registered.get() {
+                    let readd = self.accept(idx);
+                    if readd {
+                        self.add_source(idx);
+                    }
                 }
             }
 

ntex-server/src/net/test.rs

@@ -59,17 +59,19 @@ where
                 .workers(1)
                 .disable_signals()
                 .run();
-            tx.send((system, local_addr, server))
-                .expect("Failed to send Server to TestServer");
+
+            ntex_rt::spawn(async move {
+                ntex_util::time::sleep(ntex_util::time::Millis(75)).await;
+                tx.send((system, local_addr, server))
+                    .expect("Failed to send Server to TestServer");
+            });
+
             Ok(())
         })
     });
 
     let (system, addr, server) = rx.recv().unwrap();
 
-    // wait for server
-    thread::sleep(std::time::Duration::from_millis(50));
-
     TestServer {
         addr,
         server,

ntex-server/src/wrk.rs

@@ -2,8 +2,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{ready, Context, Poll};
 use std::{cmp, future::poll_fn, future::Future, hash, pin::Pin, sync::Arc};
 
-use async_broadcast::{self as bus, broadcast};
 use async_channel::{unbounded, Receiver, Sender};
+use atomic_waker::AtomicWaker;
 use core_affinity::CoreId;
 
 use ntex_rt::{spawn, Arbiter};
@@ -99,10 +99,10 @@ impl<T> Worker<T> {
 
                 log::debug!("Creating server instance in {:?}", id);
                 let factory = cfg.create().await;
-                log::debug!("Server instance has been created in {:?}", id);
 
                 match create(id, rx1, rx2, factory, avail_tx).await {
                     Ok((svc, wrk)) => {
+                        log::debug!("Server instance has been created in {:?}", id);
                         run_worker(svc, wrk).await;
                     }
                     Err(e) => {
@@ -151,10 +151,8 @@ impl<T> Worker<T> {
         if self.failed.load(Ordering::Acquire) {
             WorkerStatus::Failed
         } else {
-            // cleanup updates
-            while self.avail.notify.try_recv().is_ok() {}
-
-            if self.avail.notify.recv_direct().await.is_err() {
+            self.avail.wait_for_update().await;
+            if self.avail.failed() {
                 self.failed.store(true, Ordering::Release);
             }
             self.status()
@@ -196,52 +194,85 @@ impl Future for WorkerStop {
 
 #[derive(Debug, Clone)]
 struct WorkerAvailability {
-    notify: bus::Receiver<()>,
-    available: Arc<AtomicBool>,
+    inner: Arc<Inner>,
 }
 
 #[derive(Debug, Clone)]
 struct WorkerAvailabilityTx {
-    notify: bus::Sender<()>,
-    available: Arc<AtomicBool>,
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug)]
+struct Inner {
+    waker: AtomicWaker,
+    updated: AtomicBool,
+    available: AtomicBool,
+    failed: AtomicBool,
 }
 
 impl WorkerAvailability {
     fn create() -> (Self, WorkerAvailabilityTx) {
-        let (mut tx, rx) = broadcast(16);
-        tx.set_overflow(true);
+        let inner = Arc::new(Inner {
+            waker: AtomicWaker::new(),
+            updated: AtomicBool::new(false),
+            available: AtomicBool::new(false),
+            failed: AtomicBool::new(false),
+        });
 
         let avail = WorkerAvailability {
-            notify: rx,
-            available: Arc::new(AtomicBool::new(false)),
-        };
-        let avail_tx = WorkerAvailabilityTx {
-            notify: tx,
-            available: avail.available.clone(),
+            inner: inner.clone(),
         };
+        let avail_tx = WorkerAvailabilityTx { inner };
         (avail, avail_tx)
     }
 
+    fn failed(&self) -> bool {
+        self.inner.failed.load(Ordering::Acquire)
+    }
+
     fn available(&self) -> bool {
-        self.available.load(Ordering::Acquire)
+        self.inner.available.load(Ordering::Acquire)
+    }
+
+    async fn wait_for_update(&self) {
+        poll_fn(|cx| {
+            if self.inner.updated.load(Ordering::Acquire) {
+                self.inner.updated.store(false, Ordering::Release);
+                Poll::Ready(())
+            } else {
+                self.inner.waker.register(cx.waker());
+                Poll::Pending
+            }
+        })
+        .await;
     }
 }
 
 impl WorkerAvailabilityTx {
     fn set(&self, val: bool) {
-        let old = self.available.swap(val, Ordering::Release);
-        if !old && val {
-            let _ = self.notify.try_broadcast(());
+        let old = self.inner.available.swap(val, Ordering::Release);
+        if old != val {
+            self.inner.updated.store(true, Ordering::Release);
+            self.inner.waker.wake();
         }
     }
 }
 
+impl Drop for WorkerAvailabilityTx {
+    fn drop(&mut self) {
+        self.inner.failed.store(true, Ordering::Release);
+        self.inner.updated.store(true, Ordering::Release);
+        self.inner.available.store(false, Ordering::Release);
+        self.inner.waker.wake();
+    }
+}
+
 /// Service worker
 ///
 /// Worker accepts message via unbounded channel and starts processing.
 struct WorkerSt<T, F: ServiceFactory<T>> {
     id: WorkerId,
-    rx: Pin<Box<dyn Stream<Item = T>>>,
+    rx: Receiver<T>,
     stop: Pin<Box<dyn Stream<Item = Shutdown>>>,
     factory: F,
     availability: WorkerAvailabilityTx,
@@ -253,25 +284,43 @@ where
     F: ServiceFactory<T> + 'static,
 {
     loop {
+        let mut recv = std::pin::pin!(wrk.rx.recv());
         let fut = poll_fn(|cx| {
-            ready!(svc.poll_ready(cx)?);
-
-            if let Some(item) = ready!(Pin::new(&mut wrk.rx).poll_next(cx)) {
-                let fut = svc.call(item);
-                let _ = spawn(async move {
-                    let _ = fut.await;
-                });
+            match svc.poll_ready(cx) {
+                Poll::Ready(Ok(())) => {
+                    wrk.availability.set(true);
+                }
+                Poll::Ready(Err(err)) => {
+                    wrk.availability.set(false);
+                    return Poll::Ready(Err(err));
+                }
+                Poll::Pending => {
+                    wrk.availability.set(false);
+                    return Poll::Pending;
+                }
+            }
+
+            match ready!(recv.as_mut().poll(cx)) {
+                Ok(item) => {
+                    let fut = svc.call(item);
+                    let _ = spawn(async move {
+                        let _ = fut.await;
+                    });
+                    Poll::Ready(Ok::<_, F::Error>(true))
+                }
+                Err(_) => {
+                    log::error!("Server is gone");
+                    Poll::Ready(Ok(false))
+                }
             }
-            Poll::Ready(Ok::<(), F::Error>(()))
         });
 
         match select(fut, stream_recv(&mut wrk.stop)).await {
-            Either::Left(Ok(())) => continue,
+            Either::Left(Ok(true)) => continue,
             Either::Left(Err(_)) => {
                 let _ = ntex_rt::spawn(async move {
                     svc.shutdown().await;
                 });
-                wrk.availability.set(false);
             }
             Either::Right(Some(Shutdown { timeout, result })) => {
                 wrk.availability.set(false);
@@ -285,7 +334,8 @@ where
                 stop_svc(wrk.id, svc, timeout, Some(result)).await;
                 return;
             }
-            Either::Right(None) => {
+            Either::Left(Ok(false)) | Either::Right(None) => {
+                wrk.availability.set(false);
                 stop_svc(wrk.id, svc, STOP_TIMEOUT, None).await;
                 return;
             }
@@ -295,7 +345,6 @@ where
         loop {
             match select(wrk.factory.create(()), stream_recv(&mut wrk.stop)).await {
                 Either::Left(Ok(service)) => {
-                    wrk.availability.set(true);
                     svc = Pipeline::new(service).bind();
                     break;
                 }
@@ -336,8 +385,6 @@ where
 {
     availability.set(false);
     let factory = factory?;
-
-    let rx = Box::pin(rx);
     let mut stop = Box::pin(stop);
 
     let svc = match select(factory.create(()), stream_recv(&mut stop)).await {
@@ -356,9 +403,9 @@ where
         svc,
         WorkerSt {
             id,
+            rx,
             factory,
             availability,
-            rx: Box::pin(rx),
             stop: Box::pin(stop),
         },
     ))

ntex-tls/examples/rustls-server.rs

@@ -13,9 +13,8 @@ async fn main() -> io::Result<()> {
     println!("Started openssl echp server: 127.0.0.1:8443");
 
     // load ssl keys
-    let cert_file =
-        &mut BufReader::new(File::open("../ntex-tls/examples/cert.pem").unwrap());
-    let key_file = &mut BufReader::new(File::open("../ntex-tls/examples/key.pem").unwrap());
+    let cert_file = &mut BufReader::new(File::open("../examples/cert.pem").unwrap());
+    let key_file = &mut BufReader::new(File::open("../examples/key.pem").unwrap());
     let keys = rustls_pemfile::private_key(key_file).unwrap().unwrap();
     let cert_chain = rustls_pemfile::certs(cert_file)
         .collect::<Result<Vec<_>, _>>()

ntex-tls/examples/webserver.rs

@@ -8,18 +8,18 @@ use tls_openssl::ssl::{self, SslFiletype, SslMethod};
 
 #[ntex::main]
 async fn main() -> io::Result<()> {
-    //std::env::set_var("RUST_LOG", "trace");
-    //env_logger::init();
+    std::env::set_var("RUST_LOG", "trace");
+    let _ = env_logger::try_init();
 
     println!("Started openssl web server: 127.0.0.1:8443");
 
     // load ssl keys
     let mut builder = ssl::SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
     builder
-        .set_private_key_file("../tests/key.pem", SslFiletype::PEM)
+        .set_private_key_file("./examples/key.pem", SslFiletype::PEM)
         .unwrap();
     builder
-        .set_certificate_chain_file("../tests/cert.pem")
+        .set_certificate_chain_file("./examples/cert.pem")
         .unwrap();
 
     // h2 alpn config

ntex/CHANGES.md

@@ -1,5 +1,17 @@
 # Changes
 
+## [2.12.4] - 2025-03-28
+
+* http: Return PayloadError::Incomplete on server disconnect
+
+* web: Expose WebStack for external wrapper support in downstream crates #542
+
+## [2.12.3] - 2025-03-22
+
+* web: Export web::app_service::AppService #534
+
+* http: Add delay for test server availability, could cause connect race
+
 ## [2.12.2] - 2025-03-15
 
 * http: Allow to run publish future to completion in case error

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "2.12.2"
+version = "2.12.4"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -68,11 +68,11 @@ ntex-service = "3.4"
 ntex-macros = "0.1"
 ntex-util = "2.8"
 ntex-bytes = "0.1.27"
-ntex-server = "2.7"
+ntex-server = "2.7.4"
 ntex-h2 = "1.8.6"
 ntex-rt = "0.4.27"
 ntex-io = "2.11"
-ntex-net = "2.5"
+ntex-net = "2.5.10"
 ntex-tls = "2.3"
 
 base64 = "0.22"
@@ -114,6 +114,7 @@ flate2 = { version = "1.0", optional = true }
 [dev-dependencies]
 rand = "0.8"
 time = "0.3"
+oneshot = "0.1"
 futures-util = "0.3"
 tls-openssl = { version = "0.10", package = "openssl" }
 tls-rustls = { version = "0.23", package = "rustls", features = ["ring", "std"], default-features = false }

ntex/src/http/client/h1proto.rs

@@ -1,13 +1,11 @@
-use std::{
-    future::poll_fn, io, io::Write, pin::Pin, task::Context, task::Poll, time::Instant,
-};
+use std::{future::poll_fn, io, io::Write, pin::Pin, task, task::Poll, time::Instant};
 
 use crate::http::body::{BodySize, MessageBody};
 use crate::http::error::PayloadError;
-use crate::http::h1;
 use crate::http::header::{HeaderMap, HeaderValue, HOST};
 use crate::http::message::{RequestHeadType, ResponseHead};
 use crate::http::payload::{Payload, PayloadStream};
+use crate::http::{h1, Version};
 use crate::io::{IoBoxed, RecvError};
 use crate::time::{timeout_checked, Millis};
 use crate::util::{ready, BufMut, Bytes, BytesMut, Stream};
@@ -101,7 +99,13 @@ where
             Ok((head, Payload::None))
         }
         _ => {
-            let pl: PayloadStream = Box::pin(PlStream::new(io, codec, created, pool));
+            let pl: PayloadStream = Box::pin(PlStream::new(
+                io,
+                codec,
+                created,
+                pool,
+                head.version == Version::HTTP_10,
+            ));
             Ok((head, pl.into()))
         }
     }
@@ -137,6 +141,7 @@ pub(super) struct PlStream {
     io: Option<IoBoxed>,
     codec: h1::ClientPayloadCodec,
     created: Instant,
+    http_10: bool,
     pool: Option<Acquired>,
 }
 
@@ -146,12 +151,14 @@ impl PlStream {
         codec: h1::ClientCodec,
         created: Instant,
         pool: Option<Acquired>,
+        http_10: bool,
     ) -> Self {
         PlStream {
             io: Some(io),
             codec: codec.into_payload_codec(),
             created,
             pool,
+            http_10,
         }
     }
 }
@@ -161,41 +168,46 @@ impl Stream for PlStream {
 
     fn poll_next(
         mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
+        cx: &mut task::Context<'_>,
     ) -> Poll<Option<Self::Item>> {
         let mut this = self.as_mut();
         loop {
-            return Poll::Ready(Some(
-                match ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx)) {
-                    Ok(chunk) => {
-                        if let Some(chunk) = chunk {
-                            Ok(chunk)
-                        } else {
-                            release_connection(
-                                this.io.take().unwrap(),
-                                !this.codec.keepalive(),
-                                this.created,
-                                this.pool.take(),
-                            );
-                            return Poll::Ready(None);
-                        }
+            let item = ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx));
+            return Poll::Ready(Some(match item {
+                Ok(chunk) => {
+                    if let Some(chunk) = chunk {
+                        Ok(chunk)
+                    } else {
+                        release_connection(
+                            this.io.take().unwrap(),
+                            !this.codec.keepalive(),
+                            this.created,
+                            this.pool.take(),
+                        );
+                        return Poll::Ready(None);
                     }
-                    Err(RecvError::KeepAlive) => {
-                        Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into())
+                }
+                Err(RecvError::KeepAlive) => {
+                    Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into())
+                }
+                Err(RecvError::Stop) => {
+                    Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped").into())
+                }
+                Err(RecvError::WriteBackpressure) => {
+                    ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?;
+                    continue;
+                }
+                Err(RecvError::Decoder(err)) => Err(err),
+                Err(RecvError::PeerGone(Some(err))) => {
+                    Err(PayloadError::Incomplete(Some(err)))
+                }
+                Err(RecvError::PeerGone(None)) => {
+                    if this.http_10 {
+                        return Poll::Ready(None);
                     }
-                    Err(RecvError::Stop) => {
-                        Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped")
-                            .into())
-                    }
-                    Err(RecvError::WriteBackpressure) => {
-                        ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?;
-                        continue;
-                    }
-                    Err(RecvError::Decoder(err)) => Err(err),
-                    Err(RecvError::PeerGone(Some(err))) => Err(err.into()),
-                    Err(RecvError::PeerGone(None)) => return Poll::Ready(None),
-                },
-            ));
+                    Err(PayloadError::Incomplete(None))
+                }
+            }));
         }
     }
 }

ntex/src/http/client/response.rs

@@ -387,8 +387,8 @@ impl Future for ReadBody {
         let this = self.get_mut();
 
         loop {
-            return match Pin::new(&mut this.stream).poll_next(cx)? {
-                Poll::Ready(Some(chunk)) => {
+            return match Pin::new(&mut this.stream).poll_next(cx) {
+                Poll::Ready(Some(Ok(chunk))) => {
                     if this.limit > 0 && (this.buf.len() + chunk.len()) > this.limit {
                         Poll::Ready(Err(PayloadError::Overflow))
                     } else {
@@ -397,6 +397,7 @@ impl Future for ReadBody {
                     }
                 }
                 Poll::Ready(None) => Poll::Ready(Ok(this.buf.split().freeze())),
+                Poll::Ready(Some(Err(err))) => Poll::Ready(Err(err)),
                 Poll::Pending => {
                     if this.timeout.poll_elapsed(cx).is_ready() {
                         Poll::Ready(Err(PayloadError::Incomplete(Some(

ntex/src/http/test.rs

@@ -11,7 +11,7 @@ use crate::server::Server;
 use crate::service::ServiceFactory;
 #[cfg(feature = "ws")]
 use crate::ws::{error::WsClientError, WsClient, WsConnection};
-use crate::{rt::System, time::Millis, time::Seconds, util::Bytes};
+use crate::{rt::System, time::sleep, time::Millis, time::Seconds, util::Bytes};
 
 use super::client::{Client, ClientRequest, ClientResponse, Connector};
 use super::error::{HttpError, PayloadError};
@@ -244,10 +244,15 @@ where
                 .workers(1)
                 .disable_signals()
                 .run();
-            tx.send((system, srv, local_addr)).unwrap();
+
+            crate::rt::spawn(async move {
+                sleep(Millis(125)).await;
+                tx.send((system, srv, local_addr)).unwrap();
+            });
             Ok(())
         })
     });
+    thread::sleep(std::time::Duration::from_millis(150));
 
     let (system, server, addr) = rx.recv().unwrap();
 

ntex/src/web/config.rs

@@ -68,7 +68,7 @@ pub struct ServiceConfig<Err = DefaultError> {
 }
 
 impl<Err: ErrorRenderer> ServiceConfig<Err> {
-    pub(crate) fn new() -> Self {
+    pub fn new() -> Self {
         Self {
             services: Vec::new(),
             state: Extensions::new(),
@@ -132,7 +132,7 @@ mod tests {
     use crate::http::{Method, StatusCode};
     use crate::util::Bytes;
     use crate::web::test::{call_service, init_service, read_body, TestRequest};
-    use crate::web::{self, App, HttpRequest, HttpResponse};
+    use crate::web::{self, App, DefaultError, HttpRequest, HttpResponse};
 
     #[crate::rt_test]
     async fn test_configure_state() {
@@ -205,4 +205,11 @@ mod tests {
         let resp = call_service(&srv, req).await;
         assert_eq!(resp.status(), StatusCode::OK);
     }
+
+    #[test]
+    fn test_new_service_config() {
+        let cfg: ServiceConfig<DefaultError> = ServiceConfig::new();
+        assert!(cfg.services.is_empty());
+        assert!(cfg.external.is_empty());
+    }
 }

ntex/src/web/mod.rs

@@ -82,7 +82,7 @@ mod route;
 mod scope;
 mod server;
 mod service;
-mod stack;
+pub mod stack;
 pub mod test;
 pub mod types;
 mod util;
@@ -128,6 +128,7 @@ pub mod dev {
     //! The purpose of this module is to alleviate imports of many common
     //! traits by adding a glob import to the top of ntex::web heavy modules:
 
+    pub use crate::web::app_service::AppService;
     pub use crate::web::config::AppConfig;
     pub use crate::web::info::ConnectionInfo;
     pub use crate::web::rmap::ResourceMap;

ntex/src/web/test.rs

@@ -697,7 +697,10 @@ where
             .set_tag("test", "WEB-SRV")
             .run();
 
-            tx.send((System::current(), srv, local_addr)).unwrap();
+            crate::rt::spawn(async move {
+                sleep(Millis(125)).await;
+                tx.send((System::current(), srv, local_addr)).unwrap();
+            });
             Ok(())
         })
     });

ntex/tests/http_awc_client.rs

@@ -508,19 +508,21 @@ async fn test_client_gzip_encoding_large() {
 async fn test_client_gzip_encoding_large_random() {
     let data = rand::thread_rng()
         .sample_iter(&rand::distributions::Alphanumeric)
-        .take(100_000)
+        .take(1_048_500)
         .map(char::from)
         .collect::<String>();
 
     let srv = test::server(|| {
-        App::new().service(web::resource("/").route(web::to(|data: Bytes| async move {
-            let mut e = GzEncoder::new(Vec::new(), Compression::default());
-            e.write_all(&data).unwrap();
-            let data = e.finish().unwrap();
-            HttpResponse::Ok()
-                .header("content-encoding", "gzip")
-                .body(data)
-        })))
+        App::new()
+            .state(web::types::PayloadConfig::default().limit(1_048_576))
+            .service(web::resource("/").route(web::to(|data: Bytes| async move {
+                let mut e = GzEncoder::new(Vec::new(), Compression::default());
+                e.write_all(&data).unwrap();
+                let data = e.finish().unwrap();
+                HttpResponse::Ok()
+                    .header("content-encoding", "gzip")
+                    .body(data)
+            })))
     });
 
     // client request
@@ -528,7 +530,7 @@ async fn test_client_gzip_encoding_large_random() {
     assert!(response.status().is_success());
 
     // read response
-    let bytes = response.body().await.unwrap();
+    let bytes = response.body().limit(1_048_576).await.unwrap();
     assert_eq!(bytes, Bytes::from(data));
 }
 

ntex/tests/http_openssl.rs

@@ -1,5 +1,6 @@
 #![cfg(feature = "openssl")]
-use std::{io, sync::atomic::AtomicUsize, sync::atomic::Ordering, sync::Arc};
+use std::io;
+use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex};
 
 use futures_util::stream::{once, Stream, StreamExt};
 use tls_openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod};
@@ -424,11 +425,12 @@ async fn test_h2_service_error() {
     assert_eq!(bytes, Bytes::from_static(b"error"));
 }
 
-struct SetOnDrop(Arc<AtomicUsize>);
+struct SetOnDrop(Arc<AtomicUsize>, Arc<Mutex<Option<::oneshot::Sender<()>>>>);
 
 impl Drop for SetOnDrop {
     fn drop(&mut self) {
         self.0.fetch_add(1, Ordering::Relaxed);
+        let _ = self.1.lock().unwrap().take().unwrap().send(());
     }
 }
 
@@ -436,17 +438,20 @@ impl Drop for SetOnDrop {
 async fn test_h2_client_drop() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
+    let (tx, rx) = ::oneshot::channel();
+    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
+        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build()
             .h2(move |req: Request| {
-                let count = count.clone();
+                let st = SetOnDrop(count.clone(), tx.clone());
                 async move {
-                    let _st = SetOnDrop(count);
                     assert!(req.peer_addr().is_some());
                     assert_eq!(req.version(), Version::HTTP_2);
-                    sleep(Seconds(100)).await;
+                    sleep(Seconds(30)).await;
+                    drop(st);
                     Ok::<_, io::Error>(Response::Ok().finish())
                 }
             })
@@ -454,9 +459,9 @@ async fn test_h2_client_drop() -> io::Result<()> {
             .map_err(|_| ())
     });
 
-    let result = timeout(Millis(250), srv.srequest(Method::GET, "/").send()).await;
+    let result = timeout(Millis(1500), srv.srequest(Method::GET, "/").send()).await;
     assert!(result.is_err());
-    sleep(Millis(150)).await;
+    let _ = timeout(Millis(1500), rx).await;
     assert_eq!(count.load(Ordering::Relaxed), 1);
     Ok(())
 }
@@ -539,13 +544,19 @@ async fn test_ws_transport() {
 async fn test_h2_graceful_shutdown() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
+    let (tx, rx) = ::oneshot::channel();
+    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
+        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build()
             .h2(move |_| {
                 let count = count.clone();
                 count.fetch_add(1, Ordering::Relaxed);
+                if count.load(Ordering::Relaxed) == 2 {
+                    let _ = tx.lock().unwrap().take().unwrap().send(());
+                }
                 async move {
                     sleep(Millis(1000)).await;
                     count.fetch_sub(1, Ordering::Relaxed);
@@ -566,7 +577,7 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> {
         let _ = req.send().await.unwrap();
         sleep(Millis(100000)).await;
     });
-    sleep(Millis(150)).await;
+    let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let (tx, rx) = oneshot::channel();
@@ -574,8 +585,6 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> {
         srv.stop().await;
         let _ = tx.send(());
     });
-    sleep(Millis(150)).await;
-    assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);

ntex/tests/http_server.rs

@@ -1,4 +1,4 @@
-use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc};
+use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex};
 use std::{io, io::Read, io::Write, net};
 
 use futures_util::future::{self, FutureExt};
@@ -723,11 +723,12 @@ async fn test_h1_service_error() {
     assert_eq!(bytes, Bytes::from_static(b"error"));
 }
 
-struct SetOnDrop(Arc<AtomicUsize>);
+struct SetOnDrop(Arc<AtomicUsize>, Option<::oneshot::Sender<()>>);
 
 impl Drop for SetOnDrop {
     fn drop(&mut self) {
         self.0.fetch_add(1, Ordering::Relaxed);
+        let _ = self.1.take().unwrap().send(());
     }
 }
 
@@ -735,24 +736,28 @@ impl Drop for SetOnDrop {
 async fn test_h1_client_drop() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
+    let (tx, rx) = ::oneshot::channel();
+    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
+        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().h1(move |req: Request| {
+            let tx = tx.clone();
             let count = count.clone();
             async move {
-                let _st = SetOnDrop(count);
+                let _st = SetOnDrop(count, tx.lock().unwrap().take());
                 assert!(req.peer_addr().is_some());
                 assert_eq!(req.version(), Version::HTTP_11);
-                sleep(Millis(500)).await;
+                sleep(Millis(50000)).await;
                 Ok::<_, io::Error>(Response::Ok().finish())
             }
         })
     });
 
-    let result = timeout(Millis(100), srv.request(Method::GET, "/").send()).await;
+    let result = timeout(Millis(1500), srv.request(Method::GET, "/").send()).await;
     assert!(result.is_err());
-    sleep(Millis(1000)).await;
+    let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 1);
     Ok(())
 }
@@ -761,12 +766,18 @@ async fn test_h1_client_drop() -> io::Result<()> {
 async fn test_h1_gracefull_shutdown() {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
+    let (tx, rx) = ::oneshot::channel();
+    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
+        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().h1(move |_: Request| {
             let count = count.clone();
             count.fetch_add(1, Ordering::Relaxed);
+            if count.load(Ordering::Relaxed) == 2 {
+                let _ = tx.lock().unwrap().take().unwrap().send(());
+            }
             async move {
                 sleep(Millis(1000)).await;
                 count.fetch_sub(1, Ordering::Relaxed);
@@ -781,7 +792,7 @@ async fn test_h1_gracefull_shutdown() {
     let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n");
 
-    sleep(Millis(150)).await;
+    let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let (tx, rx) = oneshot::channel();
@@ -789,8 +800,6 @@ async fn test_h1_gracefull_shutdown() {
         srv.stop().await;
         let _ = tx.send(());
     });
-    sleep(Millis(150)).await;
-    assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);
@@ -800,12 +809,18 @@ async fn test_h1_gracefull_shutdown() {
 async fn test_h1_gracefull_shutdown_2() {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
+    let (tx, rx) = ::oneshot::channel();
+    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
+        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().finish(move |_: Request| {
             let count = count.clone();
             count.fetch_add(1, Ordering::Relaxed);
+            if count.load(Ordering::Relaxed) == 2 {
+                let _ = tx.lock().unwrap().take().unwrap().send(());
+            }
             async move {
                 sleep(Millis(1000)).await;
                 count.fetch_sub(1, Ordering::Relaxed);
@@ -820,17 +835,14 @@ async fn test_h1_gracefull_shutdown_2() {
     let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n");
 
-    sleep(Millis(150)).await;
-    assert_eq!(count.load(Ordering::Relaxed), 2);
+    let _ = rx.await;
+    assert_eq!(count.load(Ordering::Acquire), 2);
 
     let (tx, rx) = oneshot::channel();
     rt::spawn(async move {
         srv.stop().await;
         let _ = tx.send(());
     });
-    sleep(Millis(150)).await;
-    assert_eq!(count.load(Ordering::Relaxed), 2);
-
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);
 }