Fix #13449: add 'blocking' parameter to sched.scheduler.run() so that the scheduler can be used in non-blocking applications

2 files changed, 15 insertions, 1 deletions

diff --git a/Lib/sched.py b/Lib/sched.py
index 5c4a7b6..5292fcf 100644
--- a/Lib/sched.py
+++ b/Lib/sched.py
@@ -91,8 +91,10 @@ class scheduler:
         with self._lock:
             return not self._queue
 
-    def run(self):
+    def run(self, blocking=True):
         """Execute events until the queue is empty.
+        If blocking is False executes the scheduled events due to
+        expire soonest (if any) and then return.
 
         When there is a positive delay until the first event, the
         delay function is called and the event is left in the queue;
@@ -123,6 +125,8 @@ class scheduler:
                 time, priority, action, argument, kwargs = checked_event = q[0]
                 now = timefunc()
                 if now < time:
+                    if not blocking:
+                        return
                     delayfunc(time - now)
                 else:
                     event = pop(q)
diff --git a/Lib/test/test_sched.py b/Lib/test/test_sched.py
index f5e159d..ae82f94 100644
--- a/Lib/test/test_sched.py
+++ b/Lib/test/test_sched.py
@@ -86,6 +86,16 @@ class TestCase(unittest.TestCase):
         scheduler.run()
         self.assertEqual(flag, [None])
 
+    def test_run_non_blocking(self):
+        l = []
+        fun = lambda x: l.append(x)
+        scheduler = sched.scheduler(time.time, time.sleep)
+        for x in [10, 9, 8, 7, 6]:
+            scheduler.enter(x, 1, fun, (x,))
+        scheduler.run(blocking=False)
+        self.assertEqual(l, [])
+
+
 def test_main():
     support.run_unittest(TestCase)