Merge with latest which_linden tip

AUTHORS

@@ -6,7 +6,7 @@ Original Authors
 Contributors
 ------------
 * AG Projects
-* Chris Atlee
+* Chris AtLee
 * R\. Tyler Ballance
 * Denis Bilenko
 * Mike Barton
@@ -18,6 +18,7 @@ Contributors
 * radix
 * Tavis Rudd
 * Sergey Shepelev
+* Chuck Thier
 
 Linden Lab Contributors
 -----------------------
@@ -34,7 +35,6 @@ Thanks To
 ---------
 * AdamKG, giving the hint that invalid argument errors were introduced post-0.9.0
 * Luke Tucker, bug report regarding wsgi + webob
-* Chuck Thier, reporting a bug in processes.py
 * Taso Du Val, reproing an exception squelching bug, saving children's lives  ;-)
 * Luci Stanescu, for reporting twisted hub bug
 * Marcus Cavanaugh, for test case code that has been incredibly useful in tracking down bugs

NEWS

@@ -1,3 +1,22 @@
+0.9.6
+=====
+* new EVENTLET_HUB environment variable allows you to select a hub without code
+* improved GreenSocket and GreenPipe compatibility with stdlib
+* bugfixes on GreenSocket and GreenPipe objects
+* code coverage increased across the board
+* Queue resizing
+* internal DeprecationWarnings largely eliminated
+* tpool is now reentrant (i.e., can call tpool.execute(tpool.execute(foo)))
+* more reliable access to unpatched modules reduces some race conditions when monkeypatching
+* completely threading-compatible corolocal implementation, plus tests and enthusiastic adoption
+* tests stomp on each others' toes less
+* performance improvements in timers, hubs, greenpool
+* Greenlet-aware profile module courtesy of CCP
+* support for select26 module's epoll
+* better PEP-8 compliance and import cleanup
+* new eventlet.serve convenience function for easy TCP servers
+
+
 0.9.5
 =====
 * support psycopg in db_pool

doc/basic_usage.rst

@@ -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`.

eventlet/__init__.py

@@ -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
+    connect = convenience.connect
-    listen = greenio.listen
+    listen = convenience.listen
+    serve = convenience.serve
+    StopServe = convenience.StopServe
 
     getcurrent = greenlet.getcurrent
     

eventlet/convenience.py

@@ -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

eventlet/greenio.py

@@ -534,106 +534,3 @@ def shutdown_safe(sock):
         if get_errno(e) != 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

tests/api_test.py

@@ -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:

tests/convenience_test.py

@@ -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")
+
+        
+        

tests/db_pool_test.py

@@ -603,16 +603,17 @@ class Psycopg2ConnectionPool(object):
         super(Psycopg2ConnectionPool, self).tearDown()
 
     def create_db(self):
+        dbname = 'test%s' % os.getpid()
+        self._auth['database'] = dbname
         try:
             self.drop_db()
         except Exception:
             pass
         auth = self._auth.copy()
+        auth.pop('database')  # can't create if you're connecting to it
         conn = self._dbmodule.connect(**auth)
         conn.set_isolation_level(0)
         db = conn.cursor()
-        dbname = 'test%s' % os.getpid()
-        self._auth['database'] = dbname
         db.execute("create database "+dbname)
         db.close()
         del db

tests/greenio_test.py

@@ -1,8 +1,10 @@
-from tests import LimitedTestCase, skip_with_pyevent, main
+import socket as _orig_sock
+from tests import LimitedTestCase, skip_with_pyevent, main, skipped
 from eventlet import event
 from eventlet import greenio
 from eventlet import debug
 from eventlet.green import socket
+from eventlet.green import time
 from eventlet.green.socket import GreenSSLObject
 import errno
 
@@ -480,7 +482,7 @@ class TestGreenIo(LimitedTestCase):
 class TestGreenIoLong(LimitedTestCase):
     TEST_TIMEOUT=10  # the test here might take a while depending on the OS
     @skip_with_pyevent
-    def test_multiple_readers(self):
+    def test_multiple_readers(self, clibufsize=False):
         recvsize = 2 * min_buf_size()
         sendsize = 10 * recvsize
         # test that we can have multiple coroutines reading
@@ -505,17 +507,22 @@ class TestGreenIoLong(LimitedTestCase):
             try:
                 c1 = eventlet.spawn(reader, sock, results1)
                 c2 = eventlet.spawn(reader, sock, results2)
-                c1.wait()
+                try:
-                c2.wait()
+                    c1.wait()
+                    c2.wait()
+                finally:
+                    c1.kill()
+                    c2.kill()
             finally:
-                c1.kill()
-                c2.kill()
                 sock.close()
 
         server_coro = eventlet.spawn(server)
         client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         client.connect(('127.0.0.1', listener.getsockname()[1]))
-        bufsized(client)
+        if clibufsize:
+            bufsized(client, size=sendsize)
+        else:
+            bufsized(client)
         client.sendall('*' * sendsize)
         client.close()
         server_coro.wait()
@@ -523,59 +530,104 @@ class TestGreenIoLong(LimitedTestCase):
         self.assert_(len(results1) > 0)
         self.assert_(len(results2) > 0)
 
+    @skipped  # by rdw because it fails but it's not clear how to make it pass
+    @skip_with_pyevent
+    def test_multiple_readers2(self):
+        self.test_multiple_readers(clibufsize=True)
 
-class TestServe(LimitedTestCase):
+class TestGreenIoStarvation(LimitedTestCase):    
-    def setUp(self):
+    # fixme: this doesn't succeed, because of eventlet's predetermined
-        super(TestServe, self).setUp()
+    # ordering.  two processes, one with server, one with client eventlets
-        from eventlet import debug
+    # might be more reliable?
-        debug.hub_exceptions(False)
     
-    def tearDown(self):
+    TEST_TIMEOUT=300  # the test here might take a while depending on the OS
-        super(TestServe, self).tearDown()
+    @skipped  # by rdw, because it fails but it's not clear how to make it pass
-        from eventlet import debug
+    @skip_with_pyevent
-        debug.hub_exceptions(True)
+    def test_server_starvation(self, sendloops=15):
+        recvsize = 2 * min_buf_size()
+        sendsize = 10000 * recvsize
         
-    def test_exiting_server(self):
+        results = [[] for i in xrange(5)]
-        # tests that the server closes the client sock on handle() exit
-        def closer(sock,addr):
-            pass
         
-        l = eventlet.listen(('localhost', 0))
+        listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        gt = eventlet.spawn(greenio.serve, l, closer)
+        listener.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR, 1)
-        client = eventlet.connect(('localhost', l.getsockname()[1]))
+        listener.bind(('127.0.0.1', 0))
-        client.sendall('a')
+        port = listener.getsockname()[1]
-        self.assertEqual('', client.recv(100))
+        listener.listen(50)
-        gt.kill()
         
+        base_time = time.time()
         
-    def test_excepting_server(self):
+        def server(my_results):
-        # tests that the server closes the client sock on handle() exception
+            (sock, addr) = listener.accept()
-        def crasher(sock,addr):
-            x = sock.recv(1024)
-            0/0
             
-        l = eventlet.listen(('localhost', 0))
+            datasize = 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):
+            t1 = None
-        # tests that the server closes the client sock on handle() exception
+            t2 = None
-        def crasher(sock,addr):
+            try:
-            x = sock.recv(1024)
+                while True:
-            sock.close()
+                    data = sock.recv(recvsize)
-            0/0
+                    if not t1:
+                        t1 = time.time() - base_time
+                    if data == '':
+                        t2 = time.time() - base_time
+                        my_results.append(datasize)
+                        my_results.append((t1,t2))
+                        break
+                    datasize += len(data)
+            finally:
+                sock.close()
 
-        l = eventlet.listen(('localhost', 0))
+        def client():
-        gt = eventlet.spawn(greenio.serve, l, crasher)
+            pid = os.fork()
-        client = eventlet.connect(('localhost', l.getsockname()[1]))
+            if pid:
-        client.sendall('a')
+                return pid
-        self.assertRaises(ZeroDivisionError, gt.wait)
-        self.assertEqual('', client.recv(100))
     
+            client = _orig_sock.socket(socket.AF_INET, socket.SOCK_STREAM)
+            client.connect(('127.0.0.1', port))
 
+            bufsized(client, size=sendsize)
+
+            for i in range(sendloops):
+                client.sendall('*' * sendsize)
+            client.close()
+            os._exit(0)
+
+        clients = []
+        servers = []
+        for r in results:
+            servers.append(eventlet.spawn(server, r))
+        for r in results:
+            clients.append(client())
+
+        for s in servers:
+            s.wait()
+        for c in clients:
+            os.waitpid(c, 0)
+
+        listener.close()
+
+        # now test that all of the server receive intervals overlap, and
+        # that there were no errors.
+        for r in results:
+            assert len(r) == 2, "length is %d not 2!: %s\n%s" % (len(r), r, results)
+            assert r[0] == sendsize * sendloops
+            assert len(r[1]) == 2
+            assert r[1][0] is not None
+            assert r[1][1] is not None
+
+        starttimes = sorted(r[1][0] for r in results)
+        endtimes = sorted(r[1][1] for r in results)
+        runlengths = sorted(r[1][1] - r[1][0] for r in results)
+
+        # assert that the last task started before the first task ended
+        # (our no-starvation condition)
+        assert starttimes[-1] < endtimes[0], "Not overlapping: starts %s ends %s" % (starttimes, endtimes)
+
+        maxstartdiff = starttimes[-1] - starttimes[0]
+
+        assert maxstartdiff * 2 < runlengths[0], "Largest difference in starting times more than twice the shortest running time!"
+        assert runlengths[0] * 2 > runlengths[-1], "Longest runtime more than twice as long as shortest!"
 
 if __name__ == '__main__':
     main()

tests/test__socket_errors.py

@@ -1,4 +1,5 @@
 import unittest
+import socket as _original_sock
 from eventlet import api
 from eventlet.green import socket
 
@@ -20,5 +21,36 @@ class TestSocketErrors(unittest.TestCase):
             assert code in [111, 61, 10061], (code, text)
             assert 'refused' in text.lower(), (code, text)
 
+    def test_timeout_real_socket(self):
+        """ Test underlying socket behavior to ensure correspondence
+            between green sockets and the underlying socket module. """
+        return self.test_timeout(socket=_original_sock)
+        
+    def test_timeout(self, socket=socket):
+        """ Test that the socket timeout exception works correctly. """
+        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        server.bind(('127.0.0.1', 0))
+        server.listen(1)
+        port = server.getsockname()[1]
+
+        s = socket.socket()
+        
+        s.connect(('127.0.0.1', port))
+
+        cs, addr = server.accept()
+        cs.settimeout(1)
+        try:
+            try:
+                cs.recv(1024)
+                self.fail("Should have timed out")
+            except socket.timeout, ex:
+                assert hasattr(ex, 'args')
+                assert len(ex.args) == 1
+                assert ex.args[0] == 'timed out'
+        finally:
+            s.close()
+            cs.close()
+            server.close()
+
 if __name__=='__main__':
     unittest.main()