Repository: pola-rs/polars
File: crates/polars-utils/src/live_timer.rs (Rust)

Code before:
            self.state_ns.store(new_state_ns, Ordering::Release); // See (2).
        }
    }

    /// Returns true if this timer should be destroyed.
    fn stop_session(&self) -> bool {
        let mut stopped = false;
        let mut stopped_at_ns = u64::MAX;
        let mut orig_state_ns = u64::MAX;

        // (1) Acquire and Release to ensure we load state_ns from previous start_session, if necessary.
        self.refcount
            .fetch_update(Ordering::Release, Ordering::Acquire, |rc| {
                if rc == SESSION_UNIT {
                    return None; // We're the sole reference, can just destroy.
                }

                // Stop or re-start the timer if necessary.
                let should_stop = rc & SESSION_MASK == SESSION_UNIT;
                if should_stop && !stopped {
                    if stopped_at_ns == u64::MAX {
                        orig_state_ns = self.state_ns.load(Ordering::Relaxed);
                        stopped_at_ns = self.base_timestamp.elapsed().as_nanos() as u64;
                    }
                    let new_state_ns = stopped_at_ns.saturating_sub(orig_state_ns & !TICKING_BIT);
                    self.state_ns.store(new_state_ns, Ordering::Release); // See (2).
                    stopped = true;
                } else if !should_stop && stopped {
                    self.state_ns.store(orig_state_ns, Ordering::Release); // See (2).
                    stopped = false;
                }

                Some(rc - SESSION_UNIT)
            })
            .is_err()
    }

    fn total_time_live_ns(&self) -> u64 {
        // (2) Loading this with Acquire improves monotonicity in our elapsed() call, forcing it (barring platform bugs)
        // to be >= than the stored state.
        let state_ns = self.state_ns.load(Ordering::Acquire);
        let active_time = if state_ns & TICKING_BIT == 0 {
            state_ns
        } else {
            let now_ns = self.base_timestamp.elapsed().as_nanos() as u64;
            now_ns.saturating_sub(state_ns & !TICKING_BIT)
        };

        // Needed to ensure monotonicity, our load above interleaved with stops/restarts
        // could otherwise be non-monotonic.
        self.max_live_ns.fetch_max(active_time, Ordering::Relaxed)

Review comment (suggestion):
I think this store can be `Relaxed`, as `total_time_live_ns` would never measure `elapsed()` if it reads this value.


Code after:
            self.state_ns.store(new_state_ns, Ordering::Release); // See (2).
        }
    }

    /// Returns true if this timer should be destroyed.
    fn stop_session(&self) -> bool {
        let mut stopped = false;
        let mut stopped_at_ns = u64::MAX;
        let mut orig_state_ns = u64::MAX;

        // (1) Acquire and Release to ensure we load state_ns from previous start_session, if necessary.
        self.refcount
            .fetch_update(Ordering::Release, Ordering::Acquire, |rc| {
                if rc == SESSION_UNIT {
                    return None; // We're the sole reference, can just destroy.
                }

                // Stop or re-start the timer if necessary.
                let should_stop = rc & SESSION_MASK == SESSION_UNIT;
                if should_stop && !stopped {
                    if stopped_at_ns == u64::MAX {
                        orig_state_ns = self.state_ns.load(Ordering::Relaxed);
                        stopped_at_ns = self.base_timestamp.elapsed().as_nanos() as u64;
                    }
                    let new_state_ns = stopped_at_ns.saturating_sub(orig_state_ns & !TICKING_BIT);
                    self.state_ns.store(new_state_ns, Ordering::Relaxed);
                    stopped = true;
                } else if !should_stop && stopped {
                    self.state_ns.store(orig_state_ns, Ordering::Release); // See (2).
                    stopped = false;
                }

                Some(rc - SESSION_UNIT)
            })
            .is_err()
    }

    fn total_time_live_ns(&self) -> u64 {
        // (2) Acquire ensures the elapsed() call by this function is sequenced after the elapsed()
        // call that state_ns value was calculated against if it is currently ticking.
        let state_ns = self.state_ns.load(Ordering::Acquire);
        let active_time = if state_ns & TICKING_BIT == 0 {
            state_ns
        } else {
            let now_ns = self.base_timestamp.elapsed().as_nanos() as u64;
            now_ns.saturating_sub(state_ns & !TICKING_BIT)
        };

        // Needed to ensure monotonicity, our load above interleaved with stops/restarts
        // could otherwise be non-monotonic.
        self.max_live_ns.fetch_max(active_time, Ordering::Relaxed)