wip

ntex-net/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [2.5.9] - 2025-03-27
+
+* Handle closed sockets
+
 ## [2.5.8] - 2025-03-25
 
 * Update neon runtime

ntex-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-net"
-version = "2.5.8"
+version = "2.5.9"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]

ntex-net/src/rt_polling/driver.rs

@@ -70,7 +70,6 @@ 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,

ntex-server/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [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.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Server for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -24,7 +24,6 @@ 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"

ntex-server/src/wrk.rs

@@ -4,7 +4,6 @@ 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};
@@ -40,7 +39,6 @@ pub struct Worker<T> {
     id: WorkerId,
     tx1: Sender<T>,
     tx2: Sender<Shutdown>,
-    waker: Arc<AtomicWaker>,
     avail: WorkerAvailability,
     failed: Arc<AtomicBool>,
 }
@@ -87,8 +85,6 @@ 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 || {
@@ -104,7 +100,7 @@ impl<T> Worker<T> {
                 log::debug!("Creating server instance in {:?}", id);
                 let factory = cfg.create().await;
 
-                match create(id, rx1, waker2, rx2, factory, avail_tx).await {
+                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;
@@ -121,7 +117,6 @@ impl<T> Worker<T> {
             id,
             tx1,
             tx2,
-            waker,
             avail,
             failed: Arc::new(AtomicBool::new(false)),
         }
@@ -137,7 +132,6 @@ 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())
     }
 
@@ -183,7 +177,6 @@ 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(),
         }
@@ -251,7 +244,6 @@ struct WorkerSt<T, F: ServiceFactory<T>> {
     rx: Receiver<T>,
     stop: Pin<Box<dyn Stream<Item = Shutdown>>>,
     factory: F,
-    waker: Arc<AtomicWaker>,
     availability: WorkerAvailabilityTx,
 }
 
@@ -263,8 +255,6 @@ where
     loop {
         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?;
@@ -352,7 +342,6 @@ 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,
@@ -382,7 +371,6 @@ where
         WorkerSt {
             id,
             rx,
-            waker,
             factory,
             availability,
             stop: Box::pin(stop),