mirrors/ntex
1
0
Fork 0
mirror of https://github.com/ntex-rs/ntex.git synced 2025-04-04 13:27:39 +03:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Nikolay Kim 2025-03-28 19:55:50 +01:00
parent 3455bec5bd
commit da46ac20e2
2 changed files with 30 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

4
ntex-net/CHANGES.md
View file

@ -1,5 +1,9 @@
# Changes # Changes
## [2.5.10] - 2025-03-28
* Better closed sockets handling
## [2.5.9] - 2025-03-27 ## [2.5.9] - 2025-03-27
* Handle closed sockets * Handle closed sockets

67
ntex-net/src/rt_polling/driver.rs
View file

@ -1,5 +1,5 @@
use std::os::fd::{AsRawFd, RawFd}; 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::driver::{DriverApi, Event, Handler};
use ntex_neon::{syscall, Runtime}; use ntex_neon::{syscall, Runtime};
@ -18,8 +18,6 @@ bitflags::bitflags! {
struct Flags: u8 { struct Flags: u8 {
const RD = 0b0000_0001; const RD = 0b0000_0001;
const WR = 0b0000_0010; const WR = 0b0000_0010;
const ERROR = 0b0000_0100;
const CLOSED = 0b0000_1000;
} }
} }
@ -109,16 +107,14 @@ impl<T> Handler for StreamOpsHandler<T> {
let item = &mut streams[id]; let item = &mut streams[id];
log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev); log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev);
if item.flags.contains(Flags::CLOSED) { if item.io.is_none() {
return; return;
} }
// handle HUP // handle HUP
if ev.is_interrupt() { if ev.is_interrupt() {
if item.io.take().is_some() { item.context.stopped(None);
item.context.stopped(None); close(id as u32, item, &self.inner.api, None, true);
close(id as u32, item, &self.inner.api, true);
}
return; return;
} }
@ -178,9 +174,7 @@ impl<T> Handler for StreamOpsHandler<T> {
item.fd, item.fd,
item.io.is_some() item.io.is_some()
); );
if item.io.take().is_some() { close(id, &mut item, &self.inner.api, None, true);
close(id, &mut item, &self.inner.api, false);
}
} }
} }
self.inner.delayd_drop.set(false); self.inner.delayd_drop.set(false);
@ -191,20 +185,14 @@ impl<T> Handler for StreamOpsHandler<T> {
fn error(&mut self, id: usize, err: io::Error) { fn error(&mut self, id: usize, err: io::Error) {
self.inner.with(|streams| { self.inner.with(|streams| {
if let Some(item) = streams.get_mut(id) { if let Some(item) = streams.get_mut(id) {
if !item.flags.contains(Flags::ERROR) { log::debug!(
log::debug!( "{}: FD is failed ({}) {:?}, err: {:?}",
"{}: FD is failed ({}) {:?}, err: {:?}", item.tag(),
item.tag(), id,
id, item.fd,
item.fd, err
err );
); close(id as u32, item, &self.inner.api, Some(err), false);
item.flags.insert(Flags::ERROR);
if item.io.take().is_some() {
item.context.stopped(Some(err));
close(id as u32, item, &self.inner.api, true);
}
}
} }
}) })
} }
@ -226,15 +214,20 @@ fn close<T>(
id: u32, id: u32,
item: &mut StreamItem<T>, item: &mut StreamItem<T>,
api: &DriverApi, api: &DriverApi,
error: bool, error: Option<io::Error>,
shutdown: bool,
) -> Option<ntex_rt::JoinHandle<io::Result<i32>>> { ) -> Option<ntex_rt::JoinHandle<io::Result<i32>>> {
if !item.flags.contains(Flags::CLOSED) { 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; let fd = item.fd;
item.flags.insert(Flags::CLOSED);
api.detach(fd, id); api.detach(fd, id);
Some(ntex_rt::spawn_blocking(move || { Some(ntex_rt::spawn_blocking(move || {
if !error { if shutdown {
syscall!(libc::shutdown(fd, libc::SHUT_RDWR)); let _ = syscall!(libc::shutdown(fd, libc::SHUT_RDWR));
} }
syscall!(libc::close(fd)) syscall!(libc::close(fd))
})) }))
@ -248,16 +241,10 @@ impl<T> StreamCtl<T> {
let id = self.id as usize; let id = self.id as usize;
let fut = self.inner.with(|streams| { let fut = self.inner.with(|streams| {
let item = &mut streams[id]; let item = &mut streams[id];
if let Some(io) = item.io.take() { close(self.id, item, &self.inner.api, None, false)
log::debug!("{}: Closing ({}), {:?}", item.tag(), id, item.fd);
std::mem::forget(io);
Some(close(self.id, item, &self.inner.api, false))
} else {
None
}
}); });
async move { async move {
if let Some(Some(fut)) = fut { if let Some(fut) = fut {
fut.await fut.await
.map_err(|e| io::Error::new(io::ErrorKind::Other, e)) .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
.and_then(crate::helpers::pool_io_err)?; .and_then(crate::helpers::pool_io_err)?;
@ -371,9 +358,7 @@ impl<T> Drop for StreamCtl<T> {
item.fd, item.fd,
item.io.is_some() item.io.is_some()
); );
if item.io.take().is_some() { close(self.id, &mut item, &self.inner.api, None, true);
close(self.id, &mut item, &self.inner.api, true);
}
} }
self.inner.streams.set(Some(streams)); self.inner.streams.set(Some(streams));
} else { } else {