Avoid calling super during code-paths used by FileObjectThread.

The two-arg version of super depends on a module global, making it flaky during interpreter shutdown. Seen on Python 2: https://travis-ci.org/gevent/gevent/jobs/636366809#L501

Also some cleanups to subprocess handling during tests. Avoid a ref cycle.

src/gevent/_fileobjectcommon.py

@@ -23,13 +23,15 @@ from gevent.lock import Semaphore, DummySemaphore
 class cancel_wait_ex(IOError):
 
     def __init__(self):
-        super(cancel_wait_ex, self).__init__(
+        IOError.__init__(
+            self,
             EBADF, 'File descriptor was closed in another greenlet')
 
 class FileObjectClosed(IOError):
 
     def __init__(self):
-        super(FileObjectClosed, self).__init__(
+        IOError.__init__(
+            self,
             EBADF, 'Bad file descriptor (FileObject was closed)')
 
 class UniversalNewlineBytesWrapper(io.TextIOWrapper):
@@ -408,7 +410,7 @@ class FileObjectBlock(FileObjectBase):
 
     def __init__(self, fobj, *args, **kwargs):
         descriptor = OpenDescriptor(fobj, *args, **kwargs)
-        super(FileObjectBlock, self).__init__(descriptor.open(), descriptor.closefd)
+        FileObjectBase.__init__(self, descriptor.open(), descriptor.closefd)
 
     def _do_close(self, fobj, closefd):
         fobj.close()
@@ -457,7 +459,7 @@ class FileObjectThread(FileObjectBase):
             raise TypeError('Expected a Semaphore or boolean, got %r' % type(self.lock))
 
         self.__io_holder = [descriptor.open()] # signal for _wrap_method
-        super(FileObjectThread, self).__init__(self.__io_holder[0], descriptor.closefd)
+        FileObjectBase.__init__(self, self.__io_holder[0], descriptor.closefd)
 
     def _do_close(self, fobj, closefd):
         self.__io_holder[0] = None # for _wrap_method
@@ -489,7 +491,7 @@ class FileObjectThread(FileObjectBase):
                     reraise(*exc_info)
 
     def _do_delegate_methods(self):
-        super(FileObjectThread, self)._do_delegate_methods()
+        FileObjectBase._do_delegate_methods(self)
         # if not hasattr(self, 'read1') and 'r' in getattr(self._io, 'mode', ''):
         #     self.read1 = self.read
         self.__io_holder[0] = self._io

src/gevent/testing/util.py

@@ -10,6 +10,7 @@ import time
 
 from . import six
 from gevent._config import validate_bool
+from gevent.monkey import get_original
 
 # pylint: disable=broad-except,attribute-defined-outside-init
 
@@ -22,6 +23,11 @@ QUIET = validate_bool(os.environ.get('GEVENTTEST_QUIET', '0'))
 
 
 class Popen(subprocess.Popen):
+    """
+    Depending on when we're imported and if the process has been monkey-patched,
+    this could use cooperative or native Popen.
+    """
+    timer = None # a threading.Timer instance
 
     def __enter__(self):
         return self
@@ -145,7 +151,7 @@ def killpg(pid):
 
 def kill_processtree(pid):
     ignore_msg = 'ERROR: The process "%s" not found.' % pid
-    err = subprocess.Popen('taskkill /F /PID %s /T' % pid, stderr=subprocess.PIPE).communicate()[1]
+    err = Popen('taskkill /F /PID %s /T' % pid, stderr=subprocess.PIPE).communicate()[1]
     if err and err.strip() not in [ignore_msg, '']:
         log('%r', err)
 
@@ -170,6 +176,7 @@ def _kill(popen):
 def kill(popen):
     if popen.timer is not None:
         popen.timer.cancel()
+        popen.timer = None
     if popen.poll() is not None:
         return
     popen.was_killed = True
@@ -248,9 +255,9 @@ def start(command, quiet=False, **kwargs):
     popen.name = name
     popen.setpgrp_enabled = preexec_fn is not None
     popen.was_killed = False
-    popen.timer = None
     if timeout is not None:
-        t = threading.Timer(timeout, kill, args=(popen, ))
+        t = get_original('threading', 'Timer')(timeout, kill, args=(popen, ))
+        popen.timer = t
         t.setDaemon(True)
         t.start()
         popen.timer = t
@@ -393,6 +400,7 @@ def run(command, **kwargs): # pylint:disable=too-many-locals
             result = popen.poll()
     finally:
         kill(popen)
+        assert popen.timer is None
 
 
     failed = bool(result)

src/gevent/threadpool.py

@@ -127,6 +127,7 @@ class _WorkerGreenlet(RawGreenlet):
 
     def run(self):
         # pylint:disable=too-many-branches
+        task = None
         try:
             while 1: # tiny bit faster than True on Py2
                 self.__fixup_hub_before_block()
@@ -141,7 +142,11 @@ class _WorkerGreenlet(RawGreenlet):
                     task = None
                     self._task_queue.task_done()
         except Exception as e: # pylint:disable=broad-except
-            print("Failed to run worker thread", e, file=self._stderr)
+            print(
+                "Failed to run worker thread. Task=%r Exception=%s%r" % (
+                    task, e, e
+                ),
+                file=self._stderr)
         finally:
             # Re-check for the hub in case the task created it but then
             # failed.