Merge remote-tracking branch 'origin/dmw'

- #418
- 2.6/3.x compat
- earlier bugfix fallout

docs/changelog.rst

@@ -235,6 +235,11 @@ Core Library
   was leaked on every RPC, due to a list of strong references keeping alive any
   handler ever registered for disconnect notification.
 
+* `#418 <https://github.com/dw/mitogen/issues/418>`_: the
+  :func:`mitogen.parent.iter_read` helper would leak poller FDs, because
+  execution of its :keyword:`finally` block was delayed on Python 3. Now
+  callers explicitly close the generator when finished.
+
 * `16ca111e <https://github.com/dw/mitogen/commit/16ca111e>`_: handle OpenSSH
   7.5 permission denied prompts when ``~/.ssh/config`` rewrites are present.
 

mitogen/core.py

@@ -648,7 +648,7 @@ class Message(object):
 
     def _throw_dead(self):
         if len(self.data):
-            raise ChannelError(self.data.decode(errors='replace'))
+            raise ChannelError(self.data.decode('utf-8', 'replace'))
         elif self.src_id == mitogen.context_id:
             raise ChannelError(ChannelError.local_msg)
         else:

mitogen/doas.py

@@ -80,12 +80,7 @@ class Stream(mitogen.parent.Stream):
     password_incorrect_msg = 'doas password is incorrect'
     password_required_msg = 'doas password is required'
 
-    def _connect_bootstrap(self):
-        it = mitogen.parent.iter_read(
-            fds=[self.receive_side.fd, self.diag_stream.receive_side.fd],
-            deadline=self.connect_deadline,
-        )
-
+    def _connect_input_loop(self, it):
         password_sent = False
         for buf in it:
             LOG.debug('%r: received %r', self, buf)
@@ -106,3 +101,13 @@ class Stream(mitogen.parent.Stream):
                 )
                 password_sent = True
         raise mitogen.core.StreamError('bootstrap failed')
+
+    def _connect_bootstrap(self):
+        it = mitogen.parent.iter_read(
+            fds=[self.receive_side.fd, self.diag_stream.receive_side.fd],
+            deadline=self.connect_deadline,
+        )
+        try:
+            self._connect_input_loop(it)
+        finally:
+            it.close()

mitogen/parent.py

@@ -532,12 +532,16 @@ def discard_until(fd, s, deadline):
     :raises mitogen.core.StreamError:
         Attempt to read past end of file.
     """
-    for buf in iter_read([fd], deadline):
-        if IOLOG.level == logging.DEBUG:
-            for line in buf.splitlines():
-                IOLOG.debug('discard_until: discarding %r', line)
-        if buf.endswith(s):
-            return
+    it = iter_read([fd], deadline)
+    try:
+        for buf in it:
+            if IOLOG.level == logging.DEBUG:
+                for line in buf.splitlines():
+                    IOLOG.debug('discard_until: discarding %r', line)
+            if buf.endswith(s):
+                return
+    finally:
+        it.close()  # ensure Poller.close() is called.
 
 
 def _upgrade_broker(broker):

mitogen/ssh.py

@@ -264,13 +264,7 @@ class Stream(mitogen.parent.Stream):
         # with ours.
         raise HostKeyError(self.hostkey_config_msg)
 
-    def _connect_bootstrap(self):
-        fds = [self.receive_side.fd]
-        if self.diag_stream is not None:
-            fds.append(self.diag_stream.receive_side.fd)
-
-        it = mitogen.parent.iter_read(fds=fds, deadline=self.connect_deadline)
-
+    def _connect_input_loop(self, it):
         password_sent = False
         for buf, partial in filter_debug(self, it):
             LOG.debug('%r: received %r', self, buf)
@@ -302,3 +296,14 @@ class Stream(mitogen.parent.Stream):
                 password_sent = True
 
         raise mitogen.core.StreamError('bootstrap failed')
+
+    def _connect_bootstrap(self):
+        fds = [self.receive_side.fd]
+        if self.diag_stream is not None:
+            fds.append(self.diag_stream.receive_side.fd)
+
+        it = mitogen.parent.iter_read(fds=fds, deadline=self.connect_deadline)
+        try:
+            self._connect_input_loop(it)
+        finally:
+            it.close()

mitogen/su.py

@@ -87,12 +87,8 @@ class Stream(mitogen.parent.Stream):
     password_incorrect_msg = 'su password is incorrect'
     password_required_msg = 'su password is required'
 
-    def _connect_bootstrap(self):
+    def _connect_input_loop(self, it):
         password_sent = False
-        it = mitogen.parent.iter_read(
-            fds=[self.receive_side.fd],
-            deadline=self.connect_deadline,
-        )
 
         for buf in it:
             LOG.debug('%r: received %r', self, buf)
@@ -112,4 +108,15 @@ class Stream(mitogen.parent.Stream):
                     mitogen.core.to_text(self.password + '\n').encode('utf-8')
                 )
                 password_sent = True
+
         raise mitogen.core.StreamError('bootstrap failed')
+
+    def _connect_bootstrap(self):
+        it = mitogen.parent.iter_read(
+            fds=[self.receive_side.fd],
+            deadline=self.connect_deadline,
+        )
+        try:
+            self._connect_input_loop(it)
+        finally:
+            it.close()

mitogen/sudo.py

@@ -116,10 +116,6 @@ class Stream(mitogen.parent.Stream):
     create_child = staticmethod(mitogen.parent.hybrid_tty_create_child)
     child_is_immediate_subprocess = False
 
-    #: Once connected, points to the corresponding DiagLogStream, allowing it to
-    #: be disconnected at the same time this stream is being torn down.
-    tty_stream = None
-
     sudo_path = 'sudo'
     username = 'root'
     password = None
@@ -173,16 +169,8 @@ class Stream(mitogen.parent.Stream):
     password_incorrect_msg = 'sudo password is incorrect'
     password_required_msg = 'sudo password is required'
 
-    def _connect_bootstrap(self):
-        fds = [self.receive_side.fd]
-        if self.diag_stream is not None:
-            fds.append(self.diag_stream.receive_side.fd)
-
+    def _connect_input_loop(self, it):
         password_sent = False
-        it = mitogen.parent.iter_read(
-            fds=fds,
-            deadline=self.connect_deadline,
-        )
 
         for buf in it:
             LOG.debug('%r: received %r', self, buf)
@@ -194,8 +182,24 @@ class Stream(mitogen.parent.Stream):
                     raise PasswordError(self.password_required_msg)
                 if password_sent:
                     raise PasswordError(self.password_incorrect_msg)
-                self.tty_stream.transmit_side.write(
+                self.diag_stream.transmit_side.write(
                     mitogen.core.to_text(self.password + '\n').encode('utf-8')
                 )
                 password_sent = True
+
         raise mitogen.core.StreamError('bootstrap failed')
+
+    def _connect_bootstrap(self):
+        fds = [self.receive_side.fd]
+        if self.diag_stream is not None:
+            fds.append(self.diag_stream.receive_side.fd)
+
+        it = mitogen.parent.iter_read(
+            fds=fds,
+            deadline=self.connect_deadline,
+        )
+
+        try:
+            self._connect_input_loop(it)
+        finally:
+            it.close()

tests/bench/roundtrip.py

@@ -5,13 +5,22 @@ Measure latency of local RPC.
 import mitogen
 import time
 
+import ansible_mitogen.process
+ansible_mitogen.process.setup_gil()
+
+try:
+    xrange
+except NameError:
+    xrange = range
+
 def do_nothing():
     pass
 
 @mitogen.main()
 def main(router):
     f = router.fork()
+    f.call(do_nothing)
     t0 = time.time()
-    for x in range(1000):
+    for x in xrange(20000):
         f.call(do_nothing)
-    print '++', int(1e6 * ((time.time() - t0) / (1.0+x))), 'usec'
+    print('++', int(1e6 * ((time.time() - t0) / (1.0+x))), 'usec')

tests/data/stubs/stub-lxc-info.py

@@ -1,4 +1,4 @@
 #!/usr/bin/env python
 # Mainly for use in stubconnections/kubectl.yml
 
-print 'PID: 1'
+print('PID: 1')

tests/data/stubs/stub-lxc.py

@@ -5,7 +5,7 @@ import os
 
 # setns.py fetching leader PID.
 if sys.argv[1] == 'info':
-    print 'Pid: 1'
+    print('Pid: 1')
     sys.exit(0)
 
 os.environ['ORIGINAL_ARGV'] = repr(sys.argv)

tests/fork_test.py

@@ -16,14 +16,14 @@ import plain_old_module
 
 def _find_ssl_linux():
     s = testlib.subprocess__check_output(['ldd', _ssl.__file__])
-    for line in s.splitlines():
+    for line in s.decode().splitlines():
         bits = line.split()
         if bits[0].startswith('libssl'):
             return bits[2]
 
 def _find_ssl_darwin():
     s = testlib.subprocess__check_output(['otool', '-l', _ssl.__file__])
-    for line in s.splitlines():
+    for line in s.decode().splitlines():
         bits = line.split()
         if bits[0] == 'name' and 'libssl' in bits[1]:
             return bits[1]

tests/image_prep/_container_setup.yml

@@ -2,6 +2,7 @@
 - hosts: all
   strategy: linear
   gather_facts: false
+  become: true
   tasks:
     - raw: >
         if ! python -c ''; then

tests/poller_test.py

@@ -43,7 +43,7 @@ class SockMixin(object):
         """Make `fd` unwriteable."""
         while True:
             try:
-                os.write(fd, 'x'*4096)
+                os.write(fd, mitogen.core.b('x')*4096)
             except OSError:
                 e = sys.exc_info()[1]
                 if e.args[0] == errno.EAGAIN:

tests/testlib.py

@@ -20,7 +20,7 @@ import mitogen.utils
 try:
     import faulthandler
 except ImportError:
-    pass
+    faulthandler = None
 
 try:
     import urlparse