actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2024-11-24 00:01:11 +01:00
Code Issues Releases Wiki Activity

Replace calls to Pin::new_unchecked with pin_project.

Browse Source 
This is a breaking change, as it changes some public methods to take
`Pin<&mut Self>` rather than `&mut self`.

This brings these methods into line with `Stream::poll_next`, which also
takes a `Pin<&mut Self>`
This commit is contained in:
Aaron Hill 2020-03-04 11:48:19 -05:00
parent 693d5132a9
commit c41b5d8dd4
No known key found for this signature in database
GPG Key ID: B4087E510E98B164
7 changed files with 116 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
actix-codec/Cargo.toml
View File

@ -24,3 +24,4 @@ futures-sink = "0.3.1"
tokio = { version = "0.2.4", default-features=false }
tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
log = "0.4"
pin-project = "0.4.8"

75
actix-codec/src/framed.rs
View File

@ -5,6 +5,7 @@ use std::{fmt, io};
use bytes::{Buf, BytesMut};
use futures_core::{ready, Stream};
use futures_sink::Sink;
use pin_project::pin_project;
use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
@ -20,7 +21,9 @@ bitflags::bitflags! {
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
/// the `Encoder` and `Decoder` traits to encode and decode frames.
#[pin_project]
pub struct Framed<T, U> {
#[pin]
io: T,
codec: U,
flags: Flags,
@ -28,8 +31,6 @@ pub struct Framed<T, U> {
write_buf: BytesMut,
}
impl<T, U> Unpin for Framed<T, U> {}
impl<T, U> Framed<T, U>
where
T: AsyncRead + AsyncWrite,
@ -185,17 +186,18 @@ impl<T, U> Framed<T, U> {
impl<T, U> Framed<T, U> {
/// Serialize item and Write to the inner buffer
pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
pub fn write(mut self: Pin<&mut Self>, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
where
T: AsyncWrite,
U: Encoder,
{
let remaining = self.write_buf.capacity() - self.write_buf.len();
let this = self.as_mut().project();
let remaining = this.write_buf.capacity() - this.write_buf.len();
if remaining < LW {
self.write_buf.reserve(HW - remaining);
this.write_buf.reserve(HW - remaining);
}
self.codec.encode(item, &mut self.write_buf)?;
this.codec.encode(item, this.write_buf)?;
Ok(())
}
@ -207,21 +209,22 @@ impl<T, U> Framed<T, U> {
}
/// Try to read underlying I/O stream and decode item.
pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
where
T: AsyncRead,
U: Decoder,
{
loop {
let mut this = self.as_mut().project();
// Repeatedly call `decode` or `decode_eof` as long as it is
// "readable". Readable is defined as not having returned `None`. If
// the upstream has returned EOF, and the decoder is no longer
// readable, it can be assumed that the decoder will never become
// readable again, at which point the stream is terminated.
if self.flags.contains(Flags::READABLE) {
if self.flags.contains(Flags::EOF) {
match self.codec.decode_eof(&mut self.read_buf) {
if this.flags.contains(Flags::READABLE) {
if this.flags.contains(Flags::EOF) {
match this.codec.decode_eof(&mut this.read_buf) {
Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
Ok(None) => return Poll::Ready(None),
Err(e) => return Poll::Ready(Some(Err(e))),
@ -230,7 +233,7 @@ impl<T, U> Framed<T, U> {
log::trace!("attempting to decode a frame");
match self.codec.decode(&mut self.read_buf) {
match this.codec.decode(&mut this.read_buf) {
Ok(Some(frame)) => {
log::trace!("frame decoded from buffer");
return Poll::Ready(Some(Ok(frame)));
@ -239,45 +242,44 @@ impl<T, U> Framed<T, U> {
_ => (), // Need more data
}
self.flags.remove(Flags::READABLE);
this.flags.remove(Flags::READABLE);
}
debug_assert!(!self.flags.contains(Flags::EOF));
debug_assert!(!this.flags.contains(Flags::EOF));
// Otherwise, try to read more data and try again. Make sure we've got room
let remaining = self.read_buf.capacity() - self.read_buf.len();
let remaining = this.read_buf.capacity() - this.read_buf.len();
if remaining < LW {
self.read_buf.reserve(HW - remaining)
this.read_buf.reserve(HW - remaining)
}
let cnt = match unsafe {
Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf)
} {
let cnt = match this.io.poll_read_buf(cx, &mut this.read_buf) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
Poll::Ready(Ok(cnt)) => cnt,
};
if cnt == 0 {
self.flags.insert(Flags::EOF);
this.flags.insert(Flags::EOF);
}
self.flags.insert(Flags::READABLE);
this.flags.insert(Flags::READABLE);
}
}
/// Flush write buffer to underlying I/O stream.
pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite,
U: Encoder,
{
let mut this = self.as_mut().project();
log::trace!("flushing framed transport");
while !self.write_buf.is_empty() {
log::trace!("writing; remaining={}", self.write_buf.len());
while !this.write_buf.is_empty() {
log::trace!("writing; remaining={}", this.write_buf.len());
let n = ready!(unsafe {
Pin::new_unchecked(&mut self.io).poll_write(cx, &self.write_buf)
})?;
let n = ready!(
this.io.as_mut().poll_write(cx, this.write_buf)
)?;
if n == 0 {
return Poll::Ready(Err(io::Error::new(
@ -288,26 +290,25 @@ impl<T, U> Framed<T, U> {
}
// remove written data
self.write_buf.advance(n);
this.write_buf.advance(n);
}
// Try flushing the underlying IO
ready!(unsafe { Pin::new_unchecked(&mut self.io).poll_flush(cx) })?;
ready!(this.io.poll_flush(cx))?;
log::trace!("framed transport flushed");
Poll::Ready(Ok(()))
}
/// Flush write buffer and shutdown underlying I/O stream.
pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where
T: AsyncWrite,
U: Encoder,
{
unsafe {
ready!(Pin::new_unchecked(&mut self.io).poll_flush(cx))?;
ready!(Pin::new_unchecked(&mut self.io).poll_shutdown(cx))?;
}
let mut this = self.as_mut().project();
ready!(this.io.as_mut().poll_flush(cx))?;
ready!(this.io.as_mut().poll_shutdown(cx))?;
Poll::Ready(Ok(()))
}
}
@ -319,7 +320,7 @@ where
{
type Item = Result<U::Item, U::Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.next_item(cx)
}
}
@ -341,21 +342,21 @@ where
}
fn start_send(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
item: <U as Encoder>::Item,
) -> Result<(), Self::Error> {
self.write(item)
}
fn poll_flush(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> {
self.flush(cx)
}
fn poll_close(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> {
self.close(cx)

13
actix-ioframe/src/connect.rs
View File

@ -42,6 +42,7 @@ where
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
pub(crate) state: St,
pub(crate) out: Option<Out>,
#[pin]
pub(crate) framed: Framed<Io, Codec>,
}
@ -97,8 +98,8 @@ where
{
type Error = <Codec as Encoder>::Error;
fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.framed.is_write_ready() {
fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.as_mut().project().framed.is_write_ready() {
Poll::Ready(Ok(()))
} else {
Poll::Pending
@ -112,11 +113,11 @@ where
self.project().framed.write(item)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.get_mut().framed.flush(cx)
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.as_mut().project().framed.flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.get_mut().framed.close(cx)
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.as_mut().project().framed.close(cx)
}
}

76
actix-ioframe/src/dispatcher.rs
View File

@ -6,6 +6,7 @@ use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::Service;
use actix_utils::mpsc;
use futures::Stream;
use pin_project::pin_project;
use log::debug;
use crate::error::ServiceError;
@ -15,6 +16,7 @@ type Response<U> = <U as Encoder>::Item;
/// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service.
#[pin_project]
pub(crate) struct Dispatcher<S, T, U, Out>
where
S: Service<Request = Request<U>, Response = Option<Response<U>>>,
@ -29,6 +31,7 @@ where
service: S,
sink: Option<Out>,
state: FramedState<S, U>,
#[pin]
framed: Framed<T, U>,
rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
}
@ -90,26 +93,27 @@ where
<U as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <U as Encoder>::Item> + Unpin,
{
fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
loop {
match self.service.poll_ready(cx) {
let this = self.as_mut().project();
match this.service.poll_ready(cx) {
Poll::Ready(Ok(_)) => {
let item = match self.framed.next_item(cx) {
let item = match this.framed.next_item(cx) {
Poll::Ready(Some(Ok(el))) => el,
Poll::Ready(Some(Err(err))) => {
self.state = FramedState::FramedError(ServiceError::Decoder(err));
*this.state = FramedState::FramedError(ServiceError::Decoder(err));
return true;
}
Poll::Pending => return false,
Poll::Ready(None) => {
log::trace!("Client disconnected");
self.state = FramedState::Stopping;
*this.state = FramedState::Stopping;
return true;
}
};
let tx = self.rx.sender();
let fut = self.service.call(item);
let tx = this.rx.sender();
let fut = this.service.call(item);
actix_rt::spawn(async move {
let item = fut.await;
let item = match item {
@ -122,7 +126,7 @@ where
}
Poll::Pending => return false,
Poll::Ready(Err(err)) => {
self.state = FramedState::Error(ServiceError::Service(err));
*this.state = FramedState::Error(ServiceError::Service(err));
return true;
}
}
@ -130,37 +134,38 @@ where
}
/// write to framed object
fn poll_write(&mut self, cx: &mut Context<'_>) -> bool {
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
loop {
while !self.framed.is_write_buf_full() {
match Pin::new(&mut self.rx).poll_next(cx) {
let mut this = self.as_mut().project();
while !this.framed.is_write_buf_full() {
match Pin::new(&mut this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => {
if let Err(err) = self.framed.write(msg) {
self.state = FramedState::FramedError(ServiceError::Encoder(err));
if let Err(err) = this.framed.as_mut().write(msg) {
*this.state = FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
continue;
}
Poll::Ready(Some(Err(err))) => {
self.state = FramedState::Error(ServiceError::Service(err));
*this.state = FramedState::Error(ServiceError::Service(err));
return true;
}
Poll::Ready(None) | Poll::Pending => (),
}
if self.sink.is_some() {
match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) {
if this.sink.is_some() {
match Pin::new(this.sink.as_mut().unwrap()).poll_next(cx) {
Poll::Ready(Some(msg)) => {
if let Err(err) = self.framed.write(msg) {
self.state =
if let Err(err) = this.framed.as_mut().write(msg) {
*this.state =
FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
continue;
}
Poll::Ready(None) => {
let _ = self.sink.take();
self.state = FramedState::FlushAndStop;
let _ = this.sink.take();
*this.state = FramedState::FlushAndStop;
return true;
}
Poll::Pending => (),
@ -169,13 +174,13 @@ where
break;
}
if !self.framed.is_write_buf_empty() {
match self.framed.flush(cx) {
if !this.framed.is_write_buf_empty() {
match this.framed.as_mut().flush(cx) {
Poll::Pending => break,
Poll::Ready(Ok(_)) => (),
Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err);
self.state = FramedState::FramedError(ServiceError::Encoder(err));
*this.state = FramedState::FramedError(ServiceError::Encoder(err));
return true;
}
}
@ -187,13 +192,14 @@ where
}
pub(crate) fn poll(
&mut self,
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), ServiceError<S::Error, U>>> {
match self.state {
let mut this = self.as_mut().project();
match this.state {
FramedState::Processing => loop {
let read = self.poll_read(cx);
let write = self.poll_write(cx);
let read = self.as_mut().poll_read(cx);
let write = self.as_mut().poll_write(cx);
if read || write {
continue;
} else {
@ -202,18 +208,18 @@ where
},
FramedState::Error(_) => {
// flush write buffer
if !self.framed.is_write_buf_empty() {
if let Poll::Pending = self.framed.flush(cx) {
if !this.framed.is_write_buf_empty() {
if let Poll::Pending = this.framed.flush(cx) {
return Poll::Pending;
}
}
Poll::Ready(Err(self.state.take_error()))
Poll::Ready(Err(this.state.take_error()))
}
FramedState::FlushAndStop => {
// drain service responses
match Pin::new(&mut self.rx).poll_next(cx) {
match Pin::new(this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => {
if self.framed.write(msg).is_err() {
if this.framed.as_mut().write(msg).is_err() {
return Poll::Ready(Ok(()));
}
}
@ -222,8 +228,8 @@ where
}
// flush io
if !self.framed.is_write_buf_empty() {
match self.framed.flush(cx) {
if !this.framed.is_write_buf_empty() {
match this.framed.flush(cx) {
Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err);
}
@ -235,7 +241,7 @@ where
};
Poll::Ready(Ok(()))
}
FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())),
FramedState::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
FramedState::Stopping => Poll::Ready(Ok(())),
}
}

4
actix-ioframe/src/service.rs
View File

@ -357,7 +357,7 @@ where
{
Connect(#[pin] C::Future, Rc<T>),
Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
Dispatcher(Dispatcher<T::Service, Io, Codec, Out>),
Dispatcher(#[pin] Dispatcher<T::Service, Io, Codec, Out>),
}
impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
@ -408,7 +408,7 @@ where
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
}
}
FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
FramedServiceImplResponseInner::Dispatcher(fut) => {
Either::Right(fut.poll(cx))
}
}

2
actix-rt/src/arbiter.rs
View File

@ -181,7 +181,7 @@ impl Arbiter {
// because the executor boxes the future again, but works for now
Q.with(move |cell| {
cell.borrow_mut()
.push(unsafe { Pin::new_unchecked(Box::alloc().init(future)) })
.push(Pin::from(Box::alloc().init(future)))
});
}
});

47
actix-utils/src/framed.rs
View File

@ -77,6 +77,7 @@ where
{
service: S,
state: State<S, U>,
#[pin]
framed: Framed<T, U>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
@ -169,7 +170,7 @@ where
&mut self.framed
}
fn poll_read(&mut self, cx: &mut Context<'_>) -> bool
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
where
S: Service<Request = Request<U>, Response = Response<U>>,
S::Error: 'static,
@ -180,29 +181,30 @@ where
<U as Encoder>::Error: std::fmt::Debug,
{
loop {
match self.service.poll_ready(cx) {
let this = self.as_mut().project();
match this.service.poll_ready(cx) {
Poll::Ready(Ok(_)) => {
let item = match self.framed.next_item(cx) {
let item = match this.framed.next_item(cx) {
Poll::Ready(Some(Ok(el))) => el,
Poll::Ready(Some(Err(err))) => {
self.state = State::FramedError(DispatcherError::Decoder(err));
*this.state = State::FramedError(DispatcherError::Decoder(err));
return true;
}
Poll::Pending => return false,
Poll::Ready(None) => {
self.state = State::Stopping;
*this.state = State::Stopping;
return true;
}
};
let tx = self.tx.clone();
actix_rt::spawn(self.service.call(item).map(move |item| {
let tx = this.tx.clone();
actix_rt::spawn(this.service.call(item).map(move |item| {
let _ = tx.send(item.map(Message::Item));
}));
}
Poll::Pending => return false,
Poll::Ready(Err(err)) => {
self.state = State::Error(DispatcherError::Service(err));
*this.state = State::Error(DispatcherError::Service(err));
return true;
}
}
@ -210,7 +212,7 @@ where
}
/// write to framed object
fn poll_write(&mut self, cx: &mut Context<'_>) -> bool
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
where
S: Service<Request = Request<U>, Response = Response<U>>,
S::Error: 'static,
@ -221,33 +223,34 @@ where
<U as Encoder>::Error: std::fmt::Debug,
{
loop {
while !self.framed.is_write_buf_full() {
match Pin::new(&mut self.rx).poll_next(cx) {
let mut this = self.as_mut().project();
while !this.framed.is_write_buf_full() {
match Pin::new(&mut this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(Message::Item(msg)))) => {
if let Err(err) = self.framed.write(msg) {
self.state = State::FramedError(DispatcherError::Encoder(err));
if let Err(err) = this.framed.as_mut().write(msg) {
*this.state = State::FramedError(DispatcherError::Encoder(err));
return true;
}
}
Poll::Ready(Some(Ok(Message::Close))) => {
self.state = State::FlushAndStop;
*this.state = State::FlushAndStop;
return true;
}
Poll::Ready(Some(Err(err))) => {
self.state = State::Error(DispatcherError::Service(err));
*this.state = State::Error(DispatcherError::Service(err));
return true;
}
Poll::Ready(None) | Poll::Pending => break,
}
}
if !self.framed.is_write_buf_empty() {
match self.framed.flush(cx) {
if !this.framed.is_write_buf_empty() {
match this.framed.flush(cx) {
Poll::Pending => break,
Poll::Ready(Ok(_)) => (),
Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err);
self.state = State::FramedError(DispatcherError::Encoder(err));
*this.state = State::FramedError(DispatcherError::Encoder(err));
return true;
}
}
@ -279,7 +282,7 @@ where
return match this.state {
State::Processing => {
if self.poll_read(cx) || self.poll_write(cx) {
if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
continue;
} else {
Poll::Pending
@ -287,12 +290,12 @@ where
}
State::Error(_) => {
// flush write buffer
if !self.framed.is_write_buf_empty() {
if let Poll::Pending = self.framed.flush(cx) {
if !this.framed.is_write_buf_empty() {
if let Poll::Pending = this.framed.flush(cx) {
return Poll::Pending;
}
}
Poll::Ready(Err(self.state.take_error()))
Poll::Ready(Err(this.state.take_error()))
}
State::FlushAndStop => {
if !this.framed.is_write_buf_empty() {