Repository: tokio-rs/tokio
File: tokio-util/src/sync/mpsc.rs (Rust)

Code before:

impl<T> PollSenderFuture<T> {
    /// Create with an empty inner future with no `Send` bound.
    fn empty() -> Self {
        // We don't use `make_acquire_future` here because our relaxed bounds on `T` are not
        // compatible with the transitive bounds required by `Sender<T>`.
        Self(ReusableBoxFuture::new(async { unreachable!() }))
    }
}

impl<T: Send> PollSenderFuture<T> {
    /// Create with an empty inner future.
    fn new() -> Self {
        let v = InnerFuture::new(make_acquire_future(None));
        // This is safe because `make_acquire_future(None)` is actually `'static`
        Self(unsafe { mem::transmute::<InnerFuture<'_, T>, InnerFuture<'static, T>>(v) })
    }

    /// Poll the inner future.
    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Result<OwnedPermit<T>, PollSendError<T>>> {
        self.0.poll(cx)
    }

    /// Replace the inner future.
    fn set(&mut self, sender: Option<Sender<T>>) {
        // This is safe because `make_acquire_future(sender)` is bounded by the lifetime of `T`, which inner is already bound by
        let inner: &mut InnerFuture<'_, T> =
            unsafe { &mut *(&mut self.0 as *mut InnerFuture<'static, T>).cast() };
        inner.set(make_acquire_future(sender));
    }
}

impl<T: Send> PollSender<T> {
    /// Creates a new `PollSender`.
    pub fn new(sender: Sender<T>) -> Self {
        Self {
            sender: Some(sender.clone()),
            state: State::Idle(sender),
            acquire: PollSenderFuture::new(),
        }
    }

    fn take_state(&mut self) -> State<T> {
        mem::replace(&mut self.state, State::Closed)
    }

    /// Attempts to prepare the sender to receive a value.
    ///
    /// This method must be called and return `Poll::Ready(Ok(()))` prior to each call to
    /// `send_item`.
    ///

Review comment (suggestion):
```suggestion
        // SAFETY: The `make_acquire_future(sender)` future must not exist after the type `T`
        // becomes invalid, and this casts away the type-level lifetime check for that. However, the
        // inner future is never moved out of this `PollSenderFuture<T>`, so the future will not
        // live longer than the `PollSenderFuture<T>` lives. A `PollSenderFuture<T>` is guaranteed
        // to not exist after the type `T` becomes invalid, because it is annotated with a `T`, so
        // this is ok.
```


Code after:

impl<T> PollSenderFuture<T> {
    /// Create with an empty inner future with no `Send` bound.
    fn empty() -> Self {
        // We don't use `make_acquire_future` here because our relaxed bounds on `T` are not
        // compatible with the transitive bounds required by `Sender<T>`.
        Self(ReusableBoxFuture::new(async { unreachable!() }))
    }
}

impl<T: Send> PollSenderFuture<T> {
    /// Create with an empty inner future.
    fn new() -> Self {
        let v = InnerFuture::new(make_acquire_future(None));
        // This is safe because `make_acquire_future(None)` is actually `'static`
        Self(unsafe { mem::transmute::<InnerFuture<'_, T>, InnerFuture<'static, T>>(v) })
    }

    /// Poll the inner future.
    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Result<OwnedPermit<T>, PollSendError<T>>> {
        self.0.poll(cx)
    }

    /// Replace the inner future.
    fn set(&mut self, sender: Option<Sender<T>>) {
        let inner: *mut InnerFuture<'static, T> = &mut self.0;
        let inner: *mut InnerFuture<'_, T> = inner.cast();
        // SAFETY: The `make_acquire_future(sender)` future must not exist after the type `T`
        // becomes invalid, and this casts away the type-level lifetime check for that. However, the
        // inner future is never moved out of this `PollSenderFuture<T>`, so the future will not
        // live longer than the `PollSenderFuture<T>` lives. A `PollSenderFuture<T>` is guaranteed
        // to not exist after the type `T` becomes invalid, because it is annotated with a `T`, so
        // this is ok.
        let inner = unsafe { &mut *inner };
        inner.set(make_acquire_future(sender));
    }
}

impl<T: Send> PollSender<T> {
    /// Creates a new `PollSender`.
    pub fn new(sender: Sender<T>) -> Self {
        Self {
            sender: Some(sender.clone()),
            state: State::Idle(sender),
            acquire: PollSenderFuture::new(),
        }
    }

    fn take_state(&mut self) -> State<T> {
        mem::replace(&mut self.state, State::Closed)
    }