asyncio: Improve canceled timer handles cleanup. Closes issue #22448.
Patch by Joshua Moore-Oliva.
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index db13250..5aaf58f 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -40,6 +40,13 @@
# Argument for default thread pool executor creation.
_MAX_WORKERS = 5
+# Minimum number of _scheduled timer handles before cleanup of
+# cancelled handles is performed.
+_MIN_SCHEDULED_TIMER_HANDLES = 100
+
+# Minimum fraction of _scheduled timer handles that are cancelled
+# before cleanup of cancelled handles is performed.
+_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5
def _format_handle(handle):
cb = handle._callback
@@ -145,6 +152,7 @@
class BaseEventLoop(events.AbstractEventLoop):
def __init__(self):
+ self._timer_cancelled_count = 0
self._closed = False
self._ready = collections.deque()
self._scheduled = []
@@ -349,6 +357,7 @@
if timer._source_traceback:
del timer._source_traceback[-1]
heapq.heappush(self._scheduled, timer)
+ timer._scheduled = True
return timer
def call_soon(self, callback, *args):
@@ -964,16 +973,19 @@
assert isinstance(handle, events.Handle), 'A Handle is required here'
if handle._cancelled:
return
- if isinstance(handle, events.TimerHandle):
- heapq.heappush(self._scheduled, handle)
- else:
- self._ready.append(handle)
+ assert not isinstance(handle, events.TimerHandle)
+ self._ready.append(handle)
def _add_callback_signalsafe(self, handle):
"""Like _add_callback() but called from a signal handler."""
self._add_callback(handle)
self._write_to_self()
+ def _timer_handle_cancelled(self, handle):
+ """Notification that a TimerHandle has been cancelled."""
+ if handle._scheduled:
+ self._timer_cancelled_count += 1
+
def _run_once(self):
"""Run one full iteration of the event loop.
@@ -981,9 +993,26 @@
schedules the resulting callbacks, and finally schedules
'call_later' callbacks.
"""
- # Remove delayed calls that were cancelled from head of queue.
- while self._scheduled and self._scheduled[0]._cancelled:
- heapq.heappop(self._scheduled)
+
+ # Remove delayed calls that were cancelled if their number is too high
+ sched_count = len(self._scheduled)
+ if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
+ self._timer_cancelled_count / sched_count >
+ _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
+ for handle in self._scheduled:
+ if handle._cancelled:
+ handle._scheduled = False
+
+ self._scheduled = [x for x in self._scheduled if not x._cancelled]
+ self._timer_cancelled_count = 0
+
+ heapq.heapify(self._scheduled)
+ else:
+ # Remove delayed calls that were cancelled from head of queue.
+ while self._scheduled and self._scheduled[0]._cancelled:
+ self._timer_cancelled_count -= 1
+ handle = heapq.heappop(self._scheduled)
+ handle._scheduled = False
timeout = None
if self._ready:
@@ -1024,6 +1053,7 @@
if handle._when >= end_time:
break
handle = heapq.heappop(self._scheduled)
+ handle._scheduled = False
self._ready.append(handle)
# This is the only place where callbacks are actually *called*.
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index b7cc351..806218f 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -105,14 +105,15 @@
return '<%s>' % ' '.join(info)
def cancel(self):
- self._cancelled = True
- if self._loop.get_debug():
- # Keep a representation in debug mode to keep callback and
- # parameters. For example, to log the warning "Executing <Handle
- # ...> took 2.5 second"
- self._repr = repr(self)
- self._callback = None
- self._args = None
+ if not self._cancelled:
+ self._cancelled = True
+ if self._loop.get_debug():
+ # Keep a representation in debug mode to keep callback and
+ # parameters. For example, to log the warning
+ # "Executing <Handle...> took 2.5 second"
+ self._repr = repr(self)
+ self._callback = None
+ self._args = None
def _run(self):
try:
@@ -134,7 +135,7 @@
class TimerHandle(Handle):
"""Object returned by timed callback registration methods."""
- __slots__ = ['_when']
+ __slots__ = ['_scheduled', '_when']
def __init__(self, when, callback, args, loop):
assert when is not None
@@ -142,6 +143,7 @@
if self._source_traceback:
del self._source_traceback[-1]
self._when = when
+ self._scheduled = False
def _repr_info(self):
info = super()._repr_info()
@@ -180,6 +182,11 @@
equal = self.__eq__(other)
return NotImplemented if equal is NotImplemented else not equal
+ def cancel(self):
+ if not self._cancelled:
+ self._loop._timer_handle_cancelled(self)
+ super().cancel()
+
class AbstractServer:
"""Abstract server returned by create_server()."""
@@ -238,6 +245,10 @@
# Methods scheduling callbacks. All these return Handles.
+ def _timer_handle_cancelled(self, handle):
+ """Notification that a TimerHandle has been cancelled."""
+ raise NotImplementedError
+
def call_soon(self, callback, *args):
return self.call_later(0, callback, *args)