rework Ticker (fix #416)

examples/multi-tree-ext.rs

@@ -1,12 +1,14 @@
-use console::style;
-use indicatif::{MultiProgress, MultiProgressAlignment, ProgressBar, ProgressStyle};
-use once_cell::sync::Lazy;
-use rand::{rngs::ThreadRng, Rng, RngCore};
 use std::fmt::Debug;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
+
+use console::style;
+use indicatif::{MultiProgress, MultiProgressAlignment, ProgressBar, ProgressStyle};
+use once_cell::sync::Lazy;
+use rand::rngs::ThreadRng;
+use rand::{Rng, RngCore};
 use structopt::StructOpt;
 
 #[derive(Debug, Clone)]

examples/multi-tree.rs

@@ -1,11 +1,13 @@
-use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
-use once_cell::sync::Lazy;
-use rand::{rngs::ThreadRng, Rng, RngCore};
 use std::fmt::Debug;
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::Duration;
 
+use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
+use once_cell::sync::Lazy;
+use rand::rngs::ThreadRng;
+use rand::{Rng, RngCore};
+
 #[derive(Debug, Clone)]
 enum Action {
     AddProgressBar(usize),

examples/tokio.rs

@@ -1,5 +1,6 @@
-use indicatif::ProgressBar;
 use std::time::Duration;
+
+use indicatif::ProgressBar;
 use tokio::runtime;
 use tokio::time::interval;
 

examples/yarnish.rs

@@ -1,10 +1,10 @@
-use rand::seq::SliceRandom;
-use rand::Rng;
 use std::thread;
 use std::time::{Duration, Instant};
 
 use console::{style, Emoji};
 use indicatif::{HumanDuration, MultiProgress, ProgressBar, ProgressStyle};
+use rand::seq::SliceRandom;
+use rand::Rng;
 
 static PACKAGES: &[&str] = &[
     "fs-events",

rustfmt.toml

@@ -0,0 +1,2 @@
+imports_granularity = "Module"
+group_imports = "StdExternalCrate"

src/lib.rs

@@ -237,9 +237,8 @@ pub use crate::in_memory::InMemoryTerm;
 pub use crate::iter::{ProgressBarIter, ProgressIterator};
 pub use crate::multi::{MultiProgress, MultiProgressAlignment};
 pub use crate::progress_bar::{ProgressBar, WeakProgressBar};
+#[cfg(feature = "rayon")]
+pub use crate::rayon::ParallelProgressIterator;
 pub use crate::state::{ProgressFinish, ProgressState};
 pub use crate::style::ProgressStyle;
 pub use crate::term_like::TermLike;
-
-#[cfg(feature = "rayon")]
-pub use crate::rayon::ParallelProgressIterator;

src/progress_bar.rs

@@ -1,15 +1,17 @@
 use std::borrow::Cow;
-use std::io;
-use std::sync::MutexGuard;
-use std::sync::{Arc, Mutex, Weak};
+#[cfg(test)]
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
 use std::time::{Duration, Instant};
-use std::{fmt, mem};
+use std::{fmt, io, mem, thread};
+
+#[cfg(test)]
+use once_cell::sync::Lazy;
 
 use crate::draw_target::ProgressDrawTarget;
-use crate::state::{AtomicPosition, BarState, ProgressFinish, Reset, Ticker};
+use crate::state::{AtomicPosition, BarState, ProgressFinish, Reset};
 use crate::style::ProgressStyle;
-use crate::ProgressState;
-use crate::{ProgressBarIter, ProgressIterator};
+use crate::{ProgressBarIter, ProgressIterator, ProgressState};
 
 /// A progress bar or spinner
 ///
@@ -19,6 +21,7 @@ use crate::{ProgressBarIter, ProgressIterator};
 pub struct ProgressBar {
     state: Arc<Mutex<BarState>>,
     pos: Arc<AtomicPosition>,
+    ticker: Arc<Mutex<Option<Ticker>>>,
 }
 
 impl fmt::Debug for ProgressBar {
@@ -51,6 +54,7 @@ impl ProgressBar {
         ProgressBar {
             state: Arc::new(Mutex::new(BarState::new(len, draw_target, pos.clone()))),
             pos,
+            ticker: Arc::new(Mutex::new(None)),
         }
     }
 
@@ -132,19 +136,35 @@ impl ProgressBar {
     /// When steady ticks are enabled, calling [`ProgressBar::tick()`] on a progress bar does not
     /// have any effect.
     pub fn enable_steady_tick(&self, interval: Duration) {
-        Ticker::spawn(&self.state, interval)
+        if interval.is_zero() {
+            return;
+        }
+
+        self.stop_and_replace_ticker(Some(interval));
     }
 
     /// Undoes [`ProgressBar::enable_steady_tick()`]
     pub fn disable_steady_tick(&self) {
-        self.state().ticker = None;
+        self.stop_and_replace_ticker(None);
+    }
+
+    fn stop_and_replace_ticker(&self, interval: Option<Duration>) {
+        let mut ticker_state = self.ticker.lock().unwrap();
+        if let Some(ticker) = ticker_state.take() {
+            ticker.stop();
+        }
+
+        *ticker_state = interval.map(|interval| Ticker::new(interval, &self.state));
     }
 
     /// Manually ticks the spinner or progress bar
     ///
     /// This automatically happens on any other change to a progress bar.
     pub fn tick(&self) {
-        self.state().tick(Instant::now())
+        // Only tick if a `Ticker` isn't installed
+        if self.ticker.lock().unwrap().is_none() {
+            self.state().tick(Instant::now())
+        }
     }
 
     /// Advances the position of the progress bar by `delta`
@@ -226,6 +246,7 @@ impl ProgressBar {
         WeakProgressBar {
             state: Arc::downgrade(&self.state),
             pos: Arc::downgrade(&self.pos),
+            ticker: Arc::downgrade(&self.ticker),
         }
     }
 
@@ -489,6 +510,7 @@ impl ProgressBar {
 pub struct WeakProgressBar {
     state: Weak<Mutex<BarState>>,
     pos: Weak<AtomicPosition>,
+    ticker: Weak<Mutex<Option<Ticker>>>,
 }
 
 impl WeakProgressBar {
@@ -506,10 +528,102 @@ impl WeakProgressBar {
     pub fn upgrade(&self) -> Option<ProgressBar> {
         let state = self.state.upgrade()?;
         let pos = self.pos.upgrade()?;
-        Some(ProgressBar { state, pos })
+        let ticker = self.ticker.upgrade()?;
+        Some(ProgressBar { state, pos, ticker })
+    }
+}
+
+pub(crate) struct Ticker {
+    stopping: Arc<(Mutex<bool>, Condvar)>,
+    join_handle: Option<thread::JoinHandle<()>>,
+}
+
+impl Drop for Ticker {
+    fn drop(&mut self) {
+        self.stop();
+        self.join_handle.take().map(|handle| handle.join());
     }
 }
 
+#[cfg(test)]
+static TICKER_RUNNING: AtomicBool = AtomicBool::new(false);
+
+impl Ticker {
+    pub(crate) fn new(interval: Duration, bar_state: &Arc<Mutex<BarState>>) -> Self {
+        debug_assert!(!interval.is_zero());
+
+        // A `Mutex<bool>` is used as a flag to indicate whether the ticker was requested to stop.
+        // The `Condvar` is used a notification mechanism: when the ticker is dropped, we notify
+        // the thread and interrupt the ticker wait.
+        #[allow(clippy::mutex_atomic)]
+        let stopping = Arc::new((Mutex::new(false), Condvar::new()));
+        let control = TickerControl {
+            stopping: stopping.clone(),
+            state: Arc::downgrade(bar_state),
+        };
+
+        let join_handle = thread::spawn(move || control.run(interval));
+        Self {
+            stopping,
+            join_handle: Some(join_handle),
+        }
+    }
+
+    pub(crate) fn stop(&self) {
+        *self.stopping.0.lock().unwrap() = true;
+        self.stopping.1.notify_one();
+    }
+}
+
+struct TickerControl {
+    stopping: Arc<(Mutex<bool>, Condvar)>,
+    state: Weak<Mutex<BarState>>,
+}
+
+impl TickerControl {
+    fn run(&self, interval: Duration) {
+        #[cfg(test)]
+        TICKER_RUNNING.store(true, Ordering::SeqCst);
+
+        while let Some(arc) = self.state.upgrade() {
+            let mut state = arc.lock().unwrap();
+            if state.state.is_finished() {
+                break;
+            }
+
+            if state.state.tick != 0 {
+                state.state.tick = state.state.tick.saturating_add(1);
+            }
+
+            state.draw(false, Instant::now()).ok();
+
+            drop(state); // Don't forget to drop the lock before sleeping
+            drop(arc); // Also need to drop Arc otherwise BarState won't be dropped
+
+            // Wait for `interval` but return early if we are notified to stop
+            let (_, result) = self
+                .stopping
+                .1
+                .wait_timeout_while(self.stopping.0.lock().unwrap(), interval, |stopped| {
+                    !*stopped
+                })
+                .unwrap();
+
+            // If the wait didn't time out, it means we were notified to stop
+            if !result.timed_out() {
+                break;
+            }
+        }
+
+        #[cfg(test)]
+        TICKER_RUNNING.store(false, Ordering::SeqCst);
+    }
+}
+
+// Tests using the global TICKER_RUNNING flag need to be serialized
+#[cfg(test)]
+static TICKER_TEST: Lazy<Mutex<()>> = Lazy::new(Mutex::default);
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -574,4 +688,42 @@ mod tests {
         io::copy(&mut reader, &mut writer).unwrap();
         assert_eq!(writer.it, bytes);
     }
+
+    #[test]
+    fn ticker_thread_terminates_on_drop() {
+        let _guard = TICKER_TEST.lock().unwrap();
+        assert!(!TICKER_RUNNING.load(Ordering::SeqCst));
+
+        let pb = ProgressBar::new_spinner();
+        pb.enable_steady_tick(Duration::from_millis(50));
+
+        // Give the thread time to start up
+        thread::sleep(Duration::from_millis(250));
+
+        assert!(TICKER_RUNNING.load(Ordering::SeqCst));
+
+        drop(pb);
+        assert!(!TICKER_RUNNING.load(Ordering::SeqCst));
+    }
+
+    #[test]
+    fn ticker_thread_terminates_on_drop_2() {
+        let _guard = TICKER_TEST.lock().unwrap();
+        assert!(!TICKER_RUNNING.load(Ordering::SeqCst));
+
+        let pb = ProgressBar::new_spinner();
+        pb.enable_steady_tick(Duration::from_millis(50));
+        let pb2 = pb.clone();
+
+        // Give the thread time to start up
+        thread::sleep(Duration::from_millis(250));
+
+        assert!(TICKER_RUNNING.load(Ordering::SeqCst));
+
+        drop(pb);
+        assert!(TICKER_RUNNING.load(Ordering::SeqCst));
+
+        drop(pb2);
+        assert!(!TICKER_RUNNING.load(Ordering::SeqCst));
+    }
 }

src/rayon.rs

@@ -1,10 +1,10 @@
-use crate::{ProgressBar, ProgressBarIter};
-use rayon::iter::{
-    plumbing::{Consumer, Folder, Producer, ProducerCallback, UnindexedConsumer},
-    IndexedParallelIterator, ParallelIterator,
-};
 use std::convert::TryFrom;
 
+use rayon::iter::plumbing::{Consumer, Folder, Producer, ProducerCallback, UnindexedConsumer};
+use rayon::iter::{IndexedParallelIterator, ParallelIterator};
+
+use crate::{ProgressBar, ProgressBarIter};
+
 /// Wraps a Rayon parallel iterator.
 ///
 /// See [`ProgressIterator`](trait.ProgressIterator.html) for method
@@ -210,10 +210,10 @@ impl<S: Send, T: ParallelIterator<Item = S>> ParallelIterator for ProgressBarIte
 
 #[cfg(test)]
 mod test {
-    use crate::ProgressStyle;
-    use crate::{ParallelProgressIterator, ProgressBar, ProgressBarIter};
     use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 
+    use crate::{ParallelProgressIterator, ProgressBar, ProgressBarIter, ProgressStyle};
+
     #[test]
     fn it_can_wrap_a_parallel_iterator() {
         let v = vec![1, 2, 3];