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

Optimize io layout (#388)

Browse source
This commit is contained in:
Nikolay Kim 2024-07-30 20:07:00 +02:00 • committed by GitHub
parent d81e089059
commit f5bad7bfa0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 215 additions and 168 deletions
Download patch file Download diff file Expand all files Collapse all files

4
ntex-io/CHANGES.md
View file

@ -1,5 +1,9 @@
# Changes # Changes
## [2.1.0] - 2024-07-30
* Optimize `Io` layout
## [2.0.0] - 2024-05-28 ## [2.0.0] - 2024-05-28
* Use async fn for Service::ready() and Service::shutdown() * Use async fn for Service::ready() and Service::shutdown()

4
ntex-io/Cargo.toml
View file

@ -1,6 +1,6 @@
[package] [package]
name = "ntex-io" name = "ntex-io"
version = "2.0.0" version = "2.1.0"
authors = ["ntex contributors <team@ntex.rs>"] authors = ["ntex contributors <team@ntex.rs>"]
description = "Utilities for encoding and decoding frames" description = "Utilities for encoding and decoding frames"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]
@ -18,7 +18,7 @@ path = "src/lib.rs"
[dependencies] [dependencies]
ntex-codec = "0.6.2" ntex-codec = "0.6.2"
ntex-bytes = "0.1.24" ntex-bytes = "0.1.24"
ntex-util = "2.0" ntex-util = "2.2"
ntex-service = "3.0" ntex-service = "3.0"
bitflags = "2" bitflags = "2"

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

@ -295,6 +295,12 @@ pub struct ReadBuf<'a> {
} }
impl<'a> ReadBuf<'a> { impl<'a> ReadBuf<'a> {
#[inline]
/// Get io tag
pub fn tag(&self) -> &'static str {
self.io.tag()
}
#[inline] #[inline]
/// Get number of newly added bytes /// Get number of newly added bytes
pub fn nbytes(&self) -> usize { pub fn nbytes(&self) -> usize {
@ -433,6 +439,12 @@ pub struct WriteBuf<'a> {
} }
impl<'a> WriteBuf<'a> { impl<'a> WriteBuf<'a> {
#[inline]
/// Get io tag
pub fn tag(&self) -> &'static str {
self.io.tag()
}
#[inline] #[inline]
/// Initiate graceful io stream shutdown /// Initiate graceful io stream shutdown
pub fn want_shutdown(&self) { pub fn want_shutdown(&self) {

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

@ -726,7 +726,6 @@ mod tests {
#[ntex::test] #[ntex::test]
async fn test_basic() { async fn test_basic() {
let _ = env_logger::try_init();
let (client, server) = IoTest::create(); let (client, server) = IoTest::create();
client.remote_buffer_cap(1024); client.remote_buffer_cap(1024);
client.write("GET /test HTTP/1\r\n\r\n"); client.write("GET /test HTTP/1\r\n\r\n");

2
ntex-io/src/filter.rs
View file

@ -26,7 +26,7 @@ pub(crate) struct NullFilter;
const NULL: NullFilter = NullFilter; const NULL: NullFilter = NullFilter;
impl NullFilter { impl NullFilter {
pub(super) fn get() -> &'static dyn Filter { pub(super) const fn get() -> &'static dyn Filter {
&NULL &NULL
} }
} }

343
ntex-io/src/io.rs
View file

@ -1,4 +1,4 @@
use std::cell::Cell; use std::cell::{Cell, UnsafeCell};
use std::future::{poll_fn, Future}; use std::future::{poll_fn, Future};
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use std::{fmt, hash, io, marker, mem, ops, pin::Pin, ptr, rc::Rc}; use std::{fmt, hash, io, marker, mem, ops, pin::Pin, ptr, rc::Rc};
@ -50,12 +50,13 @@ bitflags::bitflags! {
} }
/// Interface object to underlying io stream /// Interface object to underlying io stream
pub struct Io<F = Base>(pub(super) IoRef, FilterItem<F>); pub struct Io<F = Base>(UnsafeCell<IoRef>, marker::PhantomData<F>);
#[derive(Clone)] #[derive(Clone)]
pub struct IoRef(pub(super) Rc<IoState>); pub struct IoRef(pub(super) Rc<IoState>);
pub(crate) struct IoState { pub(crate) struct IoState {
filter: FilterPtr,
pub(super) flags: Cell<Flags>, pub(super) flags: Cell<Flags>,
pub(super) pool: Cell<PoolRef>, pub(super) pool: Cell<PoolRef>,
pub(super) disconnect_timeout: Cell<Seconds>, pub(super) disconnect_timeout: Cell<Seconds>,
@ -64,7 +65,6 @@ pub(crate) struct IoState {
pub(super) write_task: LocalWaker, pub(super) write_task: LocalWaker,
pub(super) dispatch_task: LocalWaker, pub(super) dispatch_task: LocalWaker,
pub(super) buffer: Stack, pub(super) buffer: Stack,
pub(super) filter: Cell<&'static dyn Filter>,
pub(super) handle: Cell<Option<Box<dyn Handle>>>, pub(super) handle: Cell<Option<Box<dyn Handle>>>,
pub(super) timeout: Cell<TimerHandle>, pub(super) timeout: Cell<TimerHandle>,
pub(super) tag: Cell<&'static str>, pub(super) tag: Cell<&'static str>,
@ -75,6 +75,10 @@ pub(crate) struct IoState {
const DEFAULT_TAG: &str = "IO"; const DEFAULT_TAG: &str = "IO";
impl IoState { impl IoState {
pub(super) fn filter(&self) -> &dyn Filter {
self.filter.filter.get()
}
pub(super) fn insert_flags(&self, f: Flags) { pub(super) fn insert_flags(&self, f: Flags) {
let mut flags = self.flags.get(); let mut flags = self.flags.get();
flags.insert(f); flags.insert(f);
@ -171,7 +175,7 @@ impl fmt::Debug for IoState {
let res = f let res = f
.debug_struct("IoState") .debug_struct("IoState")
.field("flags", &self.flags) .field("flags", &self.flags)
.field("pool", &self.pool) .field("filter", &self.filter.is_set())
.field("disconnect_timeout", &self.disconnect_timeout) .field("disconnect_timeout", &self.disconnect_timeout)
.field("timeout", &self.timeout) .field("timeout", &self.timeout)
.field("error", &err) .field("error", &err)
@ -193,27 +197,21 @@ impl Io {
/// Create `Io` instance in specific memory pool. /// Create `Io` instance in specific memory pool.
pub fn with_memory_pool<I: IoStream>(io: I, pool: PoolRef) -> Self { pub fn with_memory_pool<I: IoStream>(io: I, pool: PoolRef) -> Self {
let inner = Rc::new(IoState { let inner = Rc::new(IoState {
filter: FilterPtr::null(),
pool: Cell::new(pool), pool: Cell::new(pool),
flags: Cell::new(Flags::empty()), flags: Cell::new(Flags::empty()),
error: Cell::new(None), error: Cell::new(None),
disconnect_timeout: Cell::new(Seconds(1)),
dispatch_task: LocalWaker::new(), dispatch_task: LocalWaker::new(),
read_task: LocalWaker::new(), read_task: LocalWaker::new(),
write_task: LocalWaker::new(), write_task: LocalWaker::new(),
buffer: Stack::new(), buffer: Stack::new(),
filter: Cell::new(NullFilter::get()),
handle: Cell::new(None), handle: Cell::new(None),
timeout: Cell::new(TimerHandle::default()), timeout: Cell::new(TimerHandle::default()),
disconnect_timeout: Cell::new(Seconds(1)),
on_disconnect: Cell::new(None), on_disconnect: Cell::new(None),
tag: Cell::new(DEFAULT_TAG), tag: Cell::new(DEFAULT_TAG),
}); });
inner.filter.update(Base::new(IoRef(inner.clone())));
let filter = Box::new(Base::new(IoRef(inner.clone())));
let filter_ref: &'static dyn Filter = unsafe {
let filter: &dyn Filter = filter.as_ref();
std::mem::transmute(filter)
};
inner.filter.replace(filter_ref);
let io_ref = IoRef(inner); let io_ref = IoRef(inner);
@ -221,7 +219,7 @@ impl Io {
let hnd = io.start(ReadContext::new(&io_ref), WriteContext::new(&io_ref)); let hnd = io.start(ReadContext::new(&io_ref), WriteContext::new(&io_ref));
io_ref.0.handle.set(hnd); io_ref.0.handle.set(hnd);
Io(io_ref, FilterItem::with_filter(filter)) Io(UnsafeCell::new(io_ref), marker::PhantomData)
} }
} }
@ -229,23 +227,28 @@ impl<F> Io<F> {
#[inline] #[inline]
/// Set memory pool /// Set memory pool
pub fn set_memory_pool(&self, pool: PoolRef) { pub fn set_memory_pool(&self, pool: PoolRef) {
self.0 .0.buffer.set_memory_pool(pool); self.st().buffer.set_memory_pool(pool);
self.0 .0.pool.set(pool); self.st().pool.set(pool);
} }
#[inline] #[inline]
/// Set io disconnect timeout in millis /// Set io disconnect timeout in millis
pub fn set_disconnect_timeout(&self, timeout: Seconds) { pub fn set_disconnect_timeout(&self, timeout: Seconds) {
self.0 .0.disconnect_timeout.set(timeout); self.st().disconnect_timeout.set(timeout);
} }
#[inline] #[inline]
/// Clone current io object. /// Clone current io object.
/// ///
/// Current io object becomes closed. /// Current io object becomes closed.
pub fn take(&mut self) -> Self { pub fn take(&self) -> Self {
Self(UnsafeCell::new(self.take_io_ref()), marker::PhantomData)
}
fn take_io_ref(&self) -> IoRef {
let inner = Rc::new(IoState { let inner = Rc::new(IoState {
pool: self.0 .0.pool.clone(), filter: FilterPtr::null(),
pool: self.st().pool.clone(),
flags: Cell::new( flags: Cell::new(
Flags::DSP_STOP Flags::DSP_STOP
| Flags::IO_STOPPED | Flags::IO_STOPPED
@ -258,16 +261,12 @@ impl<F> Io<F> {
read_task: LocalWaker::new(), read_task: LocalWaker::new(),
write_task: LocalWaker::new(), write_task: LocalWaker::new(),
buffer: Stack::new(), buffer: Stack::new(),
filter: Cell::new(NullFilter::get()),
handle: Cell::new(None), handle: Cell::new(None),
timeout: Cell::new(TimerHandle::default()), timeout: Cell::new(TimerHandle::default()),
on_disconnect: Cell::new(None), on_disconnect: Cell::new(None),
tag: Cell::new(DEFAULT_TAG), tag: Cell::new(DEFAULT_TAG),
}); });
unsafe { mem::replace(&mut *self.0.get(), IoRef(inner)) }
let state = mem::replace(&mut self.0, IoRef(inner));
let filter = mem::replace(&mut self.1, FilterItem::null());
Self(state, filter)
} }
} }
@ -276,18 +275,26 @@ impl<F> Io<F> {
#[doc(hidden)] #[doc(hidden)]
/// Get current state flags /// Get current state flags
pub fn flags(&self) -> Flags { pub fn flags(&self) -> Flags {
self.0 .0.flags.get() self.st().flags.get()
} }
#[inline] #[inline]
/// Get instance of `IoRef` /// Get instance of `IoRef`
pub fn get_ref(&self) -> IoRef { pub fn get_ref(&self) -> IoRef {
self.0.clone() self.io_ref().clone()
}
fn st(&self) -> &IoState {
unsafe { &(*self.0.get()).0 }
}
fn io_ref(&self) -> &IoRef {
unsafe { &*self.0.get() }
} }
/// Get current io error /// Get current io error
fn error(&self) -> Option<io::Error> { fn error(&self) -> Option<io::Error> {
self.0 .0.error.take() self.st().error.take()
} }
} }
@ -295,39 +302,42 @@ impl<F: FilterLayer, T: Filter> Io<Layer<F, T>> {
#[inline] #[inline]
/// Get referece to a filter /// Get referece to a filter
pub fn filter(&self) -> &F { pub fn filter(&self) -> &F {
&self.1.filter().0 &self.st().filter.filter::<Layer<F, T>>().0
} }
} }
impl<F: Filter> Io<F> { impl<F: Filter> Io<F> {
#[inline] #[inline]
/// Convert current io stream into sealed version /// Convert current io stream into sealed version
pub fn seal(mut self) -> Io<Sealed> { pub fn seal(self) -> Io<Sealed> {
let (filter, filter_ref) = self.1.seal(); let state = self.take_io_ref();
self.0 .0.filter.replace(filter_ref); state.0.filter.seal::<F>();
Io(self.0.clone(), filter)
Io(UnsafeCell::new(state), marker::PhantomData)
} }
#[inline] #[inline]
/// Map current filter with new one /// Map current filter with new one
pub fn add_filter<U>(mut self, nf: U) -> Io<Layer<U, F>> pub fn add_filter<U>(self, nf: U) -> Io<Layer<U, F>>
where where
U: FilterLayer, U: FilterLayer,
{ {
let state = self.take_io_ref();
// add layer to buffers // add layer to buffers
if U::BUFFERS { if U::BUFFERS {
// Safety: .add_layer() only increases internal buffers // Safety: .add_layer() only increases internal buffers
// there is no api that holds references into buffers storage // there is no api that holds references into buffers storage
// all apis first removes buffer from storage and then work with it // all apis first removes buffer from storage and then work with it
unsafe { &mut *(Rc::as_ptr(&self.0 .0) as *mut IoState) } unsafe { &mut *(Rc::as_ptr(&state.0) as *mut IoState) }
.buffer .buffer
.add_layer(); .add_layer();
} }
// replace current filter // replace current filter
let (filter, filter_ref) = self.1.add_filter(nf); state.0.filter.add_filter::<F, U>(nf);
self.0 .0.filter.replace(filter_ref);
Io(self.0.clone(), filter) Io(UnsafeCell::new(state), marker::PhantomData)
} }
} }
@ -381,9 +391,10 @@ impl<F> Io<F> {
#[inline] #[inline]
/// Pause read task /// Pause read task
pub fn pause(&self) { pub fn pause(&self) {
if !self.0.flags().contains(Flags::RD_PAUSED) { let st = self.st();
self.0 .0.read_task.wake(); if !st.flags.get().contains(Flags::RD_PAUSED) {
self.0 .0.insert_flags(Flags::RD_PAUSED); st.read_task.wake();
st.insert_flags(Flags::RD_PAUSED);
} }
} }
@ -435,18 +446,19 @@ impl<F> Io<F> {
/// `Poll::Ready(Ok(None))` if io stream is disconnected /// `Poll::Ready(Ok(None))` if io stream is disconnected
/// `Some(Poll::Ready(Err(e)))` if an error is encountered. /// `Some(Poll::Ready(Err(e)))` if an error is encountered.
pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Option<()>>> { pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Option<()>>> {
let mut flags = self.0 .0.flags.get(); let st = self.st();
let mut flags = st.flags.get();
if flags.contains(Flags::IO_STOPPED) { if flags.contains(Flags::IO_STOPPED) {
Poll::Ready(self.error().map(Err).unwrap_or(Ok(None))) Poll::Ready(self.error().map(Err).unwrap_or(Ok(None)))
} else { } else {
self.0 .0.dispatch_task.register(cx.waker()); st.dispatch_task.register(cx.waker());
let ready = flags.contains(Flags::RD_READY); let ready = flags.contains(Flags::RD_READY);
if flags.intersects(Flags::RD_BUF_FULL | Flags::RD_PAUSED) { if flags.intersects(Flags::RD_BUF_FULL | Flags::RD_PAUSED) {
flags.remove(Flags::RD_READY | Flags::RD_BUF_FULL | Flags::RD_PAUSED); flags.remove(Flags::RD_READY | Flags::RD_BUF_FULL | Flags::RD_PAUSED);
self.0 .0.read_task.wake(); st.read_task.wake();
self.0 .0.flags.set(flags); st.flags.set(flags);
if ready { if ready {
Poll::Ready(Ok(Some(()))) Poll::Ready(Ok(Some(())))
} else { } else {
@ -454,7 +466,7 @@ impl<F> Io<F> {
} }
} else if ready { } else if ready {
flags.remove(Flags::RD_READY); flags.remove(Flags::RD_READY);
self.0 .0.flags.set(flags); st.flags.set(flags);
Poll::Ready(Ok(Some(()))) Poll::Ready(Ok(Some(())))
} else { } else {
Poll::Pending Poll::Pending
@ -484,10 +496,11 @@ impl<F> Io<F> {
let ready = self.poll_read_ready(cx); let ready = self.poll_read_ready(cx);
if ready.is_pending() { if ready.is_pending() {
if self.0 .0.remove_flags(Flags::RD_FORCE_READY) { let st = self.st();
if st.remove_flags(Flags::RD_FORCE_READY) {
Poll::Ready(Ok(Some(()))) Poll::Ready(Ok(Some(())))
} else { } else {
self.0 .0.insert_flags(Flags::RD_FORCE_READY); st.insert_flags(Flags::RD_FORCE_READY);
Poll::Pending Poll::Pending
} }
} else { } else {
@ -538,14 +551,15 @@ impl<F> Io<F> {
if decoded.item.is_some() { if decoded.item.is_some() {
Ok(decoded) Ok(decoded)
} else { } else {
let flags = self.flags(); let st = self.st();
let flags = st.flags.get();
if flags.contains(Flags::IO_STOPPED) { if flags.contains(Flags::IO_STOPPED) {
Err(RecvError::PeerGone(self.error())) Err(RecvError::PeerGone(self.error()))
} else if flags.contains(Flags::DSP_STOP) { } else if flags.contains(Flags::DSP_STOP) {
self.0 .0.remove_flags(Flags::DSP_STOP); st.remove_flags(Flags::DSP_STOP);
Err(RecvError::Stop) Err(RecvError::Stop)
} else if flags.contains(Flags::DSP_TIMEOUT) { } else if flags.contains(Flags::DSP_TIMEOUT) {
self.0 .0.remove_flags(Flags::DSP_TIMEOUT); st.remove_flags(Flags::DSP_TIMEOUT);
Err(RecvError::KeepAlive) Err(RecvError::KeepAlive)
} else if flags.contains(Flags::WR_BACKPRESSURE) { } else if flags.contains(Flags::WR_BACKPRESSURE) {
Err(RecvError::WriteBackpressure) Err(RecvError::WriteBackpressure)
@ -579,20 +593,20 @@ impl<F> Io<F> {
if flags.contains(Flags::IO_STOPPED) { if flags.contains(Flags::IO_STOPPED) {
Poll::Ready(self.error().map(Err).unwrap_or(Ok(()))) Poll::Ready(self.error().map(Err).unwrap_or(Ok(())))
} else { } else {
let inner = &self.0 .0; let st = self.st();
let len = inner.buffer.write_destination_size(); let len = st.buffer.write_destination_size();
if len > 0 { if len > 0 {
if full { if full {
inner.insert_flags(Flags::WR_WAIT); st.insert_flags(Flags::WR_WAIT);
inner.dispatch_task.register(cx.waker()); st.dispatch_task.register(cx.waker());
return Poll::Pending; return Poll::Pending;
} else if len >= inner.pool.get().write_params_high() << 1 { } else if len >= st.pool.get().write_params_high() << 1 {
inner.insert_flags(Flags::WR_BACKPRESSURE); st.insert_flags(Flags::WR_BACKPRESSURE);
inner.dispatch_task.register(cx.waker()); st.dispatch_task.register(cx.waker());
return Poll::Pending; return Poll::Pending;
} }
} }
inner.remove_flags(Flags::WR_WAIT | Flags::WR_BACKPRESSURE); st.remove_flags(Flags::WR_WAIT | Flags::WR_BACKPRESSURE);
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} }
} }
@ -600,7 +614,8 @@ impl<F> Io<F> {
#[inline] #[inline]
/// Gracefully shutdown io stream /// Gracefully shutdown io stream
pub fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> { pub fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let flags = self.flags(); let st = self.st();
let flags = st.flags.get();
if flags.intersects(Flags::IO_STOPPED) { if flags.intersects(Flags::IO_STOPPED) {
if let Some(err) = self.error() { if let Some(err) = self.error() {
@ -610,12 +625,12 @@ impl<F> Io<F> {
} }
} else { } else {
if !flags.contains(Flags::IO_STOPPING_FILTERS) { if !flags.contains(Flags::IO_STOPPING_FILTERS) {
self.0 .0.init_shutdown(); st.init_shutdown();
} }
self.0 .0.read_task.wake(); st.read_task.wake();
self.0 .0.write_task.wake(); st.write_task.wake();
self.0 .0.dispatch_task.register(cx.waker()); st.dispatch_task.register(cx.waker());
Poll::Pending Poll::Pending
} }
} }
@ -628,7 +643,7 @@ impl<F> Io<F> {
self.pause(); self.pause();
let result = self.poll_status_update(cx); let result = self.poll_status_update(cx);
if !result.is_pending() { if !result.is_pending() {
self.0 .0.dispatch_task.register(cx.waker()); self.st().dispatch_task.register(cx.waker());
} }
result result
} }
@ -636,19 +651,20 @@ impl<F> Io<F> {
#[inline] #[inline]
/// Wait for status updates /// Wait for status updates
pub fn poll_status_update(&self, cx: &mut Context<'_>) -> Poll<IoStatusUpdate> { pub fn poll_status_update(&self, cx: &mut Context<'_>) -> Poll<IoStatusUpdate> {
let flags = self.flags(); let st = self.st();
let flags = st.flags.get();
if flags.intersects(Flags::IO_STOPPED | Flags::IO_STOPPING) { if flags.intersects(Flags::IO_STOPPED | Flags::IO_STOPPING) {
Poll::Ready(IoStatusUpdate::PeerGone(self.error())) Poll::Ready(IoStatusUpdate::PeerGone(self.error()))
} else if flags.contains(Flags::DSP_STOP) { } else if flags.contains(Flags::DSP_STOP) {
self.0 .0.remove_flags(Flags::DSP_STOP); st.remove_flags(Flags::DSP_STOP);
Poll::Ready(IoStatusUpdate::Stop) Poll::Ready(IoStatusUpdate::Stop)
} else if flags.contains(Flags::DSP_TIMEOUT) { } else if flags.contains(Flags::DSP_TIMEOUT) {
self.0 .0.remove_flags(Flags::DSP_TIMEOUT); st.remove_flags(Flags::DSP_TIMEOUT);
Poll::Ready(IoStatusUpdate::KeepAlive) Poll::Ready(IoStatusUpdate::KeepAlive)
} else if flags.contains(Flags::WR_BACKPRESSURE) { } else if flags.contains(Flags::WR_BACKPRESSURE) {
Poll::Ready(IoStatusUpdate::WriteBackpressure) Poll::Ready(IoStatusUpdate::WriteBackpressure)
} else { } else {
self.0 .0.dispatch_task.register(cx.waker()); st.dispatch_task.register(cx.waker());
Poll::Pending Poll::Pending
} }
} }
@ -656,14 +672,14 @@ impl<F> Io<F> {
#[inline] #[inline]
/// Register dispatch task /// Register dispatch task
pub fn poll_dispatch(&self, cx: &mut Context<'_>) { pub fn poll_dispatch(&self, cx: &mut Context<'_>) {
self.0 .0.dispatch_task.register(cx.waker()); self.st().dispatch_task.register(cx.waker());
} }
} }
impl<F> AsRef<IoRef> for Io<F> { impl<F> AsRef<IoRef> for Io<F> {
#[inline] #[inline]
fn as_ref(&self) -> &IoRef { fn as_ref(&self) -> &IoRef {
&self.0 self.io_ref()
} }
} }
@ -672,20 +688,20 @@ impl<F> Eq for Io<F> {}
impl<F> PartialEq for Io<F> { impl<F> PartialEq for Io<F> {
#[inline] #[inline]
fn eq(&self, other: &Self) -> bool { fn eq(&self, other: &Self) -> bool {
self.0.eq(&other.0) self.io_ref().eq(other.io_ref())
} }
} }
impl<F> hash::Hash for Io<F> { impl<F> hash::Hash for Io<F> {
#[inline] #[inline]
fn hash<H: hash::Hasher>(&self, state: &mut H) { fn hash<H: hash::Hasher>(&self, state: &mut H) {
self.0.hash(state); self.io_ref().hash(state);
} }
} }
impl<F> fmt::Debug for Io<F> { impl<F> fmt::Debug for Io<F> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Io").field("state", &self.0).finish() f.debug_struct("Io").field("state", self.st()).finish()
} }
} }
@ -694,28 +710,28 @@ impl<F> ops::Deref for Io<F> {
#[inline] #[inline]
fn deref(&self) -> &Self::Target { fn deref(&self) -> &Self::Target {
&self.0 self.io_ref()
} }
} }
impl<F> Drop for Io<F> { impl<F> Drop for Io<F> {
fn drop(&mut self) { fn drop(&mut self) {
let st = self.st();
self.stop_timer(); self.stop_timer();
// filter must be dropped, it is unsafe if st.filter.is_set() {
// and wont be dropped without special attention // filter is unsafe and must be dropped explicitly,
if self.1.is_set() { // and wont be dropped without special attention
if !self.0.flags().contains(Flags::IO_STOPPED) { if !st.flags.get().contains(Flags::IO_STOPPED) {
log::trace!( log::trace!(
"{}: Io is dropped, force stopping io streams {:?}", "{}: Io is dropped, force stopping io streams {:?}",
self.tag(), st.tag.get(),
self.0.flags() st.flags.get()
); );
} }
self.force_close(); self.force_close();
self.1.drop_filter(); st.filter.drop_filter::<F>();
self.0 .0.filter.set(NullFilter::get());
} }
} }
} }
@ -727,6 +743,7 @@ const KIND_UNMASK: u8 = !KIND_MASK;
const KIND_MASK_USIZE: usize = 0b11; const KIND_MASK_USIZE: usize = 0b11;
const KIND_UNMASK_USIZE: usize = !KIND_MASK_USIZE; const KIND_UNMASK_USIZE: usize = !KIND_MASK_USIZE;
const SEALED_SIZE: usize = mem::size_of::<Sealed>(); const SEALED_SIZE: usize = mem::size_of::<Sealed>();
const NULL: [u8; SEALED_SIZE] = [0u8; SEALED_SIZE];
#[cfg(target_endian = "little")] #[cfg(target_endian = "little")]
const KIND_IDX: usize = 0; const KIND_IDX: usize = 0;
@ -734,37 +751,45 @@ const KIND_IDX: usize = 0;
#[cfg(target_endian = "big")] #[cfg(target_endian = "big")]
const KIND_IDX: usize = SEALED_SIZE - 1; const KIND_IDX: usize = SEALED_SIZE - 1;
struct FilterItem<F> { struct FilterPtr {
data: [u8; SEALED_SIZE], data: Cell<[u8; SEALED_SIZE]>,
_t: marker::PhantomData<F>, filter: Cell<&'static dyn Filter>,
} }
impl<F> FilterItem<F> { impl FilterPtr {
fn null() -> Self { const fn null() -> Self {
Self { Self {
data: [0; SEALED_SIZE], data: Cell::new(NULL),
_t: marker::PhantomData, filter: Cell::new(NullFilter::get()),
} }
} }
fn with_filter(f: Box<F>) -> Self { fn update<F: Filter>(&self, filter: F) {
let mut slf = Self { if self.is_set() {
data: [0; SEALED_SIZE], panic!("Filter is set, must be dropped first");
_t: marker::PhantomData, }
};
let filter = Box::new(filter);
unsafe { let mut data = NULL;
let ptr = &mut slf.data as *mut _ as *mut *mut F; unsafe {
ptr.write(Box::into_raw(f)); let filter_ref: &'static dyn Filter = {
slf.data[KIND_IDX] |= KIND_PTR; let f: &dyn Filter = filter.as_ref();
std::mem::transmute(f)
};
self.filter.set(filter_ref);
let ptr = &mut data as *mut _ as *mut *mut F;
ptr.write(Box::into_raw(filter));
data[KIND_IDX] |= KIND_PTR;
self.data.set(data);
} }
slf
} }
/// Get filter, panic if it is not filter /// Get filter, panic if it is not filter
fn filter(&self) -> &F { fn filter<F: Filter>(&self) -> &F {
if self.data[KIND_IDX] & KIND_PTR != 0 { let data = self.data.get();
let ptr = &self.data as *const _ as *const *mut F; if data[KIND_IDX] & KIND_PTR != 0 {
let ptr = &data as *const _ as *const *mut F;
unsafe { unsafe {
let p = (ptr.read() as *const _ as usize) & KIND_UNMASK_USIZE; let p = (ptr.read() as *const _ as usize) & KIND_UNMASK_USIZE;
(p as *const F as *mut F).as_ref().unwrap() (p as *const F as *mut F).as_ref().unwrap()
@ -775,87 +800,99 @@ impl<F> FilterItem<F> {
} }
/// Get filter, panic if it is not set /// Get filter, panic if it is not set
fn take_filter(&mut self) -> Box<F> { fn take_filter<F>(&self) -> Box<F> {
if self.data[KIND_IDX] & KIND_PTR != 0 { let mut data = self.data.get();
self.data[KIND_IDX] &= KIND_UNMASK; if data[KIND_IDX] & KIND_PTR != 0 {
let ptr = &mut self.data as *mut _ as *mut *mut F; data[KIND_IDX] &= KIND_UNMASK;
let ptr = &mut data as *mut _ as *mut *mut F;
unsafe { Box::from_raw(*ptr) } unsafe { Box::from_raw(*ptr) }
} else { } else {
panic!( panic!(
"Wrong filter item {:?} expected: {:?}", "Wrong filter item {:?} expected: {:?}",
self.data[KIND_IDX], KIND_PTR data[KIND_IDX], KIND_PTR
); );
} }
} }
/// Get sealed, panic if it is already sealed /// Get sealed, panic if it is already sealed
fn take_sealed(&mut self) -> Sealed { fn take_sealed(&self) -> Sealed {
if self.data[KIND_IDX] & KIND_SEALED != 0 { let mut data = self.data.get();
self.data[KIND_IDX] &= KIND_UNMASK;
let ptr = &mut self.data as *mut _ as *mut Sealed; if data[KIND_IDX] & KIND_SEALED != 0 {
data[KIND_IDX] &= KIND_UNMASK;
let ptr = &mut data as *mut _ as *mut Sealed;
unsafe { ptr.read() } unsafe { ptr.read() }
} else { } else {
panic!( panic!(
"Wrong filter item {:?} expected: {:?}", "Wrong filter item {:?} expected: {:?}",
self.data[KIND_IDX], KIND_SEALED data[KIND_IDX], KIND_SEALED
); );
} }
} }
fn is_set(&self) -> bool { fn is_set(&self) -> bool {
self.data[KIND_IDX] & KIND_MASK != 0 self.data.get()[KIND_IDX] & KIND_MASK != 0
} }
fn drop_filter(&mut self) { fn drop_filter<F>(&self) {
if self.data[KIND_IDX] & KIND_PTR != 0 { let data = self.data.get();
self.take_filter();
} else if self.data[KIND_IDX] & KIND_SEALED != 0 { if data[KIND_IDX] & KIND_MASK != 0 {
self.take_sealed(); if data[KIND_IDX] & KIND_PTR != 0 {
self.take_filter::<F>();
} else if data[KIND_IDX] & KIND_SEALED != 0 {
self.take_sealed();
}
self.data.set(NULL);
self.filter.set(NullFilter::get());
} }
} }
} }
impl<F: Filter> FilterItem<F> { impl FilterPtr {
fn add_filter<T: FilterLayer>( fn add_filter<F: Filter, T: FilterLayer>(&self, new: T) {
&mut self, let mut data = NULL;
new: T, let filter = Box::new(Layer::new(new, *self.take_filter::<F>()));
) -> (FilterItem<Layer<T, F>>, &'static dyn Filter) { unsafe {
let filter = Box::new(Layer::new(new, *self.take_filter())); let filter_ref: &'static dyn Filter = {
let filter_ref: &'static dyn Filter = { let f: &dyn Filter = filter.as_ref();
let filter: &dyn Filter = filter.as_ref(); std::mem::transmute(f)
unsafe { std::mem::transmute(filter) } };
}; self.filter.set(filter_ref);
(FilterItem::with_filter(filter), filter_ref)
let ptr = &mut data as *mut _ as *mut *mut Layer<T, F>;
ptr.write(Box::into_raw(filter));
data[KIND_IDX] |= KIND_PTR;
self.data.set(data);
}
} }
fn seal(&mut self) -> (FilterItem<Sealed>, &'static dyn Filter) { fn seal<F: Filter>(&self) {
let filter = if self.data[KIND_IDX] & KIND_PTR != 0 { let mut data = self.data.get();
Sealed(Box::new(*self.take_filter()))
} else if self.data[KIND_IDX] & KIND_SEALED != 0 { let filter = if data[KIND_IDX] & KIND_PTR != 0 {
Sealed(Box::new(*self.take_filter::<F>()))
} else if data[KIND_IDX] & KIND_SEALED != 0 {
self.take_sealed() self.take_sealed()
} else { } else {
panic!( panic!(
"Wrong filter item {:?} expected: {:?}", "Wrong filter item {:?} expected: {:?}",
self.data[KIND_IDX], KIND_PTR data[KIND_IDX], KIND_PTR
); );
}; };
let filter_ref: &'static dyn Filter = {
let filter: &dyn Filter = filter.0.as_ref();
unsafe { std::mem::transmute(filter) }
};
let mut slf = FilterItem {
data: [0; SEALED_SIZE],
_t: marker::PhantomData,
};
unsafe { unsafe {
let ptr = &mut slf.data as *mut _ as *mut Sealed; let filter_ref: &'static dyn Filter = {
let f: &dyn Filter = filter.0.as_ref();
std::mem::transmute(f)
};
self.filter.set(filter_ref);
let ptr = &mut data as *mut _ as *mut Sealed;
ptr.write(filter); ptr.write(filter);
slf.data[KIND_IDX] |= KIND_SEALED; data[KIND_IDX] |= KIND_SEALED;
self.data.set(data);
} }
(slf, filter_ref)
} }
} }
@ -943,7 +980,7 @@ mod tests {
let server = Io::new(server); let server = Io::new(server);
assert!(server.eq(&server)); assert!(server.eq(&server));
assert!(server.0.eq(&server.0)); assert!(server.io_ref().eq(server.io_ref()));
assert!(format!("{:?}", Flags::IO_STOPPED).contains("IO_STOPPED")); assert!(format!("{:?}", Flags::IO_STOPPED).contains("IO_STOPPED"));
assert!(Flags::IO_STOPPED == Flags::IO_STOPPED); assert!(Flags::IO_STOPPED == Flags::IO_STOPPED);
@ -957,16 +994,16 @@ mod tests {
let server = Io::new(server); let server = Io::new(server);
server.0 .0.notify_timeout(); server.st().notify_timeout();
let err = server.recv(&BytesCodec).await.err().unwrap(); let err = server.recv(&BytesCodec).await.err().unwrap();
assert!(format!("{:?}", err).contains("Timeout")); assert!(format!("{:?}", err).contains("Timeout"));
server.0 .0.insert_flags(Flags::DSP_STOP); server.st().insert_flags(Flags::DSP_STOP);
let err = server.recv(&BytesCodec).await.err().unwrap(); let err = server.recv(&BytesCodec).await.err().unwrap();
assert!(format!("{:?}", err).contains("Dispatcher stopped")); assert!(format!("{:?}", err).contains("Dispatcher stopped"));
client.write(TEXT); client.write(TEXT);
server.0 .0.insert_flags(Flags::WR_BACKPRESSURE); server.st().insert_flags(Flags::WR_BACKPRESSURE);
let item = server.recv(&BytesCodec).await.ok().unwrap().unwrap(); let item = server.recv(&BytesCodec).await.ok().unwrap().unwrap();
assert_eq!(item, TEXT); assert_eq!(item, TEXT);
} }
@ -1016,7 +1053,7 @@ mod tests {
let (client, server) = IoTest::create(); let (client, server) = IoTest::create();
let f = DropFilter { p: p.clone() }; let f = DropFilter { p: p.clone() };
let _ = format!("{:?}", f); let _ = format!("{:?}", f);
let mut io = Io::new(server).add_filter(f); let io = Io::new(server).add_filter(f);
client.remote_buffer_cap(1024); client.remote_buffer_cap(1024);
client.write(TEXT); client.write(TEXT);

14
ntex-io/src/ioref.rs
View file

@ -21,9 +21,9 @@ impl IoRef {
} }
#[inline] #[inline]
/// Get memory pool /// Get current filter
pub(crate) fn filter(&self) -> &dyn Filter { pub(crate) fn filter(&self) -> &dyn Filter {
self.0.filter.get() self.0.filter()
} }
#[inline] #[inline]
@ -192,10 +192,7 @@ impl IoRef {
F: FnOnce(&WriteBuf<'_>) -> R, F: FnOnce(&WriteBuf<'_>) -> R,
{ {
let result = self.0.buffer.write_buf(self, 0, f); let result = self.0.buffer.write_buf(self, 0, f);
self.0 self.0.filter().process_write_buf(self, &self.0.buffer, 0)?;
.filter
.get()
.process_write_buf(self, &self.0.buffer, 0)?;
Ok(result) Ok(result)
} }
@ -206,10 +203,7 @@ impl IoRef {
F: FnOnce(&mut BytesVec) -> R, F: FnOnce(&mut BytesVec) -> R,
{ {
let result = self.0.buffer.with_write_source(self, f); let result = self.0.buffer.with_write_source(self, f);
self.0 self.0.filter().process_write_buf(self, &self.0.buffer, 0)?;
.filter
.get()
.process_write_buf(self, &self.0.buffer, 0)?;
Ok(result) Ok(result)
} }

3
ntex-tls/src/openssl/mod.rs
View file

@ -167,11 +167,12 @@ impl FilterLayer for SslFilter {
Ok(new_bytes) Ok(new_bytes)
} }
Err(ref e) if e.code() == ssl::ErrorCode::ZERO_RETURN => { Err(ref e) if e.code() == ssl::ErrorCode::ZERO_RETURN => {
log::debug!("{}: SSL Error: {:?}", buf.tag(), e);
buf.want_shutdown(); buf.want_shutdown();
Ok(new_bytes) Ok(new_bytes)
} }
Err(e) => { Err(e) => {
log::trace!("SSL Error: {:?}", e); log::trace!("{}: SSL Error: {:?}", buf.tag(), e);
Err(map_to_ioerr(e)) Err(map_to_ioerr(e))
} }
}; };