Fix timers firing up to 1ms early

tests/test_base.py

@@ -220,6 +220,24 @@ def cb():
             finished = int(round(self.loop.time() * 1000))
             self.assertGreaterEqual(finished - started, 69)
 
+    def test_call_later_not_early(self):
+        # Refs #739: asyncio.sleep()/call_later() must never fire before
+        # the requested delay has actually elapsed, as measured by a
+        # wall-clock source independent of the loop's own (millisecond-
+        # truncated) internal clock. The underlying bug only manifests
+        # when the loop's internal clock happens to be sampled close to
+        # a millisecond boundary, so run enough iterations to make a
+        # regression here reliably observable.
+        async def main():
+            delay = 0.01
+            for _ in range(300):
+                started = time.monotonic()
+                await asyncio.sleep(delay)
+                elapsed = time.monotonic() - started
+                self.assertGreaterEqual(elapsed, delay)
+
+        self.loop.run_until_complete(main())
+
     def test_call_at(self):
         if (os.environ.get('TRAVIS_OS_NAME')
                 or os.environ.get('GITHUB_WORKFLOW')):

tests/test_process.py

@@ -887,6 +887,7 @@ class TestProto:
         def __init__(self):
             self.lost = 0
             self.stages = []
+            self.connection_lost_fut = asyncio.Future()
 
         def connection_made(self, transport):
             self.stages.append(('CM', transport))
@@ -905,6 +906,8 @@ def process_exited(self):
         def connection_lost(self, exc):
             self.stages.append(('CL', self.lost, exc))
             self.lost += 1
+            if not self.connection_lost_fut.done():
+                self.connection_lost_fut.set_result(None)
 
     async def run_sub(self, **kwargs):
         return await self.loop.subprocess_shell(
@@ -962,7 +965,16 @@ def test_process_delayed_stdio__not_paused__no_stdin(self):
                 stdin=None,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE))
-        self.loop.run_until_complete(transport._wait())
+        # transport._wait() only waits for the child process to be reaped
+        # (SIGCHLD/waitpid); it says nothing about whether the stdout pipe
+        # has been drained and closed yet. Process exit and pipe EOF are
+        # delivered to libuv via independent kernel mechanisms (a signal
+        # and epoll readiness, respectively), so they can be observed in
+        # separate event loop iterations. Waiting on the transport's own
+        # connection_lost (which uvloop only fires once *both* the process
+        # has exited *and* all pipes have disconnected) avoids asserting on
+        # proto.stages before that sequence has actually finished.
+        self.loop.run_until_complete(proto.connection_lost_fut)
         self.assertEqual(transport.get_returncode(), 0)
         self.assertIsNot(transport, None)
         self.assertEqual(

uvloop/handles/timer.pyx

@@ -50,9 +50,19 @@ cdef class UVTimer(UVHandle):
             uv.uv_update_time(self._loop.uvloop)  # void
             self.start_t = uv.uv_now(self._loop.uvloop)
 
+            # libuv's internal clock (loop->time, just refreshed above) is
+            # the current monotonic time truncated ("floored") to whole
+            # milliseconds, discarding up to just under 1ms. uv_timer_start()
+            # schedules the callback for loop->time + timeout, so without
+            # compensation the timer can fire up to ~1ms sooner than
+            # `self.timeout` milliseconds of actual wall-clock time have
+            # elapsed. Pad the timeout given to libuv by 1ms so the timer
+            # never fires earlier than requested; self.timeout (and thus
+            # get_when()/TimerHandle.when()) is left untouched so it keeps
+            # reporting the originally requested deadline.
             err = uv.uv_timer_start(<uv.uv_timer_t*>self._handle,
                                     __uvtimer_callback,
-                                    self.timeout, 0)
+                                    self.timeout + 1, 0)
             if err < 0:
                 exc = convert_error(err)
                 self._fatal_error(exc, True)