Merge remote-tracking branch 'origin/549-open-files'

* origin/549-open-files:
  issue #603: Revert "ci: update to Ansible 2.8.3"
  Fix unit_Test.ClientTest following 108015a
  service: clean up log messages, especially at shutdown
  remove unused imports flagged by lgtm
  [linear2]: merge fallout flaggged by LGTM
  issue #549: docs: update Changelog
  issue #549: increase open file limit automatically if possible
  ansible: improve process.py docs
  docs: remove old list link.
  docs: migrate email list
  docs: changelog tweaks
  parent: decode logged stdout as UTF-8.
  scripts: import affin.sh
  ci: update to Ansible 2.8.3
  tests: terraform tweaks
  unix: include more IO in the try/except for connection failure
  tests: update gcloud.py to match terraform config
  tests: hide ugly error during Ansible tests
  tests/ansible/gcloud: terraform conf for load testing
  ansible: gracefully handle failure to connect to MuxProcess
  ansible: fix affinity tests for 5ae45f6612390bbc888b65964fb5c218feed1679
  ansible: pin per-CPU muxes to their corresponding CPU
  ansible: reap mux processes on shut down

Author: dw

ansible_mitogen/affinity.py

@@ -142,7 +142,7 @@ def assign_controller(self):
         Assign the Ansible top-level policy to this process.
         """
 
-    def assign_muxprocess(self):
+    def assign_muxprocess(self, index):
         """
         Assign the MuxProcess policy to this process.
         """
@@ -224,7 +224,7 @@ def _balance(self):
         ))
 
     def _set_cpu(self, cpu):
-        self._set_affinity(1 << cpu)
+        self._set_affinity(1 << (cpu % self.cpu_count))
 
     def _clear(self):
         all_cpus = (1 << self.cpu_count) - 1
@@ -236,8 +236,8 @@ def assign_controller(self):
         else:
             self._balance()
 
-    def assign_muxprocess(self):
-        self._set_cpu(0)
+    def assign_muxprocess(self, index):
+        self._set_cpu(index)
 
     def assign_worker(self):
         self._balance()

ansible_mitogen/connection.py

@@ -37,16 +37,13 @@
 import sys
 import time
 
-import jinja2.runtime
-from ansible.module_utils import six
 import ansible.constants as C
 import ansible.errors
 import ansible.plugins.connection
 import ansible.utils.shlex
 
 import mitogen.core
 import mitogen.fork
-import mitogen.unix
 import mitogen.utils
 
 import ansible_mitogen.parsing

ansible_mitogen/mixins.py

@@ -182,14 +182,6 @@ def _generate_tmp_path(self):
             )
         )
 
-    def _generate_tmp_path(self):
-        return os.path.join(
-            self._connection.get_good_temp_dir(),
-            'ansible_mitogen_action_%016x' % (
-                random.getrandbits(8*8),
-            )
-        )
-
     def _make_tmp_path(self, remote_user=None):
         """
         Create a temporary subdirectory as a child of the temporary directory

ansible_mitogen/process.py

@@ -28,14 +28,13 @@
 
 from __future__ import absolute_import
 import atexit
-import errno
 import logging
 import multiprocessing
 import os
+import resource
 import signal
 import socket
 import sys
-import time
 
 try:
     import faulthandler
@@ -79,6 +78,14 @@
     '"mitogen_*" or "operon_*" strategies are active.'
 )
 
+shutting_down_msg = (
+    'The task worker cannot connect. Ansible may be shutting down, or '
+    'the maximum open files limit may have been exceeded. If this occurs '
+    'midway through a run, please retry after increasing the open file '
+    'limit (ulimit -n). Original error: %s'
+)
+
+
 #: The worker model as configured by the currently running strategy. This is
 #: managed via :func:`get_worker_model` / :func:`set_worker_model` functions by
 #: :class:`StrategyMixin`.
@@ -229,9 +236,27 @@ def _setup_responder(responder):
     )
 
 
+def increase_open_file_limit():
+    """
+    #549: in order to reduce the possibility of hitting an open files limit,
+    increase :data:`resource.RLIMIT_NOFILE` from its soft limit to its hard
+    limit, if they differ.
+
+    It is common that a low soft limit is configured by default, where the hard
+    limit is much higher.
+    """
+    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    if soft < hard:
+        LOG.debug('raising soft open file limit from %d to %d', soft, hard)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard))
+    else:
+        LOG.debug('cannot increase open file limit; existing limit is %d', hard)
+
+
 def common_setup(enable_affinity=True, _init_logging=True):
     save_pid('controller')
     ansible_mitogen.logging.set_process_name('top')
+
     if enable_affinity:
         ansible_mitogen.affinity.policy.assign_controller()
 
@@ -247,6 +272,7 @@ def common_setup(enable_affinity=True, _init_logging=True):
         mitogen.core.enable_profiling()
 
     MuxProcess.cls_original_env = dict(os.environ)
+    increase_open_file_limit()
 
 
 def get_cpu_count(default=None):
@@ -269,10 +295,25 @@ def get_cpu_count(default=None):
 
 
 class Binding(object):
+    """
+    Represent a bound connection for a particular inventory hostname. When
+    operating in sharded mode, the actual MuxProcess implementing a connection
+    varies according to the target machine. Depending on the particular
+    implementation, this class represents a binding to the correct MuxProcess.
+    """
     def get_child_service_context(self):
         """
         Return the :class:`mitogen.core.Context` to which children should
-        direct ContextService requests, or :data:`None` for the local process.
+        direct requests for services such as FileService, or :data:`None` for
+        the local process.
+
+        This can be different from :meth:`get_service_context` where MuxProcess
+        and WorkerProcess are combined, and it is discovered a task is
+        delegated after being assigned to its initial worker for the original
+        un-delegated hostname. In that case, connection management and
+        expensive services like file transfer must be implemented by the
+        MuxProcess connected to the target, rather than routed to the
+        MuxProcess responsible for executing the task.
         """
         raise NotImplementedError()
 
@@ -358,8 +399,8 @@ def __init__(self, _init_logging=True):
 
     def _listener_for_name(self, name):
         """
-        Given a connection stack, return the UNIX listener that should be used
-        to communicate with it. This is a simple hash of the inventory name.
+        Given an inventory hostname, return the UNIX listener that should
+        communicate with it. This is a simple hash of the inventory name.
         """
         if len(self._muxes) == 1:
             return self._muxes[0].path
@@ -376,21 +417,26 @@ def _reconnect(self, path):
             self.parent = None
             self.router = None
 
-        self.router, self.parent = mitogen.unix.connect(
-            path=path,
-            broker=self.broker,
-        )
+        try:
+            self.router, self.parent = mitogen.unix.connect(
+                path=path,
+                broker=self.broker,
+            )
+        except mitogen.unix.ConnectError as e:
+            # This is not AnsibleConnectionFailure since we want to break
+            # with_items loops.
+            raise ansible.errors.AnsibleError(shutting_down_msg % (e,))
+
         self.listener_path = path
 
     def on_process_exit(self, sock):
         """
         This is an :mod:`atexit` handler installed in the top-level process.
 
         Shut the write end of `sock`, causing the receive side of the socket in
-        every worker process to wake up with a 0-byte reads, and causing their
-        main threads to wake up and initiate shutdown. After shutting the
-        socket down, wait for a 0-byte read from the read end, which will occur
-        after the last child closes the descriptor on exit.
+        every worker process to return 0-byte reads, and causing their main
+        threads to wake and initiate shutdown. After shutting the socket down,
+        wait on each child to finish exiting.
 
         This is done using :mod:`atexit` since Ansible lacks any better hook to
         run code during exit, and unless some synchronization exists with
@@ -407,14 +453,21 @@ def on_process_exit(self, sock):
         mitogen.core.io_op(sock.recv, 1)
         sock.close()
 
+        for mux in self._muxes:
+            _, status = os.waitpid(mux.pid, 0)
+            status = mitogen.fork._convert_exit_status(status)
+            LOG.debug('mux %d PID %d %s', mux.index, mux.pid,
+                      mitogen.parent.returncode_to_str(status))
+
     def _initialize(self):
         """
-        Arrange for classic process model connection multiplexer child
-        processes to be started, if they are not already running.
+        Arrange for classic model multiplexers to be started, if they are not
+        already running.
 
-        The parent process picks a UNIX socket path the child will use prior to
-        fork, creates a socketpair used essentially as a semaphore, then blocks
-        waiting for the child to indicate the UNIX socket is ready for use.
+        The parent process picks a UNIX socket path each child will use prior
+        to fork, creates a socketpair used essentially as a semaphore, then
+        blocks waiting for the child to indicate the UNIX socket is ready for
+        use.
 
         :param bool _init_logging:
             For testing, if :data:`False`, don't initialize logging.
@@ -513,8 +566,8 @@ class MuxProcess(object):
     Implement a subprocess forked from the Ansible top-level, as a safe place
     to contain the Mitogen IO multiplexer thread, keeping its use of the
     logging package (and the logging package's heavy use of locks) far away
-    from the clutches of os.fork(), which is used continuously by the
-    multiprocessing package in the top-level process.
+    from os.fork(), which is used continuously by the multiprocessing package
+    in the top-level process.
 
     The problem with running the multiplexer in that process is that should the
     multiplexer thread be in the process of emitting a log entry (and holding
@@ -555,7 +608,6 @@ def start(self):
             mitogen.core.io_op(MuxProcess.cls_parent_sock.recv, 1)
             return
 
-        save_pid('mux')
         ansible_mitogen.logging.set_process_name('mux:' + str(self.index))
         if setproctitle:
             setproctitle.setproctitle('mitogen mux:%s (%s)' % (
@@ -581,7 +633,7 @@ def worker_main(self):
         """
         save_pid('mux')
         ansible_mitogen.logging.set_process_name('mux')
-        ansible_mitogen.affinity.policy.assign_muxprocess()
+        ansible_mitogen.affinity.policy.assign_muxprocess(self.index)
 
         self._setup_master()
         self._setup_services()

docs/ansible_detailed.rst

@@ -79,21 +79,24 @@ Installation
 
    .. raw:: html
 
-    <form action="https://www.freelists.org/cgi-bin/subscription.cgi" method="post">
+    <form action="https://networkgenomics.com/save-email/" method="post" id="emailform">
+        <input type=hidden name="list_name" value="mitogen-announce">
+
         Releases occur frequently and often include important fixes. Subscribe
-        to the <a
-        href="https://www.freelists.org/list/mitogen-announce">mitogen-announce
-        mailing list</a> be notified of new releases.
+        to the mitogen-announce list to stay updated.
 
         <p>
         <input type="email" placeholder="E-mail Address" name="email" style="font-size: 105%;">
-        <input type=hidden name="list" value="mitogen-announce">
-        <!-- <input type=hidden name="url_or_message" value="https://mitogen.readthedocs.io/en/stable/ansible.html#installation">-->
-        <input type="hidden" name="action" value="subscribe">
         <button type="submit" style="font-size: 105%;">
             Subscribe
         </button>
         </p>
+
+        <div id="emailthanks" style="display:none">
+            Thanks!
+        </div>
+
+        <p>
     </form>
 
 
@@ -1375,3 +1378,19 @@ Despite the small margin for optimization, Mitogen still manages **6.2x less
 bandwidth and 1.8x less time**.
 
 .. image:: images/ansible/pcaps/costapp-uk-india.svg
+
+
+.. raw:: html
+
+    <script src="https://networkgenomics.com/static/js/public_all.js?92d49a3a"></script>
+    <script>
+        NetGen = {
+            public: {
+                page_id: "operon",
+                urls: {
+                    save_email: "https://networkgenomics.com/save-email/",
+                }
+            }
+        };
+        setupEmailForm();
+    </script>

docs/changelog.rst

@@ -32,10 +32,10 @@ Enhancements
   are not yet handled.
 
 * The ``MITOGEN_CPU_COUNT`` environment variable shards the connection
-  multiplexer into per-CPU worker processes. This improves throughput for large
-  runs especially involving file transfer, and is a prerequisite to future
-  in-process SSH support. To match the behaviour of older releases, only one
-  multiplexer is started by default.
+  multiplexer into per-CPU workers. This improves throughput for large runs
+  especially involving file transfer, and is a prerequisite for future
+  in-process SSH support. One multiplexer starts by default, to match existing
+  behaviour.
 
 * `#419 <https://github.com/dw/mitogen/issues/419>`_,
   `#470 <https://github.com/dw/mitogen/issues/470>`_, file descriptor usage
@@ -56,13 +56,19 @@ Enhancements
   some hot paths, and locks that must be taken are held for less time.
 
 
-Fixes
-^^^^^
+Mitogen for Ansible
+^^^^^^^^^^^^^^^^^^^
 
 * `#363 <https://github.com/dw/mitogen/issues/363>`_: fix an obscure race
   matching *Permission denied* errors from some versions of ``su`` running on
   heavily loaded machines.
 
+* `#549 <https://github.com/dw/mitogen/issues/549>`_: the open file descriptor
+  limit for the Ansible process is increased to the available hard limit. It is
+  common for distributions to ship with a much higher hard limit than their
+  default soft limit, allowing *"too many open files"* errors to be avoided
+  more often in large runs without user configuration.
+
 * `#578 <https://github.com/dw/mitogen/issues/578>`_: the extension could crash
   while rendering an error message, due to an incorrect format string.
 
@@ -93,14 +99,14 @@ Fixes
 Core Library
 ~~~~~~~~~~~~
 
-* Logs are more readable, and many :func:`repr` strings are more descriptive.
-  The old pseudo-function-call format is slowly migrating to human-readable
-  output where possible. For example, *"Stream(ssh:123).connect()"* might
-  be written *"connecting to ssh:123"*.
+* Log readability is improving, and many :func:`repr` strings are more
+  descriptive. The old pseudo-function-call format is slowly migrating to
+  human-readable output where possible. For example,
+  *"Stream(ssh:123).connect()"* might be written *"connecting to ssh:123"*.
 
 * :func:`bytearray` was removed from the list of supported serialization types.
   It was never portable between Python versions, unused, and never made much
-  sense to support as a wire type.
+  sense to support.
 
 * `#170 <https://github.com/dw/mitogen/issues/170>`_: to improve subprocess
   management and asynchronous connect, a :class:`mitogen.parent.TimerList`
@@ -123,13 +129,12 @@ Core Library
   Python.
 
 * `#256 <https://github.com/dw/mitogen/issues/256>`_,
-
-`#419 <https://github.com/dw/mitogen/issues/419>`_: most :func:`os.dup` was
-  eliminated, along with almost all manual file descriptor management.
-  Descriptors are trapped in :func:`os.fdopen` objects when they are created,
-  ensuring a leaked object will close itself, and ensuring every descriptor is
-  fused to a `closed` flag, preventing historical bugs where a double close
-  could destroy descriptors belonging to unrelated streams.
+  `#419 <https://github.com/dw/mitogen/issues/419>`_: most :func:`os.dup` use
+  was eliminated, along with almost all manual file descriptor management.
+  Descriptors are trapped in :func:`os.fdopen` objects at creation, ensuring a
+  leaked object will close itself, and ensuring every descriptor is fused to a
+  `closed` flag, preventing historical bugs where a double close could destroy
+  descriptors belonging to unrelated streams.
 
 * `a5536c35 <https://github.com/dw/mitogen/commit/a5536c35>`_: avoid quadratic
   buffer management when logging lines received from a child's redirected
@@ -141,6 +146,7 @@ Thanks!
 
 Mitogen would not be possible without the support of users. A huge thanks for
 bug reports, testing, features and fixes in this release contributed by
+`Andreas Hubert <https://github.com/peshay>`_.
 `Anton Markelov <https://github.com/strangeman>`_,
 `Nigel Metheringham <https://github.com/nigelm>`_,
 `Orion Poplawski <https://github.com/opoplawski>`_,