From 1f55440515e9e3e40fe0c29fbf79b4a997ac6194 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 17 Feb 2021 21:37:53 -0500
Subject: [PATCH] add Worker::step_or_park_while

---
 timely/src/worker.rs | 34 ++++++++++++++++++++++++++++++++++
 1 file changed, 34 insertions(+)

diff --git a/timely/src/worker.rs b/timely/src/worker.rs
index 62196b970..a07b9e4e4 100644
--- a/timely/src/worker.rs
+++ b/timely/src/worker.rs
@@ -392,6 +392,11 @@ impl<A: Allocate> Worker<A> {
 
     /// Calls `self.step()` as long as `func` evaluates to true.
     ///
+    /// This method will continually execute even if there is not work
+    /// for the worker to perform. Consider using the similar method
+    /// `Self::step_or_park_while(duration)` to allow the worker to yield
+    /// control if that is appropriate.
+    ///
     /// # Examples
     ///
     /// ```
@@ -414,6 +419,35 @@ impl<A: Allocate> Worker<A> {
         while func() { self.step(); }
     }
 
+    /// Calls `self.step_or_park(duration)` as long as `func` evaluates to true.
+    ///
+    /// This method may yield whenever there is no work to perform, as performed
+    /// by `Self::step_or_park()`. Please consult the documentation for further
+    /// information about that method and its behavior. In particular, the method
+    /// can park the worker indefinitely, if no new work re-awakens the worker.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// timely::execute_from_args(::std::env::args(), |worker| {
+    ///
+    ///     use timely::dataflow::operators::{ToStream, Inspect, Probe};
+    ///
+    ///     let probe =
+    ///     worker.dataflow::<usize,_,_>(|scope| {
+    ///         (0 .. 10)
+    ///             .to_stream(scope)
+    ///             .inspect(|x| println!("{:?}", x))
+    ///             .probe()
+    ///     });
+    ///
+    ///     worker.step_or_park_while(None, || probe.less_than(&0));
+    /// });
+    /// ```
+    pub fn step_or_park_while<F: FnMut()->bool>(&mut self, duration: Option<Duration>, mut func: F) {
+        while func() { self.step_or_park(duration); }
+    }
+
     /// The index of the worker out of its peers.
     ///
     /// # Examples