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

Refactor ntex-io (#164)

Browse source 
* Refactor Io and Filter types
This commit is contained in:
Nikolay Kim 2023-01-23 14:42:00 +06:00 • committed by GitHub
parent 8cbd8758a5
commit 83d05d81ef
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
55 changed files with 1615 additions and 1495 deletions
Download patch file Download diff file Expand all files Collapse all files

6
ntex-tokio/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex-tokio"
version = "0.2.0"
version = "0.2.1"
authors = ["ntex contributors <team@ntex.rs>"]
description = "tokio intergration for ntex framework"
keywords = ["network", "framework", "async", "futures"]
@ -16,8 +16,8 @@ name = "ntex_tokio"
path = "src/lib.rs"
[dependencies]
ntex-bytes = "0.1.18"
ntex-io = "0.2.0"
ntex-bytes = "0.1.19"
ntex-io = "0.2.1"
ntex-util = "0.2.0"
log = "0.4"
pin-project-lite = "0.2"

123
ntex-tokio/src/io.rs
View file

@ -54,73 +54,42 @@ impl Future for ReadTask {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_ref();
loop {
this.state.with_buf(|buf, hw, lw| {
match ready!(this.state.poll_ready(cx)) {
ReadStatus::Ready => {
let pool = this.state.memory_pool();
let mut io = this.io.borrow_mut();
let mut buf = self.state.get_read_buf();
let (hw, lw) = pool.read_params().unpack();
// read data from socket
let mut new_bytes = 0;
let mut close = false;
let mut pending = false;
let mut io = this.io.borrow_mut();
loop {
// make sure we've got room
let remaining = buf.remaining_mut();
if remaining < lw {
buf.reserve(hw - remaining);
}
match poll_read_buf(Pin::new(&mut *io), cx, &mut buf) {
Poll::Pending => {
pending = true;
break;
}
return match poll_read_buf(Pin::new(&mut *io), cx, buf) {
Poll::Pending => Poll::Pending,
Poll::Ready(Ok(n)) => {
if n == 0 {
log::trace!("tokio stream is disconnected");
close = true;
Poll::Ready(Ok(()))
} else if buf.len() < hw {
continue;
} else {
new_bytes += n;
if new_bytes <= hw {
continue;
}
Poll::Pending
}
break;
}
Poll::Ready(Err(err)) => {
log::trace!("read task failed on io {:?}", err);
drop(io);
this.state.release_read_buf(buf, new_bytes);
this.state.close(Some(err));
return Poll::Ready(());
Poll::Ready(Err(err))
}
}
};
}
drop(io);
if new_bytes == 0 && close {
this.state.close(None);
return Poll::Ready(());
}
this.state.release_read_buf(buf, new_bytes);
return if close {
this.state.close(None);
Poll::Ready(())
} else if pending {
Poll::Pending
} else {
continue;
};
}
ReadStatus::Terminate => {
log::trace!("read task is instructed to shutdown");
return Poll::Ready(());
Poll::Ready(Ok(()))
}
}
}
})
}
}
@ -269,14 +238,14 @@ impl Future for WriteTask {
if read_buf.filled().is_empty() =>
{
this.state.close(None);
log::trace!("write task is stopped");
log::trace!("tokio write task is stopped");
return Poll::Ready(());
}
Poll::Pending => {
*count += read_buf.filled().len() as u16;
if *count > 4096 {
log::trace!(
"write task is stopped, too much input"
"tokio write task is stopped, too much input"
);
this.state.close(None);
return Poll::Ready(());
@ -344,7 +313,7 @@ pub(super) fn flush_io<T: AsyncRead + AsyncWrite + Unpin>(
}
}
}
log::trace!("flushed {} bytes", written);
// log::trace!("flushed {} bytes", written);
// remove written data
let result = if written == len {
@ -501,18 +470,11 @@ mod unixstream {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_ref();
loop {
this.state.with_buf(|buf, hw, lw| {
match ready!(this.state.poll_ready(cx)) {
ReadStatus::Ready => {
let pool = this.state.memory_pool();
let mut io = this.io.borrow_mut();
let mut buf = self.state.get_read_buf();
let (hw, lw) = pool.read_params().unpack();
// read data from socket
let mut new_bytes = 0;
let mut close = false;
let mut pending = false;
let mut io = this.io.borrow_mut();
loop {
// make sure we've got room
let remaining = buf.remaining_mut();
@ -520,54 +482,31 @@ mod unixstream {
buf.reserve(hw - remaining);
}
match poll_read_buf(Pin::new(&mut *io), cx, &mut buf) {
Poll::Pending => {
pending = true;
break;
}
return match poll_read_buf(Pin::new(&mut *io), cx, buf) {
Poll::Pending => Poll::Pending,
Poll::Ready(Ok(n)) => {
if n == 0 {
log::trace!("unix stream is disconnected");
close = true;
log::trace!("tokio unix stream is disconnected");
Poll::Ready(Ok(()))
} else if buf.len() < hw {
continue;
} else {
new_bytes += n;
if new_bytes <= hw {
continue;
}
Poll::Pending
}
break;
}
Poll::Ready(Err(err)) => {
log::trace!("read task failed on io {:?}", err);
drop(io);
this.state.release_read_buf(buf, new_bytes);
this.state.close(Some(err));
return Poll::Ready(());
log::trace!("unix stream read task failed {:?}", err);
Poll::Ready(Err(err))
}
}
};
}
drop(io);
if new_bytes == 0 && close {
this.state.close(None);
return Poll::Ready(());
}
this.state.release_read_buf(buf, new_bytes);
return if close {
this.state.close(None);
Poll::Ready(())
} else if pending {
Poll::Pending
} else {
continue;
};
}
ReadStatus::Terminate => {
log::trace!("read task is instructed to shutdown");
return Poll::Ready(());
Poll::Ready(Ok(()))
}
}
}
})
}
}
@ -735,10 +674,6 @@ pub fn poll_read_buf<T: AsyncRead>(
cx: &mut Context<'_>,
buf: &mut BytesVec,
) -> 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>]) };