Add gthreadless (greenlet / deferred integration) and fix some module names

that conflicted with toplevel modules.
This commit is contained in:
radix
2008-03-19 20:38:10 -05:00
parent 95d45c89ab
commit 2d3eb63488
7 changed files with 151 additions and 7 deletions

View File

@@ -200,12 +200,12 @@ def get_default_hub():
pass pass
import select import select
if hasattr(select, 'poll'): if 0:#hasattr(select, 'poll'):
import eventlet.hubs.poll import eventlet.hubs.poll
return eventlet.hubs.poll return eventlet.hubs.poll
else: else:
import eventlet.hubs.select import eventlet.hubs.selects
return eventlet.hubs.select return eventlet.hubs.selects
def use_hub(mod=None): def use_hub(mod=None):

135
eventlet/gthreadless.py Normal file
View File

@@ -0,0 +1,135 @@
import greenlet
greenlet.main = greenlet.getcurrent() # WTF did greenlet.main go?
from twisted.internet import defer, reactor
def _desc(g):
if isinstance(g, DebugGreenlet):
if hasattr(g, 'name'):
desc = "<%s %s" % (g.name, hex(id(g)))
else:
desc = "<NO NAME!? %s" % (hex(id(g)), )
else:
desc = "<%s" % (hex(id(g)),)
if g is greenlet.main:
desc += " (main)"
desc += ">"
return desc
class DebugGreenlet(greenlet.greenlet):
__slots__ = ('name',)
def __init__(self, func, name=None):
super(DebugGreenlet, self).__init__(func)
self.name = name
def switch(self, *args, **kwargs):
current = greenlet.getcurrent()
#print "%s -> %s" % (_desc(current), _desc(self))
return super(DebugGreenlet, self).switch(*args, **kwargs)
def deferredGreenlet(func):
"""
I am a function decorator for functions that call blockOn. The
function I return will call the original function inside of a
greenlet, and return a Deferred.
TODO: Do a hack so the name of 'replacement' is the name of 'func'.
"""
def replacement(*args, **kwargs):
d = defer.Deferred()
def greenfunc(*args, **kwargs):
try:
d.callback(func(*args, **kwargs))
except:
d.errback()
g = greenlet.greenlet(greenfunc)
crap = g.switch(*args, **kwargs)
return d
return replacement
class CalledFromMain(Exception):
pass
class _IAmAnException(object):
def __init__(self, f):
self.f = f
def blockOn(d, desc=None):
"""
Use me in non-main greenlets to wait for a Deferred to fire.
"""
g = greenlet.getcurrent()
if g is greenlet.main:
raise CalledFromMain("You cannot call blockOn from the main greenlet.")
## Note ##
# Notice that this code catches and ignores GreenletExit. The
# greenlet mechanism sends a GreenletExit at a blocking greenlet if
# there is no chance that the greenlet will be fired by anyone
# else -- that is, no other greenlets have a reference to the one
# that's blocking.
# This is often the case with blockOn. When someone blocks on a
# Deferred, these callbacks are added to it. When the deferred
# fires, we make the blockOn() call finish -- we resume the
# blocker. At that point, the Deferred chain is irrelevant; it
# makes no sense for any other callbacks to be called. The
# Deferred, then, will likely be garbage collected and thus all
# references to our greenlet will be lost -- and thus it will have
# GreenletExit fired.
def cb(r):
try:
# This callback might be fired immediately when added
# and switching to the current greenlet seems to do nothing
# (ie. we will never actually return to the function we called
# blockOn from), so we make the call happen later in the main greenlet
# instead, if the current greenlet is the same as the one we are swithcing
# to.
if g == greenlet.getcurrent():
reactor.callLater(0, g.switch, r)
else:
g.switch(r)
except greenlet.GreenletExit:
pass
def eb(f):
try:
g.switch(_IAmAnException(f))
except greenlet.GreenletExit:
pass
d.addCallbacks(cb, eb)
x = g.parent.switch()
if isinstance(x, _IAmAnException):
x.f.raiseException()
return x
class GreenletWrapper(object):
"""Wrap an object which presents an asynchronous interface (via Deferreds).
The wrapped object will present the same interface, but all methods will
return results, rather than Deferreds.
When a Deferred would otherwise be returned, a greenlet is created and then
control is switched back to the main greenlet. When the Deferred fires,
control is switched back to the created greenlet and execution resumes with
the result.
"""
def __init__(self, wrappee):
self.wrappee = wrappee
def __getattribute__(self, name):
wrappee = super(GreenletWrapper, self).__getattribute__('wrappee')
original = getattr(wrappee, name)
if callable(original):
def wrapper(*a, **kw):
result = original(*a, **kw)
if isinstance(result, defer.Deferred):
return blockOn(result)
return result
return wrapper
return original

View File

@@ -203,7 +203,7 @@ class BaseHub(object):
def add_timer(self, timer): def add_timer(self, timer):
scheduled_time = self.clock() + timer.seconds scheduled_time = self.clock() + timer.seconds
self._add_absolute_timer(scheduled_time, timer) self._add_absolute_timer(scheduled_time, timer)
timer.greenlet = current_greenlet timer.greenlet = greenlet.getcurrent()
self.track_timer(timer) self.track_timer(timer)
return scheduled_time return scheduled_time

View File

@@ -29,6 +29,7 @@ import socket
import errno import errno
import traceback import traceback
from time import sleep from time import sleep
import time
from eventlet import greenlib from eventlet import greenlib
from eventlet.hubs import hub from eventlet.hubs import hub

View File

@@ -91,28 +91,36 @@ class EventletReactor(posixbase.PosixReactorBase):
api.get_hub().abort() api.get_hub().abort()
def addReader(self, reader): def addReader(self, reader):
print "NEW READER", reader.fileno()
fileno = reader.fileno() fileno = reader.fileno()
self._readers[fileno] = reader self._readers[fileno] = reader
api.get_hub().add_descriptor(fileno, read=self._got_read) api.get_hub().add_descriptor(fileno, read=self._got_read)
def _got_read(self, fileno): def _got_read(self, fileno):
print "got read on", fileno, self._readers[fileno]
api.get_hub().add_descriptor(fileno, read=self._got_read)
self._readers[fileno].doRead() self._readers[fileno].doRead()
def addWriter(self, writer): def addWriter(self, writer):
print "NEW WRITER", writer.fileno()
fileno = writer.fileno() fileno = writer.fileno()
self._writers[fileno] = writer self._writers[fileno] = writer
api.get_hub().add_descriptor(fileno, write=self._got_write) api.get_hub().add_descriptor(fileno, write=self._got_write)
def _got_write(self, fileno): def _got_write(self, fileno):
print "got write on", fileno, self._writers[fileno]
api.get_hub().add_descriptor(fileno, write=self._got_write)
self._writers[fileno].doWrite() self._writers[fileno].doWrite()
def removeReader(self, reader): def removeReader(self, reader):
print "removing reader", reader.fileno()
fileno = reader.fileno() fileno = reader.fileno()
if fileno in self._readers: if fileno in self._readers:
self._readers.pop(fileno) self._readers.pop(fileno)
api.get_hub().remove_descriptor(fileno) api.get_hub().remove_descriptor(fileno)
def removeWriter(self, writer): def removeWriter(self, writer):
print "removing writer", writer.fileno()
fileno = writer.fileno() fileno = writer.fileno()
if fileno in self._writers: if fileno in self._writers:
self._writers.pop(fileno) self._writers.pop(fileno)
@@ -122,7 +130,7 @@ class EventletReactor(posixbase.PosixReactorBase):
return self._removeAll(self._readers.values(), self._writers.values()) return self._removeAll(self._readers.values(), self._writers.values())
def emulate(): def install():
if not _working: if not _working:
raise RuntimeError, "Can't use support.twisted because zope.interface is not installed." raise RuntimeError, "Can't use support.twisted because zope.interface is not installed."
reactor = EventletReactor() reactor = EventletReactor()
@@ -130,5 +138,5 @@ def emulate():
installReactor(reactor) installReactor(reactor)
__all__ = ['emulate'] __all__ = ['install']