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

wip

Browse source
This commit is contained in:
Nikolay Kim 2025-03-27 10:12:14 +01:00
parent ea74503736
commit a8db7de953
9 changed files with 84 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -46,6 +46,7 @@ ntex-compio = { path = "ntex-compio" }
ntex-tokio = { path = "ntex-tokio" }
ntex-neon = { git = "https://github.com/ntex-rs/neon.git", branch = "iour-fix" }
#ntex-neon = { path = "../dev/neon" }
[workspace.dependencies]
async-task = "4.5.0"

2
ntex-net/src/helpers.rs
View file

@ -49,7 +49,7 @@ async fn connect_inner(
let (sender, rx) = channel();
crate::rt_impl::connect::ConnectOps::current().connect(fd, addr, sender)?;
crate::rt_impl::connect::ConnectOps::current().connect("-", fd, addr, sender)?;
rx.await
.map_err(|_| io::Error::new(io::ErrorKind::Other, "IO Driver is gone"))

12
ntex-net/src/rt_polling/connect.rs
View file

@ -22,6 +22,7 @@ struct ConnectOpsBatcher {
struct Item {
fd: RawFd,
tag: &'static str,
sender: Sender<io::Result<()>>,
}
@ -49,6 +50,7 @@ impl ConnectOps {
pub(crate) fn connect(
&self,
tag: &'static str,
fd: RawFd,
addr: SockAddr,
sender: Sender<io::Result<()>>,
@ -59,10 +61,12 @@ impl ConnectOps {
res?;
}
let item = Item { fd, sender };
let item = Item { tag, fd, sender };
let id = self.0.connects.borrow_mut().insert(item);
self.0.api.attach(fd, id as u32, Some(Event::writable(0)));
self.0
.api
.attach(tag, fd, id as u32, Some(Event::writable(0)));
Ok(id)
}
}
@ -93,7 +97,7 @@ impl Handler for ConnectOpsBatcher {
Err(io::Error::from_raw_os_error(err))
};
self.inner.api.detach(item.fd, id as u32);
self.inner.api.detach(item.tag, item.fd, id as u32);
let _ = item.sender.send(res);
}
}
@ -105,7 +109,7 @@ impl Handler for ConnectOpsBatcher {
if connects.contains(id) {
let item = connects.remove(id);
let _ = item.sender.send(Err(err));
self.inner.api.detach(item.fd, id as u32);
self.inner.api.detach(item.tag, item.fd, id as u32);
}
}
}

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

@ -16,8 +16,9 @@ pub(crate) struct StreamCtl<T> {
bitflags::bitflags! {
#[derive(Copy, Clone, Debug)]
struct Flags: u8 {
const RD = 0b0000_0001;
const WR = 0b0000_0010;
const RD = 0b0000_0001;
const WR = 0b0000_0010;
const CLOSED = 0b0000_0100;
}
}
@ -69,6 +70,7 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl<T> {
let fd = io.as_raw_fd();
let tag = context.tag();
let stream = self.0.with(move |streams| {
let item = StreamItem {
fd,
@ -84,6 +86,7 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
});
self.0.api.attach(
tag,
fd,
stream.id,
Some(Event::new(0, false, false).with_interrupt()),
@ -100,19 +103,22 @@ impl<T> Clone for StreamOps<T> {
impl<T> Handler for StreamOpsHandler<T> {
fn event(&mut self, id: usize, ev: Event) {
log::debug!("FD event {:?} event: {:?}", id, ev);
self.inner.with(|streams| {
if !streams.contains(id) {
return;
}
let item = &mut streams[id];
log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev);
if item.flags.contains(Flags::CLOSED) {
return;
}
// handle HUP
if ev.is_interrupt() {
item.context.stopped(None);
if item.io.take().is_some() {
close(id as u32, item.fd, &self.inner.api);
close(id as u32, item, &self.inner.api);
}
return;
}
@ -157,7 +163,9 @@ impl<T> Handler for StreamOpsHandler<T> {
renew_ev.writable = true;
}
self.inner.api.modify(item.fd, id as u32, renew_ev);
self.inner
.api
.modify(item.tag(), item.fd, id as u32, renew_ev);
// delayed drops
if self.inner.delayd_drop.get() {
@ -165,7 +173,7 @@ impl<T> Handler for StreamOpsHandler<T> {
let item = &mut streams[id as usize];
item.ref_count -= 1;
if item.ref_count == 0 {
let item = streams.remove(id as usize);
let mut item = streams.remove(id as usize);
log::debug!(
"{}: Drop ({}), {:?}, has-io: {}",
item.tag(),
@ -174,7 +182,7 @@ impl<T> Handler for StreamOpsHandler<T> {
item.io.is_some()
);
if item.io.is_some() {
close(id, item.fd, &self.inner.api);
close(id, &mut item, &self.inner.api);
}
}
}
@ -189,7 +197,7 @@ impl<T> Handler for StreamOpsHandler<T> {
log::debug!("FD is failed ({}) {:?}, err: {:?}", id, item.fd, err);
item.context.stopped(Some(err));
if item.io.take().is_some() {
close(id as u32, item.fd, &self.inner.api);
close(id as u32, item, &self.inner.api);
}
}
})
@ -208,8 +216,14 @@ impl<T> StreamOpsInner<T> {
}
}
fn close(id: u32, fd: RawFd, api: &DriverApi) -> ntex_rt::JoinHandle<io::Result<i32>> {
api.detach(fd, id);
fn close<T>(
id: u32,
item: &mut StreamItem<T>,
api: &DriverApi,
) -> ntex_rt::JoinHandle<io::Result<i32>> {
let fd = item.fd;
item.flags.insert(Flags::CLOSED);
api.detach(item.tag(), fd, id);
ntex_rt::spawn_blocking(move || {
syscall!(libc::shutdown(fd, libc::SHUT_RDWR))?;
syscall!(libc::close(fd))
@ -219,16 +233,16 @@ fn close(id: u32, fd: RawFd, api: &DriverApi) -> ntex_rt::JoinHandle<io::Result<
impl<T> StreamCtl<T> {
pub(crate) fn close(self) -> impl Future<Output = io::Result<()>> {
let id = self.id as usize;
let (io, fd) = self
.inner
.with(|streams| (streams[id].io.take(), streams[id].fd));
let fut = if let Some(io) = io {
log::debug!("Closing ({}), {:?}", id, fd);
std::mem::forget(io);
Some(close(self.id, fd, &self.inner.api))
} else {
None
};
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
}
});
async move {
if let Some(fut) = fut {
fut.await
@ -313,7 +327,7 @@ impl<T> StreamCtl<T> {
}
}
self.inner.api.modify(item.fd, self.id, event);
self.inner.api.modify(item.tag(), item.fd, self.id, event);
})
}
}
@ -336,7 +350,7 @@ impl<T> Drop for StreamCtl<T> {
let id = self.id as usize;
streams[id].ref_count -= 1;
if streams[id].ref_count == 0 {
let item = streams.remove(id);
let mut item = streams.remove(id);
log::debug!(
"{}: Drop io ({}), {:?}, has-io: {}",
item.tag(),
@ -345,7 +359,7 @@ impl<T> Drop for StreamCtl<T> {
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);
}
}
self.inner.streams.set(Some(streams));

1
ntex-server/Cargo.toml
View file

@ -24,6 +24,7 @@ ntex-util = "2.8"
async-channel = "2"
async-broadcast = "0.7"
atomic-waker = "1.1"
core_affinity = "0.8"
polling = "3.3"
log = "0.4"

2
ntex-server/src/manager.rs
View file

@ -184,8 +184,10 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
if self.next > self.workers.len() {
self.next = self.workers.len() - 1;
}
println!("--------- SENDING ITEM");
match self.workers[self.next].send(item) {
Ok(()) => {
println!("--------- ITEM SENT");
self.next = (self.next + 1) % self.workers.len();
break;
}

3
ntex-server/src/net/accept.rs
View file

@ -386,11 +386,14 @@ impl Accept {
io,
token: info.token,
};
println!("------- ACCEPTED {:?}", msg);
if let Err(msg) = self.srv.process(msg) {
log::trace!("Server is unavailable");
self.backlog.push_back(msg);
self.backpressure(true);
return false;
} else {
println!("------- SENT ACCEPTED");
}
}
Ok(None) => return true,

22
ntex-server/src/wrk.rs
View file

@ -5,6 +5,7 @@ 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 core_affinity::CoreId;
use atomic_waker::AtomicWaker;
use ntex_rt::{spawn, Arbiter};
use ntex_service::{Pipeline, PipelineBinding, Service, ServiceFactory};
@ -39,6 +40,7 @@ pub struct Worker<T> {
id: WorkerId,
tx1: Sender<T>,
tx2: Sender<Shutdown>,
waker: Arc<AtomicWaker>,
avail: WorkerAvailability,
failed: Arc<AtomicBool>,
}
@ -85,6 +87,8 @@ impl<T> Worker<T> {
{
let (tx1, rx1) = unbounded();
let (tx2, rx2) = unbounded();
let waker = Arc::new(AtomicWaker::new());
let waker2 = waker.clone();
let (avail, avail_tx) = WorkerAvailability::create();
Arbiter::default().exec_fn(move || {
@ -100,7 +104,7 @@ impl<T> Worker<T> {
log::debug!("Creating server instance in {:?}", id);
let factory = cfg.create().await;
match create(id, rx1, rx2, factory, avail_tx).await {
match create(id, rx1, waker2, rx2, factory, avail_tx).await {
Ok((svc, wrk)) => {
log::debug!("Server instance has been created in {:?}", id);
run_worker(svc, wrk).await;
@ -117,6 +121,7 @@ impl<T> Worker<T> {
id,
tx1,
tx2,
waker,
avail,
failed: Arc::new(AtomicBool::new(false)),
}
@ -132,6 +137,7 @@ impl<T> Worker<T> {
/// Returns `Ok` if message got accepted by the worker.
/// Otherwise return message back as `Err`
pub fn send(&self, msg: T) -> Result<(), T> {
self.waker.wake();
self.tx1.try_send(msg).map_err(|msg| msg.into_inner())
}
@ -177,6 +183,7 @@ impl<T> Clone for Worker<T> {
id: self.id,
tx1: self.tx1.clone(),
tx2: self.tx2.clone(),
waker: self.waker.clone(),
avail: self.avail.clone(),
failed: self.failed.clone(),
}
@ -244,6 +251,7 @@ struct WorkerSt<T, F: ServiceFactory<T>> {
rx: Receiver<T>,
stop: Pin<Box<dyn Stream<Item = Shutdown>>>,
factory: F,
waker: Arc<AtomicWaker>,
availability: WorkerAvailabilityTx,
}
@ -258,6 +266,8 @@ where
//println!("------- run worker {:?}", wrk.id);
let mut recv = std::pin::pin!(wrk.rx.recv());
let fut = poll_fn(|cx| {
wrk.waker.register(cx.waker());
match svc.poll_ready(cx) {
Poll::Ready(res) => {
res?;
@ -265,14 +275,14 @@ where
}
Poll::Pending => {
wrk.availability.set(false);
return Poll::Pending
return Poll::Pending;
}
}
//println!("------- waiting socket {:?}", wrk.id);
println!("------- waiting socket {:?}", wrk.id);
match ready!(recv.as_mut().poll(cx)) {
Ok(item) => {
//println!("------- got {:?}", wrk.id);
println!("------- got {:?}", wrk.id);
let fut = svc.call(item);
let _ = spawn(async move {
@ -281,7 +291,7 @@ where
Poll::Ready(Ok::<_, F::Error>(true))
}
Err(_) => {
//println!("------- failed {:?}", wrk.id);
println!("------- failed {:?}", wrk.id);
log::error!("Server is gone");
Poll::Ready(Ok(false))
@ -350,6 +360,7 @@ async fn stop_svc<T, F>(
async fn create<T, F>(
id: WorkerId,
rx: Receiver<T>,
waker: Arc<AtomicWaker>,
stop: Receiver<Shutdown>,
factory: Result<F, ()>,
availability: WorkerAvailabilityTx,
@ -379,6 +390,7 @@ where
WorkerSt {
id,
rx,
waker,
factory,
availability,
stop: Box::pin(stop),

24
ntex/tests/http_awc_client.rs
View file

@ -156,7 +156,7 @@ async fn test_form() {
assert!(response.status().is_success());
}
#[ntex::test]
//#[ntex::test]
async fn test_timeout() {
let srv = test::server(|| {
App::new().service(web::resource("/").route(web::to(|| async {
@ -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));
}