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

allow to replace async runtime

Browse source
This commit is contained in:
Nikolay Kim 2021-12-18 11:46:11 +06:00
parent aa5f6e4b55
commit b8a8e98c1c
26 changed files with 529 additions and 604 deletions
Download patch file Download diff file Expand all files Collapse all files

15
ntex-io/Cargo.toml
View file

@ -18,25 +18,26 @@ path = "src/lib.rs"
[features]
default = ["tokio"]
# tokio support
tokio = ["tok-io"]
# tokio traits support
tokio-traits = ["tok-io/net"]
# tokio runtime support
tokio = ["tok-io/net"]
[dependencies]
bitflags = "1.3"
fxhash = "0.2.1"
ntex-codec = "0.5.1"
ntex-codec = "0.6.0"
ntex-bytes = "0.1.7"
ntex-util = "0.1.2"
ntex-service = "0.2.1"
log = "0.4"
pin-project-lite = "0.2"
tok-io = { version = "1", package = "tokio", default-features = false, features = ["net"], optional = true }
backtrace = "*"
tok-io = { version = "1", package = "tokio", default-features = false, optional = true }
[dev-dependencies]
ntex = "0.5.0-b.0"
futures = "0.3"
rand = "0.8"
env_logger = "0.9"
env_logger = "0.9"

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

@ -1,15 +1,13 @@
//! Framed transport dispatcher
use std::{
cell::Cell, future::Future, pin::Pin, rc::Rc, task::Context, task::Poll, time,
};
use std::{cell::Cell, future, pin::Pin, rc::Rc, task::Context, task::Poll, time};
use ntex_bytes::Pool;
use ntex_codec::{Decoder, Encoder};
use ntex_service::{IntoService, Service};
use ntex_util::future::Either;
use ntex_util::time::{now, Seconds};
use ntex_util::{future::Either, spawn};
use super::{DispatchItem, IoBoxed, ReadRef, Timer, WriteRef};
use super::{rt::spawn, DispatchItem, IoBoxed, ReadRef, Timer, WriteRef};
type Response<U> = <U as Encoder>::Item;
@ -178,7 +176,7 @@ where
}
}
impl<S, U> Future for Dispatcher<S, U>
impl<S, U> future::Future for Dispatcher<S, U>
where
S: Service<Request = DispatchItem<U>, Response = Option<Response<U>>> + 'static,
U: Decoder + Encoder + 'static,

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

@ -1,17 +1,19 @@
use std::{any::Any, any::TypeId, fmt, future::Future, io, task::Context, task::Poll};
pub mod testing;
pub mod types;
mod dispatcher;
mod filter;
mod state;
mod tasks;
mod time;
pub mod types;
mod utils;
#[cfg(feature = "tokio")]
#[cfg(any(feature = "tokio-traits", feature = "tokio"))]
mod tokio_impl;
#[cfg(any(feature = "tokio"))]
mod tokio_rt;
use ntex_bytes::BytesMut;
use ntex_codec::{Decoder, Encoder};
@ -128,6 +130,13 @@ where
}
}
pub mod rt {
//! async runtime helpers
#[cfg(feature = "tokio")]
pub use crate::tokio_rt::*;
}
#[cfg(test)]
mod tests {
use super::*;

273
ntex-io/src/testing.rs
View file

@ -37,7 +37,7 @@ pub struct IoTest {
}
bitflags::bitflags! {
struct Flags: u8 {
struct IoTestFlags: u8 {
const FLUSHED = 0b0000_0001;
const CLOSED = 0b0000_0010;
}
@ -61,35 +61,35 @@ struct State {
struct Channel {
buf: BytesMut,
buf_cap: usize,
flags: Flags,
flags: IoTestFlags,
waker: AtomicWaker,
read: IoState,
write: IoState,
read: IoTestState,
write: IoTestState,
}
impl Channel {
fn is_closed(&self) -> bool {
self.flags.contains(Flags::CLOSED)
self.flags.contains(IoTestFlags::CLOSED)
}
}
impl Default for Flags {
impl Default for IoTestFlags {
fn default() -> Self {
Flags::empty()
IoTestFlags::empty()
}
}
#[derive(Debug)]
enum IoState {
enum IoTestState {
Ok,
Pending,
Close,
Err(io::Error),
}
impl Default for IoState {
impl Default for IoTestState {
fn default() -> Self {
IoState::Ok
IoTestState::Ok
}
}
@ -139,19 +139,19 @@ impl IoTest {
/// Set read to Pending state
pub fn read_pending(&self) {
self.remote.lock().unwrap().borrow_mut().read = IoState::Pending;
self.remote.lock().unwrap().borrow_mut().read = IoTestState::Pending;
}
/// Set read to error
pub fn read_error(&self, err: io::Error) {
let channel = self.remote.lock().unwrap();
channel.borrow_mut().read = IoState::Err(err);
channel.borrow_mut().read = IoTestState::Err(err);
channel.borrow().waker.wake();
}
/// Set write error on remote side
pub fn write_error(&self, err: io::Error) {
self.local.lock().unwrap().borrow_mut().write = IoState::Err(err);
self.local.lock().unwrap().borrow_mut().write = IoTestState::Err(err);
self.remote.lock().unwrap().borrow().waker.wake();
}
@ -180,7 +180,7 @@ impl IoTest {
{
let guard = self.remote.lock().unwrap();
let mut remote = guard.borrow_mut();
remote.read = IoState::Close;
remote.read = IoTestState::Close;
remote.waker.wake();
log::trace!("close remote socket");
}
@ -256,13 +256,13 @@ impl IoTest {
}
match mem::take(&mut ch.read) {
IoState::Ok => Poll::Pending,
IoState::Close => {
ch.read = IoState::Close;
IoTestState::Ok => Poll::Pending,
IoTestState::Close => {
ch.read = IoTestState::Close;
Poll::Ready(Ok(0))
}
IoState::Pending => Poll::Pending,
IoState::Err(e) => Poll::Ready(Err(e)),
IoTestState::Pending => Poll::Pending,
IoTestState::Err(e) => Poll::Ready(Err(e)),
}
}
@ -275,12 +275,12 @@ impl IoTest {
let mut ch = guard.borrow_mut();
match mem::take(&mut ch.write) {
IoState::Ok => {
IoTestState::Ok => {
let cap = cmp::min(buf.len(), ch.buf_cap);
if cap > 0 {
ch.buf.extend(&buf[..cap]);
ch.buf_cap -= cap;
ch.flags.remove(Flags::FLUSHED);
ch.flags.remove(IoTestFlags::FLUSHED);
ch.waker.wake();
Poll::Ready(Ok(cap))
} else {
@ -297,8 +297,8 @@ impl IoTest {
Poll::Pending
}
}
IoState::Close => Poll::Ready(Ok(0)),
IoState::Pending => {
IoTestState::Close => Poll::Ready(Ok(0)),
IoTestState::Pending => {
*self
.local
.lock()
@ -311,7 +311,7 @@ impl IoTest {
.borrow_mut() = Some(cx.waker().clone());
Poll::Pending
}
IoState::Err(e) => Poll::Ready(Err(e)),
IoTestState::Err(e) => Poll::Ready(Err(e)),
}
}
}
@ -346,125 +346,15 @@ impl Drop for IoTest {
}
}
#[cfg(feature = "tokio")]
mod tokio {
use std::task::{Context, Poll};
use std::{cmp, io, mem, pin::Pin};
use tok_io::io::{AsyncRead, AsyncWrite, ReadBuf};
use super::{Flags, IoState, IoTest};
impl AsyncRead for IoTest {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let guard = self.local.lock().unwrap();
let mut ch = guard.borrow_mut();
*ch.waker.0.lock().unwrap().borrow_mut() = Some(cx.waker().clone());
if !ch.buf.is_empty() {
let size = std::cmp::min(ch.buf.len(), buf.remaining());
let b = ch.buf.split_to(size);
buf.put_slice(&b);
return Poll::Ready(Ok(()));
}
match mem::take(&mut ch.read) {
IoState::Ok => Poll::Pending,
IoState::Close => {
ch.read = IoState::Close;
Poll::Ready(Ok(()))
}
IoState::Pending => Poll::Pending,
IoState::Err(e) => Poll::Ready(Err(e)),
}
}
}
impl AsyncWrite for IoTest {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let guard = self.remote.lock().unwrap();
let mut ch = guard.borrow_mut();
match mem::take(&mut ch.write) {
IoState::Ok => {
let cap = cmp::min(buf.len(), ch.buf_cap);
if cap > 0 {
ch.buf.extend(&buf[..cap]);
ch.buf_cap -= cap;
ch.flags.remove(Flags::FLUSHED);
ch.waker.wake();
Poll::Ready(Ok(cap))
} else {
*self
.local
.lock()
.unwrap()
.borrow_mut()
.waker
.0
.lock()
.unwrap()
.borrow_mut() = Some(cx.waker().clone());
Poll::Pending
}
}
IoState::Close => Poll::Ready(Ok(0)),
IoState::Pending => {
*self
.local
.lock()
.unwrap()
.borrow_mut()
.waker
.0
.lock()
.unwrap()
.borrow_mut() = Some(cx.waker().clone());
Poll::Pending
}
IoState::Err(e) => Poll::Ready(Err(e)),
}
}
fn poll_flush(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<io::Result<()>> {
self.local
.lock()
.unwrap()
.borrow_mut()
.flags
.insert(Flags::CLOSED);
Poll::Ready(Ok(()))
}
}
}
impl IoStream for IoTest {
fn start(self, read: ReadContext, write: WriteContext) -> Option<Box<dyn Handle>> {
let io = Rc::new(self);
ntex_util::spawn(ReadTask {
crate::rt::spawn(ReadTask {
io: io.clone(),
state: read,
});
ntex_util::spawn(WriteTask {
crate::rt::spawn(WriteTask {
io: io.clone(),
state: write,
st: IoWriteState::Processing(None),
@ -615,7 +505,7 @@ impl Future for WriteTask {
.unwrap()
.borrow_mut()
.flags
.insert(Flags::CLOSED);
.insert(IoTestFlags::CLOSED);
this.state.close(None);
Poll::Ready(())
}
@ -652,7 +542,7 @@ impl Future for WriteTask {
.unwrap()
.borrow_mut()
.flags
.insert(Flags::CLOSED);
.insert(IoTestFlags::CLOSED);
*st = Shutdown::Stopping;
continue;
}
@ -752,6 +642,113 @@ pub(super) fn flush_io(
}
}
#[cfg(any(feature = "tokio", feature = "tokio-traits"))]
mod tokio_impl {
use tok_io::io::{AsyncRead, AsyncWrite, ReadBuf};
use super::*;
impl AsyncRead for IoTest {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let guard = self.local.lock().unwrap();
let mut ch = guard.borrow_mut();
*ch.waker.0.lock().unwrap().borrow_mut() = Some(cx.waker().clone());
if !ch.buf.is_empty() {
let size = std::cmp::min(ch.buf.len(), buf.remaining());
let b = ch.buf.split_to(size);
buf.put_slice(&b);
return Poll::Ready(Ok(()));
}
match mem::take(&mut ch.read) {
IoTestState::Ok => Poll::Pending,
IoTestState::Close => {
ch.read = IoTestState::Close;
Poll::Ready(Ok(()))
}
IoTestState::Pending => Poll::Pending,
IoTestState::Err(e) => Poll::Ready(Err(e)),
}
}
}
impl AsyncWrite for IoTest {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let guard = self.remote.lock().unwrap();
let mut ch = guard.borrow_mut();
match mem::take(&mut ch.write) {
IoTestState::Ok => {
let cap = cmp::min(buf.len(), ch.buf_cap);
if cap > 0 {
ch.buf.extend(&buf[..cap]);
ch.buf_cap -= cap;
ch.flags.remove(IoTestFlags::FLUSHED);
ch.waker.wake();
Poll::Ready(Ok(cap))
} else {
*self
.local
.lock()
.unwrap()
.borrow_mut()
.waker
.0
.lock()
.unwrap()
.borrow_mut() = Some(cx.waker().clone());
Poll::Pending
}
}
IoTestState::Close => Poll::Ready(Ok(0)),
IoTestState::Pending => {
*self
.local
.lock()
.unwrap()
.borrow_mut()
.waker
.0
.lock()
.unwrap()
.borrow_mut() = Some(cx.waker().clone());
Poll::Pending
}
IoTestState::Err(e) => Poll::Ready(Err(e)),
}
}
fn poll_flush(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<io::Result<()>> {
self.local
.lock()
.unwrap()
.borrow_mut()
.flags
.insert(IoTestFlags::CLOSED);
Poll::Ready(Ok(()))
}
}
}
#[cfg(test)]
#[allow(clippy::redundant_clone)]
mod tests {

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

@ -2,10 +2,10 @@ use std::{
cell::RefCell, collections::BTreeMap, collections::HashSet, rc::Rc, time::Instant,
};
use ntex_util::spawn;
use ntex_util::time::{now, sleep, Millis};
use super::state::{IoRef, IoStateInner};
use crate::rt::spawn;
use crate::state::{IoRef, IoStateInner};
pub struct Timer(Rc<RefCell<Inner>>);

43
ntex-io/src/tokio_impl.rs
View file

@ -1,12 +1,12 @@
use std::task::{Context, Poll};
use std::{any, cell::RefCell, cmp, future::Future, io, pin::Pin, rc::Rc};
use std::{any, cell::RefCell, cmp, future::Future, io, mem, pin::Pin, rc::Rc};
use ntex_bytes::{Buf, BufMut};
use ntex_bytes::{Buf, BufMut, BytesMut};
use ntex_util::time::{sleep, Sleep};
use tok_io::io::{AsyncRead, AsyncWrite, ReadBuf};
use tok_io::net::TcpStream;
use super::{
use crate::{
types, Filter, Handle, Io, IoBoxed, IoStream, ReadContext, WriteContext,
WriteReadiness,
};
@ -71,7 +71,7 @@ impl Future for ReadTask {
buf.reserve(hw - remaining);
}
match ntex_codec::poll_read_buf(Pin::new(&mut *io), cx, &mut buf) {
match poll_read_buf(Pin::new(&mut *io), cx, &mut buf) {
Poll::Pending => break,
Poll::Ready(Ok(n)) => {
if n == 0 {
@ -505,8 +505,7 @@ mod unixstream {
buf.reserve(hw - remaining);
}
match ntex_codec::poll_read_buf(Pin::new(&mut *io), cx, &mut buf)
{
match poll_read_buf(Pin::new(&mut *io), cx, &mut buf) {
Poll::Pending => break,
Poll::Ready(Ok(n)) => {
if n == 0 {
@ -699,3 +698,35 @@ mod unixstream {
}
}
}
pub fn poll_read_buf<T: AsyncRead>(
io: Pin<&mut T>,
cx: &mut Context<'_>,
buf: &mut BytesMut,
) -> Poll<io::Result<usize>> {
if !buf.has_remaining_mut() {
return Poll::Ready(Ok(0));
}
let n = {
let dst =
unsafe { &mut *(buf.chunk_mut() as *mut _ as *mut [mem::MaybeUninit<u8>]) };
let mut buf = ReadBuf::uninit(dst);
let ptr = buf.filled().as_ptr();
if io.poll_read(cx, &mut buf)?.is_pending() {
return Poll::Pending;
}
// Ensure the pointer does not change from under us
assert_eq!(ptr, buf.filled().as_ptr());
buf.filled().len()
};
// Safety: This is guaranteed to be the number of initialized (and read)
// bytes due to the invariants provided by `ReadBuf::filled`.
unsafe {
buf.advance_mut(n);
}
Poll::Ready(Ok(n))
}

37
ntex-io/src/tokio_rt.rs Normal file
View file

@ -0,0 +1,37 @@
//! async net providers
use ntex_util::future::lazy;
use std::future::Future;
/// 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) -> tok_io::task::JoinHandle<F::Output>
where
F: Future + 'static,
{
tok_io::task::spawn_local(f)
}
/// 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) -> tok_io::task::JoinHandle<R::Output>
where
F: FnOnce() -> R + 'static,
R: Future + 'static,
{
spawn(async move {
let r = lazy(|_| f()).await;
r.await
})
}