diff --git a/AUTHORS b/AUTHORS index cd5ef65..1b6d574 100644 --- a/AUTHORS +++ b/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 diff --git a/NEWS b/NEWS index 8b14602..1f079ce 100644 --- a/NEWS +++ b/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 diff --git a/doc/basic_usage.rst b/doc/basic_usage.rst index c7a68e1..db242b3 100644 --- a/doc/basic_usage.rst +++ b/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 ` 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 ` for more details. +Greenthread Control +----------------------- + .. function:: eventlet.sleep(seconds=0) Suspends the current greenthread and allows others a chance to process. See :func:`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 ` for more on how to use these. @@ -53,6 +55,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 ` for more details. + +Patching Functions +--------------------- .. function:: eventlet.import_patched(modulename, *additional_modules, **kw_additional_modules) @@ -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`. diff --git a/eventlet/__init__.py b/eventlet/__init__.py index 1434293..e82a15a 100644 --- a/eventlet/__init__.py +++ b/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 - listen = greenio.listen + connect = convenience.connect + listen = convenience.listen + serve = convenience.serve + StopServe = convenience.StopServe getcurrent = greenlet.getcurrent diff --git a/eventlet/convenience.py b/eventlet/convenience.py new file mode 100644 index 0000000..b737242 --- /dev/null +++ b/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 diff --git a/eventlet/greenio.py b/eventlet/greenio.py index f5dfb7f..686febd 100644 --- a/eventlet/greenio.py +++ b/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 diff --git a/tests/api_test.py b/tests/api_test.py index 0300316..a41ffc8 100644 --- a/tests/api_test.py +++ b/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: diff --git a/tests/convenience_test.py b/tests/convenience_test.py new file mode 100644 index 0000000..ca561d9 --- /dev/null +++ b/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") + + + \ No newline at end of file diff --git a/tests/db_pool_test.py b/tests/db_pool_test.py index 398708f..c0b1093 100644 --- a/tests/db_pool_test.py +++ b/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 diff --git a/tests/greenio_test.py b/tests/greenio_test.py index 8a3e019..d9c626c 100644 --- a/tests/greenio_test.py +++ b/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() - c2.wait() + try: + 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): - def setUp(self): - super(TestServe, self).setUp() - from eventlet import debug - debug.hub_exceptions(False) +class TestGreenIoStarvation(LimitedTestCase): + # fixme: this doesn't succeed, because of eventlet's predetermined + # ordering. two processes, one with server, one with client eventlets + # might be more reliable? + + TEST_TIMEOUT=300 # the test here might take a while depending on the OS + @skipped # by rdw, because it fails but it's not clear how to make it pass + @skip_with_pyevent + def test_server_starvation(self, sendloops=15): + recvsize = 2 * min_buf_size() + sendsize = 10000 * recvsize - def tearDown(self): - super(TestServe, self).tearDown() - from eventlet import debug - debug.hub_exceptions(True) + results = [[] for i in xrange(5)] - def test_exiting_server(self): - # tests that the server closes the client sock on handle() exit - def closer(sock,addr): - pass + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR, 1) + listener.bind(('127.0.0.1', 0)) + port = listener.getsockname()[1] + listener.listen(50) + + base_time = time.time() + + def server(my_results): + (sock, addr) = listener.accept() - 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 + datasize = 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)) + t1 = None + t2 = None + try: + while True: + data = sock.recv(recvsize) + 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() - 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)) + def client(): + pid = os.fork() + if pid: + return pid + + 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() diff --git a/tests/test__socket_errors.py b/tests/test__socket_errors.py index 91fe3b0..ad2ad21 100644 --- a/tests/test__socket_errors.py +++ b/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()