import asyncio import collections import time def measure_accuracy_cb(fut, t0): dt = time.perf_counter() - t0 fut.set_result(dt) @asyncio.coroutine def measure_accuracy(deltas, nruns=10): for run in range(1, nruns+1): for unit, format, multiply in ( (1, "%.3f ms", 1e3), # seconds (1e-1, "%.1f ms", 1e3), # milliseconds (1e-2, "%.1f ms", 1e3), # milliseconds (1e-3, "%.1f ms", 1e3), # milliseconds (1e-4, "%.1f us", 1e6), # milliseconds (1e-6, "%.1f us", 1e6), # microsecond (1e-9, "%.4f us", 1e6), # microsecond ): for factor in (0.1, 0.5, 1.0, 1.5, 2.0, 3.0, 4.0, 5.0): fut = asyncio.Future() delay = unit * factor print("Test %s (run #%s/%s)" % (delay, run, nruns)) t0 = time.perf_counter() loop.call_later(delay, measure_accuracy_cb, fut, t0) yield from asyncio.wait_for(fut, None) dt = fut.result() - delay deltas[(delay, format, multiply)].append(dt) loop = asyncio.get_event_loop() deltas = collections.defaultdict(list) loop.run_until_complete(measure_accuracy(deltas)) print("") print("[ asyncio accuracy ]") all_dt = [] for key, dt in sorted(deltas.items()): delay, format, multiply = key dt_min = min(dt) dt_max = max(dt) percent_min = dt_min * 100. / delay percent_max = dt_max * 100. / delay delay = format % (delay * multiply) dt_min = format % (dt_min * multiply) dt_max = format % (dt_max * multiply) print("call_later(%s): difference in [%s; %s], [%+.1f%%, %+.1f%%]" % (delay, dt_min, dt_max, percent_min, percent_max)) all_dt.extend(dt) dt_min = min(all_dt) dt_max = max(all_dt) print("Global accuracy in [%.1f us; %.1f ms]" % (dt_min * 1e6, dt_max * 1e3)) print("") if hasattr(loop, '_granularity'): print("Loop granularity: %.1f ms" % (loop._granularity * 1e3)) if hasattr(loop._selector, 'resolution'): print("Selector resolution: %.1f ms" % (loop._selector.resolution * 1e3)) print("Event loop: %s" % loop.__class__.__name__) print("Selector: %s" % loop._selector.__class__.__name__) loop.close()