pypy compatibility

.travis.yml

@@ -24,8 +24,11 @@ matrix:
 install:
   - sudo apt-get update -qq
   - sudo apt-get install -qq libssl-dev libmysqlclient-dev libpq-dev libzmq3-dev
-  - pip install --upgrade tox setuptools virtualenv
+  - pip install --upgrade pip setuptools tox virtualenv
 before_script:
   - "export EVENTLET_DB_TEST_AUTH='{\"psycopg2\": {\"user\": \"postgres\"}, \"MySQLdb\": {\"passwd\": \"\", \"host\": \"localhost\", \"user\": \"root\"}}'"
 script:
-  - tox -e $TOX_ENV
+  - tox -v -v -e $TOX_ENV
+after_failure:
+  - for X in .tox/$TOX_ENV/log/*; do echo "$X\n"; cat "$X"; echo "\n\n"; done
+  - echo "pip.log\n"; cat $HOME/.pip/pip.log

eventlet/greenio.py

@@ -391,6 +391,9 @@ class _SocketDuckForFd(object):
                     raise IOError(*e.args)
 
     def __del__(self):
+        self._close()
+
+    def _close(self):
         try:
             os.close(self._fileno)
         except:
@@ -401,11 +404,15 @@ class _SocketDuckForFd(object):
         return "%s:%d" % (self.__class__.__name__, self._fileno)
 
     if "__pypy__" in sys.builtin_module_names:
+        _refcount = 0
+
         def _reuse(self):
-            pass
+            self._refcount += 1
 
         def _drop(self):
-            pass
+            self._refcount -= 1
+            if self._refcount == 0:
+                self._close()
 
 
 def _operationOnClosedFile(*args, **kwargs):

eventlet/hubs/kqueue.py

@@ -41,7 +41,7 @@ class Hub(BaseHub):
     def _control(self, events, max_events, timeout):
         try:
             return self.kqueue.control(events, max_events, timeout)
-        except OSError:
+        except (OSError, IOError):
             # have we forked?
             if os.getpid() != self._pid:
                 self._reinit_kqueue()

tests/__init__.py

@@ -2,6 +2,7 @@
 from __future__ import print_function
 
 import errno
+import gc
 import os
 try:
     import resource
@@ -181,6 +182,8 @@ class LimitedTestCase(unittest.TestCase):
             signal.alarm(self.previous_alarm[1])
 
         tpool.killall()
+        gc.collect()
+        eventlet.sleep(0)
         verify_hub_empty()
 
     def assert_less_than(self, a,b,msg=None):

tests/db_pool_test.py

@@ -27,6 +27,7 @@ class DBTester(object):
         );""")
         connection.commit()
         cursor.close()
+        connection.close()
 
     def tearDown(self):
         if self.connection:
@@ -180,7 +181,7 @@ class DBConnectionPool(DBTester):
         self.pool.put(conn2)
 
     def test_visibility_from_other_connections(self):
-        self.pool = self.create_pool(3)
+        self.pool = self.create_pool(max_size=3)
         conn = self.pool.get()
         conn2 = self.pool.get()
         curs = conn.cursor()
@@ -213,7 +214,7 @@ class DBConnectionPool(DBTester):
     def test_two_simultaneous_connections(self):
         # timing-sensitive test, disabled until we come up with a better
         # way to do this
-        self.pool = self.create_pool(2)
+        self.pool = self.create_pool(max_size=2)
         conn = self.pool.get()
         self.set_up_dummy_table(conn)
         self.fill_up_table(conn)
@@ -629,7 +630,7 @@ class Psycopg2ConnectionPool(object):
         db = conn.cursor()
         db.execute("create database "+dbname)
         db.close()
-        del db
+        conn.close()
 
     def drop_db(self):
         auth = self._auth.copy()
@@ -639,7 +640,7 @@ class Psycopg2ConnectionPool(object):
         db = conn.cursor()
         db.execute("drop database "+self._auth['database'])
         db.close()
-        del db
+        conn.close()
 
 
 class TestPsycopg2Base(TestCase):

tests/greenio_test.py

@@ -2,12 +2,13 @@ import socket as _orig_sock
 from tests import LimitedTestCase, skip_with_pyevent, main, skipped, s2b, skip_if, skip_on_windows
 from eventlet import event, greenio, debug
 from eventlet.hubs import get_hub
-from eventlet.green import select, socket, time
+from eventlet.green import select, socket, time, ssl
 from eventlet.support import get_errno
 
 import array
 import errno
 import eventlet
+import gc
 import fcntl
 import os
 import sys
@@ -321,6 +322,7 @@ class TestGreenSocket(LimitedTestCase):
                 conn = conn.makefile('w')
                 conn.write('hello\n')
                 conn.close()
+                gc.collect()
                 self.assertWriteToClosedFileRaises(conn)
             finally:
                 listener.close()
@@ -423,16 +425,11 @@ class TestGreenSocket(LimitedTestCase):
             test_sendall_impl(how_many)
 
     def test_wrap_socket(self):
-        try:
-            import ssl
-        except ImportError:
-            pass  # pre-2.6
-        else:
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            sock.bind(('127.0.0.1', 0))
-            sock.listen(50)
-            ssl.wrap_socket(sock)
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.bind(('127.0.0.1', 0))
+        sock.listen(50)
+        ssl.wrap_socket(sock)
 
     def test_timeout_and_final_write(self):
         # This test verifies that a write on a socket that we've
@@ -535,7 +532,9 @@ class TestGreenSocket(LimitedTestCase):
 
     def test_invalid_connection(self):
         # find an unused port by creating a socket then closing it
-        port = eventlet.listen(('127.0.0.1', 0)).getsockname()[1]
+        listening_socket = eventlet.listen(('127.0.0.1', 0))
+        port = listening_socket.getsockname()[1]
+        listening_socket.close()
         self.assertRaises(socket.error, eventlet.connect, ('127.0.0.1', port))
 
     def test_zero_timeout_and_back(self):

tests/greenthread_test.py

@@ -101,7 +101,7 @@ class Spawn(LimitedTestCase, Asserts):
         gt.wait()
         self.assertEquals(called, [True])
 
-class SpawnAfter(LimitedTestCase, Asserts):
+class SpawnAfter(Spawn):
     def test_basic(self):
         gt = greenthread.spawn_after(0.1, passthru, 20)
         self.assertEquals(gt.wait(), ((20,), {}))

tests/test__refcount.py

@@ -63,6 +63,7 @@ def run_interaction(run_client):
 def run_and_check(run_client):
     w = run_interaction(run_client=run_client)
     clear_sys_exc_info()
+    gc.collect()
     if w():
         print(pformat(gc.get_referrers(w())))
         for x in gc.get_referrers(w()):

tests/thread_test.py

@@ -1,3 +1,4 @@
+import gc
 import weakref
 
 from eventlet.green import thread
@@ -109,5 +110,8 @@ class Locals(LimitedTestCase):
             p.spawn(do_something, i)
         p.waitall()
         del p
+        gc.collect()
+        eventlet.sleep(0)
+        gc.collect()
         # at this point all our coros have terminated
         self.assertEqual(len(refs), 1)

tests/timeout_test_with_statement.py

@@ -1,5 +1,6 @@
 """Tests with-statement behavior of Timeout class."""
 
+import gc
 import sys
 import time
 import unittest
@@ -109,6 +110,7 @@ class Test(LimitedTestCase):
         with Timeout(DELAY*2, err):
             sleep(DELAY)
         del err
+        gc.collect()
         assert not err_ref(), repr(err_ref())
 
     def test_nested_timeout(self):

tests/wsgi_test.py

@@ -10,6 +10,7 @@ from unittest import main
 
 from eventlet import greenio
 from eventlet import event
+from eventlet import hubs
 from eventlet.green import socket as greensocket
 from eventlet import wsgi
 from eventlet.support import get_errno
@@ -154,6 +155,7 @@ def read_http(sock):
     if not response_line:
         raise ConnectionClosed(response_line)
 
+
     header_lines = []
     while True:
         line = fd.readline()
@@ -161,6 +163,7 @@ def read_http(sock):
             break
         else:
             header_lines.append(line)
+
     headers = dict()
     for x in header_lines:
         x = x.strip()
@@ -170,6 +173,7 @@ def read_http(sock):
         assert key.lower() not in headers, "%s header duplicated" % key
         headers[key.lower()] = value
 
+
     if CONTENT_LENGTH in headers:
         num = int(headers[CONTENT_LENGTH])
         body = fd.read(num)
@@ -255,6 +259,7 @@ class TestHttpd(_TestBase):
         fd.flush()
         read_http(sock)
         fd.close()
+        sock.close()
 
     def test_003_passing_non_int_to_read(self):
         # This should go in greenio_test
@@ -356,6 +361,7 @@ class TestHttpd(_TestBase):
         response_line_test,_,_ = read_http(sock)
         self.assertEqual(response_line_200,response_line_test)
         fd.close()
+        sock.close()
 
     def test_009_chunked_response(self):
         self.site.application = chunked_app
@@ -579,6 +585,7 @@ class TestHttpd(_TestBase):
         response_line, headers, body = read_http(sock)
         self.assert_('connection' in headers)
         self.assertEqual('keep-alive', headers['connection'])
+        sock.close()
 
     def test_019_fieldstorage_compat(self):
         def use_fieldstorage(environ, start_response):
@@ -744,6 +751,7 @@ class TestHttpd(_TestBase):
         self.assert_(header_lines[0].startswith('HTTP/1.1 200 OK'))
         self.assertEquals(fd.read(7), 'testing')
         fd.close()
+        sock.close()
 
     def test_025_accept_errors(self):
         from eventlet import debug
@@ -800,6 +808,7 @@ class TestHttpd(_TestBase):
         read_http(sock)
 
         self.assertEqual(wsgi.HttpProtocol.minimum_chunk_size, start_size)
+        sock.close()
 
     def test_error_in_chunked_closes_connection(self):
         # From http://rhodesmill.org/brandon/2013/chunked-wsgi/
@@ -845,6 +854,7 @@ class TestHttpd(_TestBase):
         fd.write('PUT /a HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n10\r\n0123456789abcdef\r\n0\r\n\r\n')
         fd.flush()
         read_http(sock)
+        sock.close()
 
     @skip_if_no_ssl
     def test_028_ssl_handshake_errors(self):
@@ -1357,6 +1367,7 @@ class TestChunkedInput(_TestBase):
         self.assertEquals(read_http(fd)[-1], "this is ch")
 
         self.ping(fd)
+        fd.close()
 
     def test_short_read_with_zero_content_length(self):
         body = self.body()
@@ -1366,6 +1377,7 @@ class TestChunkedInput(_TestBase):
         self.assertEquals(read_http(fd)[-1], "this is ch")
 
         self.ping(fd)
+        fd.close()
 
     def test_short_read(self):
         body = self.body()
@@ -1376,6 +1388,7 @@ class TestChunkedInput(_TestBase):
         self.assertEquals(read_http(fd)[-1], "this is ch")
 
         self.ping(fd)
+        fd.close()
 
     def test_dirt(self):
         body = self.body(dirt="; here is dirt\0bla")
@@ -1386,6 +1399,7 @@ class TestChunkedInput(_TestBase):
         self.assertEquals(read_http(fd)[-1], "pong")
 
         self.ping(fd)
+        fd.close()
 
     def test_chunked_readline(self):
         body = self.body()
@@ -1394,6 +1408,7 @@ class TestChunkedInput(_TestBase):
         fd = self.connect()
         fd.sendall(req)
         self.assertEquals(read_http(fd)[-1], 'this is chunked\nline 2\nline3')
+        fd.close()
 
     def test_chunked_readline_wsgi_override_minimum_chunk_size(self):
 

tests/zmq_test.py

@@ -13,6 +13,9 @@ except ImportError:
     zmq = {}    # for systems lacking zmq, skips tests instead of barfing
 
 
+RECV_ON_CLOSED_SOCKET_ERRNOS = (zmq.ENOTSUP, zmq.ENOTSOCK)
+
+
 def zmq_supported(_):
     try:
         import zmq
@@ -49,12 +52,20 @@ class TestUpstreamDownStream(LimitedTestCase):
         self.sockets = None
         self.context.destroy(0)
 
-    def assertRaisesErrno(self, errno, func, *args):
+    def assertRaisesErrno(self, errnos, func, *args):
         try:
             func(*args)
         except zmq.ZMQError as e:
-            self.assertEqual(e.errno, errno, "wrong error raised, expected '%s' \
-got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
+            if not hasattr(errnos, '__iter__'):
+                errnos = (errnos,)
+
+            if e.errno not in errnos:
+                raise AssertionError(
+                    "wrong error raised, expected one of ['%s'], got '%s'" % (
+                        ", ".join("%s" % zmq.ZMQError(errno) for errno in errnos),
+                        zmq.ZMQError(e.errno)
+                    ),
+                )
         else:
             self.fail("Function did not raise any error")
 
@@ -92,8 +103,8 @@ got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
 
         rep.close()
         req.close()
-        self.assertRaisesErrno(zmq.ENOTSUP, rep.recv)
-        self.assertRaisesErrno(zmq.ENOTSUP, req.send, 'test')
+        self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, rep.recv)
+        self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, 'test')
 
     @skip_unless(zmq_supported)
     def test_close_xsocket_raises_enotsup(self):
@@ -101,8 +112,8 @@ got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
 
         rep.close()
         req.close()
-        self.assertRaisesErrno(zmq.ENOTSUP, rep.recv)
-        self.assertRaisesErrno(zmq.ENOTSUP, req.send, 'test')
+        self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, rep.recv)
+        self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, 'test')
 
     @skip_unless(zmq_supported)
     def test_send_1k_req_rep(self):
@@ -378,7 +389,7 @@ got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
         done2 = event.Event()
 
         def rx(e):
-            self.assertRaisesErrno(zmq.ENOTSUP, receiver.recv)
+            self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, receiver.recv)
             e.send()
 
         spawn(rx, done1)

tox.ini

@@ -15,7 +15,7 @@ envlist =
     py26selects,py26poll,py26epolls,
     py27selects,py27poll,py27epolls,
     py33selects,py33poll,py33epolls,
-    pypyselects,pypypoll,pypyepoll
+    pypyselects,pypypoll,pypyepolls,
 
 [testenv]
 downloadcache = {toxworkdir}/pip_download_cache