openstack/deb-python-eventlet
Code Issues Proposed changes

New module convenience; moved convenience functions in there. Sectioned off the basic_usage document a little differently to highlight the convenience functions. Wrote a bunch more serve tests.

Browse Source
This commit is contained in:
Ryan Williams
2010-02-25 17:55:52 -05:00
parent 4d9ef644c2
commit 74c3c8fe9e
7 changed files with 251 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
doc/basic_usage.rst
View File

@@ -15,6 +15,9 @@ The design goal for Eventlet's API is simplicity and readability. You should be
Though Eventlet has many modules, much of the most-used stuff is accessible simply by doing ``import eventlet``. Here's a quick summary of the functionality available in the ``eventlet`` module, with links to more verbose documentation on each.
Greenthread Spawn
-----------------------
.. function:: eventlet.spawn(func, *args, **kw)
This launches a greenthread to call *func*. Spawning off multiple greenthreads gets work done in parallel. The return value from ``spawn`` is a :class:`greenthread.GreenThread` object, which can be used to retrieve the return value of *func*. See :func:`spawn <eventlet.greenthread.spawn>` for more details.
@@ -27,14 +30,13 @@ Though Eventlet has many modules, much of the most-used stuff is accessible simp
Spawns *func* after *seconds* have elapsed; a delayed version of :func:`spawn`. To abort the spawn and prevent *func* from being called, call :meth:`GreenThread.cancel` on the return value of :func:`spawn_after`. See :func:`spawn_after <eventlet.greenthread.spawn_after>` for more details.
Greenthread Control
-----------------------
.. function:: eventlet.sleep(seconds=0)
Suspends the current greenthread and allows others a chance to process. See :func:`sleep <eventlet.greenthread.sleep>` for more details.
.. autofunction:: eventlet.connect
.. autofunction:: eventlet.listen
.. class:: eventlet.GreenPool
Pools control concurrency. It's very common in applications to want to consume only a finite amount of memory, or to restrict the amount of connections that one part of the code holds open so as to leave more for the rest, or to behave consistently in the face of unpredictable input data. GreenPools provide this control. See :class:`GreenPool <eventlet.greenpool.GreenPool>` for more on how to use these.
@@ -54,6 +56,9 @@ Though Eventlet has many modules, much of the most-used stuff is accessible simp
Timeout objects are context managers, and so can be used in with statements.
See :class:`Timeout <eventlet.timeout.Timeout>` for more details.
Patching Functions
---------------------
.. function:: eventlet.import_patched(modulename, *additional_modules, **kw_additional_modules)
Imports a module in a way that ensures that the module uses "green" versions of the standard library modules, so that everything works nonblockingly. The only required argument is the name of the module to be imported. For more information see :ref:`import-green`.
@@ -62,6 +67,15 @@ Though Eventlet has many modules, much of the most-used stuff is accessible simp
Globally patches certain system modules to be greenthread-friendly. The keyword arguments afford some control over which modules are patched. If *all* is True, then all modules are patched regardless of the other arguments. If it's False, then the rest of the keyword arguments control patching of specific subsections of the standard library. Most patch the single module of the same name (os, time, select). The exceptions are socket, which also patches the ssl module if present; and thread, which patches thread, threading, and Queue. It's safe to call monkey_patch multiple times. For more information see :ref:`monkey-patch`.
Network Convenience Functions
------------------------------
.. autofunction:: eventlet.connect
.. autofunction:: eventlet.listen
.. autofunction:: eventlet.serve
.. autofunction:: eventlet.StopServe
These are the basic primitives of Eventlet; there are a lot more out there in the other Eventlet modules; check out the :doc:`modules`.

8
eventlet/__init__.py
View File

@@ -7,7 +7,7 @@ try:
from eventlet import queue
from eventlet import timeout
from eventlet import patcher
from eventlet import greenio
from eventlet import convenience
import greenlet
sleep = greenthread.sleep
@@ -27,8 +27,10 @@ try:
import_patched = patcher.import_patched
monkey_patch = patcher.monkey_patch
connect = greenio.connect
listen = greenio.listen
connect = convenience.connect
listen = convenience.listen
serve = convenience.serve
StopServe = convenience.StopServe
getcurrent = greenlet.getcurrent

114
eventlet/convenience.py Normal file
View File

@@ -0,0 +1,114 @@
import sys
from eventlet import greenio
from eventlet import greenthread
from eventlet import greenpool
from eventlet.green import socket
from eventlet.support import greenlets as greenlet
def connect(addr, family=socket.AF_INET, bind=None):
"""Convenience function for opening client sockets.
:param addr: Address of the server to connect to. For TCP sockets, this is a (host, port) tuple.
:param family: Socket family, optional. See :mod:`socket` documentation for available families.
:param bind: Local address to bind to, optional.
:return: The connected green socket object.
"""
sock = socket.socket(family, socket.SOCK_STREAM)
if bind is not None:
sock.bind(bind)
sock.connect(addr)
return sock
def listen(addr, family=socket.AF_INET, backlog=50):
"""Convenience function for opening server sockets. This
socket can be used in :func:`~eventlet.serve` or a custom ``accept()`` loop.
Sets SO_REUSEADDR on the socket to save on annoyance.
:param addr: Address to listen on. For TCP sockets, this is a (host, port) tuple.
:param family: Socket family, optional. See :mod:`socket` documentation for available families.
:param backlog: The maximum number of queued connections. Should be at least 1; the maximum value is system-dependent.
:return: The listening green socket object.
"""
sock = socket.socket(family, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(addr)
sock.listen(backlog)
return sock
class StopServe(Exception):
"""Exception class used for quitting :func:`~eventlet.serve` gracefully."""
pass
def _stop_checker(t, server_gt, conn):
try:
try:
t.wait()
finally:
conn.close()
except greenlet.GreenletExit:
pass
except Exception:
greenthread.kill(server_gt, *sys.exc_info())
def serve(sock, handle, concurrency=1000):
"""Runs a server on the supplied socket. Calls the function *handle* in a
separate greenthread for every incoming client connection. *handle* takes
two arguments: the client socket object, and the client address::
def myhandle(client_sock, client_addr):
print "client connected", client_addr
eventlet.serve(eventlet.listen(('127.0.0.1', 9999)), myhandle)
Returning from *handle* closes the client socket.
:func:`serve` blocks the calling greenthread; it won't return until
the server completes. If you desire an immediate return,
spawn a new greenthread for :func:`serve`.
Any uncaught exceptions raised in *handle* are raised as exceptions
from :func:`serve`, terminating the server, so be sure to be aware of the
exceptions your application can raise. The return value of *handle* is
ignored.
Raise a :class:`~eventlet.StopServe` exception to gracefully terminate the
server -- that's the only way to get the server() function to return rather
than raise.
The value in *concurrency* controls the maximum number of
greenthreads that will be open at any time handling requests. When
the server hits the concurrency limit, it stops accepting new
connections until the existing ones complete.
"""
pool = greenpool.GreenPool(concurrency)
server_gt = greenthread.getcurrent()
while True:
try:
conn, addr = sock.accept()
gt = pool.spawn(handle, conn, addr)
gt.link(_stop_checker, server_gt, conn)
conn, addr, gt = None, None, None
except StopServe:
return
def wrap_ssl(sock, keyfile=None, certfile=None, server_side=False,
cert_reqs=None, ssl_version=None, ca_certs=None,
do_handshake_on_connect=True, suppress_ragged_eofs=True):
"""Convenience function for converting a regular socket into an SSL
socket. Has the same interface as :func:`ssl.wrap_socket`, but
works on 2.5 or earlier, using PyOpenSSL.
The preferred idiom is to call wrap_ssl directly on the creation
method, e.g., ``wrap_ssl(connect(addr))`` or
``wrap_ssl(listen(addr), server_side=True)``. This way there is
no "naked" socket sitting around to accidentally corrupt the SSL
session.
:return Green SSL object.
"""
pass

103
eventlet/greenio.py
View File

@@ -509,106 +509,3 @@ def shutdown_safe(sock):
if e[0] != errno.ENOTCONN:
raise
def connect(addr, family=socket.AF_INET, bind=None):
"""Convenience function for opening client sockets.
:param addr: Address of the server to connect to. For TCP sockets, this is a (host, port) tuple.
:param family: Socket family, optional. See :mod:`socket` documentation for available families.
:param bind: Local address to bind to, optional.
:return: The connected green socket object.
"""
sock = GreenSocket(family, socket.SOCK_STREAM)
if bind is not None:
sock.bind(bind)
sock.connect(addr)
return sock
def listen(addr, family=socket.AF_INET, backlog=50):
"""Convenience function for opening server sockets. This
socket can be used in an ``accept()`` loop.
Sets SO_REUSEADDR on the socket to save on annoyance.
:param addr: Address to listen on. For TCP sockets, this is a (host, port) tuple.
:param family: Socket family, optional. See :mod:`socket` documentation for available families.
:param backlog: The maximum number of queued connections. Should be at least 1; the maximum value is system-dependent.
:return: The listening green socket object.
"""
sock = GreenSocket(family, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(addr)
sock.listen(backlog)
return sock
def wrap_ssl(sock, keyfile=None, certfile=None, server_side=False,
cert_reqs=None, ssl_version=None, ca_certs=None,
do_handshake_on_connect=True, suppress_ragged_eofs=True):
"""Convenience function for converting a regular socket into an SSL
socket. Has the same interface as :func:`ssl.wrap_socket`, but
works on 2.5 or earlier, using PyOpenSSL.
The preferred idiom is to call wrap_ssl directly on the creation
method, e.g., ``wrap_ssl(connect(addr))`` or
``wrap_ssl(listen(addr), server_side=True)``. This way there is
no "naked" socket sitting around to accidentally corrupt the SSL
session.
:return Green SSL object.
"""
pass
class StopServe(Exception): pass
def serve(sock, handle, concurrency=1000):
"""Runs a server on the supplied socket. Calls the function
*handle* in a separate greenthread for every incoming request with
two arguments: the client socket object, and the client address::
def myhandle(client_sock, client_addr):
print "client connected", client_addr
eventlet.serve(eventlet.listen(('127.0.0.1', 9999)), myhandle)
Returning from *handle* closes the client socket.
:func:`serve` blocks the calling greenthread; it won't return until
the server completes. If you desire an immediate return,
spawn a new greenthread for :func:`serve`.
The *handle* function must raise a StopServe exception to
gracefully terminate the server -- that's the only way to get the
server() function to return. Any other uncaught exceptions raised
in *handle* are raised as exceptions from :func:`serve`, so be
sure to do a good job catching exceptions that your application
raises. The return value of *handle* is ignored.
The value in *concurrency* controls the maximum number of
greenthreads that will be open at any time handling requests. When
the server hits the concurrency limit, it stops accepting new
connections until the existing ones complete.
"""
from eventlet import greenpool
from eventlet import greenthread
pool = greenpool.GreenPool(concurrency)
server_thread = greenthread.getcurrent()
def stop_checker(t, server_thread, conn):
try:
t.wait()
except greenthread.greenlet.GreenletExit:
pass
except Exception:
conn.close()
server_thread.throw(*sys.exc_info())
while True:
try:
conn, addr = sock.accept()
pool.spawn(handle, conn, addr).link(stop_checker, server_thread, conn)
conn, addr = None, None
except StopServe:
return

17
tests/api_test.py
View File

@@ -4,6 +4,7 @@ import socket
from unittest import TestCase, main
import warnings
import eventlet
warnings.simplefilter('ignore', DeprecationWarning)
from eventlet import api
warnings.simplefilter('default', DeprecationWarning)
@@ -30,7 +31,7 @@ class TestApi(TestCase):
private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')
def test_tcp_listener(self):
socket = greenio.listen(('0.0.0.0', 0))
socket = eventlet.listen(('0.0.0.0', 0))
assert socket.getsockname()[0] == '0.0.0.0'
socket.close()
@@ -47,10 +48,10 @@ class TestApi(TestCase):
finally:
listenfd.close()
server = greenio.listen(('0.0.0.0', 0))
server = eventlet.listen(('0.0.0.0', 0))
api.spawn(accept_once, server)
client = greenio.connect(('127.0.0.1', server.getsockname()[1]))
client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
fd = client.makefile()
client.close()
assert fd.readline() == 'hello\n'
@@ -76,7 +77,7 @@ class TestApi(TestCase):
self.private_key_file)
api.spawn(accept_once, server)
raw_client = greenio.connect(('127.0.0.1', server.getsockname()[1]))
raw_client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
client = util.wrap_ssl(raw_client)
fd = socket._fileobject(client, 'rb', 8192)
@@ -93,7 +94,7 @@ class TestApi(TestCase):
def test_001_trampoline_timeout(self):
from eventlet import coros
server_sock = greenio.listen(('127.0.0.1', 0))
server_sock = eventlet.listen(('127.0.0.1', 0))
bound_port = server_sock.getsockname()[1]
def server(sock):
client, addr = sock.accept()
@@ -101,7 +102,7 @@ class TestApi(TestCase):
server_evt = spawn(server, server_sock)
api.sleep(0)
try:
desc = greenio.connect(('127.0.0.1', bound_port))
desc = eventlet.connect(('127.0.0.1', bound_port))
api.trampoline(desc, read=True, write=False, timeout=0.001)
except api.TimeoutError:
pass # test passed
@@ -112,7 +113,7 @@ class TestApi(TestCase):
check_hub()
def test_timeout_cancel(self):
server = greenio.listen(('0.0.0.0', 0))
server = eventlet.listen(('0.0.0.0', 0))
bound_port = server.getsockname()[1]
done = [False]
@@ -122,7 +123,7 @@ class TestApi(TestCase):
conn.close()
def go():
desc = greenio.connect(('127.0.0.1', bound_port))
desc = eventlet.connect(('127.0.0.1', bound_port))
try:
api.trampoline(desc, read=True, timeout=0.1)
except api.TimeoutError:

105
tests/convenience_test.py Normal file
View File

@@ -0,0 +1,105 @@
import eventlet
from eventlet import event
from tests import LimitedTestCase
class TestServe(LimitedTestCase):
def setUp(self):
super(TestServe, self).setUp()
from eventlet import debug
debug.hub_exceptions(False)
def tearDown(self):
super(TestServe, self).tearDown()
from eventlet import debug
debug.hub_exceptions(True)
def test_exiting_server(self):
# tests that the server closes the client sock on handle() exit
def closer(sock,addr):
pass
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, closer)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertEqual('', client.recv(100))
gt.kill()
def test_excepting_server(self):
# tests that the server closes the client sock on handle() exception
def crasher(sock,addr):
x = sock.recv(1024)
0/0
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertRaises(ZeroDivisionError, gt.wait)
self.assertEqual('', client.recv(100))
def test_excepting_server_already_closed(self):
# same as above but with explicit clsoe before crash
def crasher(sock,addr):
x = sock.recv(1024)
sock.close()
0/0
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertRaises(ZeroDivisionError, gt.wait)
self.assertEqual('', client.recv(100))
def test_called_for_each_connection(self):
hits = [0]
def counter(sock, addr):
hits[0]+=1
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, counter)
for i in xrange(100):
client = eventlet.connect(('localhost', l.getsockname()[1]))
self.assertEqual('', client.recv(100))
gt.kill()
self.assertEqual(100, hits[0])
def test_blocking(self):
l = eventlet.listen(('localhost', 0))
x = eventlet.with_timeout(0.01,
eventlet.serve, l, lambda c,a: None,
timeout_value="timeout")
self.assertEqual(x, "timeout")
def test_raising_stopserve(self):
def stopit(conn, addr):
raise eventlet.StopServe()
l = eventlet.listen(('localhost', 0))
# connect to trigger a call to stopit
gt = eventlet.spawn(eventlet.connect,
('localhost', l.getsockname()[1]))
eventlet.serve(l, stopit)
gt.wait()
def test_concurrency(self):
evt = event.Event()
def waiter(sock, addr):
sock.sendall('hi')
evt.wait()
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, waiter, 5)
def test_client():
c = eventlet.connect(('localhost', l.getsockname()[1]))
# verify the client is connected by getting data
self.assertEquals('hi', c.recv(2))
return c
clients = [test_client() for i in xrange(5)]
# very next client should not get anything
x = eventlet.with_timeout(0.01,
test_client,
timeout_value="timed out")
self.assertEquals(x, "timed out")

53
tests/greenio_test.py
View File

@@ -524,58 +524,5 @@ class TestGreenIoLong(LimitedTestCase):
self.assert_(len(results2) > 0)
class TestServe(LimitedTestCase):
def setUp(self):
super(TestServe, self).setUp()
from eventlet import debug
debug.hub_exceptions(False)
def tearDown(self):
super(TestServe, self).tearDown()
from eventlet import debug
debug.hub_exceptions(True)
def test_exiting_server(self):
# tests that the server closes the client sock on handle() exit
def closer(sock,addr):
pass
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(greenio.serve, l, closer)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertEqual('', client.recv(100))
gt.kill()
def test_excepting_server(self):
# tests that the server closes the client sock on handle() exception
def crasher(sock,addr):
x = sock.recv(1024)
0/0
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(greenio.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertRaises(ZeroDivisionError, gt.wait)
self.assertEqual('', client.recv(100))
def test_excepting_server_already_closed(self):
# tests that the server closes the client sock on handle() exception
def crasher(sock,addr):
x = sock.recv(1024)
sock.close()
0/0
l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(greenio.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall('a')
self.assertRaises(ZeroDivisionError, gt.wait)
self.assertEqual('', client.recv(100))
if __name__ == '__main__':
main()