diff --git a/tests/test_base.py b/tests/test_base.py index 5506748d..fdfcca6a 100644 --- a/tests/test_base.py +++ b/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')): diff --git a/tests/test_process.py b/tests/test_process.py index 45036256..9c241c1e 100644 --- a/tests/test_process.py +++ b/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( diff --git a/uvloop/handles/timer.pyx b/uvloop/handles/timer.pyx index 86d46ef0..d6e43424 100644 --- a/uvloop/handles/timer.pyx +++ b/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(self._handle, __uvtimer_callback, - self.timeout, 0) + self.timeout + 1, 0) if err < 0: exc = convert_error(err) self._fatal_error(exc, True)