facebook/tornado — Issue #408: [feature] scheduled functions "cleaner"
Author: giampaolo
Created: 2011-12-02 09:17:33+00:00
State: open

This is part of an enhancement I did for pyftpdlib internal scheduler:
http://code.google.com/p/pyftpdlib/issues/detail?id=189
...and which I also proposed as a patch for sched stdlib module:
http://bugs.python.org/issue13451

The problem I'm proposing to solve is the following:
if your application is hugely dominated by schedules and cancels, this can be overwhelming for your memory.
Imagine to schedule 1.000.000 functions with a deadline of 1 week.
Those will be kept in memory for 1 entire week and then discarded.
The patch below attempts to mitigate this problem by re-heapifying the internal queue if the number of scheduled functions > half of the entire queue.

Note: not tested.

diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index 6fd29a0..efd086d 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -117,6 +117,7 @@ class IOLoop(object):
self._stopped = False
self._thread_ident = None
self._blocking_signal_threshold = None
+ self._cancellations = 0

# Create a pipe that we send bogus data to when we want to wake
# the I/O loop when it is idle
@@ -258,6 +259,7 @@ class IOLoop(object):
if self._timeouts[0].callback is None:
# the timeout was cancelled
heapq.heappop(self._timeouts)
+ self._cancellations -= 1
elif self._timeouts[0].deadline <= now:
timeout = heapq.heappop(self._timeouts)
self._run_callback(timeout.callback)
@@ -265,6 +267,11 @@ class IOLoop(object):
seconds = self._timeouts[0].deadline - now
poll_timeout = min(seconds, poll_timeout)
break
+ if self._cancellations > 512 \
+ and self._cancellations > (len(self._timeouts) >> 1):
+ self._cancellations = 0
+ self._timeouts = [x for x in self._timeouts if not x.cancel
+ heapq.heapify(self._timeouts)

if self._callbacks:
# If any callbacks or timeouts called add_callback,
@@ -369,6 +376,7 @@ class IOLoop(object):
# If this turns out to be a problem, we could add a garbage
# collection pass whenever there are too many dead timeouts.
timeout.callback = None
+ self._cancellations += 1

def add_callback(self, callback):
"""Calls the given callback on the next I/O loop iteration.