Repository: tornadoweb/tornado

Issue:
ioloop: PeriodicCallback executes too often on windows
## Here is the code:
import math

import logging
from crontab import CronTab
from tornado.ioloop import PeriodicCallback, IOLoop

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')


class CronTabCallback(PeriodicCallback):

    def __init__(self, callback, schedule):
        self._callback = callback
        self._crontab = CronTab(schedule)
        super(CronTabCallback, self).__init__(self.run, self._calc_callbacktime())

    def _calc_callbacktime(self, now=None):
        return math.ceil(self._crontab.next(now)) * 1000.0

    def run(self):
        return self._callback()

    def _schedule_next(self):
        self.callback_time = self._calc_callbacktime()
        logging.info('calc ---------------------')
        logging.info('delay %s' % self.callback_time)
        logging.info('last execute %s' % self._next_timeout)

        last = self._next_timeout
        super(CronTabCallback, self)._schedule_next()

        if last == self._next_timeout:
            logging.error('error !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!')

        logging.info('current: %s' % self.io_loop.time())
        logging.info('calc next: %s' % self._next_timeout)
        logging.info('--------------------------\n')


def crontab(schedule):
    def decorator(func):
        CronTabCallback(func, schedule).start()
        return func

    return decorator


@crontab('*/1 * * * *')
def run():
    logging.info('execute ... \n')


if __name__ == '__main__':
    IOLoop.current().start()

## Here is the console log
2018-03-30 11:33:00,311 - asyncio - DEBUG - Using selector: SelectSelector
2018-03-30 11:33:00,316 - root - INFO - calc ---------------------
2018-03-30 11:33:00,316 - root - INFO - delay 60000.0
2018-03-30 11:33:00,316 - root - INFO - last execute 1522380780.3169544
2018-03-30 11:33:00,316 - root - INFO - current: 1522380780.3169544
2018-03-30 11:33:00,316 - root - INFO - **calc next: 1522380840.3169544**
2018-03-30 11:33:00,316 - root - INFO - --------------------------

**2018-03-30 11:34:00,313** - root - INFO - execute ... 

2018-03-30 11:34:00,313 - root - INFO - calc ---------------------
2018-03-30 11:34:00,313 - root - INFO - delay 60000.0
2018-03-30 11:34:00,313 - root - INFO - last execute 1522380840.3169544
2018-03-30 11:34:00,313 - root - ERROR - error !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
2018-03-30 11:34:00,313 - root - INFO - current: 1522380840.3139544
2018-03-30 11:34:00,313 - root - INFO - calc next: 1522380840.3169544
2018-03-30 11:34:00,313 - root - INFO - --------------------------

2018-03-30 11:34:00,318 - root - INFO - execute ... 

2018-03-30 11:34:00,318 - root - INFO - calc ---------------------
2018-03-30 11:34:00,318 - root - INFO - delay 60000.0
2018-03-30 11:34:00,318 - root - INFO - last execute 1522380840.3169544
2018-03-30 11:34:00,318 - root - INFO - current: 1522380840.3189545
2018-03-30 11:34:00,318 - root - INFO - calc next: 1522380900.3169544
2018-03-30 11:34:00,318 - root - INFO - --------------------------

## Environment:
OS: windows 7
Python:  python 3.6
Dependent library: crontab 0.22.0

Tornado 4.5.1 python2 (ok) python3(ok)
Tornado 5.0.1 python2 (ok) python3 (linux ok, window has issue)

Discussion:

Patch:
diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index f6ec177b..48700139 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -1213,11 +1213,31 @@ class PeriodicCallback(object):
 
     def _schedule_next(self):
         if self._running:
-            current_time = self.io_loop.time()
-
-            if self._next_timeout <= current_time:
-                callback_time_sec = self.callback_time / 1000.0
-                self._next_timeout += (math.floor((current_time - self._next_timeout) /
-                                                  callback_time_sec) + 1) * callback_time_sec
-
+            self._update_next(self.io_loop.time())
             self._timeout = self.io_loop.add_timeout(self._next_timeout, self._run)
+
+    def _update_next(self, current_time):
+        callback_time_sec = self.callback_time / 1000.0
+        if self._next_timeout <= current_time:
+            # The period should be measured from the start of one call
+            # to the start of the next. If one call takes too long,
+            # skip cycles to get back to a multiple of the original
+            # schedule.
+            self._next_timeout += (math.floor((current_time - self._next_timeout) /
+                                              callback_time_sec) + 1) * callback_time_sec
+        else:
+            # If the clock moved backwards, ensure we advance the next
+            # timeout instead of recomputing the same value again.
+            # This may result in long gaps between callbacks if the
+            # clock jumps backwards by a lot, but the far more common
+            # scenario is a small NTP adjustment that should just be
+            # ignored.
+            #
+            # Note that on some systems if time.time() runs slower
+            # than time.monotonic() (most common on windows), we
+            # effectively experience a small backwards time jump on
+            # every iteration because PeriodicCallback uses
+            # time.time() while asyncio schedules callbacks using
+            # time.monotonic().
+            # https://github.com/tornadoweb/tornado/issues/2333
+            self._next_timeout += callback_time_sec


Test patch:
diff --git a/tornado/test/ioloop_test.py b/tornado/test/ioloop_test.py
index 09f71c5d..9f7c1847 100644
--- a/tornado/test/ioloop_test.py
+++ b/tornado/test/ioloop_test.py
@@ -789,6 +789,62 @@ class TestPeriodicCallback(unittest.TestCase):
         io_loop.close()
 
 
+class TestPeriodicCallbackMath(unittest.TestCase):
+    def simulate_calls(self, pc, durations):
+        """Simulate a series of calls to the PeriodicCallback.
+
+        Pass a list of call durations in seconds (negative values
+        work to simulate clock adjustments during the call, or more or
+        less equivalently, between calls). This method returns the
+        times at which each call would be made.
+        """
+        calls = []
+        now = 1000
+        pc._next_timeout = now
+        for d in durations:
+            pc._update_next(now)
+            calls.append(pc._next_timeout)
+            now = pc._next_timeout + d
+        return calls
+
+    def test_basic(self):
+        pc = PeriodicCallback(None, 10000)
+        self.assertEqual(self.simulate_calls(pc, [0] * 5),
+                         [1010, 1020, 1030, 1040, 1050])
+
+    def test_overrun(self):
+        # If a call runs for too long, we skip entire cycles to get
+        # back on schedule.
+        call_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0, 0]
+        expected = [
+            1010, 1020, 1030,  # first 3 calls on schedule
+            1050, 1070,  # next 2 delayed one cycle
+            1100, 1130,  # next 2 delayed 2 cycles
+            1170, 1210,  # next 2 delayed 3 cycles
+            1220, 1230,  # then back on schedule.
+        ]
+
+        pc = PeriodicCallback(None, 10000)
+        self.assertEqual(self.simulate_calls(pc, call_durations),
+                         expected)
+
+    def test_clock_backwards(self):
+        pc = PeriodicCallback(None, 10000)
+        # Backwards jumps are ignored, potentially resulting in a
+        # slightly slow schedule (although we assume that when
+        # time.time() and time.monotonic() are different, time.time()
+        # is getting adjusted by NTP and is therefore more accurate)
+        self.assertEqual(self.simulate_calls(pc, [-2, -1, -3, -2, 0]),
+                         [1010, 1020, 1030, 1040, 1050])
+
+        # For big jumps, we should perhaps alter the schedule, but we
+        # don't currently. This trace shows that we run callbacks
+        # every 10s of time.time(), but the first and second calls are
+        # 110s of real time apart because the backwards jump is
+        # ignored.
+        self.assertEqual(self.simulate_calls(pc, [-100, 0, 0]),
+                         [1010, 1020, 1030])
+
 class TestIOLoopConfiguration(unittest.TestCase):
     def run_python(self, *statements):
         statements = [