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

Code before:

    /// Returns `true` if senders belong to the same channel.
    pub(crate) fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner, &other.inner)
    }
}

impl<T, S> TxWeak<T, S> {
    fn new(inner: Weak<Chan<T, S>>) -> Self {
        TxWeak { inner }
    }

    pub(super) fn upgrade(&self) -> Option<Tx<T, S>> {
        let inner = self.inner.upgrade();

        if let Some(inner) = inner {
            // If we were able to upgrade, `Chan` is guaranteed to still exist,
            // even though the channel might have been closed in the meantime.
            // Need to check here whether the channel was actually closed.

            let mut tx_count = inner.tx_count.load(Relaxed);
            loop {
                // FIXME Haven't thought the orderings on the CAS through yet
                match inner
                    .tx_count
                    .compare_exchange(tx_count, tx_count + 1, SeqCst, SeqCst)
                {
                    Ok(prev_count) => {
                        if prev_count == 0 {
                            mem::drop(inner);
                            return None;
                        }

                        return Some(Tx::new(inner));
                    }
                    Err(count) => {
                        if count == 0 {
                            mem::drop(inner);
                            return None;
                        }

                        tx_count = count;
                    }
                }
            }
        } else {
            None
        }
    }
}


Review comment (suggestion):
I don't think you should increment the counter in the first place if it is already closed. (If always incrementing the counter is ok, then a simple `fetch_add` would suffice.)

Code after:
    }

    /// Wake the receive half
    pub(crate) fn wake_rx(&self) {
        self.inner.rx_waker.wake();
    }

    /// Returns `true` if senders belong to the same channel.
    pub(crate) fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner, &other.inner)
    }
}

impl<T, S: Semaphore> Tx<T, S> {
    pub(crate) fn is_closed(&self) -> bool {
        self.inner.semaphore.is_closed()
    }

    pub(crate) async fn closed(&self) {
        // In order to avoid a race condition, we first request a notification,
        // **then** check whether the semaphore is closed. If the semaphore is
        // closed the notification request is dropped.
        let notified = self.inner.notify_rx_closed.notified();

        if self.inner.semaphore.is_closed() {
            return;
        }
        notified.await;
    }
}

impl<T, S> Clone for Tx<T, S> {
    fn clone(&self) -> Tx<T, S> {
        // Using a Relaxed ordering here is sufficient as the caller holds a
        // strong ref to `self`, preventing a concurrent decrement to zero.
        self.inner.tx_count.fetch_add(1, Relaxed);

        Tx {
            inner: self.inner.clone(),
        }
    }
}

impl<T, S> Drop for Tx<T, S> {
    fn drop(&mut self) {
        if self.inner.tx_count.fetch_sub(1, AcqRel) != 1 {
            return;
        }

        // Close the list, which sends a `Close` message
        self.inner.tx.close();