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

Code before:

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        if 1 == self.shared.num_tx.fetch_sub(1, SeqCst) {
            self.close_channel();
        }
    }
}

impl<T> WeakSender<T> {
    /// Tries to convert a `WeakSender` into a [`Sender`]. This will return `Some`
    /// if there are other `Sender` instances alive and the channel wasn't
    /// previously dropped, otherwise `None` is returned.
    pub fn upgrade(&self) -> Option<Sender<T>> {
        let mut tx_count = self.shared.num_tx.load(Acquire);

        loop {
            if tx_count == 0 {
                // channel is closed so this WeakSender can not be upgraded
                return None;
            }

            match self
                .shared
                .num_tx
                .compare_exchange_weak(tx_count, tx_count + 1, AcqRel, Acquire)
            {
                Ok(_) => {
                    return Some(Sender {
                        shared: self.shared.clone(),
                    })
                }
                Err(prev_count) => tx_count = prev_count,
            }
        }
    }

    /// Returns the number of [`Sender`] handles.
    pub fn strong_count(&self) -> usize {
        self.shared.num_tx.load(SeqCst)
    }

    /// Returns the number of [`WeakSender`] handles.
    pub fn weak_count(&self) -> usize {
        self.shared.num_weak_tx.load(SeqCst)
    }
}

impl<T> Clone for WeakSender<T> {
    fn clone(&self) -> WeakSender<T> {
        let shared = self.shared.clone();

Review comment (bug):
This is sufficient. When this succeeds, the operation is an increment for which relaxed is enough. When it fails, we might exit through the zero-check, so it needs acquire.
```suggestion
                .compare_exchange_weak(tx_count, tx_count + 1, Relaxed, Acquire)
```

Code after:

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        if 1 == self.shared.num_tx.fetch_sub(1, AcqRel) {
            self.close_channel();
        }
    }
}

impl<T> WeakSender<T> {
    /// Tries to convert a `WeakSender` into a [`Sender`].
    ///
    /// This will return `Some` if there are other `Sender` instances alive and
    /// the channel wasn't previously dropped, otherwise `None` is returned.
    #[must_use]
    pub fn upgrade(&self) -> Option<Sender<T>> {
        let mut tx_count = self.shared.num_tx.load(Acquire);

        loop {
            if tx_count == 0 {
                // channel is closed so this WeakSender can not be upgraded
                return None;
            }

            match self
                .shared
                .num_tx
                .compare_exchange_weak(tx_count, tx_count + 1, Relaxed, Acquire)
            {
                Ok(_) => {
                    return Some(Sender {
                        shared: self.shared.clone(),
                    })
                }
                Err(prev_count) => tx_count = prev_count,
            }
        }
    }

    /// Returns the number of [`Sender`] handles.
    pub fn strong_count(&self) -> usize {
        self.shared.num_tx.load(Acquire)
    }

    /// Returns the number of [`WeakSender`] handles.
    pub fn weak_count(&self) -> usize {
        self.shared.num_weak_tx.load(Acquire)
    }
}

impl<T> Clone for WeakSender<T> {