diff --git a/actix-codec/Cargo.toml b/actix-codec/Cargo.toml index 9decfdc1..d647ab70 100644 --- a/actix-codec/Cargo.toml +++ b/actix-codec/Cargo.toml @@ -23,4 +23,5 @@ futures-core = "0.3.1" 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" \ No newline at end of file +log = "0.4" +pin-project = "0.4.8" diff --git a/actix-codec/src/framed.rs b/actix-codec/src/framed.rs index 146d2258..685d87c9 100644 --- a/actix-codec/src/framed.rs +++ b/actix-codec/src/framed.rs @@ -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 { + #[pin] io: T, codec: U, flags: Flags, @@ -28,8 +31,6 @@ pub struct Framed { write_buf: BytesMut, } -impl Unpin for Framed {} - impl Framed where T: AsyncRead + AsyncWrite, @@ -185,17 +186,18 @@ impl Framed { impl Framed { /// Serialize item and Write to the inner buffer - pub fn write(&mut self, item: ::Item) -> Result<(), ::Error> + pub fn write(mut self: Pin<&mut Self>, item: ::Item) -> Result<(), ::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 Framed { } /// Try to read underlying I/O stream and decode item. - pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll>> + pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll>> 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 Framed { 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 Framed { _ => (), // 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> + pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> 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 Framed { } // 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> + pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> 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; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.next_item(cx) } } @@ -341,21 +342,21 @@ where } fn start_send( - mut self: Pin<&mut Self>, + self: Pin<&mut Self>, item: ::Item, ) -> Result<(), Self::Error> { self.write(item) } fn poll_flush( - mut self: Pin<&mut Self>, + self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll> { self.flush(cx) } fn poll_close( - mut self: Pin<&mut Self>, + self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll> { self.close(cx) diff --git a/actix-ioframe/src/connect.rs b/actix-ioframe/src/connect.rs index 61705951..85248fe4 100644 --- a/actix-ioframe/src/connect.rs +++ b/actix-ioframe/src/connect.rs @@ -42,6 +42,7 @@ where pub struct ConnectResult { pub(crate) state: St, pub(crate) out: Option, + #[pin] pub(crate) framed: Framed, } @@ -97,8 +98,8 @@ where { type Error = ::Error; - fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - if self.framed.is_write_ready() { + fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + 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> { - self.get_mut().framed.flush(cx) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.as_mut().project().framed.flush(cx) } - fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.get_mut().framed.close(cx) + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.as_mut().project().framed.close(cx) } } diff --git a/actix-ioframe/src/dispatcher.rs b/actix-ioframe/src/dispatcher.rs index 3e000da6..3be8963a 100644 --- a/actix-ioframe/src/dispatcher.rs +++ b/actix-ioframe/src/dispatcher.rs @@ -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 = ::Item; /// FramedTransport - is a future that reads frames from Framed object /// and pass then to the service. +#[pin_project] pub(crate) struct Dispatcher where S: Service, Response = Option>>, @@ -29,6 +31,7 @@ where service: S, sink: Option, state: FramedState, + #[pin] framed: Framed, rx: mpsc::Receiver::Item, S::Error>>, } @@ -90,26 +93,27 @@ where ::Error: std::fmt::Debug, Out: Stream::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>> { - 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(())), } } diff --git a/actix-ioframe/src/service.rs b/actix-ioframe/src/service.rs index c9564d7a..d679b45b 100644 --- a/actix-ioframe/src/service.rs +++ b/actix-ioframe/src/service.rs @@ -357,7 +357,7 @@ where { Connect(#[pin] C::Future, Rc), Handler(#[pin] T::Future, Option>, Option), - Dispatcher(Dispatcher), + Dispatcher(#[pin] Dispatcher), } impl FramedServiceImplResponseInner @@ -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)) } } diff --git a/actix-rt/src/arbiter.rs b/actix-rt/src/arbiter.rs index aadb5f37..c8c49775 100644 --- a/actix-rt/src/arbiter.rs +++ b/actix-rt/src/arbiter.rs @@ -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))) }); } }); diff --git a/actix-utils/src/framed.rs b/actix-utils/src/framed.rs index 1bdd364f..9def5992 100644 --- a/actix-utils/src/framed.rs +++ b/actix-utils/src/framed.rs @@ -77,6 +77,7 @@ where { service: S, state: State, + #[pin] framed: Framed, rx: mpsc::Receiver::Item>, S::Error>>, tx: mpsc::Sender::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, Response = Response>, S::Error: 'static, @@ -180,29 +181,30 @@ where ::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, Response = Response>, S::Error: 'static, @@ -221,33 +223,34 @@ where ::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() {