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

Optimize readiness handling (#457)

Browse source
This commit is contained in:
Nikolay Kim 2024-11-04 18:18:15 +05:00 • committed by GitHub
parent c26b336fe5
commit ddbc4a722d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 46 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

6
ntex-service/CHANGES.md
View file

@ -1,6 +1,10 @@
# Changes
## [3.3.0] - 2024-11-02
## [3.3.1] - 2024-11-04
* Optimize readiness handling
## [3.3.0] - 2024-11-04
* Added Service::not_ready() method

2
ntex-service/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex-service"
version = "3.3.0"
version = "3.3.1"
authors = ["ntex contributors <team@ntex.rs>"]
description = "ntex service"
keywords = ["network", "framework", "async", "futures"]

21
ntex-service/src/ctx.rs
View file

@ -56,8 +56,14 @@ impl WaitersRef {
}
pub(crate) fn register(&self, idx: u32, cx: &mut Context<'_>) {
let wakers = self.get_wakers();
if let Some(last) = wakers.last() {
if idx == *last {
return;
}
}
wakers.push(idx);
self.get()[idx as usize] = Some(cx.waker().clone());
self.get_wakers().push(idx);
}
pub(crate) fn register_unready(&self, cx: &mut Context<'_>) {
@ -66,13 +72,14 @@ impl WaitersRef {
}
pub(crate) fn notify(&self) {
let indexes = self.get();
let wakers = self.get_wakers();
for idx in wakers.drain(..) {
if let Some(item) = indexes.get_mut(idx as usize) {
if let Some(waker) = item.take() {
waker.wake();
if !wakers.is_empty() {
let indexes = self.get();
for idx in wakers.drain(..) {
if let Some(item) = indexes.get_mut(idx as usize) {
if let Some(waker) = item.take() {
waker.wake();
}
}
}
}

54
ntex-service/src/pipeline.rs
View file

@ -420,21 +420,20 @@ where
let mut slf = self.as_mut();
if slf.pl.state.waiters.can_check(slf.pl.index, cx) {
if let Some(ref mut fut) = slf.fut {
match unsafe { Pin::new_unchecked(fut) }.poll(cx) {
Poll::Pending => {
slf.pl.state.waiters.register(slf.pl.index, cx);
Poll::Pending
}
Poll::Ready(res) => {
let _ = slf.fut.take();
slf.pl.state.waiters.notify();
Poll::Ready(res)
}
}
} else {
if slf.fut.is_none() {
slf.fut = Some((slf.f)(slf.pl));
self.poll(cx)
}
let fut = slf.fut.as_mut().unwrap();
match unsafe { Pin::new_unchecked(fut) }.poll(cx) {
Poll::Pending => {
slf.pl.state.waiters.register(slf.pl.index, cx);
Poll::Pending
}
Poll::Ready(res) => {
let _ = slf.fut.take();
slf.pl.state.waiters.notify();
Poll::Ready(res)
}
}
} else {
Poll::Pending
@ -460,21 +459,20 @@ where
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
let mut slf = self.as_mut();
if let Some(ref mut fut) = slf.fut {
match unsafe { Pin::new_unchecked(fut) }.poll(cx) {
Poll::Pending => {
slf.pl.state.waiters.register_unready(cx);
Poll::Pending
}
Poll::Ready(res) => {
let _ = slf.fut.take();
slf.pl.state.waiters.notify();
Poll::Ready(res)
}
}
} else {
if slf.fut.is_none() {
slf.fut = Some((slf.f)(slf.pl));
self.poll(cx)
}
let fut = slf.fut.as_mut().unwrap();
match unsafe { Pin::new_unchecked(fut) }.poll(cx) {
Poll::Pending => {
slf.pl.state.waiters.register_unready(cx);
Poll::Pending
}
Poll::Ready(res) => {
let _ = slf.fut.take();
slf.pl.state.waiters.notify();
Poll::Ready(res)
}
}
}
}