actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2025-01-19 00:31:50 +01:00
Code Issues Releases Wiki Activity

optimize InOrder service

Browse Source
This commit is contained in:
Nikolay Kim 2019-12-11 18:55:53 +06:00
parent c7a8743bf9
commit 8bb81c0768
3 changed files with 5 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
actix-utils/CHANGES.md
View File

@ -1,5 +1,9 @@
# Changes # Changes
## [1.0.1] - 2019-12-11
* Optimize InOrder service
## [1.0.0] - 2019-12-11 ## [1.0.0] - 2019-12-11
* Simplify oneshot and mpsc implementations * Simplify oneshot and mpsc implementations

2
actix-utils/Cargo.toml
View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-utils" name = "actix-utils"
version = "1.0.0" version = "1.0.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix utils - various actix net related services" description = "Actix utils - various actix net related services"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]

9
actix-utils/src/order.rs
View File

@ -4,14 +4,12 @@ use std::fmt;
use std::future::Future; use std::future::Future;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::pin::Pin; use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{IntoService, Service, Transform}; use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready}; use futures::future::{ok, Ready};
use crate::oneshot; use crate::oneshot;
use crate::task::LocalWaker;
struct Record<I, E> { struct Record<I, E> {
rx: oneshot::Receiver<Result<I, E>>, rx: oneshot::Receiver<Result<I, E>>,
@ -105,7 +103,6 @@ where
pub struct InOrderService<S: Service> { pub struct InOrderService<S: Service> {
service: S, service: S,
task: Rc<LocalWaker>,
acks: VecDeque<Record<S::Response, S::Error>>, acks: VecDeque<Record<S::Response, S::Error>>,
} }
@ -123,7 +120,6 @@ where
Self { Self {
service: service.into_service(), service: service.into_service(),
acks: VecDeque::new(), acks: VecDeque::new(),
task: Rc::new(LocalWaker::new()),
} }
} }
} }
@ -141,9 +137,6 @@ where
type Future = InOrderServiceResponse<S>; type Future = InOrderServiceResponse<S>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
// poll_ready could be called from different task
self.task.register(cx.waker());
// check acks // check acks
while !self.acks.is_empty() { while !self.acks.is_empty() {
let rec = self.acks.front_mut().unwrap(); let rec = self.acks.front_mut().unwrap();
@ -172,11 +165,9 @@ where
let (tx2, rx2) = oneshot::channel(); let (tx2, rx2) = oneshot::channel();
self.acks.push_back(Record { rx: rx1, tx: tx2 }); self.acks.push_back(Record { rx: rx1, tx: tx2 });
let task = self.task.clone();
let fut = self.service.call(request); let fut = self.service.call(request);
actix_rt::spawn(async move { actix_rt::spawn(async move {
let res = fut.await; let res = fut.await;
task.wake();
let _ = tx1.send(res); let _ = tx1.send(res);
}); });