openstack/deb-python-eventlet
Code Issues Proposed changes

removed CancellingTimersGreenlet and Hub.cancel_timers; instead LocalTimer is used which does not fire if the greenlet it's bound to is dead;

Browse Source 
* removed Hub's track_timer method
* removed Hub.track_timer's track parameter
* twistedHub also loses timer_finished method
* commented 2 tests in coros_test.py that cannot work with the new order
* eventlet now uses unchanged greenlets!
This commit is contained in:
Denis Bilenko
2009-01-11 19:47:22 +06:00
parent 5cb3ede7ac
commit f42cdd32fa
9 changed files with 100 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
eventlet/api.py
View File

@@ -39,6 +39,7 @@ __all__ = [
'ssl_listener', 'tcp_listener', 'tcp_server', 'trampoline',
'unspew', 'use_hub', 'with_timeout', 'timeout']
Greenlet = greenlet.greenlet
class TimeoutError(Exception):
"""Exception raised if an asynchronous operation times out"""
@@ -237,24 +238,6 @@ def _spawn(g):
g.switch()
class CancellingTimersGreenlet(greenlet.greenlet):
def __init__(self, run=None, parent=None, hub=None):
self._run = run
if parent is None:
parent = greenlet.getcurrent()
if hub is None:
hub = get_hub()
self.hub = hub
greenlet.greenlet.__init__(self, None, parent)
def run(self, *args, **kwargs):
try:
return self._run(*args, **kwargs)
finally:
self.hub.cancel_timers(self, quiet=True)
def spawn(function, *args, **kwds):
"""Create a new coroutine, or cooperative thread of control, within which
to execute *function*.
@@ -271,7 +254,7 @@ def spawn(function, *args, **kwds):
"""
# killable
t = None
g = CancellingTimersGreenlet(_spawn_startup)
g = Greenlet(_spawn_startup)
t = get_hub().schedule_call_global(0, _spawn, g)
g.switch(function, args, kwds, t.cancel)
return g
@@ -294,7 +277,7 @@ def call_after_global(seconds, function, *args, **kwds):
"""
# cancellable
def startup():
g = CancellingTimersGreenlet(_spawn_startup)
g = Greenlet(_spawn_startup)
g.switch(function, args, kwds)
g.switch()
t = get_hub().schedule_call_global(seconds, startup)
@@ -313,7 +296,7 @@ def call_after_local(seconds, function, *args, **kwds):
"""
# cancellable
def startup():
g = CancellingTimersGreenlet(_spawn_startup)
g = Greenlet(_spawn_startup)
g.switch(function, args, kwds)
g.switch()
t = get_hub().schedule_call_local(seconds, startup)

2
eventlet/coros.py
View File

@@ -822,7 +822,7 @@ class CoroutinePool(pools.Pool):
sender = event()
(evt, func, args, kw) = recvd
self._safe_apply(evt, func, args, kw)
api.get_hub().cancel_timers(api.getcurrent())
#api.get_hub().cancel_timers(api.getcurrent())
# Likewise, delete these variables or else they will
# be referenced by this frame until replaced by the
# next recvd, which may or may not be a long time from

3
eventlet/httpd.py
View File

@@ -511,7 +511,8 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
body=err.body)
finally:
# clean up any timers that might have been left around by the handling code
api.get_hub().cancel_timers(api.getcurrent())
pass
#api.get_hub().cancel_timers(api.getcurrent())
# throw an exception if it failed to write a body
if not request.response_written():

48
eventlet/hubs/hub.py
View File

@@ -22,15 +22,12 @@ THE SOFTWARE.
"""
import bisect
import weakref
import sys
import socket
import errno
import traceback
import time
from eventlet.support import greenlet
from eventlet.timer import Timer
from eventlet.timer import Timer, LocalTimer
_g_debug = True
@@ -51,7 +48,6 @@ class BaseHub(object):
self.stopping = False
self.running = False
self.timers = []
self.timers_by_greenlet = {}
self.next_timers = []
self.observers = {}
self.observer_modes = {
@@ -236,26 +232,12 @@ class BaseHub(object):
# the 0 placeholder makes it easy to bisect_right using (now, 1)
self.next_timers.append((when, 0, info))
def add_timer(self, timer, track=True):
def add_timer(self, timer):
scheduled_time = self.clock() + timer.seconds
self._add_absolute_timer(scheduled_time, timer)
if track:
self.track_timer(timer)
return scheduled_time
def track_timer(self, timer):
current_greenlet = greenlet.getcurrent()
timer.greenlet = current_greenlet
self.timers_by_greenlet.setdefault(
current_greenlet,
weakref.WeakKeyDictionary())[timer] = True
def timer_finished(self, timer):
try:
del self.timers_by_greenlet[timer.greenlet][timer]
if not self.timers_by_greenlet[timer.greenlet]:
del self.timers_by_greenlet[timer.greenlet]
except (KeyError, AttributeError):
pass
def timer_canceled(self, timer):
@@ -276,8 +258,8 @@ class BaseHub(object):
*args: Arguments to pass to the callable when called.
**kw: Keyword arguments to pass to the callable when called.
"""
t = Timer(seconds, cb, *args, **kw)
self.add_timer(t, track=True)
t = LocalTimer(seconds, cb, *args, **kw)
self.add_timer(t)
return t
schedule_call = schedule_call_local
@@ -291,7 +273,7 @@ class BaseHub(object):
**kw: Keyword arguments to pass to the callable when called.
"""
t = Timer(seconds, cb, *args, **kw)
self.add_timer(t, track=False)
self.add_timer(t)
return t
def fire_timers(self, when):
@@ -311,26 +293,6 @@ class BaseHub(object):
self.timer_finished(timer)
del t[:last]
def cancel_timers(self, greenlet, quiet=False):
if greenlet not in self.timers_by_greenlet:
return
for timer in self.timers_by_greenlet[greenlet].keys():
if not timer.cancelled and not timer.called and timer.seconds:
## If timer.seconds is 0, this isn't a timer, it's
## actually eventlet's silly way of specifying whether
## a coroutine is "ready to run" or not.
try:
# this might be None due to weirdness with weakrefs
timer.cancel()
except TypeError:
pass
if _g_debug and not quiet:
print 'Hub cancelling left-over timer %s' % timer
try:
del self.timers_by_greenlet[greenlet]
except KeyError:
pass
# for debugging:
def get_readers(self):

5
eventlet/hubs/libev.py
View File

@@ -29,7 +29,6 @@ import errno
import traceback
import time
from eventlet.timer import Timer
from eventlet.hubs import hub
from eventlet.support import greenlet
@@ -112,13 +111,11 @@ class Hub(hub.BaseHub):
self.interrupted = False
raise KeyboardInterrupt()
def add_timer(self, timer, track=True):
def add_timer(self, timer):
# store the pyevent timer object so that we can cancel later
eventtimer = libev.Timer(timer.seconds, 0, self._evloop, timer)
timer.impltimer = eventtimer
eventtimer.start()
if track:
self.track_timer(timer)
def timer_finished(self, timer):
try:

4
eventlet/hubs/libevent.py
View File

@@ -118,13 +118,11 @@ class Hub(hub.BaseHub):
self.interrupted = False
raise KeyboardInterrupt()
def add_timer(self, timer, track=True):
def add_timer(self, timer):
# store the pyevent timer object so that we can cancel later
eventtimer = event.timeout(timer.seconds, timer)
timer.impltimer = eventtimer
eventtimer.add()
if track:
self.track_timer(timer)
def timer_finished(self, timer):
try:

92
eventlet/hubs/twistedr.py
View File

@@ -20,13 +20,9 @@
import sys
import threading
import weakref
from twisted.internet.base import DelayedCall as TwistedDelayedCall
from eventlet.hubs.hub import _g_debug
from eventlet.support.greenlet import greenlet
import traceback
class DelayedCall(TwistedDelayedCall):
"fix DelayedCall to behave like eventlet's Timer in some respects"
@@ -37,12 +33,28 @@ class DelayedCall(TwistedDelayedCall):
return
return TwistedDelayedCall.cancel(self)
def callLater(reactor, _seconds, _f, *args, **kw):
class LocalDelayedCall(DelayedCall):
def __init__(self, *args, **kwargs):
self.greenlet = greenlet.getcurrent()
DelayedCall.__init__(self, *args, **kwargs)
def _get_cancelled(self):
if self.greenlet is None or self.greenlet.dead:
return True
return self.__dict__['cancelled']
def _set_cancelled(self, value):
self.__dict__['cancelled'] = value
cancelled = property(_get_cancelled, _set_cancelled)
def callLater(DelayedCallClass, reactor, _seconds, _f, *args, **kw):
# the same as original but creates fixed DelayedCall instance
assert callable(_f), "%s is not callable" % _f
assert sys.maxint >= _seconds >= 0, \
"%s is not greater than or equal to 0 seconds" % (_seconds,)
tple = DelayedCall(reactor.seconds() + _seconds, _f, args, kw,
tple = DelayedCallClass(reactor.seconds() + _seconds, _f, args, kw,
reactor._cancelCallLater,
reactor._moveCallLaterSooner,
seconds=reactor.seconds)
@@ -94,7 +106,6 @@ class BaseTwistedHub(object):
def __init__(self, mainloop_greenlet):
self.greenlet = mainloop_greenlet
self.waiters_by_greenlet = {}
self.timers_by_greenlet = {}
def switch(self):
assert greenlet.getcurrent() is not self.greenlet, 'Impossible to switch() from the mainloop greenlet'
@@ -137,75 +148,18 @@ class BaseTwistedHub(object):
def schedule_call_local(self, seconds, func, *args, **kwargs):
from twisted.internet import reactor
def call_with_timer_attached(*args1, **kwargs1):
try:
def call_if_greenlet_alive(*args1, **kwargs1):
if timer.greenlet.dead:
return
return func(*args1, **kwargs1)
finally:
if seconds:
self.timer_finished(timer)
timer = callLater(reactor, seconds, call_with_timer_attached, *args, **kwargs)
if seconds:
self.track_timer(timer)
timer = callLater(LocalDelayedCall, reactor, seconds, call_if_greenlet_alive, *args, **kwargs)
return timer
schedule_call = schedule_call_local
def schedule_call_global(self, seconds, func, *args, **kwargs):
from twisted.internet import reactor
return callLater(reactor, seconds, func, *args, **kwargs)
def track_timer(self, timer):
try:
current_greenlet = greenlet.getcurrent()
timer.greenlet = current_greenlet
self.timers_by_greenlet.setdefault(
current_greenlet,
weakref.WeakKeyDictionary())[timer] = True
except:
print 'track_timer failed'
traceback.print_exc()
raise
def timer_finished(self, timer):
try:
greenlet = timer.greenlet
del self.timers_by_greenlet[greenlet][timer]
if not self.timers_by_greenlet[greenlet]:
del self.timers_by_greenlet[greenlet]
except (AttributeError, KeyError):
pass
except:
print 'timer_finished failed'
traceback.print_exc()
raise
def cancel_timers(self, greenlet, quiet=False):
try:
if greenlet not in self.timers_by_greenlet:
return
for timer in self.timers_by_greenlet[greenlet].keys():
if not timer.cancelled and not timer.called and hasattr(timer, 'greenlet'):
## If timer.seconds is 0, this isn't a timer, it's
## actually eventlet's silly way of specifying whether
## a coroutine is "ready to run" or not.
## TwistedHub: I do the same, by not attaching 'greenlet' attribute to zero-timers QQQ
try:
# this might be None due to weirdness with weakrefs
timer.cancel()
except TypeError:
pass
if _g_debug and not quiet:
print 'Hub cancelling left-over timer %s' % timer
try:
del self.timers_by_greenlet[greenlet]
except KeyError:
pass
except:
print 'cancel_timers failed'
import traceback
traceback.print_exc()
if not quiet:
raise
return callLater(DelayedCall, reactor, seconds, func, *args, **kwargs)
def abort(self):
from twisted.internet import reactor

38
eventlet/timer.py
View File

@@ -22,7 +22,7 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
from eventlet.api import get_hub
from eventlet.api import get_hub, getcurrent
""" If true, captures a stack trace for each timer when constructed. This is
useful for debugging leaking timers, to find out where the timer was set up. """
@@ -40,7 +40,7 @@ class Timer(object):
This timer will not be run unless it is scheduled in a runloop by
calling timer.schedule() or runloop.add_timer(timer).
"""
self.cancelled = False
self._cancelled = False
self.seconds = seconds
self.tpl = cb, args, kw
self.called = False
@@ -49,6 +49,10 @@ class Timer(object):
self.traceback = cStringIO.StringIO()
traceback.print_stack(file=self.traceback)
@property
def cancelled(self):
return self._cancelled
def __repr__(self):
secs = getattr(self, 'seconds', None)
cb, args, kw = getattr(self, 'tpl', (None, None, None))
@@ -82,10 +86,38 @@ class Timer(object):
"""Prevent this timer from being called. If the timer has already
been called, has no effect.
"""
self.cancelled = True
self._cancelled = True
self.called = True
get_hub().timer_canceled(self)
try:
del self.tpl
except AttributeError:
pass
class LocalTimer(Timer):
def __init__(self, *args, **kwargs):
self.greenlet = getcurrent()
Timer.__init__(self, *args, **kwargs)
@property
def cancelled(self):
if self.greenlet is None or self.greenlet.dead:
return True
return self._cancelled
def __call__(self, *args):
if not self.called:
self.called = True
if self.greenlet is not None and self.greenlet.dead:
return
cb, args, kw = self.tpl
try:
cb(*args, **kw)
finally:
get_hub().timer_finished(self)
def cancel(self):
self.greenlet = None
Timer.cancel(self)

42
greentest/coros_test.py
View File

@@ -65,15 +65,18 @@ class TestEvent(tests.TestCase):
self.assertEqual(len(results), count)
def test_cancel(self):
evt = coros.event()
# close over the current coro so we can cancel it explicitly
current = api.getcurrent()
def cancel_event():
evt.cancel(current)
api.spawn(cancel_event)
self.assertRaises(coros.Cancelled, evt.wait)
# commented out, not fixed because it's unclear what event.cancel(waiter) should do
# (docstring and the code say different things) and because cancel() as implemented now
# has a bug
# def test_cancel(self):
# evt = coros.event()
# # close over the current coro so we can cancel it explicitly
# current = api.getcurrent()
# def cancel_event():
# evt.cancel(current)
# api.spawn(cancel_event)
#
# self.assertRaises(coros.Cancelled, evt.wait)
def test_reset(self):
evt = coros.event()
@@ -154,16 +157,17 @@ class TestCoroutinePool(tests.TestCase):
done.wait()
self.assertEquals(['cons1', 'prod', 'cons2'], results)
def test_timer_cancel(self):
def some_work():
t = timer.Timer(5, lambda: None)
t.schedule()
return t
pool = coros.CoroutinePool(0, 2)
worker = pool.execute(some_work)
t = worker.wait()
api.sleep(0)
self.assertEquals(t.cancelled, True)
# since CoroutinePool does not kill the greenlet, the following does not work
# def test_timer_cancel(self):
# def some_work():
# t = timer.LocalTimer(5, lambda: None)
# t.schedule()
# return t
# pool = coros.CoroutinePool(0, 2)
# worker = pool.execute(some_work)
# t = worker.wait()
# api.sleep(0)
# self.assertEquals(t.cancelled, True)
def test_reentrant(self):
pool = coros.CoroutinePool(0,1)