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

wip

Browse source
This commit is contained in:
Nikolay Kim 2025-02-28 13:41:05 -05:00
parent 60cba94f27
commit 4c9c1adece
41 changed files with 4003 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.github/workflows/cov.yml vendored
View file

@ -36,6 +36,9 @@ jobs:
- 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 (default)
run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/default,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
- name: Generate coverage report
run: cargo llvm-cov report --lcov --output-path lcov.info --ignore-filename-regex="ntex-compio|ntex-tokio|ntex-glommio|ntex-async-std"

5
.github/workflows/linux.yml vendored
View file

@ -54,6 +54,11 @@ jobs:
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 (default)
timeout-minutes: 40
run: |
cargo test --all --no-default-features --features="ntex/default-rt,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
- name: Run tests (async-std)
timeout-minutes: 40
continue-on-error: true

33
Cargo.toml
View file

@ -5,9 +5,11 @@ members = [
"ntex-bytes",
"ntex-codec",
"ntex-io",
"ntex-iodriver",
"ntex-http",
"ntex-router",
"ntex-rt",
"ntex-runtime",
"ntex-net",
"ntex-server",
"ntex-service",
@ -21,15 +23,25 @@ members = [
"ntex-tokio",
]
[workspace.package]
authors = ["ntex contributors <team@ntex.rs>"]
repository = "https://github.com/ntex-rs/ntex"
documentation = "https://docs.rs/ntex/"
license = "MIT OR Apache-2.0"
edition = "2021"
rust-version = "1.75"
[patch.crates-io]
ntex = { path = "ntex" }
ntex-bytes = { path = "ntex-bytes" }
ntex-codec = { path = "ntex-codec" }
ntex-io = { path = "ntex-io" }
ntex-iodriver = { path = "ntex-iodriver" }
ntex-net = { path = "ntex-net" }
ntex-http = { path = "ntex-http" }
ntex-router = { path = "ntex-router" }
ntex-rt = { path = "ntex-rt" }
ntex-runtime = { path = "ntex-runtime" }
ntex-server = { path = "ntex-server" }
ntex-service = { path = "ntex-service" }
ntex-tls = { path = "ntex-tls" }
@ -41,8 +53,19 @@ ntex-glommio = { path = "ntex-glommio" }
ntex-tokio = { path = "ntex-tokio" }
ntex-async-std = { path = "ntex-async-std" }
compio-buf = { path = "../dev/compio/compio-buf", optional = true }
compio-io = { path = "../dev/compio/compio-io", optional = true }
compio-net = { path = "../dev/compio/compio-net", optional = true }
compio-driver = { path = "../dev/compio/compio-driver", optional = true }
compio-runtime = { path = "../dev/compio/compio-runtime", optional = true }
[workspace.dependencies]
async-task = "4.5.0"
bitflags = "2"
cfg_aliases = "0.2.1"
cfg-if = "1.0.0"
crossbeam-channel = "0.5.8"
crossbeam-queue = "0.3.8"
futures-util = "0.3.29"
fxhash = "0.2"
libc = "0.2.164"
log = "0.4"
scoped-tls = "1.0.1"
slab = "0.4.9"
socket2 = "0.5.6"
windows-sys = "0.52.0"
thiserror = "1"

14
ntex-compio/Cargo.toml
View file

@ -21,10 +21,12 @@ ntex-bytes = "0.1"
ntex-io = "2.5"
ntex-util = "2"
ntex-rt = "0.4"
log = "0.4"
ntex-iodriver = "0.1"
compio-buf = "0.5"
compio-io = "0.5"
compio-net = "0.6"
compio-driver = "0.6"
compio-runtime = { version = "0.6", features = ["io-uring", "polling", "event"] }
bitflags = "2"
libc = "0.2.164"
log = "0.4"
slab = "0.4"
socket2 = "0.5.6"
compio-runtime = { version = "0.6", features = ["io-uring", "polling"] }

21
ntex-io/src/buf.rs
View file

@ -152,6 +152,27 @@ impl Stack {
}
}
pub(crate) fn with_read_source<F, R>(&self, io: &IoRef, f: F) -> R
where
F: FnOnce(&mut BytesVec) -> R,
{
let item = self.get_last_level();
let mut rb = item.0.take();
if rb.is_none() {
rb = Some(io.memory_pool().get_read_buf());
}
let result = f(rb.as_mut().unwrap());
if let Some(b) = rb {
if b.is_empty() {
io.memory_pool().release_read_buf(b);
} else {
item.0.set(Some(b));
}
}
result
}
pub(crate) fn with_read_destination<F, R>(&self, io: &IoRef, f: F) -> R
where
F: FnOnce(&mut BytesVec) -> R,

10
ntex-io/src/flags.rs
View file

@ -25,6 +25,8 @@ bitflags::bitflags! {
/// write task paused
const WR_PAUSED = 0b0000_0100_0000_0000;
/// wait for write completion task
const WR_TASK_WAIT = 0b0000_1000_0000_0000;
/// dispatcher is marked stopped
const DSP_STOP = 0b0001_0000_0000_0000;
@ -38,6 +40,10 @@ impl Flags {
self.intersects(Flags::IO_STOPPED)
}
pub(crate) fn is_task_waiting_for_write(&self) -> bool {
self.contains(Flags::WR_TASK_WAIT)
}
pub(crate) fn is_waiting_for_write(&self) -> bool {
self.intersects(Flags::BUF_W_MUST_FLUSH | Flags::BUF_W_BACKPRESSURE)
}
@ -46,6 +52,10 @@ impl Flags {
self.remove(Flags::BUF_W_MUST_FLUSH | Flags::BUF_W_BACKPRESSURE);
}
pub(crate) fn task_waiting_for_write_is_done(&mut self) {
self.remove(Flags::WR_TASK_WAIT);
}
pub(crate) fn is_read_buf_ready(&self) -> bool {
self.contains(Flags::BUF_R_READY)
}

4
ntex-io/src/lib.rs
View file

@ -53,7 +53,9 @@ pub trait AsyncWrite {
/// Status for read task
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub enum ReadStatus {
/// Read task is clear to proceed with read operation
Ready,
/// Terminate read task
Terminate,
}
@ -75,7 +77,7 @@ pub trait FilterLayer: fmt::Debug + 'static {
#[inline]
/// Check readiness for read operations
fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<ReadStatus> {
fn poll_read_ready(&self, waker: &mut Context<'_>) -> Poll<ReadStatus> {
Poll::Ready(ReadStatus::Ready)
}

351
ntex-io/src/tasks.rs
View file

@ -1,6 +1,6 @@
use std::{cell::Cell, fmt, future::poll_fn, io, task::Context, task::Poll};
use std::{cell::Cell, fmt, future::poll_fn, io, task::ready, task::Context, task::Poll};
use ntex_bytes::{BufMut, BytesVec};
use ntex_bytes::{Buf, BufMut, BytesVec};
use ntex_util::{future::lazy, future::select, future::Either, time::sleep, time::Sleep};
use crate::{AsyncRead, AsyncWrite, Flags, IoRef, ReadStatus, WriteStatus};
@ -19,14 +19,30 @@ impl ReadContext {
Self(io.clone(), Cell::new(None))
}
pub fn clone(&self) -> Self {
Self(self.0.clone(), Cell::new(None))
}
#[inline]
/// Io tag
pub fn tag(&self) -> &'static str {
self.0.tag()
}
#[inline]
/// Io tag
pub fn io(&self) -> IoRef {
self.0.clone()
}
#[inline]
/// Check readiness for read operations
pub fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<ReadStatus> {
self.0.filter().poll_read_ready(cx)
}
/// Wait when io get closed or preparing for close
async fn wait_for_close(&self) {
pub async fn wait_for_close(&self) {
poll_fn(|cx| {
let flags = self.0.flags();
@ -43,6 +59,111 @@ impl ReadContext {
.await
}
#[inline]
/// Get io error
pub fn set_stopped(&self, e: Option<io::Error>) {
self.0 .0.io_stopped(e);
}
/// Get read buffer
pub fn with_buf<F>(&self, f: F) -> Poll<()>
where
F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
{
let inner = &self.0 .0;
let (hw, lw) = self.0.memory_pool().read_params().unpack();
let result = inner.buffer.with_read_source(&self.0, |buf| {
// make sure we've got room
let remaining = buf.remaining_mut();
if remaining < lw {
buf.reserve(hw - remaining);
}
// call provided callback
f(buf)
});
// handle buffer changes
match result {
Poll::Ready(Ok(0)) => {
inner.io_stopped(None);
Poll::Ready(())
}
Poll::Ready(Ok(nbytes)) => {
let filter = self.0.filter();
let _ = filter
.process_read_buf(&self.0, &inner.buffer, 0, nbytes)
.and_then(|status| {
if status.nbytes > 0 {
// dest buffer has new data, wake up dispatcher
if inner.buffer.read_destination_size() >= hw {
log::trace!(
"{}: Io read buffer is too large {}, enable read back-pressure",
self.0.tag(),
nbytes
);
inner.insert_flags(Flags::BUF_R_READY | Flags::BUF_R_FULL);
} else {
inner.insert_flags(Flags::BUF_R_READY);
if nbytes >= hw {
// read task is paused because of read back-pressure
// but there is no new data in top most read buffer
// so we need to wake up read task to read more data
// otherwise read task would sleep forever
inner.read_task.wake();
}
}
log::trace!(
"{}: New {} bytes available, wakeup dispatcher",
self.0.tag(),
nbytes
);
inner.dispatch_task.wake();
} else {
if nbytes >= hw {
// read task is paused because of read back-pressure
// but there is no new data in top most read buffer
// so we need to wake up read task to read more data
// otherwise read task would sleep forever
inner.read_task.wake();
}
if inner.flags.get().contains(Flags::RD_NOTIFY) {
// in case of "notify" we must wake up dispatch task
// if we read any data from source
inner.dispatch_task.wake();
}
}
// while reading, filter wrote some data
// in that case filters need to process write buffers
// and potentialy wake write task
if status.need_write {
filter.process_write_buf(&self.0, &inner.buffer, 0)
} else {
Ok(())
}
})
.map_err(|err| {
inner.dispatch_task.wake();
inner.io_stopped(Some(err));
inner.insert_flags(Flags::BUF_R_READY);
});
Poll::Pending
}
Poll::Ready(Err(e)) => {
inner.io_stopped(Some(e));
Poll::Ready(())
}
Poll::Pending => {
if inner.flags.get().contains(Flags::IO_STOPPING_FILTERS) {
shutdown_filters(&self.0);
}
Poll::Pending
}
}
}
/// Handle read io operations
pub async fn handle<T>(&self, io: &mut T)
where
@ -166,18 +287,64 @@ impl ReadContext {
}
}
fn shutdown_filters(&self, cx: &mut Context<'_>) {
pub fn shutdown_filters(&self, cx: &mut Context<'_>) {
let st = &self.0 .0;
let filter = self.0.filter();
match filter.shutdown(&self.0, &st.buffer, 0) {
if st.flags.get().contains(Flags::IO_STOPPING_FILTERS) {
let filter = self.0.filter();
match filter.shutdown(&self.0, &st.buffer, 0) {
Ok(Poll::Ready(())) => {
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
}
Ok(Poll::Pending) => {
let flags = st.flags.get();
// check read buffer, if buffer is not consumed it is unlikely
// that filter will properly complete shutdown
if flags.contains(Flags::RD_PAUSED)
|| flags.contains(Flags::BUF_R_FULL | Flags::BUF_R_READY)
{
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
} else {
// filter shutdown timeout
let timeout = self
.1
.take()
.unwrap_or_else(|| sleep(st.disconnect_timeout.get()));
if timeout.poll_elapsed(cx).is_ready() {
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
} else {
self.1.set(Some(timeout));
}
}
}
Err(err) => {
st.io_stopped(Some(err));
}
}
if let Err(err) = filter.process_write_buf(&self.0, &st.buffer, 0) {
st.io_stopped(Some(err));
}
}
}
}
fn shutdown_filters(io: &IoRef) {
let st = &io.0;
let flags = st.flags.get();
if !flags.intersects(Flags::IO_STOPPED | Flags::IO_STOPPING) {
let filter = io.filter();
match filter.shutdown(io, &st.buffer, 0) {
Ok(Poll::Ready(())) => {
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
}
Ok(Poll::Pending) => {
let flags = st.flags.get();
// check read buffer, if buffer is not consumed it is unlikely
// that filter will properly complete shutdown
if flags.contains(Flags::RD_PAUSED)
@ -185,25 +352,13 @@ impl ReadContext {
{
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
} else {
// filter shutdown timeout
let timeout = self
.1
.take()
.unwrap_or_else(|| sleep(st.disconnect_timeout.get()));
if timeout.poll_elapsed(cx).is_ready() {
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
} else {
self.1.set(Some(timeout));
}
}
}
Err(err) => {
st.io_stopped(Some(err));
}
}
if let Err(err) = filter.process_write_buf(&self.0, &st.buffer, 0) {
if let Err(err) = filter.process_write_buf(io, &st.buffer, 0) {
st.io_stopped(Some(err));
}
}
@ -225,6 +380,10 @@ impl WriteContext {
Self(io.clone())
}
pub fn clone(&self) -> Self {
Self(self.0.clone())
}
#[inline]
/// Io tag
pub fn tag(&self) -> &'static str {
@ -236,6 +395,12 @@ impl WriteContext {
poll_fn(|cx| self.0.filter().poll_write_ready(cx)).await
}
#[inline]
/// Check readiness for write operations
pub fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<WriteStatus> {
self.0.filter().poll_write_ready(cx)
}
/// Indicate that write io task is stopped
fn close(&self, err: Option<io::Error>) {
self.0 .0.io_stopped(err);
@ -254,6 +419,62 @@ impl WriteContext {
.await
}
/// Wait when io get closed or preparing for close
pub async fn wait_for_shutdown(&self, flush_buf: bool) {
let st = &self.0 .0;
// filter shutdown timeout
let mut timeout = None;
poll_fn(|cx| {
let flags = self.0.flags();
if flags.intersects(Flags::IO_STOPPING | Flags::IO_STOPPED) {
Poll::Ready(())
} else {
st.write_task.register(cx.waker());
if flags.contains(Flags::IO_STOPPING_FILTERS) {
if timeout.is_none() {
timeout = Some(sleep(st.disconnect_timeout.get()));
}
if timeout.as_ref().unwrap().poll_elapsed(cx).is_ready() {
st.dispatch_task.wake();
st.insert_flags(Flags::IO_STOPPING);
return Poll::Ready(());
}
}
Poll::Pending
}
})
.await;
if flush_buf {
if !self.0.flags().contains(Flags::WR_PAUSED) {
st.insert_flags(Flags::WR_TASK_WAIT);
poll_fn(|cx| {
let flags = self.0.flags();
if flags.intersects(Flags::WR_PAUSED | Flags::IO_STOPPED) {
Poll::Ready(())
} else {
st.write_task.register(cx.waker());
if timeout.is_none() {
timeout = Some(sleep(st.disconnect_timeout.get()));
}
if timeout.as_ref().unwrap().poll_elapsed(cx).is_ready() {
Poll::Ready(())
} else {
Poll::Pending
}
}
})
.await;
}
}
}
/// Handle write io operations
pub async fn handle<T>(&self, io: &mut T)
where
@ -297,6 +518,94 @@ impl WriteContext {
return;
}
}
/// Get write buffer
pub fn with_buf<F>(&self, f: F) -> Poll<()>
where
F: FnOnce(&BytesVec) -> Poll<io::Result<usize>>,
{
let inner = &self.0 .0;
// call provided callback
let result = inner.buffer.with_write_destination(&self.0, |buf| {
let buf = if let Some(buf) = buf {
buf
} else {
return Poll::Ready(Ok(0));
};
if buf.is_empty() {
return Poll::Ready(Ok(0));
}
let result = ready!(f(buf));
match result {
Ok(n) => {
if n == 0 {
log::trace!(
"{}: Disconnected during flush, written {}",
self.tag(),
n
);
Poll::Ready(Err(io::Error::new(
io::ErrorKind::WriteZero,
"failed to write frame to transport",
)))
} else {
if n == buf.len() {
buf.clear();
Poll::Ready(Ok(0))
} else {
buf.advance(n);
Poll::Ready(Ok(buf.len()))
}
}
}
Err(e) => Poll::Ready(Err(e)),
}
});
let mut flags = inner.flags.get();
let result = match result {
Poll::Pending => {
flags.remove(Flags::WR_PAUSED);
Poll::Pending
}
Poll::Ready(Ok(len)) => {
// if write buffer is smaller than high watermark value, turn off back-pressure
if len == 0 {
flags.insert(Flags::WR_PAUSED);
if flags.is_task_waiting_for_write() {
flags.task_waiting_for_write_is_done();
inner.write_task.wake();
}
if flags.is_waiting_for_write() {
flags.waiting_for_write_is_done();
inner.dispatch_task.wake();
}
Poll::Ready(())
} else if flags.contains(Flags::BUF_W_BACKPRESSURE)
&& len < inner.pool.get().write_params_high() << 1
{
flags.remove(Flags::BUF_W_BACKPRESSURE);
inner.dispatch_task.wake();
Poll::Pending
} else {
Poll::Pending
}
}
Poll::Ready(Err(e)) => {
self.close(Some(e));
Poll::Ready(())
}
};
inner.flags.set(flags);
result
}
}
impl WriteContextBuf {

80
ntex-iodriver/Cargo.toml Normal file
View file

@ -0,0 +1,80 @@
[package]
name = "ntex-iodriver"
version = "0.1.0"
description = "Low-level driver for ntex"
categories = ["asynchronous"]
keywords = ["async", "iocp", "io-uring"]
edition = { workspace = true }
authors = { workspace = true }
readme = { workspace = true }
license = { workspace = true }
repository = { workspace = true }
[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]
default-target = "x86_64-unknown-linux-gnu"
targets = [
"x86_64-pc-windows-gnu",
"x86_64-unknown-linux-gnu",
"x86_64-apple-darwin",
"aarch64-apple-ios",
"aarch64-linux-android",
"x86_64-unknown-dragonfly",
"x86_64-unknown-freebsd",
"x86_64-unknown-illumos",
"x86_64-unknown-netbsd",
"x86_64-unknown-openbsd",
]
[dependencies]
bitflags = { workspace = true }
log = { workspace = true }
cfg-if = { workspace = true }
crossbeam-channel = { workspace = true }
socket2 = { workspace = true }
slab = { workspace = true }
# Windows specific dependencies
[target.'cfg(windows)'.dependencies]
aligned-array = "1.0.1"
windows-sys = { workspace = true, features = [
"Win32_Foundation",
"Win32_Networking_WinSock",
"Win32_Security",
"Win32_Storage_FileSystem",
"Win32_System_Console",
"Win32_System_IO",
"Win32_System_Pipes",
"Win32_System_SystemServices",
"Win32_System_Threading",
"Win32_System_WindowsProgramming",
] }
# Linux specific dependencies
[target.'cfg(target_os = "linux")'.dependencies]
io-uring = { version = "0.7.0", optional = true }
polling = { version = "3.3.0", optional = true }
# Other platform dependencies
[target.'cfg(all(not(target_os = "linux"), unix))'.dependencies]
polling = "3.3.0"
[target.'cfg(unix)'.dependencies]
crossbeam-channel = { workspace = true }
crossbeam-queue = { workspace = true }
libc = { workspace = true }
[build-dependencies]
cfg_aliases = { workspace = true }
[features]
default = ["io-uring"]
polling = ["dep:polling"]
io-uring-sqe128 = []
io-uring-cqe32 = []
io-uring-socket = []
iocp-global = []
iocp-wait-packet = []

1
ntex-iodriver/LICENSE-APACHE Symbolic link
View file

@ -0,0 +1 @@
../LICENSE-APACHE

1
ntex-iodriver/LICENSE-MIT Symbolic link
View file

@ -0,0 +1 @@
../LICENSE-MIT

13
ntex-iodriver/README.md Normal file
View file

@ -0,0 +1,13 @@
---
# ntex
[![MIT licensed](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/compio-rs/compio/blob/master/LICENSE)
[![crates.io](https://img.shields.io/crates/v/compio)](https://crates.io/crates/compio)
[![docs.rs](https://img.shields.io/badge/docs.rs-compio-latest)](https://docs.rs/compio)
[![Check](https://github.com/compio-rs/compio/actions/workflows/ci_check.yml/badge.svg)](https://github.com/compio-rs/compio/actions/workflows/ci_check.yml)
[![Test](https://github.com/compio-rs/compio/actions/workflows/ci_test.yml/badge.svg)](https://github.com/compio-rs/compio/actions/workflows/ci_test.yml)
[![Telegram](https://img.shields.io/badge/Telegram-compio--rs-blue?logo=telegram)](https://t.me/compio_rs)
A specialized runtime for ntex framework with IOCP/io_uring/polling support.
This crate is inspired by [compio](https://github.com/compio-rs/compio/).

128
ntex-iodriver/src/asyncify.rs Normal file
View file

@ -0,0 +1,128 @@
use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc};
use std::{fmt, time::Duration};
use crossbeam_channel::{bounded, Receiver, Sender, TrySendError};
/// An error that may be emitted when all worker threads are busy. It simply
/// returns the dispatchable value with a convenient [`fmt::Debug`] and
/// [`fmt::Display`] implementation.
#[derive(Copy, Clone, PartialEq, Eq)]
pub struct DispatchError<T>(pub T);
impl<T> DispatchError<T> {
/// Consume the error, yielding the dispatchable that failed to be sent.
pub fn into_inner(self) -> T {
self.0
}
}
impl<T> fmt::Debug for DispatchError<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
"DispatchError(..)".fmt(f)
}
}
impl<T> fmt::Display for DispatchError<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
"all threads are busy".fmt(f)
}
}
impl<T> std::error::Error for DispatchError<T> {}
type BoxedDispatchable = Box<dyn Dispatchable + Send>;
/// A trait for dispatching a closure. It's implemented for all `FnOnce() + Send
/// + 'static` but may also be implemented for any other types that are `Send`
/// and `'static`.
pub trait Dispatchable: Send + 'static {
/// Run the dispatchable
fn run(self: Box<Self>);
}
impl<F> Dispatchable for F
where
F: FnOnce() + Send + 'static,
{
fn run(self: Box<Self>) {
(*self)()
}
}
struct CounterGuard(Arc<AtomicUsize>);
impl Drop for CounterGuard {
fn drop(&mut self) {
self.0.fetch_sub(1, Ordering::AcqRel);
}
}
fn worker(
receiver: Receiver<BoxedDispatchable>,
counter: Arc<AtomicUsize>,
timeout: Duration,
) -> impl FnOnce() {
move || {
counter.fetch_add(1, Ordering::AcqRel);
let _guard = CounterGuard(counter);
while let Ok(f) = receiver.recv_timeout(timeout) {
f.run();
}
}
}
/// A thread pool to perform blocking operations in other threads.
#[derive(Debug, Clone)]
pub struct AsyncifyPool {
sender: Sender<BoxedDispatchable>,
receiver: Receiver<BoxedDispatchable>,
counter: Arc<AtomicUsize>,
thread_limit: usize,
recv_timeout: Duration,
}
impl AsyncifyPool {
/// Create [`AsyncifyPool`] with thread number limit and channel receive
/// timeout.
pub fn new(thread_limit: usize, recv_timeout: Duration) -> Self {
let (sender, receiver) = bounded(0);
Self {
sender,
receiver,
counter: Arc::new(AtomicUsize::new(0)),
thread_limit,
recv_timeout,
}
}
/// Send a dispatchable, usually a closure, to another thread. Usually the
/// user should not use it. When all threads are busy and thread number
/// limit has been reached, it will return an error with the original
/// dispatchable.
pub fn dispatch<D: Dispatchable>(&self, f: D) -> Result<(), DispatchError<D>> {
match self.sender.try_send(Box::new(f) as BoxedDispatchable) {
Ok(_) => Ok(()),
Err(e) => match e {
TrySendError::Full(f) => {
if self.counter.load(Ordering::Acquire) >= self.thread_limit {
// Safety: we can ensure the type
Err(DispatchError(*unsafe {
Box::from_raw(Box::into_raw(f).cast())
}))
} else {
std::thread::spawn(worker(
self.receiver.clone(),
self.counter.clone(),
self.recv_timeout,
));
self.sender.send(f).expect("the channel should not be full");
Ok(())
}
}
TrySendError::Disconnected(_) => {
unreachable!("receiver should not all disconnected")
}
},
}
}
}

115
ntex-iodriver/src/driver_type.rs Normal file
View file

@ -0,0 +1,115 @@
use std::sync::atomic::{AtomicU8, Ordering};
const UNINIT: u8 = u8::MAX;
const IO_URING: u8 = 0;
const POLLING: u8 = 1;
const IOCP: u8 = 2;
static DRIVER_TYPE: AtomicU8 = AtomicU8::new(UNINIT);
/// Representing underlying driver type the fusion driver is using
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum DriverType {
/// Using `polling` driver
Poll = POLLING,
/// Using `io-uring` driver
IoUring = IO_URING,
/// Using `iocp` driver
IOCP = IOCP,
}
impl DriverType {
fn from_num(n: u8) -> Self {
match n {
IO_URING => Self::IoUring,
POLLING => Self::Poll,
IOCP => Self::IOCP,
_ => unreachable!("invalid driver type"),
}
}
/// Get the underlying driver type
fn get() -> DriverType {
cfg_if::cfg_if! {
if #[cfg(windows)] {
DriverType::IOCP
} else if #[cfg(all(target_os = "linux", feature = "polling", feature = "io-uring"))] {
use io_uring::opcode::*;
// Add more opcodes here if used
const USED_OP: &[u8] = &[
Read::CODE,
Readv::CODE,
Write::CODE,
Writev::CODE,
Fsync::CODE,
Accept::CODE,
Connect::CODE,
RecvMsg::CODE,
SendMsg::CODE,
AsyncCancel::CODE,
OpenAt::CODE,
Close::CODE,
Shutdown::CODE,
// Linux kernel 5.19
#[cfg(any(
feature = "io-uring-sqe128",
feature = "io-uring-cqe32",
feature = "io-uring-socket"
))]
Socket::CODE,
];
(|| {
let uring = io_uring::IoUring::new(2)?;
let mut probe = io_uring::Probe::new();
uring.submitter().register_probe(&mut probe)?;
if USED_OP.iter().all(|op| probe.is_supported(*op)) {
std::io::Result::Ok(DriverType::IoUring)
} else {
Ok(DriverType::Poll)
}
})()
.unwrap_or(DriverType::Poll) // Should we fail here?
} else if #[cfg(all(target_os = "linux", feature = "io-uring"))] {
DriverType::IoUring
} else if #[cfg(unix)] {
DriverType::Poll
} else {
compile_error!("unsupported platform");
}
}
}
/// Get the underlying driver type and cache it. Following calls will return
/// the cached value.
pub fn current() -> DriverType {
match DRIVER_TYPE.load(Ordering::Acquire) {
UNINIT => {}
x => return DriverType::from_num(x),
}
let dev_ty = Self::get();
DRIVER_TYPE.store(dev_ty as u8, Ordering::Release);
dev_ty
}
/// Check if the current driver is `polling`
pub fn is_polling() -> bool {
Self::current() == DriverType::Poll
}
/// Check if the current driver is `io-uring`
pub fn is_iouring() -> bool {
Self::current() == DriverType::IoUring
}
/// Check if the current driver is `iocp`
pub fn is_iocp() -> bool {
Self::current() == DriverType::IOCP
}
}

223
ntex-iodriver/src/key.rs Normal file
View file

@ -0,0 +1,223 @@
use std::{io, marker::PhantomData, mem::MaybeUninit, pin::Pin, task::Waker};
use crate::{OpCode, Overlapped, PushEntry, RawFd};
/// An operation with other needed information. It should be allocated on the
/// heap. The pointer to this struct is used as `user_data`, and on Windows, it
/// is used as the pointer to `OVERLAPPED`.
///
/// `*const RawOp<dyn OpCode>` can be obtained from any `Key<T: OpCode>` by
/// first casting `Key::user_data` to `*const RawOp<()>`, then upcasted with
/// `upcast_fn`. It is done in [`Key::as_op_pin`].
#[repr(C)]
pub(crate) struct RawOp<T: ?Sized> {
header: Overlapped,
// The cancelled flag and the result here are manual reference counting. The driver holds the
// strong ref until it completes; the runtime holds the strong ref until the future is
// dropped.
cancelled: bool,
// The metadata in `*mut RawOp<dyn OpCode>`
metadata: usize,
result: PushEntry<Option<Waker>, io::Result<usize>>,
flags: u32,
op: T,
}
#[repr(C)]
union OpCodePtrRepr {
ptr: *mut RawOp<dyn OpCode>,
components: OpCodePtrComponents,
}
#[repr(C)]
#[derive(Clone, Copy)]
struct OpCodePtrComponents {
data_pointer: *mut (),
metadata: usize,
}
fn opcode_metadata<T: OpCode + 'static>() -> usize {
let mut op = MaybeUninit::<RawOp<T>>::uninit();
// SAFETY: same as `core::ptr::metadata`.
unsafe {
OpCodePtrRepr {
ptr: op.as_mut_ptr(),
}
.components
.metadata
}
}
const unsafe fn opcode_dyn_mut(ptr: *mut (), metadata: usize) -> *mut RawOp<dyn OpCode> {
OpCodePtrRepr {
components: OpCodePtrComponents {
metadata,
data_pointer: ptr,
},
}
.ptr
}
/// A typed wrapper for key of Ops submitted into driver. It doesn't free the
/// inner on dropping. Instead, the memory is managed by the proactor. The inner
/// is only freed when:
///
/// 1. The op is completed and the future asks the result. `into_inner` will be
/// called by the proactor.
/// 2. The op is completed and the future cancels it. `into_box` will be called
/// by the proactor.
#[derive(PartialEq, Eq, Hash)]
pub struct Key<T: ?Sized> {
user_data: *mut (),
_p: PhantomData<Box<RawOp<T>>>,
}
impl<T: ?Sized> Unpin for Key<T> {}
impl<T: OpCode + 'static> Key<T> {
/// Create [`RawOp`] and get the [`Key`] to it.
pub(crate) fn new(driver: RawFd, op: T) -> Self {
let header = Overlapped::new(driver);
let raw_op = Box::new(RawOp {
header,
cancelled: false,
metadata: opcode_metadata::<T>(),
result: PushEntry::Pending(None),
flags: 0,
op,
});
unsafe { Self::new_unchecked(Box::into_raw(raw_op) as _) }
}
}
impl<T: ?Sized> Key<T> {
/// Create a new `Key` with the given user data.
///
/// # Safety
///
/// Caller needs to ensure that `T` does correspond to `user_data` in driver
/// this `Key` is created with. In most cases, it is enough to let `T` be
/// `dyn OpCode`.
pub unsafe fn new_unchecked(user_data: usize) -> Self {
Self {
user_data: user_data as _,
_p: PhantomData,
}
}
/// Get the unique user-defined data.
pub fn user_data(&self) -> usize {
self.user_data as _
}
fn as_opaque(&self) -> &RawOp<()> {
// SAFETY: user_data is unique and RawOp is repr(C).
unsafe { &*(self.user_data as *const RawOp<()>) }
}
fn as_opaque_mut(&mut self) -> &mut RawOp<()> {
// SAFETY: see `as_opaque`.
unsafe { &mut *(self.user_data as *mut RawOp<()>) }
}
fn as_dyn_mut_ptr(&mut self) -> *mut RawOp<dyn OpCode> {
let user_data = self.user_data;
let this = self.as_opaque_mut();
// SAFETY: metadata from `Key::new`.
unsafe { opcode_dyn_mut(user_data, this.metadata) }
}
/// A pointer to OVERLAPPED.
#[cfg(windows)]
pub(crate) fn as_mut_ptr(&mut self) -> *mut Overlapped {
&mut self.as_opaque_mut().header
}
/// Cancel the op, decrease the ref count. The return value indicates if the
/// op is completed. If so, the op should be dropped because it is
/// useless.
pub(crate) fn set_cancelled(&mut self) -> bool {
self.as_opaque_mut().cancelled = true;
self.has_result()
}
/// Complete the op, decrease the ref count. Wake the future if a waker is
/// set. The return value indicates if the op is cancelled. If so, the
/// op should be dropped because it is useless.
pub(crate) fn set_result(&mut self, res: io::Result<usize>) -> bool {
let this = unsafe { &mut *self.as_dyn_mut_ptr() };
#[cfg(all(target_os = "linux", feature = "io-uring"))]
if let Ok(res) = res {
unsafe {
Pin::new_unchecked(&mut this.op).set_result(res);
}
}
if let PushEntry::Pending(Some(w)) =
std::mem::replace(&mut this.result, PushEntry::Ready(res))
{
w.wake();
}
this.cancelled
}
pub(crate) fn set_flags(&mut self, flags: u32) {
self.as_opaque_mut().flags = flags;
}
pub(crate) fn flags(&self) -> u32 {
self.as_opaque().flags
}
/// Whether the op is completed.
pub(crate) fn has_result(&self) -> bool {
self.as_opaque().result.is_ready()
}
/// Set waker of the current future.
pub(crate) fn set_waker(&mut self, waker: Waker) {
if let PushEntry::Pending(w) = &mut self.as_opaque_mut().result {
*w = Some(waker)
}
}
/// Get the inner [`RawOp`]. It is usually used to drop the inner
/// immediately, without knowing about the inner `T`.
///
/// # Safety
///
/// Call it only when the op is cancelled and completed, which is the case
/// when the ref count becomes zero. See doc of [`Key::set_cancelled`]
/// and [`Key::set_result`].
pub(crate) unsafe fn into_box(mut self) -> Box<RawOp<dyn OpCode>> {
Box::from_raw(self.as_dyn_mut_ptr())
}
}
impl<T> Key<T> {
/// Get the inner result if it is completed.
///
/// # Safety
///
/// Call it only when the op is completed, otherwise it is UB.
pub(crate) unsafe fn into_inner(self) -> (io::Result<usize>, T) {
let op = unsafe { Box::from_raw(self.user_data as *mut RawOp<T>) };
(op.result.take_ready().unwrap_unchecked(), op.op)
}
}
impl<T: OpCode + ?Sized> Key<T> {
/// Pin the inner op.
pub(crate) fn as_op_pin(&mut self) -> Pin<&mut dyn OpCode> {
// SAFETY: the inner won't be moved.
unsafe {
let this = &mut *self.as_dyn_mut_ptr();
Pin::new_unchecked(&mut this.op)
}
}
}
impl<T: ?Sized> std::fmt::Debug for Key<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Key({})", self.user_data())
}
}

479
ntex-iodriver/src/lib.rs Normal file
View file

@ -0,0 +1,479 @@
//! The platform-specified driver.
//! Some types differ by compilation target.
#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
#[cfg(all(
target_os = "linux",
not(feature = "io-uring"),
not(feature = "polling")
))]
compile_error!(
"You must choose at least one of these features: [\"io-uring\", \"polling\"]"
);
use std::{io, task::Poll, task::Waker, time::Duration};
mod key;
pub use key::Key;
pub mod op;
#[cfg(unix)]
#[cfg_attr(docsrs, doc(cfg(all())))]
mod unix;
#[cfg(unix)]
use unix::Overlapped;
mod asyncify;
pub use asyncify::*;
mod driver_type;
pub use driver_type::*;
cfg_if::cfg_if! {
//if #[cfg(windows)] {
// #[path = "iocp/mod.rs"]
// mod sys;
//} else if #[cfg(all(target_os = "linux", feature = "io-uring"))] {
// #[path = "iour/mod.rs"]
// mod sys;
//} else
if #[cfg(unix)] {
#[path = "poll/mod.rs"]
mod sys;
}
}
pub use sys::*;
#[cfg(windows)]
#[macro_export]
#[doc(hidden)]
macro_rules! syscall {
(BOOL, $e:expr) => {
$crate::syscall!($e, == 0)
};
(SOCKET, $e:expr) => {
$crate::syscall!($e, != 0)
};
(HANDLE, $e:expr) => {
$crate::syscall!($e, == ::windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE)
};
($e:expr, $op: tt $rhs: expr) => {{
#[allow(unused_unsafe)]
let res = unsafe { $e };
if res $op $rhs {
Err(::std::io::Error::last_os_error())
} else {
Ok(res)
}
}};
}
/// Helper macro to execute a system call
#[cfg(unix)]
#[macro_export]
#[doc(hidden)]
macro_rules! syscall {
(break $e:expr) => {
loop {
match $crate::syscall!($e) {
Ok(fd) => break ::std::task::Poll::Ready(Ok(fd as usize)),
Err(e) if e.kind() == ::std::io::ErrorKind::WouldBlock || e.raw_os_error() == Some(::libc::EINPROGRESS)
=> break ::std::task::Poll::Pending,
Err(e) if e.kind() == ::std::io::ErrorKind::Interrupted => {},
Err(e) => break ::std::task::Poll::Ready(Err(e)),
}
}
};
($e:expr, $f:ident($fd:expr)) => {
match $crate::syscall!(break $e) {
::std::task::Poll::Pending => Ok($crate::sys::Decision::$f($fd)),
::std::task::Poll::Ready(Ok(res)) => Ok($crate::sys::Decision::Completed(res)),
::std::task::Poll::Ready(Err(e)) => Err(e),
}
};
($e:expr) => {{
#[allow(unused_unsafe)]
let res = unsafe { $e };
if res == -1 {
Err(::std::io::Error::last_os_error())
} else {
Ok(res)
}
}};
}
#[macro_export]
#[doc(hidden)]
macro_rules! impl_raw_fd {
($t:ty, $it:ty, $inner:ident) => {
impl $crate::AsRawFd for $t {
fn as_raw_fd(&self) -> $crate::RawFd {
self.$inner.as_raw_fd()
}
}
#[cfg(unix)]
impl std::os::fd::FromRawFd for $t {
unsafe fn from_raw_fd(fd: $crate::RawFd) -> Self {
Self {
$inner: std::os::fd::FromRawFd::from_raw_fd(fd),
}
}
}
};
($t:ty, $it:ty, $inner:ident,file) => {
$crate::impl_raw_fd!($t, $it, $inner);
#[cfg(windows)]
impl std::os::windows::io::FromRawHandle for $t {
unsafe fn from_raw_handle(handle: std::os::windows::io::RawHandle) -> Self {
Self {
$inner: std::os::windows::io::FromRawHandle::from_raw_handle(handle),
}
}
}
#[cfg(windows)]
impl std::os::windows::io::AsRawHandle for $t {
fn as_raw_handle(&self) -> std::os::windows::io::RawHandle {
self.$inner.as_raw_handle()
}
}
};
($t:ty, $it:ty, $inner:ident,socket) => {
$crate::impl_raw_fd!($t, $it, $inner);
#[cfg(windows)]
impl std::os::windows::io::FromRawSocket for $t {
unsafe fn from_raw_socket(sock: std::os::windows::io::RawSocket) -> Self {
Self {
$inner: std::os::windows::io::FromRawSocket::from_raw_socket(sock),
}
}
}
#[cfg(windows)]
impl std::os::windows::io::AsRawSocket for $t {
fn as_raw_socket(&self) -> std::os::windows::io::RawSocket {
self.$inner.as_raw_socket()
}
}
};
}
/// The return type of [`Proactor::push`].
pub enum PushEntry<K, R> {
/// The operation is pushed to the submission queue.
Pending(K),
/// The operation is ready and returns.
Ready(R),
}
impl<K, R> PushEntry<K, R> {
/// Get if the current variant is [`PushEntry::Ready`].
pub const fn is_ready(&self) -> bool {
matches!(self, Self::Ready(_))
}
/// Take the ready variant if exists.
pub fn take_ready(self) -> Option<R> {
match self {
Self::Pending(_) => None,
Self::Ready(res) => Some(res),
}
}
/// Map the [`PushEntry::Pending`] branch.
pub fn map_pending<L>(self, f: impl FnOnce(K) -> L) -> PushEntry<L, R> {
match self {
Self::Pending(k) => PushEntry::Pending(f(k)),
Self::Ready(r) => PushEntry::Ready(r),
}
}
/// Map the [`PushEntry::Ready`] branch.
pub fn map_ready<S>(self, f: impl FnOnce(R) -> S) -> PushEntry<K, S> {
match self {
Self::Pending(k) => PushEntry::Pending(k),
Self::Ready(r) => PushEntry::Ready(f(r)),
}
}
}
/// Low-level actions of completion-based IO.
/// It owns the operations to keep the driver safe.
pub struct Proactor {
driver: Driver,
}
impl Proactor {
/// Create [`Proactor`] with 1024 entries.
pub fn new() -> io::Result<Self> {
Self::builder().build()
}
/// Create [`ProactorBuilder`] to config the proactor.
pub fn builder() -> ProactorBuilder {
ProactorBuilder::new()
}
fn with_builder(builder: &ProactorBuilder) -> io::Result<Self> {
Ok(Self {
driver: Driver::new(builder)?,
})
}
/// Attach an fd to the driver.
///
/// ## Platform specific
/// * IOCP: it will be attached to the completion port. An fd could only be
/// attached to one driver, and could only be attached once, even if you
/// `try_clone` it.
/// * io-uring & polling: it will do nothing but return `Ok(())`.
pub fn attach(&self, fd: RawFd) -> io::Result<()> {
self.driver.attach(fd)
}
/// Cancel an operation with the pushed user-defined data.
///
/// The cancellation is not reliable. The underlying operation may continue,
/// but just don't return from [`Proactor::poll`]. Therefore, although an
/// operation is cancelled, you should not reuse its `user_data`.
pub fn cancel<T: OpCode>(&self, mut op: Key<T>) -> Option<(io::Result<usize>, T)> {
if op.set_cancelled() {
// SAFETY: completed.
Some(unsafe { op.into_inner() })
} else {
None
}
}
/// Push an operation into the driver, and return the unique key, called
/// user-defined data, associated with it.
pub fn push<T: OpCode + 'static>(
&self,
op: T,
) -> PushEntry<Key<T>, (io::Result<usize>, T)> {
let mut op = self.driver.create_op(op);
match self
.driver
.push(&mut unsafe { Key::<dyn OpCode>::new_unchecked(op.user_data()) })
{
Poll::Pending => PushEntry::Pending(op),
Poll::Ready(res) => {
op.set_result(res);
// SAFETY: just completed.
PushEntry::Ready(unsafe { op.into_inner() })
}
}
}
/// Poll the driver and get completed entries.
/// You need to call [`Proactor::pop`] to get the pushed
/// operations.
pub fn poll<F: FnOnce()>(&self, timeout: Option<Duration>, f: F) -> io::Result<()> {
unsafe { self.driver.poll(timeout, f) }
}
/// Get the pushed operations from the completion entries.
///
/// # Panics
/// This function will panic if the requested operation has not been
/// completed.
pub fn pop<T>(&self, op: Key<T>) -> PushEntry<Key<T>, ((io::Result<usize>, T), u32)> {
if op.has_result() {
let flags = op.flags();
// SAFETY: completed.
PushEntry::Ready((unsafe { op.into_inner() }, flags))
} else {
PushEntry::Pending(op)
}
}
/// Update the waker of the specified op.
pub fn update_waker<T>(&self, op: &mut Key<T>, waker: Waker) {
op.set_waker(waker);
}
/// Create a notify handle to interrupt the inner driver.
pub fn handle(&self) -> NotifyHandle {
self.driver.handle()
}
pub fn register_handler<F>(&self, f: F)
where
F: FnOnce(DriverApi) -> Box<dyn op::Handler>,
{
self.driver.register_handler(f)
}
}
impl AsRawFd for Proactor {
fn as_raw_fd(&self) -> RawFd {
self.driver.as_raw_fd()
}
}
/// An completed entry returned from kernel.
#[derive(Debug)]
pub(crate) struct Entry {
user_data: usize,
result: io::Result<usize>,
flags: u32,
}
impl Entry {
pub(crate) fn new(user_data: usize, result: io::Result<usize>) -> Self {
Self {
user_data,
result,
flags: 0,
}
}
#[cfg(all(target_os = "linux", feature = "io-uring"))]
// this method only used by in io-uring driver
pub(crate) fn set_flags(&mut self, flags: u32) {
self.flags = flags;
}
/// The user-defined data returned by [`Proactor::push`].
pub fn user_data(&self) -> usize {
self.user_data
}
pub fn flags(&self) -> u32 {
self.flags
}
/// The result of the operation.
pub fn into_result(self) -> io::Result<usize> {
self.result
}
/// SAFETY: `user_data` should be a valid pointer.
pub unsafe fn notify(self) {
let user_data = self.user_data();
let mut op = Key::<()>::new_unchecked(user_data);
op.set_flags(self.flags());
if op.set_result(self.into_result()) {
// SAFETY: completed and cancelled.
let _ = op.into_box();
}
}
}
#[derive(Debug, Clone)]
enum ThreadPoolBuilder {
Create { limit: usize, recv_limit: Duration },
Reuse(AsyncifyPool),
}
impl Default for ThreadPoolBuilder {
fn default() -> Self {
Self::new()
}
}
impl ThreadPoolBuilder {
pub fn new() -> Self {
Self::Create {
limit: 256,
recv_limit: Duration::from_secs(60),
}
}
pub fn create_or_reuse(&self) -> AsyncifyPool {
match self {
Self::Create { limit, recv_limit } => AsyncifyPool::new(*limit, *recv_limit),
Self::Reuse(pool) => pool.clone(),
}
}
}
/// Builder for [`Proactor`].
#[derive(Debug, Clone)]
pub struct ProactorBuilder {
capacity: u32,
pool_builder: ThreadPoolBuilder,
sqpoll_idle: Option<Duration>,
}
impl Default for ProactorBuilder {
fn default() -> Self {
Self::new()
}
}
impl ProactorBuilder {
/// Create the builder with default config.
pub fn new() -> Self {
Self {
capacity: 1024,
pool_builder: ThreadPoolBuilder::new(),
sqpoll_idle: None,
}
}
/// Set the capacity of the inner event queue or submission queue, if
/// exists. The default value is 1024.
pub fn capacity(&mut self, capacity: u32) -> &mut Self {
self.capacity = capacity;
self
}
/// Set the thread number limit of the inner thread pool, if exists. The
/// default value is 256.
///
/// It will be ignored if `reuse_thread_pool` is set.
pub fn thread_pool_limit(&mut self, value: usize) -> &mut Self {
if let ThreadPoolBuilder::Create { limit, .. } = &mut self.pool_builder {
*limit = value;
}
self
}
/// Set the waiting timeout of the inner thread, if exists. The default is
/// 60 seconds.
///
/// It will be ignored if `reuse_thread_pool` is set.
pub fn thread_pool_recv_timeout(&mut self, timeout: Duration) -> &mut Self {
if let ThreadPoolBuilder::Create { recv_limit, .. } = &mut self.pool_builder {
*recv_limit = timeout;
}
self
}
/// Set to reuse an existing [`AsyncifyPool`] in this proactor.
pub fn reuse_thread_pool(&mut self, pool: AsyncifyPool) -> &mut Self {
self.pool_builder = ThreadPoolBuilder::Reuse(pool);
self
}
/// Force reuse the thread pool for each proactor created by this builder,
/// even `reuse_thread_pool` is not set.
pub fn force_reuse_thread_pool(&mut self) -> &mut Self {
self.reuse_thread_pool(self.create_or_get_thread_pool());
self
}
/// Create or reuse the thread pool from the config.
pub fn create_or_get_thread_pool(&self) -> AsyncifyPool {
self.pool_builder.create_or_reuse()
}
/// Set `io-uring` sqpoll idle milliseconds, when `sqpoll_idle` is set,
/// io-uring sqpoll feature will be enabled
///
/// # Notes
///
/// - Only effective when the `io-uring` feature is enabled
/// - `idle` must >= 1ms, otherwise will set sqpoll idle 0ms
/// - `idle` will be rounded down
pub fn sqpoll_idle(&mut self, idle: Duration) -> &mut Self {
self.sqpoll_idle = Some(idle);
self
}
/// Build the [`Proactor`].
pub fn build(&self) -> io::Result<Proactor> {
Proactor::with_builder(self)
}
}

75
ntex-iodriver/src/op.rs Normal file
View file

@ -0,0 +1,75 @@
//! The async operations.
//!
//! Types in this mod represents the low-level operations passed to kernel.
//! The operation itself doesn't perform anything.
//! You need to pass them to [`crate::Proactor`], and poll the driver.
use std::{io, marker::PhantomPinned, mem::ManuallyDrop, net::Shutdown};
#[cfg(unix)]
pub use crate::sys::op::{CreateSocket, Interest};
use crate::OwnedFd;
pub trait Handler {
/// Submitted interest
fn readable(&mut self, id: usize);
/// Submitted interest
fn writable(&mut self, id: usize);
/// Operation submission has failed
fn error(&mut self, id: usize, err: io::Error);
/// All events are processed, process all updates
fn commit(&mut self);
}
/// Spawn a blocking function in the thread pool.
pub struct Asyncify<F, D> {
pub(crate) f: Option<F>,
pub(crate) data: Option<D>,
_p: PhantomPinned,
}
impl<F, D> Asyncify<F, D> {
/// Create [`Asyncify`].
pub fn new(f: F) -> Self {
Self {
f: Some(f),
data: None,
_p: PhantomPinned,
}
}
pub fn into_inner(mut self) -> D {
self.data.take().expect("the data should not be None")
}
}
/// Shutdown a socket.
pub struct ShutdownSocket<S> {
pub(crate) fd: S,
pub(crate) how: Shutdown,
}
impl<S> ShutdownSocket<S> {
/// Create [`ShutdownSocket`].
pub fn new(fd: S, how: Shutdown) -> Self {
Self { fd, how }
}
}
/// Close socket fd.
pub struct CloseSocket {
pub(crate) fd: ManuallyDrop<OwnedFd>,
}
impl CloseSocket {
/// Create [`CloseSocket`].
pub fn new(fd: OwnedFd) -> Self {
Self {
fd: ManuallyDrop::new(fd),
}
}
}

480
ntex-iodriver/src/poll/mod.rs Normal file
View file

@ -0,0 +1,480 @@
pub use std::os::fd::{AsRawFd, OwnedFd, RawFd};
use std::{cell::Cell, cell::RefCell, collections::HashMap, io, rc::Rc, sync::Arc};
use std::{num::NonZeroUsize, os::fd::BorrowedFd, pin::Pin, task::Poll, time::Duration};
use crossbeam_queue::SegQueue;
use polling::{Event, Events, Poller};
use crate::{
op::Handler, op::Interest, syscall, AsyncifyPool, Entry, Key, ProactorBuilder,
};
pub(crate) mod op;
/// Abstraction of operations.
pub trait OpCode {
/// Perform the operation before submit, and return [`Decision`] to
/// indicate whether submitting the operation to polling is required.
fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision>;
/// Perform the operation after received corresponding
/// event. If this operation is blocking, the return value should be
/// [`Poll::Ready`].
fn operate(self: Pin<&mut Self>) -> Poll<io::Result<usize>>;
}
/// Result of [`OpCode::pre_submit`].
#[non_exhaustive]
pub enum Decision {
/// Instant operation, no need to submit
Completed(usize),
/// Blocking operation, needs to be spawned in another thread
Blocking,
}
bitflags::bitflags! {
#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
struct Flags: u8 {
const NEW = 0b0000_0001;
const CHANGED = 0b0000_0010;
}
}
#[derive(Debug)]
struct FdItem {
flags: Flags,
batch: usize,
read: Option<usize>,
write: Option<usize>,
}
impl FdItem {
fn new(batch: usize) -> Self {
Self {
batch,
read: None,
write: None,
flags: Flags::NEW,
}
}
fn register(&mut self, user_data: usize, interest: Interest) {
self.flags.insert(Flags::CHANGED);
match interest {
Interest::Readable => {
self.read = Some(user_data);
}
Interest::Writable => {
self.write = Some(user_data);
}
}
}
fn unregister(&mut self, int: Interest) {
let res = match int {
Interest::Readable => self.read.take(),
Interest::Writable => self.write.take(),
};
if res.is_some() {
self.flags.insert(Flags::CHANGED);
}
}
fn unregister_all(&mut self) {
if self.read.is_some() || self.write.is_some() {
self.flags.insert(Flags::CHANGED);
}
let _ = self.read.take();
let _ = self.write.take();
}
fn user_data(&mut self, interest: Interest) -> Option<usize> {
match interest {
Interest::Readable => self.read,
Interest::Writable => self.write,
}
}
fn event(&self, key: usize) -> Event {
let mut event = Event::none(key);
if self.read.is_some() {
event.readable = true;
}
if self.write.is_some() {
event.writable = true;
}
event
}
}
#[derive(Debug)]
enum InterestChange {
Register(Interest),
Unregister(Interest),
UnregisterAll,
}
#[derive(Debug)]
struct BatchChange {
fd: RawFd,
batch: usize,
user_data: usize,
interest: InterestChange,
}
pub struct DriverApi {
batch: usize,
changes: Rc<RefCell<Vec<BatchChange>>>,
}
impl DriverApi {
pub fn register(&self, fd: RawFd, user_data: usize, int: Interest) {
log::debug!(
"Register interest {:?} for {:?} user-data: {:?}",
int,
fd,
user_data
);
self.change(BatchChange {
fd,
user_data,
batch: self.batch,
interest: InterestChange::Register(int),
});
}
pub fn unregister(&self, fd: RawFd, int: Interest) {
log::debug!(
"Unregister interest {:?} for {:?} batch: {:?}",
int,
fd,
self.batch
);
self.change(BatchChange {
fd,
user_data: 0,
batch: self.batch,
interest: InterestChange::Unregister(int),
});
}
pub fn unregister_all(&self, fd: RawFd) {
self.change(BatchChange {
fd,
user_data: 0,
batch: self.batch,
interest: InterestChange::UnregisterAll,
});
}
fn change(&self, change: BatchChange) {
self.changes.borrow_mut().push(change);
}
}
/// Low-level driver of polling.
pub(crate) struct Driver {
poll: Arc<Poller>,
events: RefCell<Events>,
registry: RefCell<HashMap<RawFd, FdItem>>,
pool: AsyncifyPool,
pool_completed: Arc<SegQueue<Entry>>,
hid: Cell<usize>,
changes: Rc<RefCell<Vec<BatchChange>>>,
handlers: Cell<Option<Box<Vec<Box<dyn Handler>>>>>,
}
impl Driver {
pub fn new(builder: &ProactorBuilder) -> io::Result<Self> {
log::trace!("New poll driver");
let entries = builder.capacity as usize; // for the sake of consistency, use u32 like iour
let events = if entries == 0 {
Events::new()
} else {
Events::with_capacity(NonZeroUsize::new(entries).unwrap())
};
Ok(Self {
poll: Arc::new(Poller::new()?),
events: RefCell::new(events),
registry: RefCell::new(HashMap::default()),
pool: builder.create_or_get_thread_pool(),
pool_completed: Arc::new(SegQueue::new()),
hid: Cell::new(0),
changes: Rc::new(RefCell::new(Vec::with_capacity(16))),
handlers: Cell::new(Some(Box::new(Vec::default()))),
})
}
pub fn register_handler<F>(&self, f: F)
where
F: FnOnce(DriverApi) -> Box<dyn Handler>,
{
let id = self.hid.get();
let mut handlers = self.handlers.take().unwrap_or_default();
let api = DriverApi {
batch: id,
changes: self.changes.clone(),
};
handlers.push(f(api));
self.hid.set(id + 1);
self.handlers.set(Some(handlers));
}
pub fn create_op<T: crate::sys::OpCode + 'static>(&self, op: T) -> Key<T> {
Key::new(self.as_raw_fd(), op)
}
fn renew(
&self,
fd: BorrowedFd,
renew_event: Event,
registry: &mut HashMap<RawFd, FdItem>,
) -> io::Result<()> {
if !renew_event.readable && !renew_event.writable {
//println!("DELETE - 2");
if let Some(item) = registry.remove(&fd.as_raw_fd()) {
if !item.flags.contains(Flags::NEW) {
self.poll.delete(fd)?;
}
}
} else {
if let Some(item) = registry.get(&fd.as_raw_fd()) {
if item.flags.contains(Flags::NEW) {
//println!("ADD - 2 {:?}", fd.as_raw_fd());
unsafe { self.poll.add(&fd, renew_event)? };
return Ok(());
}
}
//println!("MODIFY - 2");
self.poll.modify(fd, renew_event)?;
}
Ok(())
}
pub fn attach(&self, _fd: RawFd) -> io::Result<()> {
Ok(())
}
pub fn push(&self, op: &mut Key<dyn crate::sys::OpCode>) -> Poll<io::Result<usize>> {
let user_data = op.user_data();
let op_pin = op.as_op_pin();
match op_pin.pre_submit()? {
Decision::Completed(res) => Poll::Ready(Ok(res)),
Decision::Blocking => self.push_blocking(user_data),
}
}
pub unsafe fn poll<F: FnOnce()>(
&self,
timeout: Option<Duration>,
f: F,
) -> io::Result<()> {
if self.poll_blocking() {
f();
self.apply_changes()?;
return Ok(());
}
let mut events = self.events.borrow_mut();
self.poll.wait(&mut events, timeout)?;
if events.is_empty() && timeout != Some(Duration::ZERO) {
return Err(io::Error::from_raw_os_error(libc::ETIMEDOUT));
}
// println!("POLL, events: {:?}", events.len());
if !events.is_empty() {
let mut registry = self.registry.borrow_mut();
let mut handlers = self.handlers.take().unwrap();
for event in events.iter() {
let user_data = event.key;
let fd = user_data as RawFd;
log::debug!(
"receive {} for {:?} {:?}",
user_data,
event,
registry.get_mut(&fd)
);
if let Some(item) = registry.get_mut(&fd) {
self.handle_batch_event(event, item, &mut handlers);
}
}
drop(registry);
self.handlers.set(Some(handlers));
}
// apply changes
self.apply_changes()?;
// complete batch handling
let mut handlers = self.handlers.take().unwrap();
for handler in handlers.iter_mut() {
handler.commit();
}
self.handlers.set(Some(handlers));
self.apply_changes()?;
// run user function
f();
// check if we have more changes from "run"
self.apply_changes()?;
Ok(())
}
fn handle_batch_event(
&self,
event: Event,
item: &mut FdItem,
handlers: &mut [Box<dyn Handler>],
) {
if event.readable {
if let Some(user_data) = item.user_data(Interest::Readable) {
handlers[item.batch].readable(user_data)
}
}
if event.writable {
if let Some(user_data) = item.user_data(Interest::Writable) {
handlers[item.batch].writable(user_data)
}
}
}
/// re-calc driver changes
unsafe fn apply_changes(&self) -> io::Result<()> {
let mut changes = self.changes.borrow_mut();
if changes.is_empty() {
return Ok(());
}
log::debug!("Apply driver changes, {:?}", changes.len());
let mut registry = self.registry.borrow_mut();
for change in &mut *changes {
let item = registry
.entry(change.fd)
.or_insert_with(|| FdItem::new(change.batch));
debug_assert!(item.batch == change.batch, "{:?} - {:?}", item, change);
match change.interest {
InterestChange::Register(int) => {
let _ = item.register(change.user_data, int);
}
InterestChange::Unregister(int) => {
let _ = item.unregister(int);
}
InterestChange::UnregisterAll => {
let _ = item.unregister_all();
}
}
}
for change in changes.drain(..) {
let result = registry.get_mut(&change.fd).and_then(|item| {
if item.flags.contains(Flags::CHANGED) {
item.flags.remove(Flags::CHANGED);
Some((
item.event(change.fd as usize),
item.flags.contains(Flags::NEW),
))
} else {
None
}
});
if let Some((event, new)) = result {
self.renew(BorrowedFd::borrow_raw(change.fd), event, &mut registry)?;
if new {
registry.get_mut(&change.fd).map(|item| {
item.flags.remove(Flags::NEW);
});
}
}
}
Ok(())
}
fn push_blocking(&self, user_data: usize) -> Poll<io::Result<usize>> {
let poll = self.poll.clone();
let completed = self.pool_completed.clone();
let mut closure = move || {
let mut op = unsafe { Key::<dyn crate::sys::OpCode>::new_unchecked(user_data) };
let op_pin = op.as_op_pin();
let res = match op_pin.operate() {
Poll::Pending => unreachable!("this operation is not non-blocking"),
Poll::Ready(res) => res,
};
completed.push(Entry::new(user_data, res));
poll.notify().ok();
};
loop {
match self.pool.dispatch(closure) {
Ok(()) => return Poll::Pending,
Err(e) => {
closure = e.0;
self.poll_blocking();
}
}
}
}
fn poll_blocking(&self) -> bool {
if self.pool_completed.is_empty() {
return false;
}
while let Some(entry) = self.pool_completed.pop() {
unsafe {
entry.notify();
}
}
true
}
pub fn handle(&self) -> NotifyHandle {
NotifyHandle::new(self.poll.clone())
}
}
impl AsRawFd for Driver {
fn as_raw_fd(&self) -> RawFd {
self.poll.as_raw_fd()
}
}
impl Drop for Driver {
fn drop(&mut self) {
for fd in self.registry.borrow().keys() {
unsafe {
let fd = BorrowedFd::borrow_raw(*fd);
self.poll.delete(fd).ok();
}
}
}
}
#[derive(Clone)]
/// A notify handle to the inner driver.
pub struct NotifyHandle {
poll: Arc<Poller>,
}
impl NotifyHandle {
fn new(poll: Arc<Poller>) -> Self {
Self { poll }
}
/// Notify the inner driver.
pub fn notify(&self) -> io::Result<()> {
self.poll.notify()
}
}

67
ntex-iodriver/src/poll/op.rs Normal file
View file

@ -0,0 +1,67 @@
use std::{io, marker::Send, os::fd::FromRawFd, os::fd::RawFd, pin::Pin, task::Poll};
use super::{syscall, AsRawFd, Decision, OpCode};
use crate::op::*;
pub use crate::unix::op::*;
impl<D, F> OpCode for Asyncify<F, D>
where
D: Send + 'static,
F: (FnOnce() -> (io::Result<usize>, D)) + Send + 'static,
{
fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision> {
Ok(Decision::Blocking)
}
fn operate(self: Pin<&mut Self>) -> Poll<io::Result<usize>> {
// Safety: self won't be moved
let this = unsafe { self.get_unchecked_mut() };
let f = this
.f
.take()
.expect("the operate method could only be called once");
let (res, data) = f();
this.data = Some(data);
Poll::Ready(res)
}
}
impl OpCode for CreateSocket {
fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision> {
Ok(Decision::Blocking)
}
fn operate(self: Pin<&mut Self>) -> Poll<io::Result<usize>> {
Poll::Ready(Ok(
syscall!(libc::socket(self.domain, self.socket_type, self.protocol))? as _,
))
}
}
impl<S: AsRawFd> OpCode for ShutdownSocket<S> {
fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision> {
Ok(Decision::Blocking)
}
fn operate(self: Pin<&mut Self>) -> Poll<io::Result<usize>> {
Poll::Ready(Ok(
syscall!(libc::shutdown(self.fd.as_raw_fd(), self.how()))? as _,
))
}
}
impl CloseSocket {
pub fn from_raw_fd(fd: RawFd) -> Self {
Self::new(unsafe { FromRawFd::from_raw_fd(fd) })
}
}
impl OpCode for CloseSocket {
fn pre_submit(self: Pin<&mut Self>) -> io::Result<Decision> {
Ok(Decision::Blocking)
}
fn operate(self: Pin<&mut Self>) -> Poll<io::Result<usize>> {
Poll::Ready(Ok(syscall!(libc::close(self.fd.as_raw_fd()))? as _))
}
}

15
ntex-iodriver/src/unix/mod.rs Normal file
View file

@ -0,0 +1,15 @@
//! This mod doesn't actually contain any driver, but meant to provide some
//! common op type and utilities for unix platform (for iour and polling).
pub(crate) mod op;
use crate::RawFd;
/// The overlapped struct for unix needn't contain extra fields.
pub(crate) struct Overlapped;
impl Overlapped {
pub fn new(_driver: RawFd) -> Self {
Self
}
}

40
ntex-iodriver/src/unix/op.rs Normal file
View file

@ -0,0 +1,40 @@
use std::net::Shutdown;
use crate::op::*;
/// The interest to poll a file descriptor.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Interest {
/// Represents a read operation.
Readable,
/// Represents a write operation.
Writable,
}
/// Create a socket.
pub struct CreateSocket {
pub(crate) domain: i32,
pub(crate) socket_type: i32,
pub(crate) protocol: i32,
}
impl CreateSocket {
/// Create [`CreateSocket`].
pub fn new(domain: i32, socket_type: i32, protocol: i32) -> Self {
Self {
domain,
socket_type,
protocol,
}
}
}
impl<S> ShutdownSocket<S> {
pub(crate) fn how(&self) -> i32 {
match self.how {
Shutdown::Write => libc::SHUT_WR,
Shutdown::Read => libc::SHUT_RD,
Shutdown::Both => libc::SHUT_RDWR,
}
}
}

18
ntex-net/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex-net"
version = "2.4.0"
version = "2.5.0"
authors = ["ntex contributors <team@ntex.rs>"]
description = "ntexwork utils for ntex framework"
keywords = ["network", "framework", "async", "futures"]
@ -30,12 +30,15 @@ glommio = ["ntex-rt/glommio", "ntex-glommio"]
# async-std runtime
async-std = ["ntex-rt/async-std", "ntex-async-std"]
# default ntex runtime
default-rt = ["ntex-rt/default-rt", "ntex-runtime", "ntex-iodriver", "slab", "socket2"]
[dependencies]
ntex-service = "3.3"
ntex-bytes = "0.1"
ntex-http = "0.1"
ntex-io = "2.8"
ntex-rt = "0.4.21"
ntex-rt = "0.4.25"
ntex-util = "2.5"
ntex-tokio = { version = "0.5.3", optional = true }
@ -43,8 +46,15 @@ ntex-compio = { version = "0.2.4", optional = true }
ntex-glommio = { version = "0.5.2", optional = true }
ntex-async-std = { version = "0.5.1", optional = true }
log = "0.4"
thiserror = "1"
ntex-runtime = { version = "0.1.0", optional = true }
ntex-iodriver = { version = "0.1.0", optional = true }
bitflags = { workspace = true }
log = { workspace = true }
libc = { workspace = true }
thiserror = { workspace = true }
slab = { workspace = true, optional = true }
socket2 = { workspace = true, optional = true }
[dev-dependencies]
ntex = "2"

20
ntex-net/src/compat.rs
View file

@ -6,10 +6,23 @@ pub use ntex_tokio::{from_tcp_stream, tcp_connect, tcp_connect_in};
#[cfg(all(unix, feature = "tokio"))]
pub use ntex_tokio::{from_unix_stream, unix_connect, unix_connect_in};
#[cfg(all(
feature = "default-rt",
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "compio"),
not(feature = "glommio")
))]
pub use crate::rt::{
from_tcp_stream, from_unix_stream, tcp_connect, tcp_connect_in, unix_connect,
unix_connect_in,
};
#[cfg(all(
feature = "compio",
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use ntex_compio::{from_tcp_stream, tcp_connect, tcp_connect_in};
@ -19,6 +32,7 @@ pub use ntex_compio::{from_tcp_stream, tcp_connect, tcp_connect_in};
feature = "compio",
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use ntex_compio::{from_unix_stream, unix_connect, unix_connect_in};
@ -27,6 +41,7 @@ pub use ntex_compio::{from_unix_stream, unix_connect, unix_connect_in};
feature = "async-std",
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use ntex_async_std::{from_tcp_stream, tcp_connect, tcp_connect_in};
@ -36,6 +51,7 @@ pub use ntex_async_std::{from_tcp_stream, tcp_connect, tcp_connect_in};
feature = "async-std",
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use ntex_async_std::{from_unix_stream, unix_connect, unix_connect_in};
@ -44,6 +60,7 @@ pub use ntex_async_std::{from_unix_stream, unix_connect, unix_connect_in};
feature = "glommio",
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "async-std")
))]
pub use ntex_glommio::{from_tcp_stream, tcp_connect, tcp_connect_in};
@ -53,6 +70,7 @@ pub use ntex_glommio::{from_tcp_stream, tcp_connect, tcp_connect_in};
feature = "glommio",
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "async-std")
))]
pub use ntex_glommio::{from_unix_stream, unix_connect, unix_connect_in};
@ -61,6 +79,7 @@ pub use ntex_glommio::{from_unix_stream, unix_connect, unix_connect_in};
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "async-std"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
mod no_rt {
@ -131,6 +150,7 @@ mod no_rt {
not(feature = "tokio"),
not(feature = "compio"),
not(feature = "async-std"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use no_rt::*;

9
ntex-net/src/lib.rs
View file

@ -8,3 +8,12 @@ pub use ntex_io::Io;
pub use ntex_rt::{spawn, spawn_blocking};
pub use self::compat::*;
#[cfg(all(
feature = "default-rt",
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "compio"),
not(feature = "glommio")
))]
mod rt;

196
ntex-net/src/rt/connect.rs Normal file
View file

@ -0,0 +1,196 @@
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6};
use std::{cell::RefCell, collections::VecDeque, io, path::Path, rc::Rc, task::Poll};
use ntex_iodriver::op::{Handler, Interest};
use ntex_iodriver::{syscall, AsRawFd, DriverApi, RawFd};
use ntex_runtime::net::{Socket, TcpStream, UnixStream};
use ntex_runtime::Runtime;
use ntex_util::channel::oneshot::{channel, Sender};
use slab::Slab;
use socket2::{Protocol, SockAddr, Type};
pub(crate) async fn connect(addr: SocketAddr) -> io::Result<TcpStream> {
let addr = SockAddr::from(addr);
let socket = if cfg!(windows) {
let bind_addr = if addr.is_ipv4() {
SockAddr::from(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0))
} else if addr.is_ipv6() {
SockAddr::from(SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, 0, 0, 0))
} else {
return Err(io::Error::new(
io::ErrorKind::AddrNotAvailable,
"Unsupported address domain.",
));
};
Socket::bind(&bind_addr, Type::STREAM, Some(Protocol::TCP)).await?
} else {
Socket::new(addr.domain(), Type::STREAM, Some(Protocol::TCP)).await?
};
let (sender, rx) = channel();
ConnectOps::current().connect(socket.as_raw_fd(), addr, sender)?;
rx.await
.map_err(|_| io::Error::new(io::ErrorKind::Other, "IO Driver is gone").into())
.and_then(|item| item)?;
Ok(TcpStream::from_socket(socket))
}
pub(crate) async fn connect_unix(path: impl AsRef<Path>) -> io::Result<UnixStream> {
let addr = SockAddr::unix(path)?;
#[cfg(windows)]
let socket = {
let new_addr = empty_unix_socket();
Socket::bind(&new_addr, Type::STREAM, None).await?
};
#[cfg(unix)]
let socket = {
use socket2::Domain;
Socket::new(Domain::UNIX, Type::STREAM, None).await?
};
let (sender, rx) = channel();
ConnectOps::current().connect(socket.as_raw_fd(), addr, sender)?;
rx.await
.map_err(|_| io::Error::new(io::ErrorKind::Other, "IO Driver is gone").into())
.and_then(|item| item)?;
Ok(UnixStream::from_socket(socket))
}
#[derive(Clone)]
pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
#[derive(Debug)]
enum Change {
Readable,
Writable,
Error(io::Error),
}
struct ConnectOpsBatcher {
feed: VecDeque<(usize, Change)>,
inner: Rc<ConnectOpsInner>,
}
struct Item {
fd: RawFd,
sender: Sender<io::Result<()>>,
}
struct ConnectOpsInner {
api: DriverApi,
connects: RefCell<Slab<Item>>,
}
impl ConnectOps {
pub(crate) fn current() -> Self {
Runtime::with_current(|rt| {
if let Some(s) = rt.get::<Self>() {
s
} else {
let mut inner = None;
rt.driver().register_handler(|api| {
let ops = Rc::new(ConnectOpsInner {
api,
connects: RefCell::new(Slab::new()),
});
inner = Some(ops.clone());
Box::new(ConnectOpsBatcher {
inner: ops,
feed: VecDeque::new(),
})
});
let s = ConnectOps(inner.unwrap());
rt.insert(s.clone());
s
}
})
}
pub(crate) fn connect(
&self,
fd: RawFd,
addr: SockAddr,
sender: Sender<io::Result<()>>,
) -> io::Result<usize> {
let result = syscall!(break libc::connect(fd, addr.as_ptr(), addr.len()));
if let Poll::Ready(res) = result {
res?;
}
let item = Item { fd, sender };
let id = self.0.connects.borrow_mut().insert(item);
self.0.api.register(fd, id, Interest::Writable);
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());
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::<libc::c_int>() 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 = 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);
}
}
}
}
}
}

353
ntex-net/src/rt/driver.rs Normal file
View file

@ -0,0 +1,353 @@
use std::{cell::Cell, collections::VecDeque, fmt, io, ptr, rc::Rc, task, task::Poll};
use ntex_iodriver::op::{Handler, Interest};
use ntex_iodriver::{syscall, AsRawFd, DriverApi, RawFd};
use ntex_runtime::Runtime;
use slab::Slab;
use ntex_bytes::BufMut;
use ntex_io::{ReadContext, WriteContext};
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<T> {
id: usize,
inner: Rc<CompioOpsInner<T>>,
}
struct TcpStreamItem<T> {
io: Option<T>,
fd: RawFd,
read: ReadContext,
write: WriteContext,
flags: Flags,
ref_count: usize,
}
pub(crate) struct CompioOps<T>(Rc<CompioOpsInner<T>>);
#[derive(Debug)]
enum Change {
Readable,
Writable,
Error(io::Error),
}
struct CompioOpsBatcher<T> {
feed: VecDeque<(usize, Change)>,
inner: Rc<CompioOpsInner<T>>,
}
struct CompioOpsInner<T> {
api: DriverApi,
feed: Cell<Option<VecDeque<usize>>>,
streams: Cell<Option<Box<Slab<TcpStreamItem<T>>>>>,
}
impl<T: AsRawFd + 'static> CompioOps<T> {
pub(crate) fn current() -> Self {
Runtime::with_current(|rt| {
if let Some(s) = rt.get::<Self>() {
s
} else {
let mut inner = None;
rt.driver().register_handler(|api| {
let ops = Rc::new(CompioOpsInner {
api,
feed: Cell::new(Some(VecDeque::new())),
streams: Cell::new(Some(Box::new(Slab::new()))),
});
inner = Some(ops.clone());
Box::new(CompioOpsBatcher {
inner: ops,
feed: VecDeque::new(),
})
});
let s = CompioOps(inner.unwrap());
rt.insert(s.clone());
s
}
})
}
pub(crate) fn register(
&self,
io: T,
read: ReadContext,
write: WriteContext,
) -> StreamCtl<T> {
let item = TcpStreamItem {
read,
write,
fd: io.as_raw_fd(),
io: Some(io),
flags: Flags::empty(),
ref_count: 1,
};
self.with(|streams| {
let id = streams.insert(item);
StreamCtl {
id,
inner: self.0.clone(),
}
})
}
fn with<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut Slab<TcpStreamItem<T>>) -> R,
{
let mut inner = self.0.streams.take().unwrap();
let result = f(&mut inner);
self.0.streams.set(Some(inner));
result
}
}
impl<T> Clone for CompioOps<T> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<T> Handler for CompioOpsBatcher<T> {
fn readable(&mut self, id: usize) {
log::debug!("FD is readable {:?}", id);
self.feed.push_back((id, Change::Readable));
}
fn writable(&mut self, id: usize) {
log::debug!("FD is writable {:?}", id);
self.feed.push_back((id, Change::Writable));
}
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)));
}
fn commit(&mut self) {
if self.feed.is_empty() {
return;
}
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.read.with_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!("FD: {:?}, BUF: {:?}", item.fd, 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.write.with_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.read.set_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);
}
}
}
}
}
// extra
let mut feed = self.inner.feed.take().unwrap();
for id in feed.drain(..) {
log::debug!("Drop io ({}), {:?}", id, streams[id].fd);
streams[id].ref_count -= 1;
if streams[id].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));
}
}
impl<T> StreamCtl<T> {
pub(crate) fn take_io(&self) -> Option<T> {
self.with(|streams| streams[self.id].io.take())
}
pub(crate) fn with_io<F, R>(&self, f: F) -> R
where
F: FnOnce(Option<&T>) -> R,
{
self.with(|streams| f(streams[self.id].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 ({}), {:?}", 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];
log::debug!("Pause io read ({}), {:?}", self.id, item.fd);
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];
log::debug!("Resume io read ({}), {:?}", 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);
}
})
}
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 ({}), {:?}", self.id, item.fd);
let result = item.write.with_buf(|buf| {
log::debug!("Writing io ({}), buf: {:?}", 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 ({}), {:?}",
self.id,
item.read.io().flags()
);
item.flags.insert(Flags::WR);
self.inner
.api
.register(item.fd, self.id, Interest::Writable);
}
}
})
}
fn with<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut Slab<TcpStreamItem<T>>) -> R,
{
let mut inner = self.inner.streams.take().unwrap();
let result = f(&mut inner);
self.inner.streams.set(Some(inner));
result
}
}
impl<T> Clone for StreamCtl<T> {
fn clone(&self) -> Self {
self.with(|streams| {
streams[self.id].ref_count += 1;
Self {
id: self.id,
inner: self.inner.clone(),
}
})
}
}
impl<T> Drop for StreamCtl<T> {
fn drop(&mut self) {
if let Some(mut streams) = self.inner.streams.take() {
log::debug!("Drop io ({}), {:?}", 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);
}
}
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));
}
}
}
impl<T> PartialEq for StreamCtl<T> {
#[inline]
fn eq(&self, other: &StreamCtl<T>) -> bool {
self.id == other.id && ptr::eq(&self.inner, &other.inner)
}
}
impl<T: fmt::Debug> fmt::Debug for StreamCtl<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.with(|streams| {
f.debug_struct("StreamCtl")
.field("id", &self.id)
.field("io", &streams[self.id].io)
.finish()
})
}
}

128
ntex-net/src/rt/io.rs Normal file
View file

@ -0,0 +1,128 @@
use std::{any, future::poll_fn, io, task::Poll};
use ntex_io::{
types, Handle, IoStream, ReadContext, ReadStatus, WriteContext, WriteStatus,
};
use ntex_runtime::{net::TcpStream, net::UnixStream, spawn};
use super::driver::{CompioOps, StreamCtl};
impl IoStream for super::TcpStream {
fn start(self, read: ReadContext, write: WriteContext) -> Option<Box<dyn Handle>> {
let io = self.0;
let ctl = CompioOps::current().register(io, read.clone(), write.clone());
let ctl2 = ctl.clone();
spawn(async move { run(ctl, read, write).await }).detach();
Some(Box::new(HandleWrapper(ctl2)))
}
}
impl IoStream for super::UnixStream {
fn start(self, read: ReadContext, write: WriteContext) -> Option<Box<dyn Handle>> {
let io = self.0;
let ctl = CompioOps::current().register(io, read.clone(), write.clone());
spawn(async move { run(ctl, read, write).await }).detach();
None
}
}
struct HandleWrapper(StreamCtl<TcpStream>);
impl Handle for HandleWrapper {
fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
if id == any::TypeId::of::<types::PeerAddr>() {
let addr = self.0.with_io(|io| io.and_then(|io| io.peer_addr().ok()));
if let Some(addr) = addr {
return Some(Box::new(types::PeerAddr(addr)));
}
}
None
}
}
trait Closable {
async fn close(self) -> io::Result<()>;
}
impl Closable for TcpStream {
async fn close(self) -> io::Result<()> {
TcpStream::close(self).await
}
}
impl Closable for UnixStream {
async fn close(self) -> io::Result<()> {
UnixStream::close(self).await
}
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum Status {
Shutdown,
Terminate,
}
async fn run<T: Closable>(ctl: StreamCtl<T>, read: ReadContext, write: WriteContext) {
// Handle io read readiness
let st = poll_fn(|cx| {
read.shutdown_filters(cx);
let read_st = read.poll_ready(cx);
let write_st = write.poll_ready(cx);
//println!("\n\n");
//println!(
// "IO2 read-st {:?}, write-st: {:?}, flags: {:?}",
// read_st,
// write_st,
// read.io().flags()
//);
//println!("\n\n");
//let read = match read.poll_ready(cx) {
let read = match read_st {
Poll::Ready(ReadStatus::Ready) => {
ctl.resume_read();
Poll::Pending
}
Poll::Ready(ReadStatus::Terminate) => Poll::Ready(()),
Poll::Pending => {
ctl.pause_read();
Poll::Pending
}
};
let write = match write_st {
Poll::Ready(WriteStatus::Ready) => {
ctl.resume_write();
Poll::Pending
}
Poll::Ready(WriteStatus::Shutdown) => Poll::Ready(Status::Shutdown),
Poll::Ready(WriteStatus::Terminate) => Poll::Ready(Status::Terminate),
Poll::Pending => Poll::Pending,
};
if read.is_pending() && write.is_pending() {
Poll::Pending
} else if write.is_ready() {
write
} else {
Poll::Ready(Status::Terminate)
}
})
.await;
ctl.resume_write();
if st == Status::Shutdown {
write.wait_for_shutdown(true).await;
} else {
write.wait_for_shutdown(false).await;
}
ctl.pause_all();
let io = ctl.take_io().unwrap();
let result = io.close().await;
read.set_stopped(result.err());
}

60
ntex-net/src/rt/mod.rs Normal file
View file

@ -0,0 +1,60 @@
#![allow(clippy::type_complexity)]
use std::{io::Result, net, net::SocketAddr};
use ntex_bytes::PoolRef;
use ntex_io::Io;
mod connect;
mod driver;
mod io;
/// Tcp stream wrapper for compio TcpStream
struct TcpStream(ntex_runtime::net::TcpStream);
/// Tcp stream wrapper for compio UnixStream
struct UnixStream(ntex_runtime::net::UnixStream);
/// Opens a TCP connection to a remote host.
pub async fn tcp_connect(addr: SocketAddr) -> Result<Io> {
let sock = connect::connect(addr).await?;
Ok(Io::new(TcpStream(sock)))
}
/// Opens a TCP connection to a remote host and use specified memory pool.
pub async fn tcp_connect_in(addr: SocketAddr, pool: PoolRef) -> Result<Io> {
let sock = connect::connect(addr).await?;
Ok(Io::with_memory_pool(TcpStream(sock), pool))
}
/// Opens a unix stream connection.
pub async fn unix_connect<'a, P>(addr: P) -> Result<Io>
where
P: AsRef<std::path::Path> + 'a,
{
let sock = connect::connect_unix(addr).await?;
Ok(Io::new(UnixStream(sock)))
}
/// Opens a unix stream connection and specified memory pool.
pub async fn unix_connect_in<'a, P>(addr: P, pool: PoolRef) -> Result<Io>
where
P: AsRef<std::path::Path> + 'a,
{
let sock = connect::connect_unix(addr).await?;
Ok(Io::with_memory_pool(UnixStream(sock), pool))
}
/// Convert std TcpStream to tokio's TcpStream
pub fn from_tcp_stream(stream: net::TcpStream) -> Result<Io> {
stream.set_nodelay(true)?;
Ok(Io::new(TcpStream(ntex_runtime::net::TcpStream::from_std(
stream,
)?)))
}
/// Convert std UnixStream to tokio's UnixStream
pub fn from_unix_stream(stream: std::os::unix::net::UnixStream) -> Result<Io> {
Ok(Io::new(UnixStream(
ntex_runtime::net::UnixStream::from_std(stream)?,
)))
}

8
ntex-rt/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex-rt"
version = "0.4.24"
version = "0.4.25"
authors = ["ntex contributors <team@ntex.rs>"]
description = "ntex runtime"
keywords = ["network", "framework", "async", "futures"]
@ -29,6 +29,9 @@ tokio = ["tok-io"]
# compio support
compio = ["compio-driver", "compio-runtime"]
# default ntex runtime
default-rt = ["ntex-runtime", "ntex-iodriver"]
# async-std support
async-std = ["async_std/unstable"]
@ -46,6 +49,9 @@ tok-io = { version = "1", package = "tokio", default-features = false, features
"net",
], optional = true }
ntex-runtime = { version = "0.1", optional = true }
ntex-iodriver = { version = "0.1", optional = true }
[target.'cfg(target_os = "linux")'.dependencies]
glomm-io = { version = "0.9", package = "glommio", optional = true }
futures-channel = { version = "0.3", optional = true }

1
ntex-rt/build.rs
View file

@ -9,6 +9,7 @@ fn main() {
"CARGO_FEATURE_TOKIO" => features.insert("tokio"),
"CARGO_FEATURE_GLOMMIO" => features.insert("glommio"),
"CARGO_FEATURE_ASYNC_STD" => features.insert("async-std"),
"CARGO_FEATURE_DEFAULT_RT" => features.insert("default-rt"),
_ => false,
};
}

131
ntex-rt/src/lib.rs
View file

@ -247,6 +247,132 @@ mod compio {
}
}
#[allow(dead_code)]
#[cfg(feature = "default-rt")]
mod default_rt {
use std::task::{ready, Context, Poll};
use std::{fmt, future::poll_fn, future::Future, pin::Pin};
use ntex_runtime::Runtime;
/// Runs the provided future, blocking the current thread until the future
/// completes.
pub fn block_on<F: Future<Output = ()>>(fut: F) {
log::info!(
"Starting compio runtime, driver {:?}",
ntex_iodriver::DriverType::current()
);
let rt = Runtime::new().unwrap();
rt.block_on(fut);
}
/// Spawns a blocking task.
///
/// The task will be spawned onto a thread pool specifically dedicated
/// to blocking tasks. This is useful to prevent long-running synchronous
/// operations from blocking the main futures executor.
pub fn spawn_blocking<F, T>(f: F) -> JoinHandle<T>
where
F: FnOnce() -> T + Send + Sync + 'static,
T: Send + 'static,
{
JoinHandle {
fut: Some(ntex_runtime::spawn_blocking(f)),
}
}
/// Spawn a future on the current thread. This does not create a new Arbiter
/// or Arbiter address, it is simply a helper for spawning futures on the current
/// thread.
///
/// # Panics
///
/// This function panics if ntex system is not running.
#[inline]
pub fn spawn<F>(f: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
let ptr = crate::CB.with(|cb| (cb.borrow().0)());
let fut = ntex_runtime::spawn(async move {
if let Some(ptr) = ptr {
let mut f = std::pin::pin!(f);
let result = poll_fn(|ctx| {
let new_ptr = crate::CB.with(|cb| (cb.borrow().1)(ptr));
let result = f.as_mut().poll(ctx);
crate::CB.with(|cb| (cb.borrow().2)(new_ptr));
result
})
.await;
crate::CB.with(|cb| (cb.borrow().3)(ptr));
result
} else {
f.await
}
});
JoinHandle { fut: Some(fut) }
}
/// Executes a future on the current thread. This does not create a new Arbiter
/// or Arbiter address, it is simply a helper for executing futures on the current
/// thread.
///
/// # Panics
///
/// This function panics if ntex system is not running.
#[inline]
pub fn spawn_fn<F, R>(f: F) -> JoinHandle<R::Output>
where
F: FnOnce() -> R + 'static,
R: Future + 'static,
{
spawn(async move { f().await })
}
#[derive(Debug, Copy, Clone)]
pub struct JoinError;
impl fmt::Display for JoinError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "JoinError")
}
}
impl std::error::Error for JoinError {}
pub struct JoinHandle<T> {
fut: Option<ntex_runtime::JoinHandle<T>>,
}
impl<T> JoinHandle<T> {
pub fn is_finished(&self) -> bool {
if let Some(hnd) = &self.fut {
hnd.is_finished()
} else {
true
}
}
}
impl<T> Drop for JoinHandle<T> {
fn drop(&mut self) {
self.fut.take().unwrap().detach();
}
}
impl<T> Future for JoinHandle<T> {
type Output = Result<T, JoinError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
Poll::Ready(
ready!(Pin::new(self.fut.as_mut().unwrap()).poll(cx))
.map_err(|_| JoinError),
)
}
}
}
#[allow(dead_code)]
#[cfg(feature = "async-std")]
mod asyncstd {
@ -473,11 +599,15 @@ pub use self::glommio::*;
#[cfg(feature = "compio")]
pub use self::compio::*;
#[cfg(feature = "default-rt")]
pub use self::default_rt::*;
#[allow(dead_code)]
#[cfg(all(
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
mod no_rt {
@ -542,6 +672,7 @@ mod no_rt {
not(feature = "tokio"),
not(feature = "async-std"),
not(feature = "compio"),
not(feature = "default-rt"),
not(feature = "glommio")
))]
pub use self::no_rt::*;

59
ntex-runtime/Cargo.toml Normal file
View file

@ -0,0 +1,59 @@
[package]
name = "ntex-runtime"
version = "0.1.0"
description = "Async runtime for ntex"
categories = ["asynchronous"]
keywords = ["async", "runtime"]
edition = { workspace = true }
authors = { workspace = true }
readme = { workspace = true }
license = { workspace = true }
repository = { workspace = true }
[package.metadata.docs.rs]
all-features = true
default-target = "x86_64-unknown-linux-gnu"
rustdoc-args = ["--cfg", "docsrs"]
targets = [
"x86_64-pc-windows-gnu",
"x86_64-unknown-linux-gnu",
"x86_64-apple-darwin",
"aarch64-apple-ios",
"aarch64-linux-android",
"x86_64-unknown-dragonfly",
"x86_64-unknown-freebsd",
"x86_64-unknown-illumos",
"x86_64-unknown-netbsd",
"x86_64-unknown-openbsd",
]
[dependencies]
ntex-iodriver = "0.1"
async-task = { workspace = true }
cfg-if = { workspace = true }
crossbeam-queue = { workspace = true }
futures-util = { workspace = true }
scoped-tls = { workspace = true }
fxhash = { workspace = true }
log = { workspace = true }
socket2 = { workspace = true, features = ["all"] }
# Windows specific dependencies
[target.'cfg(windows)'.dependencies]
windows-sys = { workspace = true, features = [
"Win32_Foundation",
"Win32_Networking_WinSock",
"Win32_System_IO",
] }
# Unix specific dependencies
[target.'cfg(unix)'.dependencies]
libc = { workspace = true }
[target.'cfg(windows)'.dev-dependencies]
windows-sys = { workspace = true, features = ["Win32_UI_WindowsAndMessaging"] }
[features]
io-uring = ["ntex-iodriver/io-uring"]
polling = ["ntex-iodriver/polling"]

13
ntex-runtime/src/lib.rs Normal file
View file

@ -0,0 +1,13 @@
//! The async runtime for ntex.
#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
// #![warn(missing_docs)]
pub mod net;
mod op;
mod rt;
pub use async_task::Task;
pub use rt::{
spawn, spawn_blocking, submit, submit_with_flags, JoinHandle, Runtime, RuntimeBuilder,
};

11
ntex-runtime/src/net/mod.rs Normal file
View file

@ -0,0 +1,11 @@
//! Network related.
//!
//! Currently, TCP/UDP/Unix socket are implemented.
mod socket;
mod tcp;
mod unix;
pub use socket::*;
pub use tcp::*;
pub use unix::*;

225
ntex-runtime/src/net/socket.rs Normal file
View file

@ -0,0 +1,225 @@
use std::{future::Future, io, mem, mem::MaybeUninit};
use ntex_iodriver::{impl_raw_fd, op::CloseSocket, op::ShutdownSocket, syscall, AsRawFd};
use socket2::{Domain, Protocol, SockAddr, Socket as Socket2, Type};
#[derive(Debug)]
pub struct Socket {
socket: Socket2,
}
impl Socket {
pub fn from_socket2(socket: Socket2) -> io::Result<Self> {
#[cfg(unix)]
{
#[cfg(not(any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "hurd",
target_os = "illumos",
target_os = "linux",
target_os = "netbsd",
target_os = "openbsd",
target_os = "espidf",
target_os = "vita",
)))]
socket.set_cloexec(true)?;
#[cfg(any(
target_os = "ios",
target_os = "macos",
target_os = "tvos",
target_os = "watchos",
))]
socket.set_nosigpipe(true)?;
// On Linux we use blocking socket
// Newer kernels have the patch that allows to arm io_uring poll mechanism for
// non blocking socket when there is no connections in listen queue
//
// https://patchwork.kernel.org/project/linux-block/patch/f999615b-205c-49b7-b272-c4e42e45e09d@kernel.dk/#22949861
if cfg!(not(all(target_os = "linux", feature = "io-uring")))
|| ntex_iodriver::DriverType::is_polling()
{
socket.set_nonblocking(true)?;
}
}
Ok(Self { socket })
}
pub fn peer_addr(&self) -> io::Result<SockAddr> {
self.socket.peer_addr()
}
pub fn local_addr(&self) -> io::Result<SockAddr> {
self.socket.local_addr()
}
#[cfg(windows)]
pub async fn new(
domain: Domain,
ty: Type,
protocol: Option<Protocol>,
) -> io::Result<Self> {
use std::panic::resume_unwind;
let socket = crate::spawn_blocking(move || Socket2::new(domain, ty, protocol))
.await
.unwrap_or_else(|e| resume_unwind(e))?;
Self::from_socket2(socket)
}
#[cfg(unix)]
pub async fn new(
domain: Domain,
ty: Type,
protocol: Option<Protocol>,
) -> io::Result<Self> {
use std::os::fd::FromRawFd;
#[allow(unused_mut)]
let mut ty: i32 = ty.into();
#[cfg(any(
target_os = "android",
target_os = "dragonfly",
target_os = "freebsd",
target_os = "fuchsia",
target_os = "hurd",
target_os = "illumos",
target_os = "linux",
target_os = "netbsd",
target_os = "openbsd",
))]
{
ty |= libc::SOCK_CLOEXEC;
}
let op = ntex_iodriver::op::CreateSocket::new(
domain.into(),
ty,
protocol.map(|p| p.into()).unwrap_or_default(),
);
let (res, _) = crate::submit(op).await;
let socket = unsafe { Socket2::from_raw_fd(res? as _) };
Self::from_socket2(socket)
}
pub async fn bind(
addr: &SockAddr,
ty: Type,
protocol: Option<Protocol>,
) -> io::Result<Self> {
let socket = Self::new(addr.domain(), ty, protocol).await?;
socket.socket.bind(addr)?;
Ok(socket)
}
pub fn listen(&self, backlog: i32) -> io::Result<()> {
self.socket.listen(backlog)
}
pub fn close(self) -> impl Future<Output = io::Result<()>> {
let op = CloseSocket::from_raw_fd(self.as_raw_fd());
let fut = crate::submit(op);
mem::forget(self);
async move {
fut.await.0?;
Ok(())
}
}
pub async fn shutdown(&self) -> io::Result<()> {
let op = ShutdownSocket::new(self.as_raw_fd(), std::net::Shutdown::Write);
crate::submit(op).await.0?;
Ok(())
}
#[cfg(unix)]
pub unsafe fn get_socket_option<T: Copy>(
&self,
level: i32,
name: i32,
) -> io::Result<T> {
let mut value: MaybeUninit<T> = MaybeUninit::uninit();
let mut len = size_of::<T>() as libc::socklen_t;
syscall!(libc::getsockopt(
self.socket.as_raw_fd(),
level,
name,
value.as_mut_ptr() as _,
&mut len
))
.map(|_| {
debug_assert_eq!(len as usize, size_of::<T>());
// SAFETY: The value is initialized by `getsockopt`.
value.assume_init()
})
}
#[cfg(windows)]
pub unsafe fn get_socket_option<T: Copy>(
&self,
level: i32,
name: i32,
) -> io::Result<T> {
let mut value: MaybeUninit<T> = MaybeUninit::uninit();
let mut len = size_of::<T>() as i32;
syscall!(
SOCKET,
windows_sys::Win32::Networking::WinSock::getsockopt(
self.socket.as_raw_fd() as _,
level,
name,
value.as_mut_ptr() as _,
&mut len
)
)
.map(|_| {
debug_assert_eq!(len as usize, size_of::<T>());
// SAFETY: The value is initialized by `getsockopt`.
value.assume_init()
})
}
#[cfg(unix)]
pub unsafe fn set_socket_option<T: Copy>(
&self,
level: i32,
name: i32,
value: &T,
) -> io::Result<()> {
syscall!(libc::setsockopt(
self.socket.as_raw_fd(),
level,
name,
value as *const _ as _,
std::mem::size_of::<T>() as _
))
.map(|_| ())
}
#[cfg(windows)]
pub unsafe fn set_socket_option<T: Copy>(
&self,
level: i32,
name: i32,
value: &T,
) -> io::Result<()> {
syscall!(
SOCKET,
windows_sys::Win32::Networking::WinSock::setsockopt(
self.socket.as_raw_fd() as _,
level,
name,
value as *const _ as _,
std::mem::size_of::<T>() as _
)
)
.map(|_| ())
}
}
impl_raw_fd!(Socket, Socket2, socket, socket);

50
ntex-runtime/src/net/tcp.rs Normal file
View file

@ -0,0 +1,50 @@
use std::{future::Future, io, net::SocketAddr};
use ntex_iodriver::impl_raw_fd;
use socket2::Socket as Socket2;
use crate::net::Socket;
/// A TCP stream between a local and a remote socket.
///
/// A TCP stream can either be created by connecting to an endpoint, via the
/// `connect` method, or by accepting a connection from a listener.
#[derive(Debug)]
pub struct TcpStream {
inner: Socket,
}
impl TcpStream {
/// Creates new TcpStream from a std::net::TcpStream.
pub fn from_std(stream: std::net::TcpStream) -> io::Result<Self> {
Ok(Self {
inner: Socket::from_socket2(Socket2::from(stream))?,
})
}
/// Creates new TcpStream from a std::net::TcpStream.
pub fn from_socket(inner: Socket) -> Self {
Self { inner }
}
/// Close the socket.
pub fn close(self) -> impl Future<Output = io::Result<()>> {
self.inner.close()
}
/// Returns the socket address of the remote peer of this TCP connection.
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.inner
.peer_addr()
.map(|addr| addr.as_socket().expect("should be SocketAddr"))
}
/// Returns the socket address of the local half of this TCP connection.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.inner
.local_addr()
.map(|addr| addr.as_socket().expect("should be SocketAddr"))
}
}
impl_raw_fd!(TcpStream, socket2::Socket, inner, socket);

98
ntex-runtime/src/net/unix.rs Normal file
View file

@ -0,0 +1,98 @@
use std::{future::Future, io};
use ntex_iodriver::impl_raw_fd;
use socket2::{SockAddr, Socket as Socket2};
use crate::net::Socket;
/// A Unix stream between two local sockets on Windows & WSL.
///
/// A Unix stream can either be created by connecting to an endpoint, via the
/// `connect` method.
#[derive(Debug)]
pub struct UnixStream {
inner: Socket,
}
impl UnixStream {
#[cfg(unix)]
/// Creates new UnixStream from a std::os::unix::net::UnixStream.
pub fn from_std(stream: std::os::unix::net::UnixStream) -> io::Result<Self> {
Ok(Self {
inner: Socket::from_socket2(Socket2::from(stream))?,
})
}
/// Creates new TcpStream from a Socket.
pub fn from_socket(inner: Socket) -> Self {
Self { inner }
}
/// Close the socket. If the returned future is dropped before polling, the
/// socket won't be closed.
pub fn close(self) -> impl Future<Output = io::Result<()>> {
self.inner.close()
}
/// Returns the socket path of the remote peer of this connection.
pub fn peer_addr(&self) -> io::Result<SockAddr> {
#[allow(unused_mut)]
let mut addr = self.inner.peer_addr()?;
#[cfg(windows)]
{
fix_unix_socket_length(&mut addr);
}
Ok(addr)
}
/// Returns the socket path of the local half of this connection.
pub fn local_addr(&self) -> io::Result<SockAddr> {
self.inner.local_addr()
}
}
impl_raw_fd!(UnixStream, socket2::Socket, inner, socket);
#[cfg(windows)]
#[inline]
fn empty_unix_socket() -> SockAddr {
use windows_sys::Win32::Networking::WinSock::{AF_UNIX, SOCKADDR_UN};
// SAFETY: the length is correct
unsafe {
SockAddr::try_init(|addr, len| {
let addr: *mut SOCKADDR_UN = addr.cast();
std::ptr::write(
addr,
SOCKADDR_UN {
sun_family: AF_UNIX,
sun_path: [0; 108],
},
);
std::ptr::write(len, 3);
Ok(())
})
}
// it is always Ok
.unwrap()
.1
}
// The peer addr returned after ConnectEx is buggy. It contains bytes that
// should not belong to the address. Luckily a unix path should not contain `\0`
// until the end. We can determine the path ending by that.
#[cfg(windows)]
#[inline]
fn fix_unix_socket_length(addr: &mut SockAddr) {
use windows_sys::Win32::Networking::WinSock::SOCKADDR_UN;
// SAFETY: cannot construct non-unix socket address in safe way.
let unix_addr: &SOCKADDR_UN = unsafe { &*addr.as_ptr().cast() };
let addr_len = match std::ffi::CStr::from_bytes_until_nul(&unix_addr.sun_path) {
Ok(str) => str.to_bytes_with_nul().len() + 2,
Err(_) => std::mem::size_of::<SOCKADDR_UN>(),
};
unsafe {
addr.set_length(addr_len as _);
}
}

39
ntex-runtime/src/op.rs Normal file
View file

@ -0,0 +1,39 @@
use std::{future::Future, io, pin::Pin, task::Context, task::Poll};
use ntex_iodriver::{Key, OpCode, PushEntry};
use crate::rt::Runtime;
#[derive(Debug)]
pub(crate) struct OpFuture<T: OpCode> {
key: Option<Key<T>>,
}
impl<T: OpCode> OpFuture<T> {
pub(crate) fn new(key: Key<T>) -> Self {
Self { key: Some(key) }
}
}
impl<T: OpCode> Future for OpFuture<T> {
type Output = ((io::Result<usize>, T), u32);
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let res = Runtime::with_current(|r| r.poll_task(cx, self.key.take().unwrap()));
match res {
PushEntry::Pending(key) => {
self.key = Some(key);
Poll::Pending
}
PushEntry::Ready(res) => Poll::Ready(res),
}
}
}
impl<T: OpCode> Drop for OpFuture<T> {
fn drop(&mut self) {
if let Some(key) = self.key.take() {
Runtime::with_current(|r| r.cancel_op(key));
}
}
}

458
ntex-runtime/src/rt.rs Normal file
View file

@ -0,0 +1,458 @@
use std::any::{Any, TypeId};
use std::collections::{HashMap, VecDeque};
use std::future::{ready, Future};
use std::task::Context;
use std::{
cell::Cell, cell::RefCell, io, panic::AssertUnwindSafe, sync::Arc, thread,
time::Duration,
};
use async_task::{Runnable, Task};
use crossbeam_queue::SegQueue;
use futures_util::{future::Either, FutureExt};
use ntex_iodriver::{
op::Asyncify, AsRawFd, Key, NotifyHandle, OpCode, Proactor, ProactorBuilder, PushEntry,
RawFd,
};
use crate::op::OpFuture;
scoped_tls::scoped_thread_local!(static CURRENT_RUNTIME: Runtime);
/// Type alias for `Task<Result<T, Box<dyn Any + Send>>>`, which resolves to an
/// `Err` when the spawned future panicked.
pub type JoinHandle<T> = Task<Result<T, Box<dyn Any + Send>>>;
pub struct RemoteHandle {
handle: NotifyHandle,
runnables: Arc<RunnableQueue>,
}
impl RemoteHandle {
/// Wake up runtime
pub fn notify(&self) {
self.handle.notify().ok();
}
/// Spawns a new asynchronous task, returning a [`Task`] for it.
///
/// Spawning a task enables the task to execute concurrently to other tasks.
/// There is no guarantee that a spawned task will execute to completion.
pub fn spawn<F: Future + Send + 'static>(&self, future: F) -> Task<F::Output> {
let runnables = self.runnables.clone();
let handle = self.handle.clone();
let schedule = move |runnable| {
runnables.schedule(runnable, &handle);
};
let (runnable, task) = unsafe { async_task::spawn_unchecked(future, schedule) };
runnable.schedule();
task
}
}
struct RunnableQueue {
id: thread::ThreadId,
idle: Cell<bool>,
local_runnables: RefCell<VecDeque<Runnable>>,
sync_runnables: SegQueue<Runnable>,
}
impl RunnableQueue {
fn new() -> Self {
Self {
id: thread::current().id(),
idle: Cell::new(true),
local_runnables: RefCell::new(VecDeque::new()),
sync_runnables: SegQueue::new(),
}
}
fn schedule(&self, runnable: Runnable, handle: &NotifyHandle) {
if self.id == thread::current().id() {
self.local_runnables.borrow_mut().push_back(runnable);
if self.idle.get() {
let _ = handle.notify();
}
} else {
self.sync_runnables.push(runnable);
handle.notify().ok();
}
}
fn run(&self, event_interval: usize) {
self.idle.set(false);
for _ in 0..event_interval {
let task = self.local_runnables.borrow_mut().pop_front();
if let Some(task) = task {
task.run();
} else {
break;
}
}
for _ in 0..event_interval {
if !self.sync_runnables.is_empty() {
if let Some(task) = self.sync_runnables.pop() {
task.run();
continue;
}
}
break;
}
self.idle.set(true);
}
fn has_tasks(&self) -> bool {
!(self.local_runnables.borrow().is_empty() && self.sync_runnables.is_empty())
}
}
/// The async runtime for ntex. It is a thread local runtime, and cannot be
/// sent to other threads.
pub struct Runtime {
driver: Proactor,
runnables: Arc<RunnableQueue>,
event_interval: usize,
data: RefCell<HashMap<TypeId, Box<dyn Any>, fxhash::FxBuildHasher>>,
}
impl Runtime {
/// Create [`Runtime`] with default config.
pub fn new() -> io::Result<Self> {
Self::builder().build()
}
/// Create a builder for [`Runtime`].
pub fn builder() -> RuntimeBuilder {
RuntimeBuilder::new()
}
fn with_builder(builder: &RuntimeBuilder) -> io::Result<Self> {
Ok(Self {
driver: builder.proactor_builder.build()?,
runnables: Arc::new(RunnableQueue::new()),
event_interval: builder.event_interval,
data: RefCell::new(HashMap::default()),
})
}
/// Try to perform a function on the current runtime, and if no runtime is
/// running, return the function back.
pub fn try_with_current<T, F: FnOnce(&Self) -> T>(f: F) -> Result<T, F> {
if CURRENT_RUNTIME.is_set() {
Ok(CURRENT_RUNTIME.with(f))
} else {
Err(f)
}
}
/// Perform a function on the current runtime.
///
/// ## Panics
///
/// This method will panic if there are no running [`Runtime`].
pub fn with_current<T, F: FnOnce(&Self) -> T>(f: F) -> T {
#[cold]
fn not_in_ntex_runtime() -> ! {
panic!("not in a ntex runtime")
}
if CURRENT_RUNTIME.is_set() {
CURRENT_RUNTIME.with(f)
} else {
not_in_ntex_runtime()
}
}
/// Get current driver
pub fn driver(&self) -> &Proactor {
&self.driver
}
/// Get handle for current runtime
pub fn handle(&self) -> RemoteHandle {
RemoteHandle {
handle: self.driver.handle(),
runnables: self.runnables.clone(),
}
}
/// Set this runtime as current runtime, and perform a function in the
/// current scope.
pub fn enter<T, F: FnOnce() -> T>(&self, f: F) -> T {
CURRENT_RUNTIME.set(self, f)
}
/// Spawns a new asynchronous task, returning a [`Task`] for it.
///
/// # Safety
///
/// The caller should ensure the captured lifetime long enough.
pub unsafe fn spawn_unchecked<F: Future>(&self, future: F) -> Task<F::Output> {
let runnables = self.runnables.clone();
let handle = self.driver.handle();
let schedule = move |runnable| {
runnables.schedule(runnable, &handle);
};
let (runnable, task) = async_task::spawn_unchecked(future, schedule);
runnable.schedule();
task
}
/// Low level API to control the runtime.
///
/// Run the scheduled tasks.
///
/// The return value indicates whether there are still tasks in the queue.
pub fn run(&self) -> bool {
self.runnables.run(self.event_interval);
self.runnables.has_tasks()
}
/// Block on the future till it completes.
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
CURRENT_RUNTIME.set(self, || {
let mut result = None;
unsafe { self.spawn_unchecked(async { result = Some(future.await) }) }.detach();
self.runnables.run(self.event_interval);
loop {
if let Some(result) = result.take() {
return result;
}
self.poll_with_driver(self.runnables.has_tasks(), || {
self.runnables.run(self.event_interval);
});
}
})
}
/// Spawns a new asynchronous task, returning a [`Task`] for it.
///
/// Spawning a task enables the task to execute concurrently to other tasks.
/// There is no guarantee that a spawned task will execute to completion.
pub fn spawn<F: Future + 'static>(&self, future: F) -> JoinHandle<F::Output> {
unsafe { self.spawn_unchecked(AssertUnwindSafe(future).catch_unwind()) }
}
/// Spawns a blocking task in a new thread, and wait for it.
///
/// The task will not be cancelled even if the future is dropped.
pub fn spawn_blocking<T: Send + 'static>(
&self,
f: impl (FnOnce() -> T) + Send + 'static,
) -> JoinHandle<T> {
let op = Asyncify::new(move || {
let res = std::panic::catch_unwind(AssertUnwindSafe(f));
(Ok(0), res)
});
// It is safe and sound to use `submit` here because the task is spawned
// immediately.
unsafe {
self.spawn_unchecked(
self.submit_with_flags(op)
.map(|(res, _)| res)
.map(|res| res.1.into_inner()),
)
}
}
/// Attach a raw file descriptor/handle/socket to the runtime.
///
/// You only need this when authoring your own high-level APIs. High-level
/// resources in this crate are attached automatically.
pub fn attach(&self, fd: RawFd) -> io::Result<()> {
self.driver.attach(fd)
}
fn submit_raw<T: OpCode + 'static>(
&self,
op: T,
) -> PushEntry<Key<T>, (io::Result<usize>, T)> {
self.driver.push(op)
}
fn submit_with_flags<T: OpCode + 'static>(
&self,
op: T,
) -> impl Future<Output = ((io::Result<usize>, T), u32)> {
match self.submit_raw(op) {
PushEntry::Pending(user_data) => Either::Left(OpFuture::new(user_data)),
PushEntry::Ready(res) => {
// submit_flags won't be ready immediately, if ready, it must be error without
// flags
Either::Right(ready((res, 0)))
}
}
}
pub(crate) fn cancel_op<T: OpCode>(&self, op: Key<T>) {
self.driver.cancel(op);
}
pub(crate) fn poll_task<T: OpCode>(
&self,
cx: &mut Context,
op: Key<T>,
) -> PushEntry<Key<T>, ((io::Result<usize>, T), u32)> {
self.driver.pop(op).map_pending(|mut k| {
self.driver.update_waker(&mut k, cx.waker().clone());
k
})
}
fn poll_with_driver<F: FnOnce()>(&self, has_tasks: bool, f: F) {
let timeout = if has_tasks {
Some(Duration::ZERO)
} else {
None
};
match self.driver.poll(timeout, f) {
Ok(()) => {}
Err(e) => match e.kind() {
io::ErrorKind::TimedOut | io::ErrorKind::Interrupted => {
log::debug!("expected error: {e}");
}
_ => panic!("{e:?}"),
},
}
}
/// Insert a type into this runtime.
pub fn insert<T: 'static>(&self, val: T) {
self.data
.borrow_mut()
.insert(TypeId::of::<T>(), Box::new(val));
}
/// Check if container contains entry
pub fn contains<T: 'static>(&self) -> bool {
self.data.borrow().contains_key(&TypeId::of::<T>())
}
/// Get a reference to a type previously inserted on this runtime.
pub fn get<T>(&self) -> Option<T>
where
T: Clone + 'static,
{
self.data
.borrow()
.get(&TypeId::of::<T>())
.and_then(|boxed| boxed.downcast_ref().cloned())
}
}
impl Drop for Runtime {
fn drop(&mut self) {
self.enter(|| {
while self.runnables.sync_runnables.pop().is_some() {}
loop {
let runnable = self.runnables.local_runnables.borrow_mut().pop_front();
if runnable.is_none() {
break;
}
}
})
}
}
impl AsRawFd for Runtime {
fn as_raw_fd(&self) -> RawFd {
self.driver.as_raw_fd()
}
}
/// Builder for [`Runtime`].
#[derive(Debug, Clone)]
pub struct RuntimeBuilder {
proactor_builder: ProactorBuilder,
event_interval: usize,
}
impl Default for RuntimeBuilder {
fn default() -> Self {
Self::new()
}
}
impl RuntimeBuilder {
/// Create the builder with default config.
pub fn new() -> Self {
Self {
proactor_builder: ProactorBuilder::new(),
event_interval: 61,
}
}
/// Replace proactor builder.
pub fn with_proactor(&mut self, builder: ProactorBuilder) -> &mut Self {
self.proactor_builder = builder;
self
}
/// Sets the number of scheduler ticks after which the scheduler will poll
/// for external events (timers, I/O, and so on).
///
/// A scheduler “tick” roughly corresponds to one poll invocation on a task.
pub fn event_interval(&mut self, val: usize) -> &mut Self {
self.event_interval = val;
self
}
/// Build [`Runtime`].
pub fn build(&self) -> io::Result<Runtime> {
Runtime::with_builder(self)
}
}
/// Spawns a new asynchronous task, returning a [`Task`] for it.
///
/// Spawning a task enables the task to execute concurrently to other tasks.
/// There is no guarantee that a spawned task will execute to completion.
///
/// ## Panics
///
/// This method doesn't create runtime. It tries to obtain the current runtime
/// by [`Runtime::with_current`].
pub fn spawn<F: Future + 'static>(future: F) -> JoinHandle<F::Output> {
Runtime::with_current(|r| r.spawn(future))
}
/// Spawns a blocking task in a new thread, and wait for it.
///
/// The task will not be cancelled even if the future is dropped.
///
/// ## Panics
///
/// This method doesn't create runtime. It tries to obtain the current runtime
/// by [`Runtime::with_current`].
pub fn spawn_blocking<T: Send + 'static>(
f: impl (FnOnce() -> T) + Send + 'static,
) -> JoinHandle<T> {
Runtime::with_current(|r| r.spawn_blocking(f))
}
/// Submit an operation to the current runtime, and return a future for it.
///
/// ## Panics
///
/// This method doesn't create runtime. It tries to obtain the current runtime
/// by [`Runtime::with_current`].
pub async fn submit<T: OpCode + 'static>(op: T) -> (io::Result<usize>, T) {
submit_with_flags(op).await.0
}
/// Submit an operation to the current runtime, and return a future for it with
/// flags.
///
/// ## Panics
///
/// This method doesn't create runtime. It tries to obtain the current runtime
/// by [`Runtime::with_current`].
pub async fn submit_with_flags<T: OpCode + 'static>(
op: T,
) -> ((io::Result<usize>, T), u32) {
Runtime::with_current(|r| r.submit_with_flags(op)).await
}

5
ntex/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex"
version = "2.11.0"
version = "2.12.0"
authors = ["ntex contributors <team@ntex.rs>"]
description = "Framework for composable network services"
readme = "README.md"
@ -54,6 +54,9 @@ async-std = ["ntex-net/async-std"]
# compio runtime
compio = ["ntex-net/compio"]
# default ntex runtime
default-rt = ["ntex-net/default-rt"]
# websocket support
ws = ["dep:sha-1"]

5
ntex/tests/server.rs
View file

@ -4,7 +4,7 @@
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
#[cfg(feature = "tokio")]
use std::{io, sync::Arc};
use std::{io::Read, net, sync::mpsc, thread, time};
use std::{io::Read, io::Write, net, sync::mpsc, thread, time};
use ntex::codec::BytesCodec;
use ntex::io::Io;
@ -72,7 +72,6 @@ async fn test_listen() {
#[ntex::test]
#[cfg(unix)]
async fn test_run() {
let _ = env_logger::try_init();
let addr = TestServer::unused_addr();
let (tx, rx) = mpsc::channel();
@ -84,7 +83,6 @@ async fn test_run() {
.workers(1)
.disable_signals()
.bind("test", addr, move |_| {
compio_driver::enable_logging();
fn_service(|io: Io| async move {
io.send(Bytes::from_static(b"test"), &BytesCodec)
.await
@ -102,6 +100,7 @@ async fn test_run() {
let mut buf = [1u8; 4];
let mut conn = net::TcpStream::connect(addr).unwrap();
conn.write(&b"test"[..]).unwrap();
let _ = conn.read_exact(&mut buf);
assert_eq!(buf, b"test"[..]);