improve MessageSync timings

Author: zariiii9003

can/io/player.py

@@ -5,7 +5,7 @@
 """
 import gzip
 import pathlib
-from time import time, sleep
+import time
 import typing
 
 from pkg_resources import iter_entry_points
@@ -132,8 +132,9 @@ def __init__(
         self.skip = skip
 
     def __iter__(self) -> typing.Generator[Message, None, None]:
-        playback_start_time = time()
+        t_wakeup = playback_start_time = time.perf_counter()
         recorded_start_time = None
+        t_skipped = 0.0
 
         for message in self.raw_messages:
 
@@ -142,15 +143,19 @@ def __iter__(self) -> typing.Generator[Message, None, None]:
                 if recorded_start_time is None:
                     recorded_start_time = message.timestamp
 
-                now = time()
-                current_offset = now - playback_start_time
-                recorded_offset_from_start = message.timestamp - recorded_start_time
-                remaining_gap = max(0.0, recorded_offset_from_start - current_offset)
-
-                sleep_period = max(self.gap, min(self.skip, remaining_gap))
+                t_wakeup = playback_start_time + (
+                    message.timestamp - t_skipped - recorded_start_time
+                )
             else:
-                sleep_period = self.gap
+                t_wakeup += self.gap
+
+            sleep_period = t_wakeup - time.perf_counter()
+
+            if self.skip and sleep_period > self.skip:
+                t_skipped += sleep_period - self.skip
+                sleep_period = self.skip
 
-            sleep(sleep_period)
+            if sleep_period > 1e-4:
+                time.sleep(sleep_period)
 
             yield message

test/test_message_sync.py

@@ -5,7 +5,7 @@
 """
 
 from copy import copy
-from time import time
+import time
 import gc
 
 import unittest
@@ -49,43 +49,40 @@ def teardown_method(self, _):
         # we need to reenable the garbage collector again
         gc.enable()
 
-    @pytest.mark.timeout(inc(0.2))
     def test_general(self):
         messages = [
             Message(timestamp=50.0),
             Message(timestamp=50.0),
             Message(timestamp=50.0 + 0.05),
-            Message(timestamp=50.0 + 0.05 + 0.08),
+            Message(timestamp=50.0 + 0.13),
             Message(timestamp=50.0),  # back in time
         ]
-        sync = MessageSync(messages, gap=0.0)
+        sync = MessageSync(messages, gap=0.0, skip=0.0)
 
-        start = time()
+        t_start = time.perf_counter()
         collected = []
         timings = []
         for message in sync:
+            t_now = time.perf_counter()
             collected.append(message)
-            now = time()
-            timings.append(now - start)
-            start = now
+            timings.append(t_now - t_start)
 
         self.assertMessagesEqual(messages, collected)
         self.assertEqual(len(timings), len(messages), "programming error in test code")
 
-        self.assertTrue(0.0 <= timings[0] < inc(0.005), str(timings[0]))
-        self.assertTrue(0.0 <= timings[1] < inc(0.005), str(timings[1]))
-        self.assertTrue(0.045 <= timings[2] < inc(0.055), str(timings[2]))
-        self.assertTrue(0.075 <= timings[3] < inc(0.085), str(timings[3]))
-        self.assertTrue(0.0 <= timings[4] < inc(0.005), str(timings[4]))
+        self.assertTrue(0.0 <= timings[0] < 0.0 + inc(0.02), str(timings[0]))
+        self.assertTrue(0.0 <= timings[1] < 0.0 + inc(0.02), str(timings[1]))
+        self.assertTrue(0.045 <= timings[2] < 0.05 + inc(0.02), str(timings[2]))
+        self.assertTrue(0.125 <= timings[3] < 0.13 + inc(0.02), str(timings[3]))
+        self.assertTrue(0.125 <= timings[4] < 0.13 + inc(0.02), str(timings[4]))
 
-    @pytest.mark.timeout(inc(0.1) * len(TEST_FEWER_MESSAGES))  # very conservative
     def test_skip(self):
         messages = copy(TEST_FEWER_MESSAGES)
         sync = MessageSync(messages, skip=0.005, gap=0.0)
 
-        before = time()
+        before = time.perf_counter()
         collected = list(sync)
-        after = time()
+        after = time.perf_counter()
         took = after - before
 
         # the handling of the messages itself also takes some time:
@@ -96,26 +93,25 @@ def test_skip(self):
 
 
 @skip_on_unreliable_platforms
-@pytest.mark.timeout(inc(0.3))
 @pytest.mark.parametrize(
-    "timestamp_1,timestamp_2", [(0.0, 0.0), (0.0, 0.01), (0.01, 0.0)]
+    "timestamp_1,timestamp_2", [(0.0, 0.0), (0.0, 0.01), (0.01, 1.5)]
 )
 def test_gap(timestamp_1, timestamp_2):
     """This method is alone so it can be parameterized."""
     messages = [
         Message(arbitration_id=0x1, timestamp=timestamp_1),
         Message(arbitration_id=0x2, timestamp=timestamp_2),
     ]
-    sync = MessageSync(messages, gap=0.1)
+    sync = MessageSync(messages, timestamps=False, gap=0.1)
 
     gc.disable()
-    before = time()
+    before = time.perf_counter()
     collected = list(sync)
-    after = time()
+    after = time.perf_counter()
     gc.enable()
     took = after - before
 
-    assert 0.1 <= took < inc(0.3)
+    assert 0.195 <= took < 0.2 + inc(0.02)
     assert messages == collected
 
 

test/test_player.py

@@ -27,7 +27,7 @@ def setUp(self) -> None:
         self.mock_virtual_bus.__enter__ = Mock(return_value=self.mock_virtual_bus)
 
         # Patch time sleep object
-        patcher_sleep = mock.patch("can.io.player.sleep", spec=True)
+        patcher_sleep = mock.patch("can.io.player.time.sleep", spec=True)
         self.MockSleep = patcher_sleep.start()
         self.addCleanup(patcher_sleep.stop)
 
@@ -60,7 +60,6 @@ def test_play_virtual(self):
             dlc=8,
             data=[0x5, 0xC, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0],
         )
-        self.assertEqual(self.MockSleep.call_count, 2)
         if sys.version_info >= (3, 8):
             # The args argument was introduced with python 3.8
             self.assertTrue(
@@ -78,16 +77,14 @@ def test_play_virtual_verbose(self):
         self.assertIn("09 08 07 06 05 04 03 02", mock_stdout.getvalue())
         self.assertIn("05 0c 00 00 00 00 00 00", mock_stdout.getvalue())
         self.assertEqual(self.mock_virtual_bus.send.call_count, 2)
-        self.assertEqual(self.MockSleep.call_count, 2)
         self.assertSuccessfulCleanup()
 
     def test_play_virtual_exit(self):
         self.MockSleep.side_effect = [None, KeyboardInterrupt]
 
         sys.argv = self.baseargs + [self.logfile]
         can.player.main()
-        self.assertEqual(self.mock_virtual_bus.send.call_count, 1)
-        self.assertEqual(self.MockSleep.call_count, 2)
+        assert self.mock_virtual_bus.send.call_count <= 2
         self.assertSuccessfulCleanup()
 
     def test_play_skip_error_frame(self):
@@ -97,7 +94,6 @@ def test_play_skip_error_frame(self):
         sys.argv = self.baseargs + ["-v", logfile]
         can.player.main()
         self.assertEqual(self.mock_virtual_bus.send.call_count, 9)
-        self.assertEqual(self.MockSleep.call_count, 12)
         self.assertSuccessfulCleanup()
 
     def test_play_error_frame(self):
@@ -107,7 +103,6 @@ def test_play_error_frame(self):
         sys.argv = self.baseargs + ["-v", "--error-frames", logfile]
         can.player.main()
         self.assertEqual(self.mock_virtual_bus.send.call_count, 12)
-        self.assertEqual(self.MockSleep.call_count, 12)
         self.assertSuccessfulCleanup()