From f5ee55d598810f56379f789b2fbcd163b8730f03 Mon Sep 17 00:00:00 2001 From: Nikolay Kim Date: Fri, 28 Mar 2025 21:06:11 +0100 Subject: [PATCH] Handle socket close for poll driver (#549) --- ntex-net/CHANGES.md | 4 ++ ntex-net/Cargo.toml | 2 +- ntex-net/src/rt_polling/driver.rs | 61 ++++++++++++++----------------- ntex-server/Cargo.toml | 2 +- ntex-server/src/manager.rs | 2 +- 5 files changed, 35 insertions(+), 36 deletions(-) diff --git a/ntex-net/CHANGES.md b/ntex-net/CHANGES.md index a16145fc..e60744ef 100644 --- a/ntex-net/CHANGES.md +++ b/ntex-net/CHANGES.md @@ -1,5 +1,9 @@ # Changes +## [2.5.10] - 2025-03-28 + +* Better closed sockets handling + ## [2.5.9] - 2025-03-27 * Handle closed sockets diff --git a/ntex-net/Cargo.toml b/ntex-net/Cargo.toml index 46cf5cc4..0142a911 100644 --- a/ntex-net/Cargo.toml +++ b/ntex-net/Cargo.toml @@ -40,7 +40,7 @@ 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.14", optional = true } +ntex-neon = { version = "0.1.15", optional = true } bitflags = { workspace = true } cfg-if = { workspace = true } diff --git a/ntex-net/src/rt_polling/driver.rs b/ntex-net/src/rt_polling/driver.rs index c179a77a..24db553d 100644 --- a/ntex-net/src/rt_polling/driver.rs +++ b/ntex-net/src/rt_polling/driver.rs @@ -1,5 +1,5 @@ use std::os::fd::{AsRawFd, RawFd}; -use std::{cell::Cell, cell::RefCell, future::Future, io, rc::Rc, task, task::Poll}; +use std::{cell::Cell, cell::RefCell, future::Future, io, mem, rc::Rc, task, task::Poll}; use ntex_neon::driver::{DriverApi, Event, Handler}; use ntex_neon::{syscall, Runtime}; @@ -18,7 +18,6 @@ bitflags::bitflags! { struct Flags: u8 { const RD = 0b0000_0001; const WR = 0b0000_0010; - const CLOSED = 0b0000_0100; } } @@ -106,18 +105,15 @@ impl Handler for StreamOpsHandler { return; } let item = &mut streams[id]; - log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev); - - if item.flags.contains(Flags::CLOSED) { + if item.io.is_none() { return; } + log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev); // handle HUP if ev.is_interrupt() { item.context.stopped(None); - if item.io.take().is_some() { - close(id as u32, item, &self.inner.api); - } + close(id as u32, item, &self.inner.api, None, true); return; } @@ -177,9 +173,7 @@ impl Handler for StreamOpsHandler { item.fd, item.io.is_some() ); - if item.io.is_some() { - close(id, &mut item, &self.inner.api); - } + close(id, &mut item, &self.inner.api, None, true); } } self.inner.delayd_drop.set(false); @@ -197,10 +191,7 @@ impl Handler for StreamOpsHandler { item.fd, err ); - item.context.stopped(Some(err)); - if item.io.take().is_some() { - close(id as u32, item, &self.inner.api); - } + close(id as u32, item, &self.inner.api, Some(err), false); } }) } @@ -222,14 +213,26 @@ fn close( id: u32, item: &mut StreamItem, api: &DriverApi, -) -> ntex_rt::JoinHandle> { - let fd = item.fd; - item.flags.insert(Flags::CLOSED); - api.detach(fd, id); - ntex_rt::spawn_blocking(move || { - syscall!(libc::shutdown(fd, libc::SHUT_RDWR))?; - syscall!(libc::close(fd)) - }) + error: Option, + shutdown: bool, +) -> Option>> { + 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 StreamCtl { @@ -237,13 +240,7 @@ impl StreamCtl { let id = self.id as usize; let fut = self.inner.with(|streams| { let item = &mut streams[id]; - if let Some(io) = item.io.take() { - log::debug!("{}: Closing ({}), {:?}", item.tag(), id, item.fd); - std::mem::forget(io); - Some(close(self.id, item, &self.inner.api)) - } else { - None - } + close(self.id, item, &self.inner.api, None, false) }); async move { if let Some(fut) = fut { @@ -360,9 +357,7 @@ impl Drop for StreamCtl { item.fd, item.io.is_some() ); - if item.io.is_some() { - close(self.id, &mut item, &self.inner.api); - } + close(self.id, &mut item, &self.inner.api, None, true); } self.inner.streams.set(Some(streams)); } else { diff --git a/ntex-server/Cargo.toml b/ntex-server/Cargo.toml index dcfa8332..a88be635 100644 --- a/ntex-server/Cargo.toml +++ b/ntex-server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex-server" -version = "2.7.3" +version = "2.7.4" authors = ["ntex contributors "] description = "Server for ntex framework" keywords = ["network", "framework", "async", "futures"] diff --git a/ntex-server/src/manager.rs b/ntex-server/src/manager.rs index f0719750..9d0bfe8d 100644 --- a/ntex-server/src/manager.rs +++ b/ntex-server/src/manager.rs @@ -180,7 +180,7 @@ impl HandleCmdState { 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) {