Python 2: Correct errors from async closed sockets.

Previously we would create a AttributeError, instead of the preferred socket.error(EBADF).

Python 3 already did the right thing.

Discovered because of bad test coverage in test__hub.py that wasn't doing what we thought.

Incremental work on increasing test coverage.

.coveragerc

@@ -37,9 +37,13 @@ exclude_lines =
     if sys.platform == 'win32':
     if mswindows:
     if is_windows:
+    if WIN:
+    self.fail
 omit =
     # local.so sometimes gets included, and it can't be parsed
     # as source, so it fails the whole process.
     # coverage 4.5 needs this specified here, 4.4.2 needed it in [run]
     *.so
     /tmp/test_*
+    # Third-party vendored code
+    src/gevent/_tblib.py

CHANGES.rst

@@ -58,6 +58,11 @@
 - Make `gevent.util.assert_switches` produce more informative messages
   when the assertion fails.
 
+- Python 2: If a `gevent.socket` was closed asynchronously (in a
+  different greenlet or a hub callback), `AttributeError` could result
+  if the socket was already in use. Now the correct socket.error
+  should be raised.
+
 1.3.7 (2018-10-12)
 ==================
 

src/gevent/__init__.py

@@ -159,8 +159,10 @@ del sys
 
 # the following makes hidden imports visible to freezing tools like
 # py2exe. see https://github.com/gevent/gevent/issues/181
+# This is not well maintained or tested, though, so it likely becomes
+# outdated on each major release.
 
-def __dependencies_for_freezing():
+def __dependencies_for_freezing(): # pragma: no cover
     # pylint:disable=unused-variable
     from gevent import core
     from gevent import resolver_thread

src/gevent/_abstract_linkable.py

@@ -181,7 +181,9 @@ class AbstractLinkable(object):
 
     def _wait_return_value(self, waited, wait_success):
         # pylint:disable=unused-argument
-        return None
+        # Subclasses should override this to return a value from _wait.
+        # By default we return None.
+        return None # pragma: no cover all extent subclasses override
 
     def _wait(self, timeout=None):
         if self.ready():

src/gevent/_hub_primitives.py

@@ -282,6 +282,10 @@ def _primitive_wait(watcher, timeout, timeout_exc, hub):
 
 # Suitable to be bound as an instance method
 def wait_on_socket(socket, watcher, timeout_exc=None):
+    if socket is None or watcher is None:
+        # test__hub TestCloseSocketWhilePolling, on Python 2; Python 3
+        # catches the EBADF differently.
+        raise ConcurrentObjectUseError("The socket has already been closed by another greenlet")
     _primitive_wait(watcher, socket.timeout,
                     timeout_exc if timeout_exc is not None else _NONE,
                     socket.hub)

src/gevent/_socket2.py

@@ -185,10 +185,9 @@ class socket(object):
     _wait = _wait_on_socket
 
     def accept(self):
-        sock = self._sock
-        while True:
+        while 1:
             try:
-                client_socket, address = sock.accept()
+                client_socket, address = self._sock.accept()
                 break
             except error as ex:
                 if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
@@ -212,10 +211,17 @@ class socket(object):
     def close(self, _closedsocket=_closedsocket):
         # This function should not reference any globals. See Python issue #808164.
 
-        # Also break any reference to the loop.io objects. Our fileno, which they were
-        # tied to, is now free to be reused, so these objects are no longer functional.
+        # Also break any reference to the loop.io objects. Our fileno,
+        # which they were tied to, is now free to be reused, so these
+        # objects are no longer functional.
         self._drop_events()
         s = self._sock
+
+        # Note that we change self._sock at this point. Methods *must not*
+        # cache `self._sock` separately from self._write_event/self._read_event,
+        # or they will be out of sync and we may get inappropriate errors.
+        # (See test__hub:TestCloseSocketWhilePolling for an example).
+
         self._sock = _closedsocket()
         if PYPY:
             s._drop()
@@ -227,16 +233,16 @@ class socket(object):
     def connect(self, address):
         if self.timeout == 0.0:
             return self._sock.connect(address)
-        sock = self._sock
-        address = _socketcommon._resolve_addr(sock, address)
+
+        address = _socketcommon._resolve_addr(self._sock, address)
 
         timer = Timeout._start_new_or_dummy(self.timeout, timeout('timed out'))
         try:
-            while True:
-                err = sock.getsockopt(SOL_SOCKET, SO_ERROR)
+            while 1:
+                err = self._sock.getsockopt(SOL_SOCKET, SO_ERROR)
                 if err:
                     raise error(err, strerror(err))
-                result = sock.connect_ex(address)
+                result = self._sock.connect_ex(address)
                 if not result or result == EISCONN:
                     break
                 elif (result in (EWOULDBLOCK, EINPROGRESS, EALREADY)) or (result == EINVAL and is_windows):
@@ -283,10 +289,9 @@ class socket(object):
         return fobj
 
     def recv(self, *args):
-        sock = self._sock  # keeping the reference so that fd is not closed during waiting
-        while True:
+        while 1:
             try:
-                return sock.recv(*args)
+                return self._sock.recv(*args)
             except error as ex:
                 if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
@@ -295,10 +300,9 @@ class socket(object):
             self._wait(self._read_event)
 
     def recvfrom(self, *args):
-        sock = self._sock
-        while True:
+        while 1:
             try:
-                return sock.recvfrom(*args)
+                return self._sock.recvfrom(*args)
             except error as ex:
                 if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
@@ -306,10 +310,9 @@ class socket(object):
             self._wait(self._read_event)
 
     def recvfrom_into(self, *args):
-        sock = self._sock
-        while True:
+        while 1:
             try:
-                return sock.recvfrom_into(*args)
+                return self._sock.recvfrom_into(*args)
             except error as ex:
                 if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
@@ -317,10 +320,9 @@ class socket(object):
             self._wait(self._read_event)
 
     def recv_into(self, *args):
-        sock = self._sock
-        while True:
+        while 1:
             try:
-                return sock.recv_into(*args)
+                return self._sock.recv_into(*args)
             except error as ex:
                 if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
@@ -328,18 +330,17 @@ class socket(object):
             self._wait(self._read_event)
 
     def send(self, data, flags=0, timeout=timeout_default):
-        sock = self._sock
         if timeout is timeout_default:
             timeout = self.timeout
         try:
-            return sock.send(data, flags)
+            return self._sock.send(data, flags)
         except error as ex:
             if ex.args[0] not in _socketcommon.GSENDAGAIN or timeout == 0.0:
                 raise
             sys.exc_clear()
             self._wait(self._write_event)
             try:
-                return sock.send(data, flags)
+                return self._sock.send(data, flags)
             except error as ex2:
                 if ex2.args[0] == EWOULDBLOCK:
                     return 0
@@ -354,16 +355,15 @@ class socket(object):
         return _socketcommon._sendall(self, data_memory, flags)
 
     def sendto(self, *args):
-        sock = self._sock
         try:
-            return sock.sendto(*args)
+            return self._sock.sendto(*args)
         except error as ex:
             if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:
                 raise
             sys.exc_clear()
             self._wait(self._write_event)
             try:
-                return sock.sendto(*args)
+                return self._sock.sendto(*args)
             except error as ex2:
                 if ex2.args[0] == EWOULDBLOCK:
                     return 0

src/gevent/testing/__init__.py

@@ -122,6 +122,12 @@ from .flaky import reraiseFlakyTestRaceCondition
 from .flaky import reraises_flaky_timeout
 from .flaky import reraises_flaky_race_condition
 
+def gc_collect_if_needed():
+    "Collect garbage if necessary for destructors to run"
+    import gc
+    if PYPY: # pragma: no cover
+        gc.collect()
+
 try:
     from unittest import mock
 except ImportError: # Python 2

src/gevent/testing/patched_tests_setup.py

@@ -466,6 +466,7 @@ if LIBUV:
             'test_socket.BufferIOTest.testRecvFromIntoBytearray',
             'test_socket.BufferIOTest.testRecvFromIntoArray',
             'test_socket.BufferIOTest.testRecvFromIntoEmptyBuffer',
+            'test_socket.BufferIOTest.testRecvFromIntoMemoryview',
         ]
 
         if PY3:

src/gevent/tests/test___monitor.py

@@ -106,7 +106,7 @@ class TestPeriodicMonitoringThread(_AbstractTestPeriodicMonitoringThread,
         self.assertRaises(ValueError, self.pmt.add_monitoring_function, lambda: None, -1)
 
         def f():
-            pass
+            "Does nothing"
 
         # Add
         self.pmt.add_monitoring_function(f, 1)

src/gevent/tests/test__environ.py

@@ -8,7 +8,7 @@ if sys.argv[1:] == []:
     os.environ['GEVENT_BACKEND'] = 'select'
     popen = subprocess.Popen([sys.executable, __file__, '1'])
     assert popen.wait() == 0, popen.poll()
-else:
+else: # pragma: no cover
     hub = gevent.get_hub()
     if 'select' in gevent.core.supported_backends():
         assert hub.loop.backend == 'select', hub.loop.backend

src/gevent/tests/test__exc_info.py

@@ -35,8 +35,8 @@ class Test(greentest.TestCase):
             g = gevent.spawn(hello, expected_error)
             g.join()
             self.assert_error(ExpectedError, expected_error)
-            if not isinstance(g.exception, ExpectedError):
-                raise g.exception
+            self.assertIsInstance(g.exception, ExpectedError)
+
             try:
                 raise
             except: # pylint:disable=bare-except

src/gevent/tests/test__greenlet.py

@@ -165,9 +165,6 @@ def return25():
     return 25
 
 
-def sleep0():
-    return sleep(0)
-
 
 class TestReturn_link(LinksTestCase):
     link_method = 'link'
@@ -748,7 +745,7 @@ class TestBasic(greentest.TestCase):
             raise ValueError("call stack is not deep enough")
         try:
             ogf = greenlet.sys_getframe
-        except AttributeError:
+        except AttributeError: # pragma: no cover
             # Must be running cython compiled
             raise unittest.SkipTest("Cannot mock when Cython compiled")
         greenlet.sys_getframe = get
@@ -806,7 +803,7 @@ class TestRef(greentest.TestCase):
 
 
 @greentest.skipOnPurePython("Needs C extension")
-class TestCExt(greentest.TestCase):
+class TestCExt(greentest.TestCase): # pragma: no cover (we only do coverage on pure-Python)
 
     def test_c_extension(self):
         self.assertEqual(greenlet.Greenlet.__module__,

src/gevent/tests/test__greenletset.py

@@ -149,13 +149,11 @@ class Test(greentest.TestCase):
         s = set()
         s.add(p1)
         s.add(p2)
-        try:
+        with self.assertRaises(Timeout):
             gevent.killall(s, timeout=0.5)
-        except Timeout:
-            for g in s:
-                assert not g.dead
-        else:
-            self.fail("Should raise timeout")
+
+        for g in s:
+            self.assertFalse(g.dead, g)
 
 
 class GreenletSubclass(gevent.Greenlet):

src/gevent/tests/test__hub.py

@@ -36,10 +36,13 @@ DELAY = 0.1
 class TestCloseSocketWhilePolling(greentest.TestCase):
 
     def test(self):
-        with self.assertRaises(Exception):
-            sock = socket.socket()
-            self._close_on_teardown(sock)
-            t = get_hub().loop.timer(0, sock.close)
+        from gevent.exceptions import ConcurrentObjectUseError
+
+        sock = socket.socket()
+        self._close_on_teardown(sock)
+        t = get_hub().loop.timer(0)
+        t.start(sock.close)
+        with self.assertRaises(socket.error):
             try:
                 sock.connect(('python.org', 81))
             finally:

src/gevent/tests/test__issue330.py

@@ -14,7 +14,7 @@ class TestSwitch(greentest.TestCase):
         self.switched_to = [False, False]
         self.caught = None
 
-    def runner(self, i):
+    def should_never_run(self, i): # pragma: no cover
         self.switched_to[i] = True
 
     def check(self, g, g2):
@@ -33,8 +33,8 @@ class TestSwitch(greentest.TestCase):
 
 
     def test_gevent_kill(self):
-        g = gevent.spawn(self.runner, 0) # create but do not switch to
-        g2 = gevent.spawn(self.runner, 1) # create but do not switch to
+        g = gevent.spawn(self.should_never_run, 0) # create but do not switch to
+        g2 = gevent.spawn(self.should_never_run, 1) # create but do not switch to
         # Using gevent.kill
         gevent.kill(g)
         gevent.kill(g2)
@@ -42,16 +42,16 @@ class TestSwitch(greentest.TestCase):
 
     def test_greenlet_kill(self):
         # killing directly
-        g = gevent.spawn(self.runner, 0)
-        g2 = gevent.spawn(self.runner, 1)
+        g = gevent.spawn(self.should_never_run, 0)
+        g2 = gevent.spawn(self.should_never_run, 1)
         g.kill()
         g2.kill()
         self.check(g, g2)
 
     def test_throw(self):
         # throwing
-        g = gevent.spawn(self.runner, 0)
-        g2 = gevent.spawn(self.runner, 1)
+        g = gevent.spawn(self.should_never_run, 0)
+        g2 = gevent.spawn(self.should_never_run, 1)
         g.throw(gevent.GreenletExit)
         g2.throw(gevent.GreenletExit)
         self.check(g, g2)

src/gevent/tests/test__issue6.py

@@ -16,7 +16,7 @@ if not sys.argv[1:]:
     #    or __package__, falling back on __name__ and __path__\n  return f(*args, **kwds)\n'
     assert err == b'' or b'sys.excepthook' in err or b'Warning' in err, (out, err, code)
 
-elif sys.argv[1:] == ['subprocess']:
+elif sys.argv[1:] == ['subprocess']: # pragma: no cover
     import gevent
     import gevent.monkey
     gevent.monkey.patch_all(sys=True)
@@ -30,5 +30,5 @@ elif sys.argv[1:] == ['subprocess']:
 
     gevent.spawn(printline).join()
 
-else:
+else: # pragma: no cover
     sys.exit('Invalid arguments: %r' % (sys.argv, ))

src/gevent/tests/test__issues461_471.py

@@ -8,7 +8,7 @@ handling of KeyboardInterrupt.
 
 import sys
 
-if sys.argv[1:] == ['subprocess']:
+if sys.argv[1:] == ['subprocess']: # pragma: no cover
     import gevent
 
     def task():

src/gevent/tests/test__local.py

@@ -277,9 +277,8 @@ class TestGeventLocal(greentest.TestCase):
 
         my_local = MyLocal()
         my_local.sentinel = None
-        if greentest.PYPY:
-            import gc
-            gc.collect()
+        greentest.gc_collect_if_needed()
+
         del created_sentinels[:]
         del deleted_sentinels[:]
 
@@ -298,8 +297,7 @@ class TestGeventLocal(greentest.TestCase):
         for g in greenlets:
             assert not g.is_alive()
         gevent.sleep() # let the callbacks run
-        if greentest.PYPY:
-            gc.collect()
+        greentest.gc_collect_if_needed()
 
         # The sentinels should be gone too
         self.assertEqual(len(deleted_sentinels), len(greenlets))
@@ -412,7 +410,7 @@ class TestLocalInterface(greentest.TestCase):
 
 
 @greentest.skipOnPurePython("Needs C extension")
-class TestCExt(greentest.TestCase):
+class TestCExt(greentest.TestCase): # pragma: no cover
 
     def test_c_extension(self):
         self.assertEqual(local.__module__,

src/gevent/tests/test__memleak.py

@@ -9,7 +9,7 @@ from gevent.timeout import Timeout
     hasattr(sys, 'gettotalrefcount'),
     "Needs debug build"
 )
-class TestQueue(TestCase):
+class TestQueue(TestCase): # pragma: no cover
     # pylint:disable=bare-except,no-member
 
     def test(self):

src/gevent/tests/test__os.py

@@ -160,7 +160,7 @@ if hasattr(os, 'fork_and_watch'):
                 # Can't assert on what the pid actually was,
                 # our testrunner may have spawned multiple children.
                 os._reap_children(0) # make the leakchecker happy
-            else:
+            else: # pragma: no cover
                 gevent.sleep(2)
                 os._exit(0)
 

src/gevent/tests/test__queue.py

@@ -140,7 +140,7 @@ class TestQueue(TestCase):
             with gevent.Timeout(0, RuntimeError()):
                 try:
                     result = q.get()
-                    evt.set(result)
+                    evt.set(result) # pragma: no cover (should have raised)
                 except RuntimeError:
                     evt.set('timed out')
 
@@ -169,7 +169,7 @@ class TestQueue(TestCase):
             with gevent.Timeout(0, RuntimeError()):
                 try:
                     result = q.get()
-                    evt.set(result)
+                    evt.set(result) # pragma: no cover (should have raised)
                 except RuntimeError:
                     evt.set('timed out')
 
@@ -188,7 +188,7 @@ class TestQueue(TestCase):
             with gevent.Timeout(0, RuntimeError()):
                 try:
                     result = q.get()
-                    evt.set(result)
+                    evt.set(result) # pragma: no cover (should have raised)
                 except RuntimeError:
                     evt.set('timed out')
 

src/gevent/tests/test__refcount_core.py

@@ -14,7 +14,7 @@ try:
     s.close()
     del s
     assert r() is None
-except AssertionError:
+except AssertionError: # pragma: no cover
     import sys
     if hasattr(sys, 'pypy_version_info'):
         # PyPy uses a non refcounted GC which may defer

src/gevent/tests/test__socket_close.py

@@ -8,7 +8,7 @@ import gevent.testing as greentest
 
 def readall(sock, _):
     while sock.recv(1024):
-        pass
+        pass # pragma: no cover we never actually send the data
     sock.close()
 
 

src/gevent/tests/test__subprocess_interrupted.py

 import sys
 
-if 'runtestcase' in sys.argv[1:]:
+if 'runtestcase' in sys.argv[1:]: # pragma: no cover
     import gevent
     import gevent.subprocess
     gevent.spawn(sys.exit, 'bye')
@@ -15,6 +15,6 @@ else:
                                      __file__, 'runtestcase'],
                                     stderr=subprocess.PIPE).communicate()
         if b'refs' in err: # Something to do with debug mode python builds?
-            assert err.startswith(b'bye'), repr(err)
+            assert err.startswith(b'bye'), repr(err) # pragma: no cover
         else:
             assert err.strip() == b'bye', repr(err)

src/gevent/timeout.py

@@ -45,7 +45,7 @@ class _FakeTimer(object):
 
     @property
     def seconds(self):
-        return None
+        "Always returns None"
 
     timer = exception = seconds