Fix filter leak during Io drop (#274)

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.3.17] - 2023-12-25
+
+* Fix filter leak during Io drop
+
 ## [0.3.16] - 2023-12-14
 
 * Better io tags handling

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "0.3.16"
+version = "0.3.17"
 authors = ["ntex contributors <[email protected]>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]

ntex-io/src/dispatcher.rs

@@ -711,7 +711,7 @@ mod tests {
             let state = Io::new(io);
             let pool = state.memory_pool().pool();
             state.set_disconnect_timeout(cfg.disconnect_timeout());
-            state.set_tag("DBG: ");
+            state.set_tag("DBG");
 
             let flags = if cfg.keepalive_timeout_secs().is_zero() {
                 super::Flags::empty()
@@ -1187,6 +1187,7 @@ mod tests {
 
     #[ntex::test]
     async fn test_read_timeout() {
+        let _ = env_logger::init();
         let (client, server) = IoTest::create();
         client.remote_buffer_cap(1024);
 

ntex-io/src/io.rs

@@ -702,13 +702,17 @@ impl<F> Drop for Io<F> {
     fn drop(&mut self) {
         self.stop_timer();
 
-        if !self.0.flags().contains(Flags::IO_STOPPED) && self.1.is_set() {
+        if !self.0.flags().contains(Flags::IO_STOPPED) {
             log::trace!(
                 "{}: Io is dropped, force stopping io streams {:?}",
                 self.tag(),
                 self.0.flags()
             );
+        }
 
+        // filter must be dropped, it is unsafe
+        // and wont be dropped without special attention
+        if self.1.is_set() {
             self.force_close();
             self.1.drop_filter();
             self.0 .0.filter.set(NullFilter::get());

ntex-io/src/tasks.rs

@@ -62,7 +62,11 @@ impl ReadContext {
                                 inner.read_task.wake();
                             }
                         }
-                        log::trace!("{}: New {} bytes available, wakeup dispatcher", self.0.tag(), nbytes);
+                        log::trace!(
+                            "{}: New {} bytes available, wakeup dispatcher",
+                            self.0.tag(),
+                            nbytes
+                        );
                         inner.dispatch_task.wake();
                     } else {
                         if nbytes >= hw {

ntex-io/src/timer.rs

@@ -128,10 +128,10 @@ pub(crate) fn register(timeout: Seconds, io: &IoRef) -> TimerHandle {
                 loop {
                     sleep(SEC).await;
                     let stop = TIMER.with(|timer| {
-                        timer.current.set(timer.current.get() + 1);
+                        let current = timer.current.get() + 1;
+                        timer.current.set(current);
 
                         // notify io dispatcher
-                        let current = timer.current.get();
                         let mut inner = timer.storage.borrow_mut();
                         while let Some(key) = inner.notifications.keys().next() {
                             let key = *key;

ntex/Cargo.toml

@@ -58,7 +58,7 @@ ntex-util = "0.3.4"
 ntex-bytes = "0.1.21"
 ntex-h2 = "0.4.4"
 ntex-rt = "0.4.11"
-ntex-io = "0.3.16"
+ntex-io = "0.3.17"
 ntex-tls = "0.3.3"
 ntex-tokio = { version = "0.3.1", optional = true }
 ntex-glommio = { version = "0.3.1", optional = true }