diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index 5c91b01d..78f2d45f 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -12,7 +12,7 @@ jobs: with: toolchain: stable - run: - cargo check --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + cargo check --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" clippy: name: Clippy @@ -24,7 +24,7 @@ jobs: toolchain: stable components: clippy - run: - cargo clippy --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + cargo clippy --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" fmt: name: Rustfmt diff --git a/.github/workflows/cov.yml b/.github/workflows/cov.yml index 409549ca..c9f7a345 100644 --- a/.github/workflows/cov.yml +++ b/.github/workflows/cov.yml @@ -8,11 +8,6 @@ jobs: env: CARGO_TERM_COLOR: always steps: - - name: Free Disk Space - uses: jlumbroso/free-disk-space@main - with: - tool-cache: true - - uses: actions/checkout@v4 - name: Install Rust run: rustup update nightly @@ -26,17 +21,17 @@ jobs: - name: Clean coverage results run: cargo llvm-cov clean --workspace - - name: Code coverage (tokio) - run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" - - - name: Code coverage (compio) - run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" - - name: Code coverage (neon) - run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Code coverage (neon-uring) - run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" + + - name: Code coverage (tokio) + run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" + + - name: Code coverage (compio) + run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Generate coverage report run: cargo llvm-cov report --lcov --output-path lcov.info --ignore-filename-regex="ntex-compio|ntex-tokio" diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index e178d634..5297364c 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -16,11 +16,6 @@ jobs: runs-on: ubuntu-latest steps: - - name: Free Disk Space - uses: jlumbroso/free-disk-space@main - with: - tool-cache: true - - uses: actions/checkout@v4 - name: Install ${{ matrix.version }} @@ -44,25 +39,25 @@ jobs: path: ~/.cargo/git key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }} - - name: Run tests (tokio) - timeout-minutes: 40 - run: | - cargo test --all --no-fail-fast --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" - - - name: Run tests (compio) - timeout-minutes: 40 - run: | - cargo test --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" - - name: Run tests (neon) timeout-minutes: 40 run: | - cargo test --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + cargo test --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Run tests (neon-uring) timeout-minutes: 40 run: | - cargo test --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + cargo test --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" + + - name: Run tests (tokio) + timeout-minutes: 40 + run: | + cargo test --all --no-fail-fast --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" + + - name: Run tests (compio) + timeout-minutes: 40 + run: | + cargo test --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Install cargo-cache continue-on-error: true diff --git a/.github/workflows/osx.yml b/.github/workflows/osx.yml index 6b0477b7..a926dd34 100644 --- a/.github/workflows/osx.yml +++ b/.github/workflows/osx.yml @@ -37,16 +37,16 @@ jobs: path: ~/.cargo/git key: ${{ matrix.version }}-aarch64-apple-darwin-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }} + - name: Run tests (neon) + timeout-minutes: 40 + run: cargo test --all --no-default-features --no-fail-fast --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" + - name: Run tests (tokio) - run: cargo test --all --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + run: cargo test --all --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Run tests (compio) timeout-minutes: 40 - run: cargo test --all --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" - - - name: Run tests (neon) - timeout-minutes: 40 - run: cargo test --all --no-default-features --no-fail-fast --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" + run: cargo test --all --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" - name: Install cargo-cache continue-on-error: true diff --git a/.github/workflows/windows.yml b/.github/workflows/windows.yml index b42e0f00..8902aa8f 100644 --- a/.github/workflows/windows.yml +++ b/.github/workflows/windows.yml @@ -63,8 +63,8 @@ jobs: - name: Run tests (tokio) run: | - cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" -- --skip test_timer + cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" -- --skip test_timer - name: Run tests (compio) run: | - cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" -- --skip test_timer + cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" -- --skip test_timer diff --git a/Cargo.toml b/Cargo.toml index 871d9de2..d9e97ef4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,7 +46,10 @@ ntex-compio = { path = "ntex-compio" } ntex-tokio = { path = "ntex-tokio" } [workspace.dependencies] +async-channel = "2" async-task = "4.5.0" +atomic-waker = "1.1" +core_affinity = "0.8" bitflags = "2" cfg_aliases = "0.2.1" cfg-if = "1.0.0" @@ -57,7 +60,8 @@ fxhash = "0.2" libc = "0.2.164" log = "0.4" io-uring = "0.7.4" -polling = "3.3.0" +oneshot = "0.1" +polling = "3.7.4" nohash-hasher = "0.2.0" scoped-tls = "1.0.1" slab = "0.4.9" diff --git a/ntex-io/CHANGES.md b/ntex-io/CHANGES.md index ff7201c9..c109a752 100644 --- a/ntex-io/CHANGES.md +++ b/ntex-io/CHANGES.md @@ -1,5 +1,9 @@ # Changes +## [2.11.1] - 2025-03-20 + +* Add readiness check support + ## [2.11.0] - 2025-03-10 * Add single io context diff --git a/ntex-io/Cargo.toml b/ntex-io/Cargo.toml index 40d4ed20..f55aa5d0 100644 --- a/ntex-io/Cargo.toml +++ b/ntex-io/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex-io" -version = "2.11.0" +version = "2.11.1" authors = ["ntex contributors "] description = "Utilities for encoding and decoding frames" keywords = ["network", "framework", "async", "futures"] @@ -28,4 +28,3 @@ pin-project-lite = "0.2" [dev-dependencies] ntex = "2" rand = "0.8" -env_logger = "0.11" diff --git a/ntex-io/src/tasks.rs b/ntex-io/src/tasks.rs index 9a4d6f94..55f99416 100644 --- a/ntex-io/src/tasks.rs +++ b/ntex-io/src/tasks.rs @@ -537,7 +537,9 @@ impl IoContext { self.0.tag(), nbytes ); - inner.dispatch_task.wake(); + if !inner.dispatch_task.wake_checked() { + log::error!("Dispatcher waker is not registered"); + } } else { if nbytes >= hw { // read task is paused because of read back-pressure @@ -722,28 +724,20 @@ impl IoContext { } /// Get read buffer - pub fn with_read_buf(&self, f: F) -> Poll<()> - where - F: FnOnce(&mut BytesVec) -> Poll>, - { - let result = self.with_read_buf_inner(f); - + pub fn is_read_ready(&self) -> bool { // check read readiness - if result.is_pending() { - if let Some(waker) = self.0 .0.read_task.take() { - let mut cx = Context::from_waker(&waker); + if let Some(waker) = self.0 .0.read_task.take() { + let mut cx = Context::from_waker(&waker); - if let Poll::Ready(ReadStatus::Ready) = - self.0.filter().poll_read_ready(&mut cx) - { - return Poll::Pending; - } + if let Poll::Ready(ReadStatus::Ready) = self.0.filter().poll_read_ready(&mut cx) + { + return true; } } - result + false } - fn with_read_buf_inner(&self, f: F) -> Poll<()> + pub fn with_read_buf(&self, f: F) -> Poll<()> where F: FnOnce(&mut BytesVec) -> Poll>, { @@ -795,7 +789,9 @@ impl IoContext { self.0.tag(), nbytes ); - inner.dispatch_task.wake(); + if !inner.dispatch_task.wake_checked() { + log::error!("Dispatcher waker is not registered"); + } } else { if nbytes >= hw { // read task is paused because of read back-pressure @@ -838,33 +834,8 @@ impl IoContext { } } - pub fn with_write_buf(&self, f: F) -> Poll<()> - where - F: FnOnce(&BytesVec) -> Poll>, - { - let result = self.with_write_buf_inner(f); - - // check write readiness - if result.is_pending() { - let inner = &self.0 .0; - if let Some(waker) = inner.write_task.take() { - let ready = self - .0 - .filter() - .poll_write_ready(&mut Context::from_waker(&waker)); - if !matches!( - ready, - Poll::Ready(WriteStatus::Ready | WriteStatus::Shutdown) - ) { - return Poll::Ready(()); - } - } - } - result - } - /// Get write buffer - fn with_write_buf_inner(&self, f: F) -> Poll<()> + pub fn with_write_buf(&self, f: F) -> Poll<()> where F: FnOnce(&BytesVec) -> Poll>, { diff --git a/ntex-macros/Cargo.toml b/ntex-macros/Cargo.toml index f6cad0e2..a5bcf67d 100644 --- a/ntex-macros/Cargo.toml +++ b/ntex-macros/Cargo.toml @@ -18,4 +18,3 @@ proc-macro2 = "^1" [dev-dependencies] ntex = "2" futures = "0.3" -env_logger = "0.11" diff --git a/ntex-net/CHANGES.md b/ntex-net/CHANGES.md index 5495a535..e60744ef 100644 --- a/ntex-net/CHANGES.md +++ b/ntex-net/CHANGES.md @@ -1,5 +1,47 @@ # Changes +## [2.5.10] - 2025-03-28 + +* Better closed sockets handling + +## [2.5.9] - 2025-03-27 + +* Handle closed sockets + +## [2.5.8] - 2025-03-25 + +* Update neon runtime + +## [2.5.7] - 2025-03-21 + +* Simplify neon poll impl + +## [2.5.6] - 2025-03-20 + +* Redesign neon poll support + +## [2.5.5] - 2025-03-17 + +* Add check for required io-uring opcodes + +* Handle io-uring cancelation + +## [2.5.4] - 2025-03-15 + +* Close FD in various case for poll driver + +## [2.5.3] - 2025-03-14 + +* Fix operation cancelation handling for poll driver + +## [2.5.2] - 2025-03-14 + +* Fix operation cancelation handling for io-uring driver + +## [2.5.1] - 2025-03-14 + +* Fix socket connect for io-uring driver + ## [2.5.0] - 2025-03-12 * Add neon runtime support diff --git a/ntex-net/Cargo.toml b/ntex-net/Cargo.toml index 994659e6..5a72d3eb 100644 --- a/ntex-net/Cargo.toml +++ b/ntex-net/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex-net" -version = "2.5.0" +version = "2.5.10" authors = ["ntex contributors "] description = "ntexwork utils for ntex framework" keywords = ["network", "framework", "async", "futures"] @@ -27,27 +27,27 @@ compio = ["ntex-rt/compio", "ntex-compio"] # neon runtime neon = ["ntex-rt/neon", "ntex-neon", "slab", "socket2"] -polling = ["ntex-neon/polling", "dep:polling"] -io-uring = ["ntex-neon/io-uring", "dep:io-uring"] +polling = ["ntex-neon/polling", "dep:polling", "socket2"] +io-uring = ["ntex-neon/io-uring", "dep:io-uring", "socket2"] [dependencies] ntex-service = "3.3" ntex-bytes = "0.1" ntex-http = "0.1" -ntex-io = "2.11" +ntex-io = "2.11.1" ntex-rt = "0.4.25" 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.3", optional = true } +ntex-neon = { version = "0.1.15", optional = true } bitflags = { workspace = true } cfg-if = { workspace = true } log = { workspace = true } libc = { workspace = true } slab = { workspace = true, optional = true } -socket2 = { workspace = true, optional = true } +socket2 = { workspace = true, optional = true, features = ["all"] } thiserror = { workspace = true } # Linux specific dependencies @@ -57,4 +57,3 @@ polling = { workspace = true, optional = true } [dev-dependencies] ntex = "2" -env_logger = "0.11" diff --git a/ntex-net/src/connect/service.rs b/ntex-net/src/connect/service.rs index 4969b9fd..9e6a0549 100644 --- a/ntex-net/src/connect/service.rs +++ b/ntex-net/src/connect/service.rs @@ -197,7 +197,7 @@ impl Future for TcpConnectorResponse { Poll::Ready(Ok(sock)) => { let req = this.req.take().unwrap(); log::trace!( - "{}: TCP connector - successfully connected to connecting to {:?} - {:?}", + "{}: TCP connector - successfully connected to {:?} - {:?}", this.tag, req.host(), sock.query::().get() diff --git a/ntex-net/src/rt_polling/connect.rs b/ntex-net/src/rt_polling/connect.rs index 1d520421..8f0f1dc9 100644 --- a/ntex-net/src/rt_polling/connect.rs +++ b/ntex-net/src/rt_polling/connect.rs @@ -1,7 +1,7 @@ use std::os::fd::{AsRawFd, RawFd}; -use std::{cell::RefCell, collections::VecDeque, io, rc::Rc, task::Poll}; +use std::{cell::RefCell, io, rc::Rc, task::Poll}; -use ntex_neon::driver::{DriverApi, Handler, Interest}; +use ntex_neon::driver::{DriverApi, Event, Handler}; use ntex_neon::{syscall, Runtime}; use ntex_util::channel::oneshot::Sender; use slab::Slab; @@ -12,13 +12,11 @@ pub(crate) struct ConnectOps(Rc); #[derive(Debug)] enum Change { - Readable, - Writable, + Event(Event), Error(io::Error), } struct ConnectOpsBatcher { - feed: VecDeque<(usize, Change)>, inner: Rc, } @@ -42,10 +40,7 @@ impl ConnectOps { connects: RefCell::new(Slab::new()), }); inner = Some(ops.clone()); - Box::new(ConnectOpsBatcher { - inner: ops, - feed: VecDeque::new(), - }) + Box::new(ConnectOpsBatcher { inner: ops }) }); ConnectOps(inner.unwrap()) @@ -67,68 +62,50 @@ impl ConnectOps { let item = Item { fd, sender }; let id = self.0.connects.borrow_mut().insert(item); - self.0.api.register(fd, id, Interest::Writable); - + self.0.api.attach(fd, id as u32, Some(Event::writable(0))); Ok(id) } } impl Handler for ConnectOpsBatcher { - fn readable(&mut self, id: usize) { - log::debug!("ConnectFD is readable {:?}", id); - self.feed.push_back((id, Change::Readable)); - } - - fn writable(&mut self, id: usize) { - log::debug!("ConnectFD is writable {:?}", id); - self.feed.push_back((id, Change::Writable)); - } - - fn error(&mut self, id: usize, err: io::Error) { - self.feed.push_back((id, Change::Error(err))); - } - - fn commit(&mut self) { - if self.feed.is_empty() { - return; - } - log::debug!("Commit connect driver changes, num: {:?}", self.feed.len()); + fn event(&mut self, id: usize, event: Event) { + log::debug!("connect-fd is readable {:?}", id); let mut connects = self.inner.connects.borrow_mut(); - for (id, change) in self.feed.drain(..) { - if connects.contains(id) { - let item = connects.remove(id); - match change { - Change::Readable => unreachable!(), - Change::Writable => { - let mut err: libc::c_int = 0; - let mut err_len = - std::mem::size_of::() as libc::socklen_t; + if connects.contains(id) { + let item = connects.remove(id); + if event.writable { + let mut err: libc::c_int = 0; + let mut err_len = std::mem::size_of::() as libc::socklen_t; - let res = syscall!(libc::getsockopt( - item.fd.as_raw_fd(), - libc::SOL_SOCKET, - libc::SO_ERROR, - &mut err as *mut _ as *mut _, - &mut err_len - )); + let res = syscall!(libc::getsockopt( + item.fd.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_ERROR, + &mut err as *mut _ as *mut _, + &mut err_len + )); - let res = if err == 0 { - res.map(|_| ()) - } else { - Err(io::Error::from_raw_os_error(err)) - }; + let res = if err == 0 { + res.map(|_| ()) + } else { + Err(io::Error::from_raw_os_error(err)) + }; - self.inner.api.unregister_all(item.fd); - let _ = item.sender.send(res); - } - Change::Error(err) => { - let _ = item.sender.send(Err(err)); - self.inner.api.unregister_all(item.fd); - } - } + self.inner.api.detach(item.fd, id as u32); + let _ = item.sender.send(res); } } } + + fn error(&mut self, id: usize, err: io::Error) { + let mut connects = self.inner.connects.borrow_mut(); + + if connects.contains(id) { + let item = connects.remove(id); + let _ = item.sender.send(Err(err)); + self.inner.api.detach(item.fd, id as u32); + } + } } diff --git a/ntex-net/src/rt_polling/driver.rs b/ntex-net/src/rt_polling/driver.rs index becc9aeb..24db553d 100644 --- a/ntex-net/src/rt_polling/driver.rs +++ b/ntex-net/src/rt_polling/driver.rs @@ -1,55 +1,53 @@ use std::os::fd::{AsRawFd, RawFd}; -use std::{cell::Cell, collections::VecDeque, 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, Handler, Interest}; +use ntex_neon::driver::{DriverApi, Event, Handler}; use ntex_neon::{syscall, Runtime}; use slab::Slab; use ntex_bytes::BufMut; use ntex_io::IoContext; -bitflags::bitflags! { - #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] - struct Flags: u8 { - const ERROR = 0b0000_0001; - const RD = 0b0000_0010; - const WR = 0b0000_0100; - } +pub(crate) struct StreamCtl { + id: u32, + inner: Rc>, } -pub(crate) struct StreamCtl { - id: usize, - inner: Rc>, +bitflags::bitflags! { + #[derive(Copy, Clone, Debug)] + struct Flags: u8 { + const RD = 0b0000_0001; + const WR = 0b0000_0010; + } } struct StreamItem { io: Option, fd: RawFd, - context: IoContext, flags: Flags, - ref_count: usize, + ref_count: u16, + context: IoContext, } pub(crate) struct StreamOps(Rc>); -#[derive(Debug)] -enum Change { - Readable, - Writable, - Error(io::Error), -} - struct StreamOpsHandler { - feed: VecDeque<(usize, Change)>, inner: Rc>, } struct StreamOpsInner { api: DriverApi, - feed: Cell>>, + delayd_drop: Cell, + feed: RefCell>, streams: Cell>>>>, } +impl StreamItem { + fn tag(&self) -> &'static str { + self.context.tag() + } +} + impl StreamOps { pub(crate) fn current() -> Self { Runtime::value(|rt| { @@ -57,14 +55,12 @@ impl StreamOps { rt.driver().register(|api| { let ops = Rc::new(StreamOpsInner { api, - feed: Cell::new(Some(VecDeque::new())), + feed: RefCell::new(Vec::new()), + delayd_drop: Cell::new(false), streams: Cell::new(Some(Box::new(Slab::new()))), }); inner = Some(ops.clone()); - Box::new(StreamOpsHandler { - inner: ops, - feed: VecDeque::new(), - }) + Box::new(StreamOpsHandler { inner: ops }) }); StreamOps(inner.unwrap()) @@ -72,30 +68,27 @@ impl StreamOps { } pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl { - let item = StreamItem { - context, - fd: io.as_raw_fd(), - io: Some(io), - flags: Flags::empty(), - ref_count: 1, - }; - self.with(|streams| { - let id = streams.insert(item); + let fd = io.as_raw_fd(); + let stream = self.0.with(move |streams| { + let item = StreamItem { + fd, + context, + io: Some(io), + ref_count: 1, + flags: Flags::empty(), + }; StreamCtl { - id, + id: streams.insert(item) as u32, inner: self.0.clone(), } - }) - } + }); - fn with(&self, f: F) -> R - where - F: FnOnce(&mut Slab>) -> R, - { - let mut inner = self.0.streams.take().unwrap(); - let result = f(&mut inner); - self.0.streams.set(Some(inner)); - result + self.0.api.attach( + fd, + stream.id, + Some(Event::new(0, false, false).with_interrupt()), + ); + stream } } @@ -106,235 +99,242 @@ impl Clone for StreamOps { } impl Handler for StreamOpsHandler { - fn readable(&mut self, id: usize) { - log::debug!("FD is readable {:?}", id); - self.feed.push_back((id, Change::Readable)); - } + fn event(&mut self, id: usize, ev: Event) { + self.inner.with(|streams| { + if !streams.contains(id) { + return; + } + let item = &mut streams[id]; + if item.io.is_none() { + return; + } + log::debug!("{}: FD event {:?} event: {:?}", item.tag(), id, ev); - fn writable(&mut self, id: usize) { - log::debug!("FD is writable {:?}", id); - self.feed.push_back((id, Change::Writable)); + // handle HUP + if ev.is_interrupt() { + item.context.stopped(None); + close(id as u32, item, &self.inner.api, None, true); + return; + } + + let mut renew_ev = Event::new(0, false, false).with_interrupt(); + + if ev.readable { + let res = item.context.with_read_buf(|buf| { + let chunk = buf.chunk_mut(); + let result = task::ready!(syscall!( + break libc::read(item.fd, chunk.as_mut_ptr() as _, chunk.len()) + )); + if let Ok(size) = result { + log::debug!("{}: data {:?}, s: {:?}", item.tag(), item.fd, size); + unsafe { buf.advance_mut(size) }; + } + Poll::Ready(result) + }); + + if res.is_pending() && item.context.is_read_ready() { + renew_ev.readable = true; + item.flags.insert(Flags::RD); + } else { + item.flags.remove(Flags::RD); + } + } else if item.flags.contains(Flags::RD) { + renew_ev.readable = true; + } + + if ev.writable { + let result = item.context.with_write_buf(|buf| { + log::debug!("{}: write {:?} s: {:?}", item.tag(), item.fd, buf.len()); + syscall!(break libc::write(item.fd, buf[..].as_ptr() as _, buf.len())) + }); + if result.is_pending() { + renew_ev.writable = true; + item.flags.insert(Flags::WR); + } else { + item.flags.remove(Flags::WR); + } + } else if item.flags.contains(Flags::WR) { + renew_ev.writable = true; + } + + self.inner.api.modify(item.fd, id as u32, renew_ev); + + // delayed drops + if self.inner.delayd_drop.get() { + for id in self.inner.feed.borrow_mut().drain(..) { + let item = &mut streams[id as usize]; + item.ref_count -= 1; + if item.ref_count == 0 { + let mut item = streams.remove(id as usize); + log::debug!( + "{}: Drop ({}), {:?}, has-io: {}", + item.tag(), + id, + item.fd, + item.io.is_some() + ); + close(id, &mut item, &self.inner.api, None, true); + } + } + self.inner.delayd_drop.set(false); + } + }); } fn error(&mut self, id: usize, err: io::Error) { - log::debug!("FD is failed {:?}, err: {:?}", id, err); - self.feed.push_back((id, Change::Error(err))); + self.inner.with(|streams| { + if let Some(item) = streams.get_mut(id) { + log::debug!( + "{}: FD is failed ({}) {:?}, err: {:?}", + item.tag(), + id, + item.fd, + err + ); + close(id as u32, item, &self.inner.api, Some(err), false); + } + }) } +} - fn commit(&mut self) { - if self.feed.is_empty() { - return; +impl StreamOpsInner { + fn with(&self, f: F) -> R + where + F: FnOnce(&mut Slab>) -> R, + { + let mut streams = self.streams.take().unwrap(); + let result = f(&mut streams); + self.streams.set(Some(streams)); + result + } +} + +fn close( + id: u32, + item: &mut StreamItem, + api: &DriverApi, + 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)); } - log::debug!("Commit changes, num: {:?}", self.feed.len()); - - let mut streams = self.inner.streams.take().unwrap(); - - for (id, change) in self.feed.drain(..) { - match change { - Change::Readable => { - let item = &mut streams[id]; - let result = item.context.with_read_buf(|buf| { - let chunk = buf.chunk_mut(); - let b = chunk.as_mut_ptr(); - Poll::Ready( - task::ready!(syscall!( - break libc::read(item.fd, b as _, chunk.len()) - )) - .inspect(|size| { - unsafe { buf.advance_mut(*size) }; - log::debug!( - "{}: {:?}, SIZE: {:?}, BUF: {:?}", - item.context.tag(), - item.fd, - size, - buf - ); - }), - ) - }); - - if result.is_pending() { - item.flags.insert(Flags::RD); - self.inner.api.register(item.fd, id, Interest::Readable); - } - } - Change::Writable => { - let item = &mut streams[id]; - let result = item.context.with_write_buf(|buf| { - let slice = &buf[..]; - syscall!( - break libc::write(item.fd, slice.as_ptr() as _, slice.len()) - ) - }); - - if result.is_pending() { - item.flags.insert(Flags::WR); - self.inner.api.register(item.fd, id, Interest::Writable); - } - } - Change::Error(err) => { - if let Some(item) = streams.get_mut(id) { - item.context.stopped(Some(err)); - if !item.flags.contains(Flags::ERROR) { - item.flags.insert(Flags::ERROR); - item.flags.remove(Flags::RD | Flags::WR); - self.inner.api.unregister_all(item.fd); - } - } - } + let fd = item.fd; + api.detach(fd, id); + Some(ntex_rt::spawn_blocking(move || { + if shutdown { + let _ = syscall!(libc::shutdown(fd, libc::SHUT_RDWR)); } - } - - // extra - let mut feed = self.inner.feed.take().unwrap(); - for id in feed.drain(..) { - let item = &mut streams[id]; - log::debug!("{}: Drop io ({}), {:?}", item.context.tag(), id, item.fd); - - item.ref_count -= 1; - if item.ref_count == 0 { - let item = streams.remove(id); - if item.io.is_some() { - self.inner.api.unregister_all(item.fd); - } - } - } - - self.inner.feed.set(Some(feed)); - self.inner.streams.set(Some(streams)); + syscall!(libc::close(fd)) + })) + } else { + None } } impl StreamCtl { - pub(crate) async fn close(self) -> io::Result<()> { - let (io, fd) = - self.with(|streams| (streams[self.id].io.take(), streams[self.id].fd)); - if let Some(io) = io { - std::mem::forget(io); - - ntex_rt::spawn_blocking(move || syscall!(libc::close(fd))) - .await - .map_err(|e| io::Error::new(io::ErrorKind::Other, e)) - .and_then(crate::helpers::pool_io_err)?; + pub(crate) fn close(self) -> impl Future> { + let id = self.id as usize; + let fut = self.inner.with(|streams| { + let item = &mut streams[id]; + close(self.id, item, &self.inner.api, None, false) + }); + async move { + if let Some(fut) = fut { + fut.await + .map_err(|e| io::Error::new(io::ErrorKind::Other, e)) + .and_then(crate::helpers::pool_io_err)?; + } + Ok(()) } - Ok(()) } pub(crate) fn with_io(&self, f: F) -> R where F: FnOnce(Option<&T>) -> R, { - self.with(|streams| f(streams[self.id].io.as_ref())) + self.inner + .with(|streams| f(streams[self.id as usize].io.as_ref())) } - pub(crate) fn pause_all(&self) { - self.with(|streams| { - let item = &mut streams[self.id]; - - if item.flags.intersects(Flags::RD | Flags::WR) { - log::debug!( - "{}: Pause all io ({}), {:?}", - item.context.tag(), - self.id, - item.fd - ); - item.flags.remove(Flags::RD | Flags::WR); - self.inner.api.unregister_all(item.fd); - } - }) - } - - pub(crate) fn pause_read(&self) { - self.with(|streams| { - let item = &mut streams[self.id]; + pub(crate) fn modify(&self, rd: bool, wr: bool) { + self.inner.with(|streams| { + let item = &mut streams[self.id as usize]; log::debug!( - "{}: Pause io read ({}), {:?}", - item.context.tag(), + "{}: Modify interest ({}), {:?} rd: {:?}, wr: {:?}", + item.tag(), self.id, - item.fd + item.fd, + rd, + wr ); - if item.flags.contains(Flags::RD) { - item.flags.remove(Flags::RD); - self.inner.api.unregister(item.fd, Interest::Readable); - } - }) - } - pub(crate) fn resume_read(&self) { - self.with(|streams| { - let item = &mut streams[self.id]; + let mut event = Event::new(0, false, false).with_interrupt(); - log::debug!( - "{}: Resume io read ({}), {:?}", - item.context.tag(), - self.id, - item.fd - ); - if !item.flags.contains(Flags::RD) { - item.flags.insert(Flags::RD); - self.inner - .api - .register(item.fd, self.id, Interest::Readable); - } - }) - } + if rd { + if item.flags.contains(Flags::RD) { + event.readable = true; + } else { + let res = item.context.with_read_buf(|buf| { + let chunk = buf.chunk_mut(); + let result = task::ready!(syscall!( + break libc::read(item.fd, chunk.as_mut_ptr() as _, chunk.len()) + )); + if let Ok(size) = result { + log::debug!( + "{}: read {:?}, s: {:?}", + item.tag(), + item.fd, + size + ); + unsafe { buf.advance_mut(size) }; + } + Poll::Ready(result) + }); - pub(crate) fn resume_write(&self) { - self.with(|streams| { - let item = &mut streams[self.id]; - - if !item.flags.contains(Flags::WR) { - log::debug!( - "{}: Resume io write ({}), {:?}", - item.context.tag(), - self.id, - item.fd - ); - let result = item.context.with_write_buf(|buf| { - log::debug!( - "{}: Writing io ({}), buf: {:?}", - item.context.tag(), - self.id, - buf.len() - ); - - let slice = &buf[..]; - syscall!(break libc::write(item.fd, slice.as_ptr() as _, slice.len())) - }); - - if result.is_pending() { - log::debug!( - "{}: Write is pending ({}), {:?}", - item.context.tag(), - self.id, - item.context.flags() - ); - - item.flags.insert(Flags::WR); - self.inner - .api - .register(item.fd, self.id, Interest::Writable); + if res.is_pending() && item.context.is_read_ready() { + event.readable = true; + item.flags.insert(Flags::RD); + } } } - }) - } - fn with(&self, f: F) -> R - where - F: FnOnce(&mut Slab>) -> R, - { - let mut inner = self.inner.streams.take().unwrap(); - let result = f(&mut inner); - self.inner.streams.set(Some(inner)); - result + if wr { + if item.flags.contains(Flags::WR) { + event.writable = true; + } else { + let result = item.context.with_write_buf(|buf| { + log::debug!( + "{}: Writing ({}), buf: {:?}", + item.tag(), + self.id, + buf.len() + ); + syscall!( + break libc::write(item.fd, buf[..].as_ptr() as _, buf.len()) + ) + }); + + if result.is_pending() { + event.writable = true; + item.flags.insert(Flags::WR); + } + } + } + + self.inner.api.modify(item.fd, self.id, event); + }) } } impl Clone for StreamCtl { fn clone(&self) -> Self { - self.with(|streams| { - streams[self.id].ref_count += 1; + self.inner.with(|streams| { + streams[self.id as usize].ref_count += 1; Self { id: self.id, inner: self.inner.clone(), @@ -346,25 +346,23 @@ impl Clone for StreamCtl { impl Drop for StreamCtl { fn drop(&mut self) { if let Some(mut streams) = self.inner.streams.take() { - log::debug!( - "{}: Drop io ({}), {:?}", - streams[self.id].context.tag(), - self.id, - streams[self.id].fd - ); - - streams[self.id].ref_count -= 1; - if streams[self.id].ref_count == 0 { - let item = streams.remove(self.id); - if item.io.is_some() { - self.inner.api.unregister_all(item.fd); - } + let id = self.id as usize; + streams[id].ref_count -= 1; + if streams[id].ref_count == 0 { + let mut item = streams.remove(id); + log::debug!( + "{}: Drop io ({}), {:?}, has-io: {}", + item.tag(), + self.id, + item.fd, + item.io.is_some() + ); + close(self.id, &mut item, &self.inner.api, None, true); } self.inner.streams.set(Some(streams)); } else { - let mut feed = self.inner.feed.take().unwrap(); - feed.push_back(self.id); - self.inner.feed.set(Some(feed)); + self.inner.delayd_drop.set(true); + self.inner.feed.borrow_mut().push(self.id); } } } diff --git a/ntex-net/src/rt_polling/io.rs b/ntex-net/src/rt_polling/io.rs index 4b7a7b1a..990dae8f 100644 --- a/ntex-net/src/rt_polling/io.rs +++ b/ntex-net/src/rt_polling/io.rs @@ -54,21 +54,26 @@ enum Status { async fn run(ctl: StreamCtl, context: IoContext) { // Handle io read readiness let st = poll_fn(|cx| { + let mut modify = false; + let mut readable = false; + let mut writable = false; let read = match context.poll_read_ready(cx) { Poll::Ready(ReadStatus::Ready) => { - ctl.resume_read(); + modify = true; + readable = true; Poll::Pending } Poll::Ready(ReadStatus::Terminate) => Poll::Ready(()), Poll::Pending => { - ctl.pause_read(); + modify = true; Poll::Pending } }; let write = match context.poll_write_ready(cx) { Poll::Ready(WriteStatus::Ready) => { - ctl.resume_write(); + modify = true; + writable = true; Poll::Pending } Poll::Ready(WriteStatus::Shutdown) => Poll::Ready(Status::Shutdown), @@ -76,6 +81,10 @@ async fn run(ctl: StreamCtl, context: IoContext) { Poll::Pending => Poll::Pending, }; + if modify { + ctl.modify(readable, writable); + } + if read.is_pending() && write.is_pending() { Poll::Pending } else if write.is_ready() { @@ -86,11 +95,7 @@ async fn run(ctl: StreamCtl, context: IoContext) { }) .await; - ctl.resume_write(); + ctl.modify(false, true); context.shutdown(st == Status::Shutdown).await; - - ctl.pause_all(); - let result = ctl.close().await; - - context.stopped(result.err()); + context.stopped(ctl.close().await.err()); } diff --git a/ntex-net/src/rt_polling/mod.rs b/ntex-net/src/rt_polling/mod.rs index 671b8493..b4fb928b 100644 --- a/ntex-net/src/rt_polling/mod.rs +++ b/ntex-net/src/rt_polling/mod.rs @@ -8,6 +8,9 @@ pub(crate) mod connect; mod driver; mod io; +#[cfg(not(target_pointer_width = "64"))] +compile_error!("Only 64bit platforms are supported"); + /// Tcp stream wrapper for neon TcpStream struct TcpStream(socket2::Socket); diff --git a/ntex-net/src/rt_uring/connect.rs b/ntex-net/src/rt_uring/connect.rs index 1d7e16a7..ea9be3e1 100644 --- a/ntex-net/src/rt_uring/connect.rs +++ b/ntex-net/src/rt_uring/connect.rs @@ -20,9 +20,11 @@ struct ConnectOpsHandler { inner: Rc, } +type Operations = RefCell, Sender>)>>; + struct ConnectOpsInner { api: DriverApi, - ops: RefCell>>>, + ops: Operations, } impl ConnectOps { @@ -30,6 +32,10 @@ impl ConnectOps { Runtime::value(|rt| { let mut inner = None; rt.driver().register(|api| { + if !api.is_supported(opcode::Connect::CODE) { + panic!("opcode::Connect is required for io-uring support"); + } + let ops = Rc::new(ConnectOpsInner { api, ops: RefCell::new(Slab::new()), @@ -47,10 +53,17 @@ impl ConnectOps { addr: SockAddr, sender: Sender>, ) -> io::Result<()> { - let id = self.0.ops.borrow_mut().insert(sender); + let addr2 = addr.clone(); + let mut ops = self.0.ops.borrow_mut(); + + // addr must be stable, neon submits ops at the end of rt turn + let addr = Box::new(addr); + let (addr_ptr, addr_len) = (addr.as_ref().as_ptr(), addr.len()); + + let id = ops.insert((addr, sender)); self.0.api.submit( id as u32, - opcode::Connect::new(Fd(fd), addr.as_ptr(), addr.len()).build(), + opcode::Connect::new(Fd(fd), addr_ptr, addr_len).build(), ); Ok(()) @@ -59,15 +72,20 @@ impl ConnectOps { impl Handler for ConnectOpsHandler { fn canceled(&mut self, user_data: usize) { - log::debug!("Op is canceled {:?}", user_data); + log::debug!("connect-op is canceled {:?}", user_data); self.inner.ops.borrow_mut().remove(user_data); } fn completed(&mut self, user_data: usize, flags: u32, result: io::Result) { - log::debug!("Op is completed {:?} result: {:?}", user_data, result); + let (addr, tx) = self.inner.ops.borrow_mut().remove(user_data); + log::debug!( + "connect-op is completed {:?} result: {:?}, addr: {:?}", + user_data, + result, + addr.as_socket() + ); - let tx = self.inner.ops.borrow_mut().remove(user_data); let _ = tx.send(result.map(|_| ())); } } diff --git a/ntex-net/src/rt_uring/driver.rs b/ntex-net/src/rt_uring/driver.rs index c51e64ea..d39d69e8 100644 --- a/ntex-net/src/rt_uring/driver.rs +++ b/ntex-net/src/rt_uring/driver.rs @@ -1,4 +1,4 @@ -use std::{cell::RefCell, fmt, io, mem, num::NonZeroU32, os, rc::Rc, task::Poll}; +use std::{cell::RefCell, io, mem, num::NonZeroU32, os, rc::Rc, task::Poll}; use io_uring::{opcode, squeue::Entry, types::Fd}; use ntex_neon::{driver::DriverApi, driver::Handler, Runtime}; @@ -13,15 +13,32 @@ pub(crate) struct StreamCtl { inner: Rc>, } +bitflags::bitflags! { + #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] + struct Flags: u8 { + const RD_CANCELING = 0b0000_0001; + const RD_REISSUE = 0b0000_0010; + const WR_CANCELING = 0b0001_0000; + const WR_REISSUE = 0b0010_0000; + } +} + struct StreamItem { io: Option, fd: Fd, context: IoContext, ref_count: usize, + flags: Flags, rd_op: Option, wr_op: Option, } +impl StreamItem { + fn tag(&self) -> &'static str { + self.context.tag() + } +} + enum Operation { Recv { id: usize, @@ -61,6 +78,16 @@ impl StreamOps { Runtime::value(|rt| { let mut inner = None; rt.driver().register(|api| { + if !api.is_supported(opcode::Recv::CODE) { + panic!("opcode::Recv is required for io-uring support"); + } + if !api.is_supported(opcode::Send::CODE) { + panic!("opcode::Send is required for io-uring support"); + } + if !api.is_supported(opcode::Close::CODE) { + panic!("opcode::Close is required for io-uring support"); + } + let mut ops = Slab::new(); ops.insert(Operation::Nop); @@ -88,6 +115,7 @@ impl StreamOps { ref_count: 1, rd_op: None, wr_op: None, + flags: Flags::empty(), }; let id = self.0.storage.borrow_mut().streams.insert(item); StreamCtl { @@ -116,12 +144,36 @@ impl Handler for StreamOpsHandler { match storage.ops.remove(user_data) { Operation::Recv { id, buf, context } => { - log::debug!("{}: Recv canceled {:?}", context.tag(), id,); + log::debug!("{}: Recv canceled {:?}", context.tag(), id); context.release_read_buf(buf); + if let Some(item) = storage.streams.get_mut(id) { + item.rd_op.take(); + item.flags.remove(Flags::RD_CANCELING); + if item.flags.contains(Flags::RD_REISSUE) { + item.flags.remove(Flags::RD_REISSUE); + + let result = storage.recv(id, Some(context)); + if let Some((id, op)) = result { + self.inner.api.submit(id, op); + } + } + } } Operation::Send { id, buf, context } => { log::debug!("{}: Send canceled: {:?}", context.tag(), id); context.release_write_buf(buf); + if let Some(item) = storage.streams.get_mut(id) { + item.wr_op.take(); + item.flags.remove(Flags::WR_CANCELING); + if item.flags.contains(Flags::WR_REISSUE) { + item.flags.remove(Flags::WR_REISSUE); + + let result = storage.send(id, Some(context)); + if let Some((id, op)) = result { + self.inner.api.submit(id, op); + } + } + } } Operation::Nop | Operation::Close { .. } => {} } @@ -145,12 +197,11 @@ impl Handler for StreamOpsHandler { // reset op reference if let Some(item) = storage.streams.get_mut(id) { log::debug!( - "{}: Recv completed {:?}, res: {:?}, buf({}): {:?}", + "{}: Recv completed {:?}, res: {:?}, buf({})", context.tag(), item.fd, result, - buf.remaining_mut(), - buf, + buf.remaining_mut() ); item.rd_op.take(); } @@ -167,21 +218,24 @@ impl Handler for StreamOpsHandler { } Operation::Send { id, buf, context } => { // reset op reference - if let Some(item) = storage.streams.get_mut(id) { + let fd = if let Some(item) = storage.streams.get_mut(id) { log::debug!( - "{}: Send completed: {:?}, res: {:?}", + "{}: Send completed: {:?}, res: {:?}, buf({})", context.tag(), item.fd, - result + result, + buf.len() ); item.wr_op.take(); - } + Some(item.fd) + } else { + None + }; // set read buf - if context - .set_write_buf(result.map(|size| size as usize), buf) - .is_pending() - { + let result = context.set_write_buf(result.map(|size| size as usize), buf); + if result.is_pending() { + log::debug!("{}: Need to send more: {:?}", context.tag(), fd); if let Some((id, op)) = storage.send(id, Some(context)) { self.inner.api.submit(id, op); } @@ -201,7 +255,7 @@ impl Handler for StreamOpsHandler { if storage.streams[id].ref_count == 0 { let mut item = storage.streams.remove(id); - log::debug!("{}: Drop io ({}), {:?}", item.context.tag(), id, item.fd); + log::debug!("{}: Drop io ({}), {:?}", item.tag(), id, item.fd); if let Some(io) = item.io.take() { mem::forget(io); @@ -224,11 +278,10 @@ impl StreamOpsStorage { if item.rd_op.is_none() { if let Poll::Ready(mut buf) = item.context.get_read_buf() { log::debug!( - "{}: Recv resume ({}), {:?} - {:?} = {:?}", - item.context.tag(), + "{}: Recv resume ({}), {:?} rem: {:?}", + item.tag(), id, item.fd, - buf, buf.remaining_mut() ); @@ -246,6 +299,8 @@ impl StreamOpsStorage { item.rd_op = NonZeroU32::new(op_id as u32); return Some((op_id as u32, op)); } + } else if item.flags.contains(Flags::RD_CANCELING) { + item.flags.insert(Flags::RD_REISSUE); } None } @@ -256,11 +311,11 @@ impl StreamOpsStorage { if item.wr_op.is_none() { if let Poll::Ready(buf) = item.context.get_write_buf() { log::debug!( - "{}: Send resume ({}), {:?} {:?}", - item.context.tag(), + "{}: Send resume ({}), {:?} len: {:?}", + item.tag(), id, item.fd, - buf + buf.len() ); let slice = buf.chunk(); @@ -277,6 +332,8 @@ impl StreamOpsStorage { item.wr_op = NonZeroU32::new(op_id as u32); return Some((op_id as u32, op)); } + } else if item.flags.contains(Flags::WR_CANCELING) { + item.flags.insert(Flags::WR_REISSUE); } None } @@ -344,13 +401,11 @@ impl StreamCtl { let item = &mut storage.streams[self.id]; if let Some(rd_op) = item.rd_op { - log::debug!( - "{}: Recv to pause ({}), {:?}", - item.context.tag(), - self.id, - item.fd - ); - self.inner.api.cancel(rd_op.get()); + if !item.flags.contains(Flags::RD_CANCELING) { + log::debug!("{}: Recv to pause ({}), {:?}", item.tag(), self.id, item.fd); + item.flags.insert(Flags::RD_CANCELING); + self.inner.api.cancel(rd_op.get()); + } } } } @@ -372,12 +427,7 @@ impl Drop for StreamCtl { if storage.streams[self.id].ref_count == 0 { let mut item = storage.streams.remove(self.id); if let Some(io) = item.io.take() { - log::debug!( - "{}: Close io ({}), {:?}", - item.context.tag(), - self.id, - item.fd - ); + log::debug!("{}: Close io ({}), {:?}", item.tag(), self.id, item.fd); mem::forget(io); let id = storage.ops.insert(Operation::Close { tx: None }); @@ -392,20 +442,3 @@ impl Drop for StreamCtl { } } } - -impl PartialEq for StreamCtl { - #[inline] - fn eq(&self, other: &StreamCtl) -> bool { - self.id == other.id && std::ptr::eq(&self.inner, &other.inner) - } -} - -impl fmt::Debug for StreamCtl { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let storage = self.inner.storage.borrow(); - f.debug_struct("StreamCtl") - .field("id", &self.id) - .field("io", &storage.streams[self.id].io) - .finish() - } -} diff --git a/ntex-net/src/rt_uring/io.rs b/ntex-net/src/rt_uring/io.rs index 2fb23a00..2f111ad7 100644 --- a/ntex-net/src/rt_uring/io.rs +++ b/ntex-net/src/rt_uring/io.rs @@ -68,7 +68,6 @@ async fn run(ctl: StreamCtl, context: IoContext) { let write = match context.poll_write_ready(cx) { Poll::Ready(WriteStatus::Ready) => { - log::debug!("{}: write ready", context.tag()); ctl.resume_write(); Poll::Pending } diff --git a/ntex-rt/CHANGES.md b/ntex-rt/CHANGES.md index f2ab4736..2afd5bd6 100644 --- a/ntex-rt/CHANGES.md +++ b/ntex-rt/CHANGES.md @@ -1,5 +1,9 @@ # Changes +## [0.4.29] - 2025-03-26 + +* Add Arbiter::get_value() helper method + ## [0.4.27] - 2025-03-14 * Add srbiters pings ttl diff --git a/ntex-rt/Cargo.toml b/ntex-rt/Cargo.toml index 0526e450..a5966d76 100644 --- a/ntex-rt/Cargo.toml +++ b/ntex-rt/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex-rt" -version = "0.4.28" +version = "0.4.29" authors = ["ntex contributors "] description = "ntex runtime" keywords = ["network", "framework", "async", "futures"] @@ -32,8 +32,8 @@ neon = ["ntex-neon"] [dependencies] async-channel = "2" futures-timer = "3.0" -log = "0.4" oneshot = "0.1" +log = "0.4" compio-driver = { version = "0.6", optional = true } compio-runtime = { version = "0.6", optional = true } @@ -42,7 +42,4 @@ tok-io = { version = "1", package = "tokio", default-features = false, features "net", ], optional = true } -ntex-neon = { version = "0.1.1", optional = true } - -[dev-dependencies] -env_logger = "0.11" +ntex-neon = { version = "0.1.14", optional = true } diff --git a/ntex-rt/src/arbiter.rs b/ntex-rt/src/arbiter.rs index 48a673ca..e20ab282 100644 --- a/ntex-rt/src/arbiter.rs +++ b/ntex-rt/src/arbiter.rs @@ -286,6 +286,25 @@ impl Arbiter { }) } + /// Get a type previously inserted to this runtime or create new one. + pub fn get_value(f: F) -> T + where + T: Clone + 'static, + F: FnOnce() -> T, + { + STORAGE.with(move |cell| { + let mut st = cell.borrow_mut(); + if let Some(boxed) = st.get(&TypeId::of::()) { + if let Some(val) = (&**boxed as &(dyn Any + 'static)).downcast_ref::() { + return val.clone(); + } + } + let val = f(); + st.insert(TypeId::of::(), Box::new(val.clone())); + val + }) + } + /// Wait for the event loop to stop by joining the underlying thread (if have Some). pub fn join(&mut self) -> thread::Result<()> { if let Some(thread_handle) = self.thread_handle.take() { @@ -355,6 +374,7 @@ mod tests { assert!(Arbiter::get_item::<&'static str, _, _>(|s| *s == "test")); assert!(Arbiter::get_mut_item::<&'static str, _, _>(|s| *s == "test")); assert!(Arbiter::contains_item::<&'static str>()); + assert!(Arbiter::get_value(|| 64u64) == 64); assert!(format!("{:?}", Arbiter::current()).contains("Arbiter")); } } diff --git a/ntex-rt/src/lib.rs b/ntex-rt/src/lib.rs index 1ffd7fe7..d5d85546 100644 --- a/ntex-rt/src/lib.rs +++ b/ntex-rt/src/lib.rs @@ -112,6 +112,8 @@ mod tokio { /// /// This function panics if ntex system is not running. #[inline] + #[doc(hidden)] + #[deprecated] pub fn spawn_fn(f: F) -> tok_io::task::JoinHandle where F: FnOnce() -> R + 'static, @@ -196,6 +198,8 @@ mod compio { /// /// This function panics if ntex system is not running. #[inline] + #[doc(hidden)] + #[deprecated] pub fn spawn_fn(f: F) -> JoinHandle where F: FnOnce() -> R + 'static, @@ -323,6 +327,8 @@ mod neon { /// /// This function panics if ntex system is not running. #[inline] + #[doc(hidden)] + #[deprecated] pub fn spawn_fn(f: F) -> Task where F: FnOnce() -> R + 'static, @@ -377,7 +383,7 @@ mod neon { impl JoinHandle { pub fn is_finished(&self) -> bool { - false + self.fut.is_none() } } diff --git a/ntex-server/CHANGES.md b/ntex-server/CHANGES.md index 7f1d8302..546a92ff 100644 --- a/ntex-server/CHANGES.md +++ b/ntex-server/CHANGES.md @@ -1,5 +1,13 @@ # Changes +## [2.7.3] - 2025-03-28 + +* Better worker availability handling + +## [2.7.2] - 2025-03-27 + +* Handle paused state + ## [2.7.1] - 2025-02-28 * Fix set core affinity out of worker start #508 diff --git a/ntex-server/Cargo.toml b/ntex-server/Cargo.toml index ed962fc0..a88be635 100644 --- a/ntex-server/Cargo.toml +++ b/ntex-server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex-server" -version = "2.7.1" +version = "2.7.4" authors = ["ntex contributors "] description = "Server for ntex framework" keywords = ["network", "framework", "async", "futures"] @@ -22,13 +22,13 @@ ntex-service = "3.4" ntex-rt = "0.4" ntex-util = "2.8" -async-channel = "2" -async-broadcast = "0.7" -core_affinity = "0.8" -polling = "3.3" -log = "0.4" -socket2 = "0.5" -oneshot = { version = "0.1", default-features = false, features = ["async"] } +async-channel = { workspace = true } +atomic-waker = { workspace = true } +core_affinity = { workspace = true } +oneshot = { workspace = true } +polling = { workspace = true } +log = { workspace = true } +socket2 = { workspace = true } [dev-dependencies] ntex = "2" diff --git a/ntex-server/src/manager.rs b/ntex-server/src/manager.rs index f0be9c40..9d0bfe8d 100644 --- a/ntex-server/src/manager.rs +++ b/ntex-server/src/manager.rs @@ -139,7 +139,6 @@ impl ServerManager { fn start_worker(mgr: ServerManager, cid: Option) { let _ = ntex_rt::spawn(async move { let id = mgr.next_id(); - let mut wrk = Worker::start(id, mgr.factory(), cid); loop { @@ -181,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) { @@ -212,10 +211,9 @@ impl HandleCmdState { match upd { Update::Available(worker) => { self.workers.push(worker); + self.workers.sort(); if self.workers.len() == 1 { self.mgr.resume(); - } else { - self.workers.sort(); } } Update::Unavailable(worker) => { @@ -234,6 +232,9 @@ impl HandleCmdState { if let Err(item) = self.workers[0].send(item) { self.backlog.push_back(item); self.workers.remove(0); + if self.workers.is_empty() { + self.mgr.pause(); + } break; } } diff --git a/ntex-server/src/net/accept.rs b/ntex-server/src/net/accept.rs index 332fc846..7694d286 100644 --- a/ntex-server/src/net/accept.rs +++ b/ntex-server/src/net/accept.rs @@ -92,12 +92,14 @@ impl AcceptLoop { /// Start accept loop pub fn start(mut self, socks: Vec<(Token, Listener)>, srv: Server) { + let (tx, rx_start) = oneshot::channel(); let (rx, poll) = self .inner .take() .expect("AcceptLoop cannot be used multiple times"); Accept::start( + tx, rx, poll, socks, @@ -105,6 +107,8 @@ impl AcceptLoop { self.notify.clone(), self.status_handler.take(), ); + + let _ = rx_start.recv(); } } @@ -121,6 +125,7 @@ impl fmt::Debug for AcceptLoop { struct Accept { poller: Arc, rx: mpsc::Receiver, + tx: Option>, sockets: Vec, srv: Server, notify: AcceptNotify, @@ -131,6 +136,7 @@ struct Accept { impl Accept { fn start( + tx: oneshot::Sender<()>, rx: mpsc::Receiver, poller: Arc, socks: Vec<(Token, Listener)>, @@ -145,11 +151,12 @@ impl Accept { .name("ntex-server accept loop".to_owned()) .spawn(move || { System::set_current(sys); - Accept::new(rx, poller, socks, srv, notify, status_handler).poll() + Accept::new(tx, rx, poller, socks, srv, notify, status_handler).poll() }); } fn new( + tx: oneshot::Sender<()>, rx: mpsc::Receiver, poller: Arc, socks: Vec<(Token, Listener)>, @@ -175,6 +182,7 @@ impl Accept { notify, srv, status_handler, + tx: Some(tx), backpressure: true, backlog: VecDeque::new(), } @@ -192,19 +200,23 @@ impl Accept { // Create storage for events let mut events = Events::with_capacity(NonZeroUsize::new(512).unwrap()); + let mut timeout = Some(Duration::ZERO); loop { - if let Err(e) = self.poller.wait(&mut events, None) { - if e.kind() == io::ErrorKind::Interrupted { - continue; - } else { + if let Err(e) = self.poller.wait(&mut events, timeout) { + if e.kind() != io::ErrorKind::Interrupted { panic!("Cannot wait for events in poller: {}", e) } + } else if timeout.is_some() { + timeout = None; + let _ = self.tx.take().unwrap().send(()); } - for event in events.iter() { - let readd = self.accept(event.key); - if readd { - self.add_source(event.key); + for idx in 0..self.sockets.len() { + if self.sockets[idx].registered.get() { + let readd = self.accept(idx); + if readd { + self.add_source(idx); + } } } diff --git a/ntex-server/src/net/test.rs b/ntex-server/src/net/test.rs index 2ddae445..1c78f5c5 100644 --- a/ntex-server/src/net/test.rs +++ b/ntex-server/src/net/test.rs @@ -59,17 +59,19 @@ where .workers(1) .disable_signals() .run(); - tx.send((system, local_addr, server)) - .expect("Failed to send Server to TestServer"); + + ntex_rt::spawn(async move { + ntex_util::time::sleep(ntex_util::time::Millis(75)).await; + tx.send((system, local_addr, server)) + .expect("Failed to send Server to TestServer"); + }); + Ok(()) }) }); let (system, addr, server) = rx.recv().unwrap(); - // wait for server - thread::sleep(std::time::Duration::from_millis(50)); - TestServer { addr, server, diff --git a/ntex-server/src/wrk.rs b/ntex-server/src/wrk.rs index a61f7731..b791817d 100644 --- a/ntex-server/src/wrk.rs +++ b/ntex-server/src/wrk.rs @@ -2,8 +2,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{ready, Context, Poll}; 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}; @@ -99,10 +99,10 @@ impl Worker { log::debug!("Creating server instance in {:?}", id); let factory = cfg.create().await; - log::debug!("Server instance has been created in {:?}", id); 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; } Err(e) => { @@ -151,10 +151,8 @@ impl Worker { if self.failed.load(Ordering::Acquire) { WorkerStatus::Failed } else { - // cleanup updates - while self.avail.notify.try_recv().is_ok() {} - - if self.avail.notify.recv_direct().await.is_err() { + self.avail.wait_for_update().await; + if self.avail.failed() { self.failed.store(true, Ordering::Release); } self.status() @@ -196,52 +194,85 @@ impl Future for WorkerStop { #[derive(Debug, Clone)] struct WorkerAvailability { - notify: bus::Receiver<()>, - available: Arc, + inner: Arc, } #[derive(Debug, Clone)] struct WorkerAvailabilityTx { - notify: bus::Sender<()>, - available: Arc, + inner: Arc, +} + +#[derive(Debug)] +struct Inner { + waker: AtomicWaker, + updated: AtomicBool, + available: AtomicBool, + failed: AtomicBool, } impl WorkerAvailability { fn create() -> (Self, WorkerAvailabilityTx) { - let (mut tx, rx) = broadcast(16); - tx.set_overflow(true); + let inner = Arc::new(Inner { + waker: AtomicWaker::new(), + updated: AtomicBool::new(false), + available: AtomicBool::new(false), + failed: AtomicBool::new(false), + }); let avail = WorkerAvailability { - notify: rx, - available: Arc::new(AtomicBool::new(false)), - }; - let avail_tx = WorkerAvailabilityTx { - notify: tx, - available: avail.available.clone(), + inner: inner.clone(), }; + let avail_tx = WorkerAvailabilityTx { inner }; (avail, avail_tx) } + fn failed(&self) -> bool { + self.inner.failed.load(Ordering::Acquire) + } + fn available(&self) -> bool { - self.available.load(Ordering::Acquire) + self.inner.available.load(Ordering::Acquire) + } + + async fn wait_for_update(&self) { + poll_fn(|cx| { + if self.inner.updated.load(Ordering::Acquire) { + self.inner.updated.store(false, Ordering::Release); + Poll::Ready(()) + } else { + self.inner.waker.register(cx.waker()); + Poll::Pending + } + }) + .await; } } impl WorkerAvailabilityTx { fn set(&self, val: bool) { - let old = self.available.swap(val, Ordering::Release); - if !old && val { - let _ = self.notify.try_broadcast(()); + let old = self.inner.available.swap(val, Ordering::Release); + if old != val { + self.inner.updated.store(true, Ordering::Release); + self.inner.waker.wake(); } } } +impl Drop for WorkerAvailabilityTx { + fn drop(&mut self) { + self.inner.failed.store(true, Ordering::Release); + self.inner.updated.store(true, Ordering::Release); + self.inner.available.store(false, Ordering::Release); + self.inner.waker.wake(); + } +} + /// Service worker /// /// Worker accepts message via unbounded channel and starts processing. struct WorkerSt> { id: WorkerId, - rx: Pin>>, + rx: Receiver, stop: Pin>>, factory: F, availability: WorkerAvailabilityTx, @@ -253,25 +284,43 @@ where F: ServiceFactory + 'static, { loop { + let mut recv = std::pin::pin!(wrk.rx.recv()); let fut = poll_fn(|cx| { - ready!(svc.poll_ready(cx)?); - - if let Some(item) = ready!(Pin::new(&mut wrk.rx).poll_next(cx)) { - let fut = svc.call(item); - let _ = spawn(async move { - let _ = fut.await; - }); + match svc.poll_ready(cx) { + Poll::Ready(Ok(())) => { + wrk.availability.set(true); + } + Poll::Ready(Err(err)) => { + wrk.availability.set(false); + return Poll::Ready(Err(err)); + } + Poll::Pending => { + wrk.availability.set(false); + return Poll::Pending; + } + } + + match ready!(recv.as_mut().poll(cx)) { + Ok(item) => { + let fut = svc.call(item); + let _ = spawn(async move { + let _ = fut.await; + }); + Poll::Ready(Ok::<_, F::Error>(true)) + } + Err(_) => { + log::error!("Server is gone"); + Poll::Ready(Ok(false)) + } } - Poll::Ready(Ok::<(), F::Error>(())) }); match select(fut, stream_recv(&mut wrk.stop)).await { - Either::Left(Ok(())) => continue, + Either::Left(Ok(true)) => continue, Either::Left(Err(_)) => { let _ = ntex_rt::spawn(async move { svc.shutdown().await; }); - wrk.availability.set(false); } Either::Right(Some(Shutdown { timeout, result })) => { wrk.availability.set(false); @@ -285,7 +334,8 @@ where stop_svc(wrk.id, svc, timeout, Some(result)).await; return; } - Either::Right(None) => { + Either::Left(Ok(false)) | Either::Right(None) => { + wrk.availability.set(false); stop_svc(wrk.id, svc, STOP_TIMEOUT, None).await; return; } @@ -295,7 +345,6 @@ where loop { match select(wrk.factory.create(()), stream_recv(&mut wrk.stop)).await { Either::Left(Ok(service)) => { - wrk.availability.set(true); svc = Pipeline::new(service).bind(); break; } @@ -336,8 +385,6 @@ where { availability.set(false); let factory = factory?; - - let rx = Box::pin(rx); let mut stop = Box::pin(stop); let svc = match select(factory.create(()), stream_recv(&mut stop)).await { @@ -356,9 +403,9 @@ where svc, WorkerSt { id, + rx, factory, availability, - rx: Box::pin(rx), stop: Box::pin(stop), }, )) diff --git a/ntex-tls/examples/rustls-server.rs b/ntex-tls/examples/rustls-server.rs index 445cffec..a80b25e2 100644 --- a/ntex-tls/examples/rustls-server.rs +++ b/ntex-tls/examples/rustls-server.rs @@ -13,9 +13,8 @@ async fn main() -> io::Result<()> { println!("Started openssl echp server: 127.0.0.1:8443"); // load ssl keys - let cert_file = - &mut BufReader::new(File::open("../ntex-tls/examples/cert.pem").unwrap()); - let key_file = &mut BufReader::new(File::open("../ntex-tls/examples/key.pem").unwrap()); + let cert_file = &mut BufReader::new(File::open("../examples/cert.pem").unwrap()); + let key_file = &mut BufReader::new(File::open("../examples/key.pem").unwrap()); let keys = rustls_pemfile::private_key(key_file).unwrap().unwrap(); let cert_chain = rustls_pemfile::certs(cert_file) .collect::, _>>() diff --git a/ntex-tls/examples/webserver.rs b/ntex-tls/examples/webserver.rs index 52867a6b..9398708e 100644 --- a/ntex-tls/examples/webserver.rs +++ b/ntex-tls/examples/webserver.rs @@ -8,18 +8,18 @@ use tls_openssl::ssl::{self, SslFiletype, SslMethod}; #[ntex::main] async fn main() -> io::Result<()> { - //std::env::set_var("RUST_LOG", "trace"); - //env_logger::init(); + std::env::set_var("RUST_LOG", "trace"); + let _ = env_logger::try_init(); println!("Started openssl web server: 127.0.0.1:8443"); // load ssl keys let mut builder = ssl::SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); builder - .set_private_key_file("../tests/key.pem", SslFiletype::PEM) + .set_private_key_file("./examples/key.pem", SslFiletype::PEM) .unwrap(); builder - .set_certificate_chain_file("../tests/cert.pem") + .set_certificate_chain_file("./examples/cert.pem") .unwrap(); // h2 alpn config diff --git a/ntex/CHANGES.md b/ntex/CHANGES.md index dc61776a..6ef4b5ef 100644 --- a/ntex/CHANGES.md +++ b/ntex/CHANGES.md @@ -1,5 +1,27 @@ # Changes +## [2.12.4] - 2025-03-28 + +* http: Return PayloadError::Incomplete on server disconnect + +* web: Expose WebStack for external wrapper support in downstream crates #542 + +## [2.12.3] - 2025-03-22 + +* web: Export web::app_service::AppService #534 + +* http: Add delay for test server availability, could cause connect race + +## [2.12.2] - 2025-03-15 + +* http: Allow to run publish future to completion in case error + +* http: Remove brotli support + +## [2.12.1] - 2025-03-14 + +* Allow to disable test logging (no-test-logging features) + ## [2.12.0] - 2025-03-12 * Add neon runtime support diff --git a/ntex/Cargo.toml b/ntex/Cargo.toml index 41c418de..0ea37469 100644 --- a/ntex/Cargo.toml +++ b/ntex/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ntex" -version = "2.12.0" +version = "2.12.4" authors = ["ntex contributors "] description = "Framework for composable network services" readme = "README.md" @@ -18,7 +18,7 @@ edition = "2021" rust-version = "1.75" [package.metadata.docs.rs] -features = ["tokio", "openssl", "rustls", "compress", "cookie", "ws", "brotli", "ntex-tls/rustls-ring"] +features = ["tokio", "openssl", "rustls", "compress", "cookie", "ws", "ntex-tls/rustls-ring"] [lib] name = "ntex" @@ -57,8 +57,8 @@ neon-uring = ["ntex-net/neon", "ntex-net/io-uring"] # websocket support ws = ["dep:sha-1"] -# brotli2 support -brotli = ["dep:brotli2"] +# disable [ntex::test] logging configuration +no-test-logging = [] [dependencies] ntex-codec = "0.6" @@ -68,11 +68,11 @@ ntex-service = "3.4" ntex-macros = "0.1" ntex-util = "2.8" ntex-bytes = "0.1.27" -ntex-server = "2.7" +ntex-server = "2.7.4" ntex-h2 = "1.8.6" ntex-rt = "0.4.27" ntex-io = "2.11" -ntex-net = "2.5" +ntex-net = "2.5.10" ntex-tls = "2.3" base64 = "0.22" @@ -109,12 +109,12 @@ tls-rustls = { version = "0.23", package = "rustls", optional = true, default-fe webpki-roots = { version = "0.26", optional = true } # compression -brotli2 = { version = "0.3.2", optional = true } flate2 = { version = "1.0", optional = true } [dev-dependencies] rand = "0.8" time = "0.3" +oneshot = "0.1" futures-util = "0.3" tls-openssl = { version = "0.10", package = "openssl" } tls-rustls = { version = "0.23", package = "rustls", features = ["ring", "std"], default-features = false } diff --git a/ntex/src/http/client/h1proto.rs b/ntex/src/http/client/h1proto.rs index 06572418..28871225 100644 --- a/ntex/src/http/client/h1proto.rs +++ b/ntex/src/http/client/h1proto.rs @@ -1,13 +1,11 @@ -use std::{ - future::poll_fn, io, io::Write, pin::Pin, task::Context, task::Poll, time::Instant, -}; +use std::{future::poll_fn, io, io::Write, pin::Pin, task, task::Poll, time::Instant}; use crate::http::body::{BodySize, MessageBody}; use crate::http::error::PayloadError; -use crate::http::h1; use crate::http::header::{HeaderMap, HeaderValue, HOST}; use crate::http::message::{RequestHeadType, ResponseHead}; use crate::http::payload::{Payload, PayloadStream}; +use crate::http::{h1, Version}; use crate::io::{IoBoxed, RecvError}; use crate::time::{timeout_checked, Millis}; use crate::util::{ready, BufMut, Bytes, BytesMut, Stream}; @@ -101,7 +99,13 @@ where Ok((head, Payload::None)) } _ => { - let pl: PayloadStream = Box::pin(PlStream::new(io, codec, created, pool)); + let pl: PayloadStream = Box::pin(PlStream::new( + io, + codec, + created, + pool, + head.version == Version::HTTP_10, + )); Ok((head, pl.into())) } } @@ -137,6 +141,7 @@ pub(super) struct PlStream { io: Option, codec: h1::ClientPayloadCodec, created: Instant, + http_10: bool, pool: Option, } @@ -146,12 +151,14 @@ impl PlStream { codec: h1::ClientCodec, created: Instant, pool: Option, + http_10: bool, ) -> Self { PlStream { io: Some(io), codec: codec.into_payload_codec(), created, pool, + http_10, } } } @@ -161,41 +168,46 @@ impl Stream for PlStream { fn poll_next( mut self: Pin<&mut Self>, - cx: &mut Context<'_>, + cx: &mut task::Context<'_>, ) -> Poll> { let mut this = self.as_mut(); loop { - return Poll::Ready(Some( - match ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx)) { - Ok(chunk) => { - if let Some(chunk) = chunk { - Ok(chunk) - } else { - release_connection( - this.io.take().unwrap(), - !this.codec.keepalive(), - this.created, - this.pool.take(), - ); - return Poll::Ready(None); - } + let item = ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx)); + return Poll::Ready(Some(match item { + Ok(chunk) => { + if let Some(chunk) = chunk { + Ok(chunk) + } else { + release_connection( + this.io.take().unwrap(), + !this.codec.keepalive(), + this.created, + this.pool.take(), + ); + return Poll::Ready(None); } - Err(RecvError::KeepAlive) => { - Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into()) + } + Err(RecvError::KeepAlive) => { + Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into()) + } + Err(RecvError::Stop) => { + Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped").into()) + } + Err(RecvError::WriteBackpressure) => { + ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?; + continue; + } + Err(RecvError::Decoder(err)) => Err(err), + Err(RecvError::PeerGone(Some(err))) => { + Err(PayloadError::Incomplete(Some(err))) + } + Err(RecvError::PeerGone(None)) => { + if this.http_10 { + return Poll::Ready(None); } - Err(RecvError::Stop) => { - Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped") - .into()) - } - Err(RecvError::WriteBackpressure) => { - ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?; - continue; - } - Err(RecvError::Decoder(err)) => Err(err), - Err(RecvError::PeerGone(Some(err))) => Err(err.into()), - Err(RecvError::PeerGone(None)) => return Poll::Ready(None), - }, - )); + Err(PayloadError::Incomplete(None)) + } + })); } } } diff --git a/ntex/src/http/client/response.rs b/ntex/src/http/client/response.rs index c68b6e73..9a450687 100644 --- a/ntex/src/http/client/response.rs +++ b/ntex/src/http/client/response.rs @@ -387,8 +387,8 @@ impl Future for ReadBody { let this = self.get_mut(); loop { - return match Pin::new(&mut this.stream).poll_next(cx)? { - Poll::Ready(Some(chunk)) => { + return match Pin::new(&mut this.stream).poll_next(cx) { + Poll::Ready(Some(Ok(chunk))) => { if this.limit > 0 && (this.buf.len() + chunk.len()) > this.limit { Poll::Ready(Err(PayloadError::Overflow)) } else { @@ -397,6 +397,7 @@ impl Future for ReadBody { } } Poll::Ready(None) => Poll::Ready(Ok(this.buf.split().freeze())), + Poll::Ready(Some(Err(err))) => Poll::Ready(Err(err)), Poll::Pending => { if this.timeout.poll_elapsed(cx).is_ready() { Poll::Ready(Err(PayloadError::Incomplete(Some( diff --git a/ntex/src/http/encoding/decoder.rs b/ntex/src/http/encoding/decoder.rs index 5a518738..45020a3a 100644 --- a/ntex/src/http/encoding/decoder.rs +++ b/ntex/src/http/encoding/decoder.rs @@ -1,7 +1,5 @@ use std::{future::Future, io, io::Write, pin::Pin, task::Context, task::Poll}; -#[cfg(feature = "brotli")] -use brotli2::write::BrotliDecoder; use flate2::write::{GzDecoder, ZlibDecoder}; use super::Writer; @@ -27,10 +25,6 @@ where #[inline] pub fn new(stream: S, encoding: ContentEncoding) -> Decoder { let decoder = match encoding { - #[cfg(feature = "brotli")] - ContentEncoding::Br => Some(ContentDecoder::Br(Box::new(BrotliDecoder::new( - Writer::new(), - )))), ContentEncoding::Deflate => Some(ContentDecoder::Deflate(Box::new( ZlibDecoder::new(Writer::new()), ))), @@ -137,25 +131,11 @@ where enum ContentDecoder { Deflate(Box>), Gzip(Box>), - #[cfg(feature = "brotli")] - Br(Box>), } impl ContentDecoder { fn feed_eof(&mut self) -> io::Result> { match self { - #[cfg(feature = "brotli")] - ContentDecoder::Br(ref mut decoder) => match decoder.flush() { - Ok(()) => { - let b = decoder.get_mut().take(); - if !b.is_empty() { - Ok(Some(b)) - } else { - Ok(None) - } - } - Err(e) => Err(e), - }, ContentDecoder::Gzip(ref mut decoder) => match decoder.try_finish() { Ok(_) => { let b = decoder.get_mut().take(); @@ -183,19 +163,6 @@ impl ContentDecoder { fn feed_data(&mut self, data: Bytes) -> io::Result> { match self { - #[cfg(feature = "brotli")] - ContentDecoder::Br(ref mut decoder) => match decoder.write_all(&data) { - Ok(_) => { - decoder.flush()?; - let b = decoder.get_mut().take(); - if !b.is_empty() { - Ok(Some(b)) - } else { - Ok(None) - } - } - Err(e) => Err(e), - }, ContentDecoder::Gzip(ref mut decoder) => match decoder.write_all(&data) { Ok(_) => { decoder.flush()?; diff --git a/ntex/src/http/encoding/encoder.rs b/ntex/src/http/encoding/encoder.rs index 7c24edf3..086fc815 100644 --- a/ntex/src/http/encoding/encoder.rs +++ b/ntex/src/http/encoding/encoder.rs @@ -1,8 +1,6 @@ //! Stream encoder use std::{fmt, future::Future, io, io::Write, pin::Pin, task::Context, task::Poll}; -#[cfg(feature = "brotli")] -use brotli2::write::BrotliEncoder; use flate2::write::{GzEncoder, ZlibEncoder}; use crate::http::body::{Body, BodySize, MessageBody, ResponseBody}; @@ -191,23 +189,11 @@ fn update_head(encoding: ContentEncoding, head: &mut ResponseHead) { enum ContentEncoder { Deflate(ZlibEncoder), Gzip(GzEncoder), - #[cfg(feature = "brotli")] - Br(BrotliEncoder), } impl ContentEncoder { fn can_encode(encoding: ContentEncoding) -> bool { - #[cfg(feature = "brotli")] - { - matches!( - encoding, - ContentEncoding::Deflate | ContentEncoding::Gzip | ContentEncoding::Br - ) - } - #[cfg(not(feature = "brotli"))] - { - matches!(encoding, ContentEncoding::Deflate | ContentEncoding::Gzip) - } + matches!(encoding, ContentEncoding::Deflate | ContentEncoding::Gzip) } fn encoder(encoding: ContentEncoding) -> Option { @@ -220,18 +206,12 @@ impl ContentEncoder { Writer::new(), flate2::Compression::fast(), ))), - #[cfg(feature = "brotli")] - ContentEncoding::Br => { - Some(ContentEncoder::Br(BrotliEncoder::new(Writer::new(), 3))) - } _ => None, } } fn take(&mut self) -> Bytes { match *self { - #[cfg(feature = "brotli")] - ContentEncoder::Br(ref mut encoder) => encoder.get_mut().take(), ContentEncoder::Deflate(ref mut encoder) => encoder.get_mut().take(), ContentEncoder::Gzip(ref mut encoder) => encoder.get_mut().take(), } @@ -239,11 +219,6 @@ impl ContentEncoder { fn finish(self) -> Result { match self { - #[cfg(feature = "brotli")] - ContentEncoder::Br(encoder) => match encoder.finish() { - Ok(writer) => Ok(writer.buf.freeze()), - Err(err) => Err(err), - }, ContentEncoder::Gzip(encoder) => match encoder.finish() { Ok(writer) => Ok(writer.buf.freeze()), Err(err) => Err(err), @@ -257,14 +232,6 @@ impl ContentEncoder { fn write(&mut self, data: &[u8]) -> Result<(), io::Error> { match *self { - #[cfg(feature = "brotli")] - ContentEncoder::Br(ref mut encoder) => match encoder.write_all(data) { - Ok(_) => Ok(()), - Err(err) => { - log::trace!("Error decoding br encoding: {}", err); - Err(err) - } - }, ContentEncoder::Gzip(ref mut encoder) => match encoder.write_all(data) { Ok(_) => Ok(()), Err(err) => { @@ -288,8 +255,6 @@ impl fmt::Debug for ContentEncoder { match self { ContentEncoder::Deflate(_) => write!(f, "ContentEncoder::Deflate"), ContentEncoder::Gzip(_) => write!(f, "ContentEncoder::Gzip"), - #[cfg(feature = "brotli")] - ContentEncoder::Br(_) => write!(f, "ContentEncoder::Br"), } } } diff --git a/ntex/src/http/h1/dispatcher.rs b/ntex/src/http/h1/dispatcher.rs index 18263583..7a2142ea 100644 --- a/ntex/src/http/h1/dispatcher.rs +++ b/ntex/src/http/h1/dispatcher.rs @@ -1,5 +1,5 @@ //! HTTP/1 protocol dispatcher -use std::{error, future, io, marker, pin::Pin, rc::Rc, task::Context, task::Poll}; +use std::{error, future, io, marker, mem, pin::Pin, rc::Rc, task::Context, task::Poll}; use crate::io::{Decoded, Filter, Io, IoStatusUpdate, RecvError}; use crate::service::{PipelineCall, Service}; @@ -144,7 +144,20 @@ where inner.send_response(res, body) } Poll::Ready(Err(err)) => inner.control(Control::err(err)), - Poll::Pending => ready!(inner.poll_request(cx)), + Poll::Pending => { + // state changed because of error. + // spawn current publish future to runtime + // so it could complete error handling + let st = ready!(inner.poll_request(cx)); + if inner.payload.is_some() { + if let State::CallPublish { fut } = + mem::replace(&mut *this.st, State::ReadRequest) + { + crate::rt::spawn(fut); + } + } + st + } }, // handle control service responses State::CallControl { fut } => match Pin::new(fut).poll(cx) { @@ -339,7 +352,7 @@ where .io .encode(Message::Item((msg, body.size())), &self.codec) .map_err(|err| { - if let Some(mut payload) = self.payload.take() { + if let Some(ref mut payload) = self.payload { payload.1.set_error(PayloadError::Incomplete(None)); } err @@ -438,7 +451,7 @@ where } fn set_payload_error(&mut self, err: PayloadError) { - if let Some(mut payload) = self.payload.take() { + if let Some(ref mut payload) = self.payload { payload.1.set_error(err); } } diff --git a/ntex/src/http/h1/payload.rs b/ntex/src/http/h1/payload.rs index 1fe5e5a5..ac3c8609 100644 --- a/ntex/src/http/h1/payload.rs +++ b/ntex/src/http/h1/payload.rs @@ -3,8 +3,7 @@ use std::rc::{Rc, Weak}; use std::task::{Context, Poll}; use std::{cell::RefCell, collections::VecDeque, pin::Pin}; -use crate::http::error::PayloadError; -use crate::{task::LocalWaker, util::Bytes, util::Stream}; +use crate::{http::error::PayloadError, task::LocalWaker, util::Bytes, util::Stream}; /// max buffer size 32k const MAX_BUFFER_SIZE: usize = 32_768; @@ -119,7 +118,7 @@ impl PayloadSender { // we check only if Payload (other side) is alive, // otherwise always return true (consume payload) if let Some(shared) = self.inner.upgrade() { - if shared.borrow().need_read { + if shared.borrow().flags.contains(Flags::NEED_READ) { PayloadStatus::Read } else { shared.borrow_mut().io_task.register(cx.waker()); @@ -131,12 +130,20 @@ impl PayloadSender { } } +bitflags::bitflags! { + #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)] + struct Flags: u8 { + const EOF = 0b0000_0001; + const ERROR = 0b0000_0010; + const NEED_READ = 0b0000_0100; + } +} + #[derive(Debug)] struct Inner { len: usize, - eof: bool, + flags: Flags, err: Option, - need_read: bool, items: VecDeque, task: LocalWaker, io_task: LocalWaker, @@ -144,12 +151,16 @@ struct Inner { impl Inner { fn new(eof: bool) -> Self { + let flags = if eof { + Flags::EOF | Flags::NEED_READ + } else { + Flags::NEED_READ + }; Inner { - eof, + flags, len: 0, err: None, items: VecDeque::new(), - need_read: true, task: LocalWaker::new(), io_task: LocalWaker::new(), } @@ -157,18 +168,23 @@ impl Inner { fn set_error(&mut self, err: PayloadError) { self.err = Some(err); + self.flags.insert(Flags::ERROR); self.task.wake() } fn feed_eof(&mut self) { - self.eof = true; + self.flags.insert(Flags::EOF); self.task.wake() } fn feed_data(&mut self, data: Bytes) { self.len += data.len(); self.items.push_back(data); - self.need_read = self.len < MAX_BUFFER_SIZE; + if self.len < MAX_BUFFER_SIZE { + self.flags.insert(Flags::NEED_READ); + } else { + self.flags.remove(Flags::NEED_READ); + } self.task.wake(); } @@ -178,19 +194,25 @@ impl Inner { ) -> Poll>> { if let Some(data) = self.items.pop_front() { self.len -= data.len(); - self.need_read = self.len < MAX_BUFFER_SIZE; + if self.len < MAX_BUFFER_SIZE { + self.flags.insert(Flags::NEED_READ); + } else { + self.flags.remove(Flags::NEED_READ); + } - if self.need_read && !self.eof { + if self.flags.contains(Flags::NEED_READ) + && !self.flags.intersects(Flags::EOF | Flags::ERROR) + { self.task.register(cx.waker()); } self.io_task.wake(); Poll::Ready(Some(Ok(data))) } else if let Some(err) = self.err.take() { Poll::Ready(Some(Err(err))) - } else if self.eof { + } else if self.flags.intersects(Flags::EOF | Flags::ERROR) { Poll::Ready(None) } else { - self.need_read = true; + self.flags.insert(Flags::NEED_READ); self.task.register(cx.waker()); self.io_task.wake(); Poll::Pending diff --git a/ntex/src/http/test.rs b/ntex/src/http/test.rs index 265ae6f8..0e4a6559 100644 --- a/ntex/src/http/test.rs +++ b/ntex/src/http/test.rs @@ -11,7 +11,7 @@ use crate::server::Server; use crate::service::ServiceFactory; #[cfg(feature = "ws")] use crate::ws::{error::WsClientError, WsClient, WsConnection}; -use crate::{rt::System, time::Millis, time::Seconds, util::Bytes}; +use crate::{rt::System, time::sleep, time::Millis, time::Seconds, util::Bytes}; use super::client::{Client, ClientRequest, ClientResponse, Connector}; use super::error::{HttpError, PayloadError}; @@ -244,14 +244,15 @@ where .workers(1) .disable_signals() .run(); - tx.send((system, srv, local_addr)).unwrap(); + + crate::rt::spawn(async move { + sleep(Millis(125)).await; + tx.send((system, srv, local_addr)).unwrap(); + }); Ok(()) }) }); - // wait for server - if std::env::var("GITHUB_ACTIONS") == Ok("true".to_string()) { - thread::sleep(std::time::Duration::from_millis(150)); - } + thread::sleep(std::time::Duration::from_millis(150)); let (system, server, addr) = rx.recv().unwrap(); diff --git a/ntex/src/lib.rs b/ntex/src/lib.rs index 93220f15..a15ee31a 100644 --- a/ntex/src/lib.rs +++ b/ntex/src/lib.rs @@ -126,6 +126,7 @@ pub mod util { #[doc(hidden)] pub fn enable_test_logging() { + #[cfg(not(feature = "no-test-logging"))] if std::env::var("NTEX_NO_TEST_LOG").is_err() { if std::env::var("RUST_LOG").is_err() { std::env::set_var("RUST_LOG", "trace"); diff --git a/ntex/src/web/config.rs b/ntex/src/web/config.rs index 91c5034c..c7edceb6 100644 --- a/ntex/src/web/config.rs +++ b/ntex/src/web/config.rs @@ -68,7 +68,7 @@ pub struct ServiceConfig { } impl ServiceConfig { - pub(crate) fn new() -> Self { + pub fn new() -> Self { Self { services: Vec::new(), state: Extensions::new(), @@ -132,7 +132,7 @@ mod tests { use crate::http::{Method, StatusCode}; use crate::util::Bytes; use crate::web::test::{call_service, init_service, read_body, TestRequest}; - use crate::web::{self, App, HttpRequest, HttpResponse}; + use crate::web::{self, App, DefaultError, HttpRequest, HttpResponse}; #[crate::rt_test] async fn test_configure_state() { @@ -205,4 +205,11 @@ mod tests { let resp = call_service(&srv, req).await; assert_eq!(resp.status(), StatusCode::OK); } + + #[test] + fn test_new_service_config() { + let cfg: ServiceConfig = ServiceConfig::new(); + assert!(cfg.services.is_empty()); + assert!(cfg.external.is_empty()); + } } diff --git a/ntex/src/web/mod.rs b/ntex/src/web/mod.rs index cf1686e0..8d9adf4d 100644 --- a/ntex/src/web/mod.rs +++ b/ntex/src/web/mod.rs @@ -82,7 +82,7 @@ mod route; mod scope; mod server; mod service; -mod stack; +pub mod stack; pub mod test; pub mod types; mod util; @@ -128,6 +128,7 @@ pub mod dev { //! The purpose of this module is to alleviate imports of many common //! traits by adding a glob import to the top of ntex::web heavy modules: + pub use crate::web::app_service::AppService; pub use crate::web::config::AppConfig; pub use crate::web::info::ConnectionInfo; pub use crate::web::rmap::ResourceMap; diff --git a/ntex/src/web/test.rs b/ntex/src/web/test.rs index d81c8338..1307ad9f 100644 --- a/ntex/src/web/test.rs +++ b/ntex/src/web/test.rs @@ -697,14 +697,13 @@ where .set_tag("test", "WEB-SRV") .run(); - tx.send((System::current(), srv, local_addr)).unwrap(); + crate::rt::spawn(async move { + sleep(Millis(125)).await; + tx.send((System::current(), srv, local_addr)).unwrap(); + }); Ok(()) }) }); - // wait for server - if std::env::var("GITHUB_ACTIONS") == Ok("true".to_string()) { - thread::sleep(std::time::Duration::from_millis(150)); - } let (system, server, addr) = rx.recv().unwrap(); diff --git a/ntex/tests/http_awc_client.rs b/ntex/tests/http_awc_client.rs index 3c953c58..bd4c7e0a 100644 --- a/ntex/tests/http_awc_client.rs +++ b/ntex/tests/http_awc_client.rs @@ -3,16 +3,14 @@ use std::io::{Read, Write}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; -use brotli2::write::BrotliEncoder; use coo_kie::Cookie; use flate2::{read::GzDecoder, write::GzEncoder, write::ZlibEncoder, Compression}; -use futures_util::stream::once; use rand::Rng; -use ntex::http::client::error::{JsonPayloadError, SendRequestError}; +use ntex::http::client::error::SendRequestError; use ntex::http::client::{Client, Connector}; use ntex::http::test::server as test_server; -use ntex::http::{header, HttpMessage, HttpService, Method}; +use ntex::http::{header, HttpMessage, HttpService}; use ntex::service::{chain_factory, map_config}; use ntex::web::dev::AppConfig; use ntex::web::middleware::Compress; @@ -510,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::(); 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 @@ -530,130 +530,10 @@ 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)); } -#[ntex::test] -async fn test_client_brotli_encoding() { - let srv = test::server(|| { - App::new().service(web::resource("/").route(web::to(|data: Bytes| async move { - let mut e = BrotliEncoder::new(Vec::new(), 5); - e.write_all(&data).unwrap(); - let data = e.finish().unwrap(); - HttpResponse::Ok() - .header("content-encoding", "br") - .body(data) - }))) - }); - - // client request - let mut response = srv.post("/").send_body(STR).await.unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes, Bytes::from_static(STR.as_ref())); -} - -#[ntex::test] -async fn test_client_brotli_encoding_large_random() { - let data = rand::thread_rng() - .sample_iter(&rand::distributions::Alphanumeric) - .take(70_000) - .map(char::from) - .collect::(); - - let srv = test::server(|| { - App::new().service(web::resource("/").route(web::to(|data: Bytes| async move { - let mut e = BrotliEncoder::new(Vec::new(), 5); - e.write_all(&data).unwrap(); - let data = e.finish().unwrap(); - HttpResponse::Ok() - .header("content-encoding", "br") - .body(data) - }))) - }); - - // client request - let mut response = srv.post("/").send_body(data.clone()).await.unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); - - // frozen request - let request = srv.post("/").timeout(Seconds(30)).freeze().unwrap(); - assert_eq!(request.get_method(), Method::POST); - assert_eq!(request.get_uri(), srv.url("/").as_str()); - let mut response = request.send_body(data.clone()).await.unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); - - // extra header - let mut response = request - .extra_header("x-test2", "222") - .send_body(data.clone()) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); - - // client stream request - let mut response = srv - .post("/") - .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from( - data.clone(), - )))) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); - - // frozen request - let request = srv.post("/").timeout(Seconds(30)).freeze().unwrap(); - let mut response = request - .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from( - data.clone(), - )))) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); - - let mut response = request - .extra_header("x-test2", "222") - .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from( - data.clone(), - )))) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data.clone())); -} - #[ntex::test] async fn test_client_deflate_encoding() { let srv = test::server(|| { diff --git a/ntex/tests/http_openssl.rs b/ntex/tests/http_openssl.rs index c91de0b8..921310a8 100644 --- a/ntex/tests/http_openssl.rs +++ b/ntex/tests/http_openssl.rs @@ -1,5 +1,6 @@ #![cfg(feature = "openssl")] -use std::{io, sync::atomic::AtomicUsize, sync::atomic::Ordering, sync::Arc}; +use std::io; +use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex}; use futures_util::stream::{once, Stream, StreamExt}; use tls_openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod}; @@ -424,11 +425,12 @@ async fn test_h2_service_error() { assert_eq!(bytes, Bytes::from_static(b"error")); } -struct SetOnDrop(Arc); +struct SetOnDrop(Arc, Arc>>>); impl Drop for SetOnDrop { fn drop(&mut self) { self.0.fetch_add(1, Ordering::Relaxed); + let _ = self.1.lock().unwrap().take().unwrap().send(()); } } @@ -436,17 +438,20 @@ impl Drop for SetOnDrop { async fn test_h2_client_drop() -> io::Result<()> { let count = Arc::new(AtomicUsize::new(0)); let count2 = count.clone(); + let (tx, rx) = ::oneshot::channel(); + let tx = Arc::new(Mutex::new(Some(tx))); let srv = test_server(move || { + let tx = tx.clone(); let count = count2.clone(); HttpService::build() .h2(move |req: Request| { - let count = count.clone(); + let st = SetOnDrop(count.clone(), tx.clone()); async move { - let _st = SetOnDrop(count); assert!(req.peer_addr().is_some()); assert_eq!(req.version(), Version::HTTP_2); - sleep(Seconds(100)).await; + sleep(Seconds(30)).await; + drop(st); Ok::<_, io::Error>(Response::Ok().finish()) } }) @@ -454,9 +459,9 @@ async fn test_h2_client_drop() -> io::Result<()> { .map_err(|_| ()) }); - let result = timeout(Millis(250), srv.srequest(Method::GET, "/").send()).await; + let result = timeout(Millis(1500), srv.srequest(Method::GET, "/").send()).await; assert!(result.is_err()); - sleep(Millis(150)).await; + let _ = timeout(Millis(1500), rx).await; assert_eq!(count.load(Ordering::Relaxed), 1); Ok(()) } @@ -539,13 +544,19 @@ async fn test_ws_transport() { async fn test_h2_graceful_shutdown() -> io::Result<()> { let count = Arc::new(AtomicUsize::new(0)); let count2 = count.clone(); + let (tx, rx) = ::oneshot::channel(); + let tx = Arc::new(Mutex::new(Some(tx))); let srv = test_server(move || { + let tx = tx.clone(); let count = count2.clone(); HttpService::build() .h2(move |_| { let count = count.clone(); count.fetch_add(1, Ordering::Relaxed); + if count.load(Ordering::Relaxed) == 2 { + let _ = tx.lock().unwrap().take().unwrap().send(()); + } async move { sleep(Millis(1000)).await; count.fetch_sub(1, Ordering::Relaxed); @@ -566,7 +577,7 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> { let _ = req.send().await.unwrap(); sleep(Millis(100000)).await; }); - sleep(Millis(150)).await; + let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 2); let (tx, rx) = oneshot::channel(); @@ -574,8 +585,6 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> { srv.stop().await; let _ = tx.send(()); }); - sleep(Millis(150)).await; - assert_eq!(count.load(Ordering::Relaxed), 2); let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 0); diff --git a/ntex/tests/http_server.rs b/ntex/tests/http_server.rs index 44512500..0227573b 100644 --- a/ntex/tests/http_server.rs +++ b/ntex/tests/http_server.rs @@ -1,4 +1,4 @@ -use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc}; +use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex}; use std::{io, io::Read, io::Write, net}; use futures_util::future::{self, FutureExt}; @@ -405,6 +405,36 @@ async fn test_http1_handle_not_consumed_payload() { assert_eq!(&data[..17], b"HTTP/1.1 200 OK\r\n"); } +/// Handle payload errors (keep-alive, disconnects) +#[ntex::test] +async fn test_http1_handle_payload_errors() { + let count = Arc::new(AtomicUsize::new(0)); + let count2 = count.clone(); + + let srv = test_server(move || { + let count = count2.clone(); + HttpService::build().h1(move |mut req: Request| { + let count = count.clone(); + async move { + let mut pl = req.take_payload(); + let result = pl.recv().await; + if result.unwrap().is_err() { + count.fetch_add(1, Ordering::Relaxed); + } + Ok::<_, io::Error>(Response::Ok().finish()) + } + }) + }); + + let mut stream = net::TcpStream::connect(srv.addr()).unwrap(); + let _ = + stream.write_all(b"GET /test/tests/test HTTP/1.1\r\ncontent-length: 99999\r\n\r\n"); + sleep(Millis(250)).await; + drop(stream); + sleep(Millis(250)).await; + assert_eq!(count.load(Ordering::Acquire), 1); +} + #[ntex::test] async fn test_content_length() { let srv = test_server(|| { @@ -693,11 +723,12 @@ async fn test_h1_service_error() { assert_eq!(bytes, Bytes::from_static(b"error")); } -struct SetOnDrop(Arc); +struct SetOnDrop(Arc, Option<::oneshot::Sender<()>>); impl Drop for SetOnDrop { fn drop(&mut self) { self.0.fetch_add(1, Ordering::Relaxed); + let _ = self.1.take().unwrap().send(()); } } @@ -705,24 +736,28 @@ impl Drop for SetOnDrop { async fn test_h1_client_drop() -> io::Result<()> { let count = Arc::new(AtomicUsize::new(0)); let count2 = count.clone(); + let (tx, rx) = ::oneshot::channel(); + let tx = Arc::new(Mutex::new(Some(tx))); let srv = test_server(move || { + let tx = tx.clone(); let count = count2.clone(); HttpService::build().h1(move |req: Request| { + let tx = tx.clone(); let count = count.clone(); async move { - let _st = SetOnDrop(count); + let _st = SetOnDrop(count, tx.lock().unwrap().take()); assert!(req.peer_addr().is_some()); assert_eq!(req.version(), Version::HTTP_11); - sleep(Seconds(100)).await; + sleep(Millis(50000)).await; Ok::<_, io::Error>(Response::Ok().finish()) } }) }); - let result = timeout(Millis(100), srv.request(Method::GET, "/").send()).await; + let result = timeout(Millis(1500), srv.request(Method::GET, "/").send()).await; assert!(result.is_err()); - sleep(Millis(250)).await; + let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 1); Ok(()) } @@ -731,12 +766,18 @@ async fn test_h1_client_drop() -> io::Result<()> { async fn test_h1_gracefull_shutdown() { let count = Arc::new(AtomicUsize::new(0)); let count2 = count.clone(); + let (tx, rx) = ::oneshot::channel(); + let tx = Arc::new(Mutex::new(Some(tx))); let srv = test_server(move || { + let tx = tx.clone(); let count = count2.clone(); HttpService::build().h1(move |_: Request| { let count = count.clone(); count.fetch_add(1, Ordering::Relaxed); + if count.load(Ordering::Relaxed) == 2 { + let _ = tx.lock().unwrap().take().unwrap().send(()); + } async move { sleep(Millis(1000)).await; count.fetch_sub(1, Ordering::Relaxed); @@ -751,7 +792,7 @@ async fn test_h1_gracefull_shutdown() { let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap(); let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n"); - sleep(Millis(150)).await; + let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 2); let (tx, rx) = oneshot::channel(); @@ -759,8 +800,6 @@ async fn test_h1_gracefull_shutdown() { srv.stop().await; let _ = tx.send(()); }); - sleep(Millis(150)).await; - assert_eq!(count.load(Ordering::Relaxed), 2); let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 0); @@ -770,12 +809,18 @@ async fn test_h1_gracefull_shutdown() { async fn test_h1_gracefull_shutdown_2() { let count = Arc::new(AtomicUsize::new(0)); let count2 = count.clone(); + let (tx, rx) = ::oneshot::channel(); + let tx = Arc::new(Mutex::new(Some(tx))); let srv = test_server(move || { + let tx = tx.clone(); let count = count2.clone(); HttpService::build().finish(move |_: Request| { let count = count.clone(); count.fetch_add(1, Ordering::Relaxed); + if count.load(Ordering::Relaxed) == 2 { + let _ = tx.lock().unwrap().take().unwrap().send(()); + } async move { sleep(Millis(1000)).await; count.fetch_sub(1, Ordering::Relaxed); @@ -790,17 +835,14 @@ async fn test_h1_gracefull_shutdown_2() { let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap(); let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n"); - sleep(Millis(150)).await; - assert_eq!(count.load(Ordering::Relaxed), 2); + let _ = rx.await; + assert_eq!(count.load(Ordering::Acquire), 2); let (tx, rx) = oneshot::channel(); rt::spawn(async move { srv.stop().await; let _ = tx.send(()); }); - sleep(Millis(150)).await; - assert_eq!(count.load(Ordering::Relaxed), 2); - let _ = rx.await; assert_eq!(count.load(Ordering::Relaxed), 0); } diff --git a/ntex/tests/web_server.rs b/ntex/tests/web_server.rs index b7bf1b75..a1ab4ace 100644 --- a/ntex/tests/web_server.rs +++ b/ntex/tests/web_server.rs @@ -1,6 +1,5 @@ use std::{future::Future, io, io::Read, io::Write, pin::Pin, task::Context, task::Poll}; -use brotli2::write::{BrotliDecoder, BrotliEncoder}; use flate2::read::GzDecoder; use flate2::write::{GzEncoder, ZlibDecoder, ZlibEncoder}; use flate2::Compression; @@ -318,36 +317,6 @@ async fn test_body_chunked_implicit() { assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref())); } -#[ntex::test] -async fn test_body_br_streaming() { - let srv = test::server_with(test::config().h1(), || { - App::new().wrap(Compress::new(ContentEncoding::Br)).service( - web::resource("/").route(web::to(move || async { - HttpResponse::Ok() - .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24)) - })), - ) - }); - - let mut response = srv - .get("/") - .header(ACCEPT_ENCODING, "br") - .no_decompress() - .send() - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - - // decode br - let mut e = BrotliDecoder::new(Vec::with_capacity(2048)); - e.write_all(bytes.as_ref()).unwrap(); - let dec = e.finish().unwrap(); - assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref())); -} - #[ntex::test] async fn test_head_binary() { let srv = test::server_with(test::config().h1(), || { @@ -422,35 +391,6 @@ async fn test_body_deflate() { assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref())); } -#[ntex::test] -async fn test_body_brotli() { - let srv = test::server_with(test::config().h1(), || { - App::new().wrap(Compress::new(ContentEncoding::Br)).service( - web::resource("/") - .route(web::to(move || async { HttpResponse::Ok().body(STR) })), - ) - }); - - // client request - let mut response = srv - .get("/") - .header(ACCEPT_ENCODING, "br") - .no_decompress() - .send() - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - - // decode brotli - let mut e = BrotliDecoder::new(Vec::with_capacity(2048)); - e.write_all(bytes.as_ref()).unwrap(); - let dec = e.finish().unwrap(); - assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref())); -} - #[ntex::test] async fn test_encoding() { let srv = test::server_with(test::config().h1(), || { @@ -644,204 +584,6 @@ async fn test_reading_deflate_encoding_large_random() { assert_eq!(bytes, Bytes::from(data)); } -#[ntex::test] -async fn test_brotli_encoding() { - let srv = test::server_with(test::config().h1(), || { - App::new().service(web::resource("/").route(web::to(move |body: Bytes| async { - HttpResponse::Ok().body(body) - }))) - }); - - let mut e = BrotliEncoder::new(Vec::new(), 5); - e.write_all(STR.as_ref()).unwrap(); - let enc = e.finish().unwrap(); - - // client request - let request = srv - .post("/") - .header(CONTENT_ENCODING, "br") - .send_body(enc.clone()); - let mut response = request.await.unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes, Bytes::from_static(STR.as_ref())); -} - -#[ntex::test] -async fn test_brotli_encoding_large() { - let data = rand::thread_rng() - .sample_iter(&Alphanumeric) - .take(320_000) - .map(char::from) - .collect::(); - - let srv = test::server_with(test::config().h1(), || { - App::new().service( - web::resource("/") - .state(web::types::PayloadConfig::new(320_000)) - .route(web::to(move |body: Bytes| async { - HttpResponse::Ok().streaming(TestBody::new(body, 10240)) - })), - ) - }); - - let mut e = BrotliEncoder::new(Vec::new(), 5); - e.write_all(data.as_ref()).unwrap(); - let enc = e.finish().unwrap(); - - // client request - let request = srv - .post("/") - .header(CONTENT_ENCODING, "br") - .send_body(enc.clone()); - let mut response = request.await.unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().limit(320_000).await.unwrap(); - assert_eq!(bytes, Bytes::from(data)); -} - -#[cfg(feature = "openssl")] -#[ntex::test] -async fn test_brotli_encoding_large_openssl() { - // load ssl keys - use tls_openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; - - let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("./tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("./tests/cert.pem") - .unwrap(); - - let data = STR.repeat(10); - let srv = test::server_with(test::config().openssl(builder.build()), move || { - App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async { - HttpResponse::Ok() - .encoding(ContentEncoding::Identity) - .body(bytes) - }))) - }); - - // body - let mut e = BrotliEncoder::new(Vec::new(), 3); - e.write_all(data.as_ref()).unwrap(); - let enc = e.finish().unwrap(); - - // client request - let mut response = srv - .post("/") - .header(CONTENT_ENCODING, "br") - .send_body(enc) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes, Bytes::from(data)); -} - -#[cfg(feature = "openssl")] -#[ntex::test] -async fn test_brotli_encoding_large_openssl_h1() { - // load ssl keys - use tls_openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; - - let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("./tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("./tests/cert.pem") - .unwrap(); - - let data = STR.repeat(10); - let srv = test::server_with(test::config().openssl(builder.build()).h1(), move || { - App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async { - HttpResponse::Ok() - .encoding(ContentEncoding::Identity) - .body(bytes) - }))) - }); - - // body - let mut e = BrotliEncoder::new(Vec::new(), 3); - e.write_all(data.as_ref()).unwrap(); - let enc = e.finish().unwrap(); - - // client request - let mut response = srv - .post("/") - .header(CONTENT_ENCODING, "br") - .send_body(enc) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes, Bytes::from(data)); -} - -#[cfg(feature = "openssl")] -#[ntex::test] -async fn test_brotli_encoding_large_openssl_h2() { - // load ssl keys - use tls_openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod}; - - let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("./tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("./tests/cert.pem") - .unwrap(); - builder.set_alpn_select_callback(|_, protos| { - const H2: &[u8] = b"\x02h2"; - const H11: &[u8] = b"\x08http/1.1"; - if protos.windows(3).any(|window| window == H2) { - Ok(b"h2") - } else if protos.windows(9).any(|window| window == H11) { - Ok(b"http/1.1") - } else { - Err(AlpnError::NOACK) - } - }); - builder.set_alpn_protos(b"\x08http/1.1\x02h2").unwrap(); - - let data = STR.repeat(10); - let srv = test::server_with(test::config().openssl(builder.build()).h2(), move || { - App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async { - HttpResponse::Ok() - .encoding(ContentEncoding::Identity) - .body(bytes) - }))) - }); - - // body - let mut e = BrotliEncoder::new(Vec::new(), 3); - e.write_all(data.as_ref()).unwrap(); - let enc = e.finish().unwrap(); - - // client request - let mut response = srv - .post("/") - .header(CONTENT_ENCODING, "br") - .send_body(enc) - .await - .unwrap(); - assert!(response.status().is_success()); - - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes, Bytes::from(data)); -} - #[cfg(all(feature = "rustls", feature = "openssl"))] #[ntex::test] async fn test_reading_deflate_encoding_large_random_rustls() {