[fidl][rust] Remove GATs from Transport trait

We ran into several edge cases with the trait solver when using GATs. By
separating the lifetime-dependent code out into a separate future, we
can avoid using GATs altogether.

Change-Id: I45832054c84d12107bf867e2db3f3140a0a04093
Reviewed-on: https://fuchsia-review.googlesource.com/c/fuchsia/+/1219444
Reviewed-by: Ian McKellar <[email protected]>
Fuchsia-Auto-Submit: David Koloski <[email protected]>
Commit-Queue: David Koloski <[email protected]>

Author: djkoloski

src/lib/fidl/rust_next/fidl_next/src/lib.rs

@@ -23,7 +23,9 @@
 
 pub use ::fidl_next_bind::*;
 pub use ::fidl_next_codec::*;
-pub use ::fidl_next_protocol::{self as protocol, ProtocolError, Transport};
+pub use ::fidl_next_protocol::{
+    self as protocol, ProtocolError, RecvFuture, SendFuture, Transport,
+};
 
 /// Fuchsia-specific FIDL extensions.
 #[cfg(target_os = "fuchsia")]

src/lib/fidl/rust_next/fidl_next_bind/src/server.rs

@@ -5,7 +5,7 @@
 use core::marker::PhantomData;
 
 use fidl_next_codec::{Encode, EncodeError};
-use fidl_next_protocol::{self as protocol, ProtocolError, Transport};
+use fidl_next_protocol::{self as protocol, ProtocolError, SendFuture, Transport};
 
 use super::{Method, ServerEnd};
 
@@ -167,7 +167,7 @@ impl<M> Responder<M> {
         self,
         server: &'s ServerSender<T, P>,
         response: &mut R,
-    ) -> Result<T::SendFuture<'s>, EncodeError>
+    ) -> Result<SendFuture<'s, T>, EncodeError>
     where
         T: Transport,
         M: Method<Protocol = P>,

src/lib/fidl/rust_next/fidl_next_protocol/src/client.rs

@@ -12,7 +12,7 @@ use std::sync::{Arc, Mutex};
 use fidl_next_codec::{Encode, EncodeError, EncoderExt};
 
 use crate::lockers::Lockers;
-use crate::{decode_header, encode_header, ProtocolError, Transport};
+use crate::{decode_header, encode_header, ProtocolError, SendFuture, Transport, TransportExt};
 
 use super::lockers::LockerError;
 
@@ -43,7 +43,7 @@ impl<T: Transport> ClientSender<T> {
         &self,
         ordinal: u64,
         request: &mut M,
-    ) -> Result<T::SendFuture<'_>, EncodeError>
+    ) -> Result<SendFuture<'_, T>, EncodeError>
     where
         M: Encode<T::SendBuffer>,
     {
@@ -80,7 +80,7 @@ impl<T: Transport> ClientSender<T> {
         txid: u32,
         ordinal: u64,
         message: &mut M,
-    ) -> Result<T::SendFuture<'_>, EncodeError>
+    ) -> Result<SendFuture<'_, T>, EncodeError>
     where
         M: Encode<T::SendBuffer>,
     {
@@ -97,8 +97,8 @@ impl<T: Transport> Clone for ClientSender<T> {
     }
 }
 
-enum ResponseFutureState<'a, T: 'a + Transport> {
-    Sending(T::SendFuture<'a>),
+enum ResponseFutureState<'a, T: Transport> {
+    Sending(SendFuture<'a, T>),
     Receiving,
     // We store the completion state locally so that we can free the locker during poll, instead of
     // waiting until the future is dropped.

src/lib/fidl/rust_next/fidl_next_protocol/src/fuchsia/channel.rs

@@ -4,7 +4,6 @@
 
 //! A transport implementation which uses Zircon channels.
 
-use core::future::Future;
 use core::mem::replace;
 use core::pin::Pin;
 use core::ptr::NonNull;
@@ -123,111 +122,21 @@ impl HandleEncoder for Buffer {
     }
 }
 
-/// A channel send future.
-#[must_use = "futures do nothing unless polled"]
-pub struct SendFuture<'s> {
-    shared: &'s Shared,
+/// The state for a channel send future.
+pub struct SendFutureState {
     buffer: Buffer,
 }
 
-impl Future for SendFuture<'_> {
-    type Output = Result<(), Status>;
-
-    fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = Pin::into_inner(self);
-
-        let result = unsafe {
-            zx_channel_write(
-                this.shared.channel.get_ref().raw_handle(),
-                0,
-                this.buffer.chunks.as_ptr().cast::<u8>(),
-                (this.buffer.chunks.len() * CHUNK_SIZE) as u32,
-                this.buffer.handles.as_ptr().cast(),
-                this.buffer.handles.len() as u32,
-            )
-        };
-
-        if result == ZX_OK {
-            // Handles were written to the channel, so we must not drop them.
-            unsafe {
-                this.buffer.handles.set_len(0);
-            }
-            Poll::Ready(Ok(()))
-        } else {
-            Poll::Ready(Err(Status::from_raw(result)))
-        }
-    }
-}
-
 /// A channel receiver.
 pub struct Receiver {
     shared: Arc<Shared>,
 }
 
-/// A channel receive future.
-#[must_use = "futures do nothing unless polled"]
-pub struct RecvFuture<'r> {
-    shared: &'r Shared,
+/// The state for a channel receive future.
+pub struct RecvFutureState {
     buffer: Option<Buffer>,
 }
 
-impl Future for RecvFuture<'_> {
-    type Output = Result<Option<RecvBuffer>, Status>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = Pin::into_inner(self);
-        let buffer = this.buffer.as_mut().unwrap();
-
-        let mut actual_bytes = 0;
-        let mut actual_handles = 0;
-
-        loop {
-            let result = unsafe {
-                zx_channel_read(
-                    this.shared.channel.get_ref().raw_handle(),
-                    0,
-                    buffer.chunks.as_mut_ptr().cast(),
-                    buffer.handles.as_mut_ptr().cast(),
-                    (buffer.chunks.capacity() * CHUNK_SIZE) as u32,
-                    buffer.handles.capacity() as u32,
-                    &mut actual_bytes,
-                    &mut actual_handles,
-                )
-            };
-
-            match result {
-                ZX_OK => {
-                    unsafe {
-                        buffer.chunks.set_len(actual_bytes as usize / CHUNK_SIZE);
-                        buffer.handles.set_len(actual_handles as usize);
-                    }
-                    return Poll::Ready(Ok(Some(RecvBuffer {
-                        buffer: this.buffer.take().unwrap(),
-                        chunks_taken: 0,
-                        handles_taken: 0,
-                    })));
-                }
-                ZX_ERR_PEER_CLOSED => return Poll::Ready(Ok(None)),
-                ZX_ERR_BUFFER_TOO_SMALL => {
-                    let min_chunks = (actual_bytes as usize).div_ceil(CHUNK_SIZE);
-                    buffer.chunks.reserve(min_chunks - buffer.chunks.capacity());
-                    buffer.handles.reserve(actual_handles as usize - buffer.handles.capacity());
-                }
-                ZX_ERR_SHOULD_WAIT => {
-                    if matches!(this.shared.channel.need_readable(cx)?, Poll::Pending) {
-                        this.shared.closed_waker.register(cx.waker());
-                        if this.shared.is_closed.load(Ordering::Relaxed) {
-                            return Poll::Ready(Ok(None));
-                        }
-                        return Poll::Pending;
-                    }
-                }
-                raw => return Poll::Ready(Err(Status::from_raw(raw))),
-            }
-        }
-    }
-}
-
 /// A channel receive buffer.
 pub struct RecvBuffer {
     buffer: Buffer,
@@ -311,26 +220,109 @@ impl Transport for Channel {
 
     type Sender = Sender;
     type SendBuffer = Buffer;
-    type SendFuture<'s> = SendFuture<'s>;
+    type SendFutureState = SendFutureState;
 
     fn acquire(_: &Self::Sender) -> Self::SendBuffer {
         Buffer::new()
     }
 
-    fn send(sender: &Self::Sender, buffer: Self::SendBuffer) -> Self::SendFuture<'_> {
-        SendFuture { shared: &sender.shared, buffer }
+    fn begin_send(_: &Self::Sender, buffer: Self::SendBuffer) -> Self::SendFutureState {
+        SendFutureState { buffer }
+    }
+
+    fn poll_send(
+        mut future_state: Pin<&mut Self::SendFutureState>,
+        _: &mut Context<'_>,
+        sender: &Self::Sender,
+    ) -> Poll<Result<(), Self::Error>> {
+        let result = unsafe {
+            zx_channel_write(
+                sender.shared.channel.get_ref().raw_handle(),
+                0,
+                future_state.buffer.chunks.as_ptr().cast::<u8>(),
+                (future_state.buffer.chunks.len() * CHUNK_SIZE) as u32,
+                future_state.buffer.handles.as_ptr().cast(),
+                future_state.buffer.handles.len() as u32,
+            )
+        };
+
+        if result == ZX_OK {
+            // Handles were written to the channel, so we must not drop them.
+            unsafe {
+                future_state.buffer.handles.set_len(0);
+            }
+            Poll::Ready(Ok(()))
+        } else {
+            Poll::Ready(Err(Status::from_raw(result)))
+        }
     }
 
     fn close(sender: &Self::Sender) {
         sender.shared.close();
     }
 
     type Receiver = Receiver;
-    type RecvFuture<'r> = RecvFuture<'r>;
+    type RecvFutureState = RecvFutureState;
     type RecvBuffer = RecvBuffer;
 
-    fn recv(receiver: &mut Self::Receiver) -> Self::RecvFuture<'_> {
-        RecvFuture { shared: &receiver.shared, buffer: Some(Buffer::new()) }
+    fn begin_recv(_: &mut Self::Receiver) -> Self::RecvFutureState {
+        RecvFutureState { buffer: Some(Buffer::new()) }
+    }
+
+    fn poll_recv(
+        mut future_state: Pin<&mut Self::RecvFutureState>,
+        cx: &mut Context<'_>,
+        receiver: &mut Self::Receiver,
+    ) -> Poll<Result<Option<Self::RecvBuffer>, Self::Error>> {
+        let buffer = future_state.buffer.as_mut().unwrap();
+
+        let mut actual_bytes = 0;
+        let mut actual_handles = 0;
+
+        loop {
+            let result = unsafe {
+                zx_channel_read(
+                    receiver.shared.channel.get_ref().raw_handle(),
+                    0,
+                    buffer.chunks.as_mut_ptr().cast(),
+                    buffer.handles.as_mut_ptr().cast(),
+                    (buffer.chunks.capacity() * CHUNK_SIZE) as u32,
+                    buffer.handles.capacity() as u32,
+                    &mut actual_bytes,
+                    &mut actual_handles,
+                )
+            };
+
+            match result {
+                ZX_OK => {
+                    unsafe {
+                        buffer.chunks.set_len(actual_bytes as usize / CHUNK_SIZE);
+                        buffer.handles.set_len(actual_handles as usize);
+                    }
+                    return Poll::Ready(Ok(Some(RecvBuffer {
+                        buffer: future_state.buffer.take().unwrap(),
+                        chunks_taken: 0,
+                        handles_taken: 0,
+                    })));
+                }
+                ZX_ERR_PEER_CLOSED => return Poll::Ready(Ok(None)),
+                ZX_ERR_BUFFER_TOO_SMALL => {
+                    let min_chunks = (actual_bytes as usize).div_ceil(CHUNK_SIZE);
+                    buffer.chunks.reserve(min_chunks - buffer.chunks.capacity());
+                    buffer.handles.reserve(actual_handles as usize - buffer.handles.capacity());
+                }
+                ZX_ERR_SHOULD_WAIT => {
+                    if matches!(receiver.shared.channel.need_readable(cx)?, Poll::Pending) {
+                        receiver.shared.closed_waker.register(cx.waker());
+                        if receiver.shared.is_closed.load(Ordering::Relaxed) {
+                            return Poll::Ready(Ok(None));
+                        }
+                        return Poll::Pending;
+                    }
+                }
+                raw => return Poll::Ready(Err(Status::from_raw(raw))),
+            }
+        }
     }
 }