tyler
/
proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

async_io: require Unpin for EitherStream and HyperAccept 

We use it with Unpin types and this way we get rid of a lot
of `unsafe` blocks.

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
This commit is contained in:
Wolfgang Bumiller 2020-11-10 14:14:43 +01:00
parent 9f9a661b1a
commit 1ecf4e6d20
1 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
src/tools/async_io.rs
View File

@ -16,18 +16,18 @@ pub enum EitherStream<L, R> {
Right(R), Right(R),
} }
impl<L: AsyncRead, R: AsyncRead> AsyncRead for EitherStream<L, R> { impl<L: AsyncRead + Unpin, R: AsyncRead + Unpin> AsyncRead for EitherStream<L, R> {
fn poll_read( fn poll_read(
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context, cx: &mut Context,
buf: &mut [u8], buf: &mut [u8],
) -> Poll<Result<usize, io::Error>> { ) -> Poll<Result<usize, io::Error>> {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_read(cx, buf) Pin::new(s).poll_read(cx, buf)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_read(cx, buf) Pin::new(s).poll_read(cx, buf)
} }
} }
} }
@ -47,51 +47,51 @@ impl<L: AsyncRead, R: AsyncRead> AsyncRead for EitherStream<L, R> {
where where
B: bytes::BufMut, B: bytes::BufMut,
{ {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_read_buf(cx, buf) Pin::new(s).poll_read_buf(cx, buf)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_read_buf(cx, buf) Pin::new(s).poll_read_buf(cx, buf)
} }
} }
} }
} }
impl<L: AsyncWrite, R: AsyncWrite> AsyncWrite for EitherStream<L, R> { impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L, R> {
fn poll_write( fn poll_write(
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context, cx: &mut Context,
buf: &[u8], buf: &[u8],
) -> Poll<Result<usize, io::Error>> { ) -> Poll<Result<usize, io::Error>> {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_write(cx, buf) Pin::new(s).poll_write(cx, buf)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_write(cx, buf) Pin::new(s).poll_write(cx, buf)
} }
} }
} }
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> { fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_flush(cx) Pin::new(s).poll_flush(cx)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_flush(cx) Pin::new(s).poll_flush(cx)
} }
} }
} }
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> { fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_shutdown(cx) Pin::new(s).poll_shutdown(cx)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_shutdown(cx) Pin::new(s).poll_shutdown(cx)
} }
} }
} }
@ -104,12 +104,12 @@ impl<L: AsyncWrite, R: AsyncWrite> AsyncWrite for EitherStream<L, R> {
where where
B: bytes::Buf, B: bytes::Buf,
{ {
match unsafe { self.get_unchecked_mut() } { match self.get_mut() {
EitherStream::Left(ref mut s) => { EitherStream::Left(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_write_buf(cx, buf) Pin::new(s).poll_write_buf(cx, buf)
} }
EitherStream::Right(ref mut s) => { EitherStream::Right(ref mut s) => {
unsafe { Pin::new_unchecked(s) }.poll_write_buf(cx, buf) Pin::new(s).poll_write_buf(cx, buf)
} }
} }
} }
@ -180,7 +180,7 @@ pub struct HyperAccept<T>(pub T);
impl<T, I> hyper::server::accept::Accept for HyperAccept<T> impl<T, I> hyper::server::accept::Accept for HyperAccept<T>
where where
T: TryStream<Ok = I>, T: TryStream<Ok = I> + Unpin,
I: AsyncRead + AsyncWrite, I: AsyncRead + AsyncWrite,
{ {
type Conn = I; type Conn = I;
@ -190,7 +190,7 @@ where
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context, cx: &mut Context,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> { ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
let this = unsafe { self.map_unchecked_mut(|this| &mut this.0) }; let this = Pin::new(&mut self.get_mut().0);
this.try_poll_next(cx) this.try_poll_next(cx)
} }
} }