More fixes to threading and subprocesses under Py3.4

gevent/monkey.py

@@ -122,6 +122,39 @@ def patch_thread(threading=True, _threading_local=True, Event=False):
         from gevent.local import local
         _threading_local.local = local
 
+    if sys.version_info[:2] >= (3, 4):
+        # Issue 18808 changes the nature of Thread.join() to use
+        # locks. This means that a greenlet spawned in the main thread
+        # (which is already running) cannot wait for the main thread---it
+        # hangs forever. We patch around this if possible. See also
+        # gevent.threading.
+        threading = __import__('threading')
+        greenlet = __import__('greenlet')
+        if threading.current_thread() == threading.main_thread():
+            main_thread = threading.main_thread()
+            _greenlet = main_thread._greenlet = greenlet.getcurrent()
+            from .hub import sleep
+
+            def join(timeout=None):
+                if threading.current_thread() is main_thread:
+                    raise RuntimeError("Cannot join current thread")
+                self = _greenlet
+                if _greenlet.dead or not main_thread.is_alive():
+                    return
+                elif timeout:
+                    raise ValueError("Cannot use a timeout to join the main thread")
+                    # XXX: Make that work
+                else:
+                    while main_thread.is_alive():
+                        sleep(0.01)
+
+            main_thread.join = join
+        else:
+            # TODO: Can we use warnings here or does that mess up monkey patching?
+            print("Monkey-patching not on the main thread; "
+                  "threading.main_thread().join() will hang from a greenlet",
+                  file=sys.stderr)
+
 
 def patch_socket(dns=True, aggressive=True):
     """Replace the standard socket object with gevent's cooperative sockets.
@@ -169,6 +202,18 @@ def patch_select(aggressive=True):
         remove_item(select, 'kqueue')
         remove_item(select, 'kevent')
 
+    if sys.version_info[:2] >= (3, 4):
+        # Python 3 wants to use `select.select` as a member function,
+        # leading to this error in selectors.py
+        #    r, w, _ = self._select(self._readers, self._writers, [], timeout)
+        #    TypeError: select() takes from 3 to 4 positional arguments but 5 were given
+        select = __import__('select')
+        selectors = __import__('selectors')
+        if selectors.SelectSelector._select is select.select:
+            def _select(self, *args, **kwargs):
+                return select.select(*args, **kwargs)
+            selectors.SelectSelector._select = _select
+
 
 def patch_subprocess():
     patch_module('subprocess')

gevent/threading.py

@@ -69,10 +69,14 @@ if sys.version_info[:2] >= (3, 4):
             try:
                 super(Thread, self).run()
             finally:
-                del self._greenlet # avoid ref cycles
+                # avoid ref cycles, but keep in __dict__ so we can
+                # distinguish the started/never-started case
+                self._greenlet = None
                 self._stop() # mark as finished
 
         def join(self, timeout=None):
+            if '_greenlet' not in self.__dict__:
+                raise RuntimeError("Cannot join an inactive thread")
             if self._greenlet is None:
                 return
             self._greenlet.join(timeout=timeout)

greentest/lock_tests.py

@@ -423,7 +423,8 @@ class BaseSemaphoreTests(BaseTestCase):
 
     def test_constructor(self):
         self.assertRaises(ValueError, self.semtype, value = -1)
-        self.assertRaises(ValueError, self.semtype, value = -sys.maxint)
+        # Py3 doesn't have sys.maxint
+        self.assertRaises(ValueError, self.semtype, value = -getattr(sys, 'maxint', getattr(sys, 'maxsize', None)))
 
     def test_acquire(self):
         sem = self.semtype(1)

greentest/test__server_pywsgi.py

@@ -30,15 +30,15 @@ class SimpleWSGIServer(pywsgi.WSGIServer):
     application = application
 
 
-internal_error_start = 'HTTP/1.1 500 Internal Server Error\n'.replace('\n', '\r\n')
-internal_error_end = '\n\nInternal Server Error'.replace('\n', '\r\n')
+internal_error_start = b'HTTP/1.1 500 Internal Server Error\n'.replace(b'\n', b'\r\n')
+internal_error_end = b'\n\nInternal Server Error'.replace(b'\n', b'\r\n')
 
-internal_error503 = '''HTTP/1.1 503 Service Unavailable
+internal_error503 = b'''HTTP/1.1 503 Service Unavailable
 Connection: close
 Content-type: text/plain
 Content-length: 31
 
-Service Temporarily Unavailable'''.replace('\n', '\r\n')
+Service Temporarily Unavailable'''.replace(b'\n', b'\r\n')
 
 
 class Settings:

greentest/test_threading_2.py

@@ -20,7 +20,7 @@ if not hasattr(threading.Thread, 'is_alive'):
     threading.Thread.is_alive = threading.Thread.isAlive
 if not hasattr(threading.Thread, 'daemon'):
     threading.Thread.daemon = property(threading.Thread.isDaemon, threading.Thread.setDaemon)
-if not hasattr(threading._Condition, 'notify_all'):
+if hasattr(threading, '_Condition') and not hasattr(threading._Condition, 'notify_all'):
     threading._Condition.notify_all = threading._Condition.notifyAll
 '''
 
@@ -305,7 +305,11 @@ class ThreadTests(unittest.TestCase):
             import subprocess
             rc = subprocess.call([sys.executable, "-c", """if 1:
 %s
-                import ctypes, sys, time, thread
+                import ctypes, sys, time
+                try:
+                    import thread
+                except ImportError:
+                    import _thread as thread # Py3
 
                 # This lock is used as a simple event variable.
                 ready = thread.allocate_lock()
@@ -354,6 +358,8 @@ class ThreadTests(unittest.TestCase):
                 stderr=subprocess.PIPE)
             stdout, stderr = p.communicate()
             stdout = stdout.strip()
+            stdout = stdout.decode('utf-8')
+            stderr = stderr.decode('utf-8')
             assert re.match('^Woke up, sleep function is: <.*?sleep.*?>$', stdout), repr(stdout)
             stderr = re.sub(r"^\[\d+ refs\]", "", stderr, re.MULTILINE).strip()
             self.assertEqual(stderr, "")
@@ -425,10 +431,10 @@ class ThreadJoinOnShutdown(unittest.TestCase):
         import subprocess
         p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
         rc = p.wait()
-        data = p.stdout.read().replace('\r', '')
-        self.assertEqual(data, "end of main\nend of thread\n")
-        self.failIf(rc == 2, "interpreter was blocked")
-        self.failUnless(rc == 0, "Unexpected error")
+        data = p.stdout.read().replace(b'\r', b'')
+        self.assertEqual(data, b"end of main\nend of thread\n")
+        self.failIf(rc == 2, b"interpreter was blocked")
+        self.failUnless(rc == 0, b"Unexpected error")
 
     def test_1_join_on_shutdown(self):
         # The usual case: on exit, wait for a non-daemon thread

known_failures.py

@@ -83,12 +83,10 @@ if PYPY:
 if PY3:
     # No idea / TODO
     FAILING_TESTS += '''
-test_threading_2.py
 test__refcount.py
 test__all__.py
 test__pywsgi.py
 test__makefile_ref.py
-test__server_pywsgi.py
 test__core_stat.py
 FLAKY test__greenio.py
 FLAKY test__socket_dns.py