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 108015aa22
  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

ansible_mitogen/affinity.py

@@ -142,7 +142,7 @@ class Policy(object):
         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 @@ class FixedPolicy(Policy):
         ))
 
     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 @@ class FixedPolicy(Policy):
         else:
             self._balance()
 
-    def assign_muxprocess(self):
+    def assign_muxprocess(self, index):
-        self._set_cpu(0)
+        self._set_cpu(index)
 
     def assign_worker(self):
         self._balance()

ansible_mitogen/connection.py

@@ -37,8 +37,6 @@ import stat
 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
@@ -46,7 +44,6 @@ 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 @@ class ActionModuleMixin(ansible.plugins.action.ActionBase):
             )
         )
 
-    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 @@ worker_model_msg = (
     '"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 @@ class ClassicWorkerModel(WorkerModel):
 
     def _listener_for_name(self, name):
         """
-        Given a connection stack, return the UNIX listener that should be used
+        Given an inventory hostname, return the UNIX listener that should
-        to communicate with it. This is a simple hash of the inventory name.
+        communicate with it. This is a simple hash of the inventory name.
         """
         if len(self._muxes) == 1:
             return self._muxes[0].path
@@ -376,10 +417,16 @@ class ClassicWorkerModel(WorkerModel):
             self.parent = None
             self.router = None
 
-        self.router, self.parent = mitogen.unix.connect(
+        try:
-            path=path,
+            self.router, self.parent = mitogen.unix.connect(
-            broker=self.broker,
+                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):
@@ -387,10 +434,9 @@ class ClassicWorkerModel(WorkerModel):
         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
+        every worker process to return 0-byte reads, and causing their main
-        main threads to wake up and initiate shutdown. After shutting the
+        threads to wake and initiate shutdown. After shutting the socket down,
-        socket down, wait for a 0-byte read from the read end, which will occur
+        wait on each child to finish exiting.
-        after the last child closes the descriptor on exit.
 
         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 @@ class ClassicWorkerModel(WorkerModel):
         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
+        Arrange for classic model multiplexers to be started, if they are not
-        processes to be started, if they are not already running.
+        already running.
 
-        The parent process picks a UNIX socket path the child will use prior to
+        The parent process picks a UNIX socket path each child will use prior
-        fork, creates a socketpair used essentially as a semaphore, then blocks
+        to fork, creates a socketpair used essentially as a semaphore, then
-        waiting for the child to indicate the UNIX socket is ready for use.
+        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
+    from os.fork(), which is used continuously by the multiprocessing package
-    multiprocessing package in the top-level process.
+    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 @@ class MuxProcess(object):
             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 @@ class MuxProcess(object):
         """
         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
+        to the mitogen-announce list to stay updated.
-        href="https://www.freelists.org/list/mitogen-announce">mitogen-announce
-        mailing list</a> be notified of new releases.
 
         <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
+  multiplexer into per-CPU workers. This improves throughput for large runs
-  runs especially involving file transfer, and is a prerequisite to future
+  especially involving file transfer, and is a prerequisite for future
-  in-process SSH support. To match the behaviour of older releases, only one
+  in-process SSH support. One multiplexer starts by default, to match existing
-  multiplexer is started by default.
+  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.
+* Log readability is improving, and many :func:`repr` strings are more
-  The old pseudo-function-call format is slowly migrating to human-readable
+  descriptive. The old pseudo-function-call format is slowly migrating to
-  output where possible. For example, *"Stream(ssh:123).connect()"* might
+  human-readable output where possible. For example,
-  be written *"connecting to ssh:123"*.
+  *"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` use
-`#419 <https://github.com/dw/mitogen/issues/419>`_: most :func:`os.dup` was
+  was eliminated, along with almost all manual file descriptor management.
-  eliminated, along with almost all manual file descriptor management.
+  Descriptors are trapped in :func:`os.fdopen` objects at creation, ensuring a
-  Descriptors are trapped in :func:`os.fdopen` objects when they are created,
+  leaked object will close itself, and ensuring every descriptor is fused to a
-  ensuring a leaked object will close itself, and ensuring every descriptor is
+  `closed` flag, preventing historical bugs where a double close could destroy
-  fused to a `closed` flag, preventing historical bugs where a double close
+  descriptors belonging to unrelated streams.
-  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>`_,

docs/conf.py

@@ -9,6 +9,7 @@ author = u'Network Genomics'
 copyright = u'2019, Network Genomics'
 exclude_patterns = ['_build']
 extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinxcontrib.programoutput']
+html_show_copyright = False
 html_show_sourcelink = False
 html_show_sphinx = False
 html_sidebars = {'**': ['globaltoc.html', 'github.html']}

mitogen/core.py

@@ -3109,7 +3109,7 @@ class Broker(object):
         if sys.version_info < (2, 6):
             # import_module() is used to avoid dep scanner.
             os_fork = import_module('mitogen.os_fork')
-            mitogen.os_fork._notice_broker_or_pool(self)
+            os_fork._notice_broker_or_pool(self)
 
     def start_receive(self, stream):
         """

mitogen/doas.py

@@ -33,7 +33,6 @@ import re
 
 import mitogen.core
 import mitogen.parent
-from mitogen.core import b
 
 
 LOG = logging.getLogger(__name__)

mitogen/parent.py

@@ -1137,7 +1137,8 @@ class BootstrapProtocol(RegexProtocol):
         return False
 
     def on_unrecognized_line_received(self, line):
-        LOG.debug('%s: stdout: %s', self.stream.name, line)
+        LOG.debug('%s: stdout: %s', self.stream.name,
+            line.decode('utf-8', 'replace'))
 
     PATTERNS = [
         (re.compile(EC0_MARKER), _on_ec0_received),

mitogen/profiler.py

@@ -57,15 +57,12 @@ Example:
 from __future__ import print_function
 import os
 import pstats
-import cProfile
 import shutil
 import subprocess
 import sys
 import tempfile
 import time
 
-import mitogen.core
-
 
 def try_merge(stats, path):
     try:

mitogen/service.py

@@ -29,6 +29,7 @@
 # !mitogen: minify_safe
 
 import grp
+import logging
 import os
 import os.path
 import pprint
@@ -41,7 +42,6 @@ import time
 import mitogen.core
 import mitogen.select
 from mitogen.core import b
-from mitogen.core import LOG
 from mitogen.core import str_rpartition
 
 try:
@@ -54,6 +54,8 @@ except NameError:
         return True
 
 
+LOG = logging.getLogger(__name__)
+
 DEFAULT_POOL_SIZE = 16
 _pool = None
 _pool_pid = None
@@ -501,7 +503,7 @@ class Pool(object):
         self._py_24_25_compat()
         self._threads = []
         for x in range(size):
-            name = 'mitogen.service.Pool.%x.worker-%d' % (id(self), x,)
+            name = 'mitogen.Pool.%04x.%d' % (id(self) & 0xffff, x,)
             thread = threading.Thread(
                 name=name,
                 target=mitogen.core._profile_hook,
@@ -608,9 +610,11 @@ class Pool(object):
         while not self.closed:
             try:
                 event = self._select.get_event()
-            except (mitogen.core.ChannelError, mitogen.core.LatchError):
+            except mitogen.core.LatchError:
-                e = sys.exc_info()[1]
+                LOG.debug('%r: graceful exit', self)
-                LOG.debug('%r: channel or latch closed, exitting: %s', self, e)
+                return
+            except mitogen.core.ChannelError:
+                LOG.debug('%r: exitting: %s', self, sys.exc_info()[1])
                 return
 
             func = self._func_by_source[event.source]
@@ -629,8 +633,8 @@ class Pool(object):
 
     def __repr__(self):
         th = threading.currentThread()
-        return 'mitogen.service.Pool(%#x, size=%d, th=%r)' % (
+        return 'Pool(%04x, size=%d, th=%r)' % (
-            id(self),
+            id(self) & 0xffff,
             len(self._threads),
             th.getName(),
         )

mitogen/ssh.py

@@ -42,7 +42,6 @@ except ImportError:
 
 import mitogen.parent
 from mitogen.core import b
-from mitogen.core import bytes_partition
 
 try:
     any

mitogen/su.py

@@ -33,7 +33,6 @@ import re
 
 import mitogen.core
 import mitogen.parent
-from mitogen.core import b
 
 try:
     any

mitogen/sudo.py

@@ -35,7 +35,6 @@ import re
 
 import mitogen.core
 import mitogen.parent
-from mitogen.core import b
 
 
 LOG = logging.getLogger(__name__)

mitogen/unix.py

@@ -48,6 +48,22 @@ import mitogen.master
 from mitogen.core import LOG
 
 
+class Error(mitogen.core.Error):
+    """
+    Base for errors raised by :mod:`mitogen.unix`.
+    """
+    pass
+
+
+class ConnectError(Error):
+    """
+    Raised when :func:`mitogen.unix.connect` fails to connect to the listening
+    socket.
+    """
+    #: UNIX error number reported by underlying exception.
+    errno = None
+
+
 def is_path_dead(path):
     s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     try:
@@ -154,9 +170,19 @@ class Listener(mitogen.core.Protocol):
 
 
 def _connect(path, broker, sock):
-    sock.connect(path)
+    try:
-    sock.send(struct.pack('>L', os.getpid()))
+        # ENOENT, ECONNREFUSED
-    mitogen.context_id, remote_id, pid = struct.unpack('>LLL', sock.recv(12))
+        sock.connect(path)
+
+        # ECONNRESET
+        sock.send(struct.pack('>L', os.getpid()))
+        mitogen.context_id, remote_id, pid = struct.unpack('>LLL', sock.recv(12))
+    except socket.error:
+        e = sys.exc_info()[1]
+        ce = ConnectError('could not connect to %s: %s', path, e.args[1])
+        ce.errno = e.args[0]
+        raise ce
+
     mitogen.parent_id = remote_id
     mitogen.parent_ids = [remote_id]
 

scripts/affin.sh

@@ -0,0 +1,4 @@
+# show process affinities for running ansible-playbook
+who="$1"
+[ ! "$who" ] && who=ansible-playbook
+for i in $(pgrep -f "$who") ; do taskset -c -p $i ; done|cut -d: -f2|sort -n |uniq -c

scripts/release-notes.py

@@ -3,7 +3,6 @@
 # Generate the fragment used to make email release announcements
 # usage: release-notes.py 0.2.6
 
-import os
 import sys
 import urllib
 import lxml.html

tests/ansible/gcloud/.gitignore

@@ -0,0 +1,2 @@
+terraform.tfstate*
+.terraform

tests/ansible/gcloud/Makefile

@@ -0,0 +1,3 @@
+
+default:
+	terraform fmt

tests/ansible/gcloud/controller.yml

@@ -1,19 +1,89 @@
 
-- hosts: controller
+- hosts: all
-  vars:
+  become: true
-    git_username: '{{ lookup("pipe", "git config --global user.name") }}'
-    git_email: '{{ lookup("pipe", "git config --global user.email") }}'
-
   tasks:
+    - apt: name={{item}} state=installed
+      with_items:
+      - openvpn
+      - tcpdump
+      - python-pip
+      - python-virtualenv
+      - strace
+      - libldap2-dev
+      - linux-perf
+      - libsasl2-dev
+      - build-essential
+      - git
+      - rsync
+
+    - file:
+        path: /etc/openvpn
+        state: directory
+
+    - copy:
+        dest: /etc/openvpn/secret
+        mode: '0600'
+        content: |
+          -----BEGIN OpenVPN Static key V1-----
+          f94005e4206828e281eb397aefd69b37
+          ebe6cd39057d5641c5d8dd539cd07651
+          557d94d0077852bd8f92b68bef927169
+          c5f0e42ac962a2cbbed35e107ffa0e71
+          1a2607c6bcd919ec5846917b20eb6684
+          c7505152815d6ed7b4420714777a3d4a
+          8edb27ca81971cba7a1e88fe3936e13b
+          85e9be6706a30cd1334836ed0f08e899
+          78942329a330392dff42e4570731ac24
+          9330358aaa6828c07ecb41fb9c498a89
+          1e0435c5a45bfed390cd2104073634ef
+          b00f9fae1d3c49ef5de51854103edac9
+          5ff39c9dfc66ae270510b2ffa74d87d2
+          9d4b3844b1e1473237bc6dc78fb03e2e
+          643ce58e667a532efceec7177367fb37
+          a16379a51e0a8c8e3ec00a59952b79d4
+          -----END OpenVPN Static key V1-----
+
+    - copy:
+        dest: /etc/openvpn/k3.conf
+        content: |
+          remote k3.botanicus.net
+          dev tun
+          ifconfig 10.18.0.1 10.18.0.2
+          secret secret
+
+    - shell: systemctl enable openvpn@k3.service
+    - shell: systemctl start openvpn@k3.service
+
     - lineinfile:
         line: "{{item}}"
         path: /etc/sysctl.conf
       register: sysctl_conf
-      become: true
       with_items:
         - "net.ipv4.ip_forward=1"
         - "kernel.perf_event_paranoid=-1"
 
+    - shell: /sbin/sysctl -p
+      when: sysctl_conf.changed
+
+    - copy:
+        dest: /etc/rc.local
+        mode: "0744"
+        content: |
+          #!/bin/bash
+          iptables -t nat -F;
+          iptables -t nat -X;
+          iptables -t nat -A POSTROUTING -j MASQUERADE;
+
+    - shell: systemctl daemon-reload
+    - shell: systemctl enable rc-local
+    - shell: systemctl start rc-local
+
+
+- hosts: all
+  vars:
+    git_username: '{{ lookup("pipe", "git config --global user.name") }}'
+    git_email: '{{ lookup("pipe", "git config --global user.email") }}'
+  tasks:
     - copy:
         src: ~/.ssh/id_gitlab
         dest: ~/.ssh/id_gitlab
@@ -23,38 +93,6 @@
         dest: ~/.ssh/config
         src: ssh_config.j2
 
-    - lineinfile:
-        line: "{{item}}"
-        path: /etc/sysctl.conf
-      become: true
-      with_items:
-      - net.ipv4.ip_forward=1
-      - kernel.perf_event_paranoid=-1
-      register: sysctl_conf
-
-    - shell: /sbin/sysctl -p
-      when: sysctl_conf.changed
-      become: true
-
-    - shell: |
-        iptables -t nat -F;
-        iptables -t nat -X;
-        iptables -t nat -A POSTROUTING -j MASQUERADE;
-      become: true
-
-    - apt: name={{item}} state=installed
-      become: true
-      with_items:
-      - python-pip
-      - python-virtualenv
-      - strace
-      - libldap2-dev
-      - linux-perf
-      - libsasl2-dev
-      - build-essential
-      - git
-      - rsync
-
     - shell: "rsync -a ~/.ssh {{inventory_hostname}}:"
       connection: local
 
@@ -119,4 +157,3 @@
         path: ~/prj/ansible/inventory/gcloud.py
         state: link
         src: ~/mitogen/tests/ansible/lib/inventory/gcloud.py
-

tests/ansible/gcloud/gce.yml

@@ -1,11 +0,0 @@
-
-- hosts: localhost
-  tasks:
-    - command: date +%Y%m%d-%H%M%S
-      register: out
-
-    - set_fact:
-        instance_name: "controller-{{out.stdout}}"
-
-    - command: >
-        gcloud compute instances create {{instance_name}} --can-ip-forward --machine-type=n1-standard-8 --preemptible --scopes=compute-ro --image-project=debian-cloud --image-family=debian-9

tests/ansible/gcloud/mitogen-load-testing.tf

@@ -0,0 +1,145 @@
+variable "node-count" {
+  default = 0
+}
+
+variable "big" {
+    default = false
+}
+
+provider "google" {
+  project = "mitogen-load-testing"
+  region  = "europe-west1"
+  zone    = "europe-west1-d"
+}
+
+resource "google_compute_instance" "controller" {
+  name = "ansible-controller"
+  machine_type = "${var.big ? "n1-highcpu-32" : "custom-1-1024"}"
+
+  allow_stopping_for_update = true
+  can_ip_forward            = true
+
+  boot_disk {
+    initialize_params {
+      image = "debian-cloud/debian-9"
+    }
+  }
+
+  scheduling {
+    preemptible       = true
+    automatic_restart = false
+  }
+
+  network_interface {
+    subnetwork    = "${google_compute_subnetwork.loadtest-subnet.self_link}"
+    access_config = {}
+  }
+
+  provisioner "local-exec" {
+    command = <<-EOF
+        ip=${google_compute_instance.controller.network_interface.0.access_config.0.nat_ip};
+        ssh-keygen -R $ip;
+        ssh-keyscan $ip >> ~/.ssh/known_hosts;
+        sed -ri -e "s/.*CONTROLLER_IP_HERE.*/    Hostname $ip/" ~/.ssh/config;
+        ansible-playbook -i $ip, controller.yml
+    EOF
+  }
+}
+
+resource "google_compute_network" "loadtest" {
+  name                    = "loadtest"
+  auto_create_subnetworks = false
+}
+
+resource "google_compute_subnetwork" "loadtest-subnet" {
+  name          = "loadtest-subnet"
+  ip_cidr_range = "10.19.0.0/16"
+  network       = "${google_compute_network.loadtest.id}"
+}
+
+resource "google_compute_firewall" "allow-all-in" {
+  name      = "allow-all-in"
+  network   = "${google_compute_network.loadtest.name}"
+  direction = "INGRESS"
+
+  allow {
+    protocol = "all"
+  }
+}
+
+resource "google_compute_firewall" "allow-all-out" {
+  name      = "allow-all-out"
+  network   = "${google_compute_network.loadtest.name}"
+  direction = "EGRESS"
+
+  allow {
+    protocol = "all"
+  }
+}
+
+resource "google_compute_route" "route-nodes-via-controller" {
+  name                   = "route-nodes-via-controller"
+  dest_range             = "0.0.0.0/0"
+  network                = "${google_compute_network.loadtest.name}"
+  next_hop_instance      = "${google_compute_instance.controller.self_link}"
+  next_hop_instance_zone = "${google_compute_instance.controller.zone}"
+  priority               = 800
+  tags                   = ["node"]
+}
+
+resource "google_compute_instance_template" "node" {
+  name         = "node"
+  tags         = ["node"]
+  machine_type = "custom-1-1024"
+
+  scheduling {
+    preemptible       = true
+    automatic_restart = false
+  }
+
+  disk {
+    source_image = "debian-cloud/debian-9"
+    auto_delete  = true
+    boot         = true
+  }
+
+  network_interface {
+    subnetwork = "${google_compute_subnetwork.loadtest-subnet.self_link}"
+  }
+}
+
+#
+# Compute Engine tops out at 1000 VMs per group
+#
+
+resource "google_compute_instance_group_manager" "nodes-a" {
+  name = "nodes-a"
+
+  base_instance_name = "node"
+  instance_template  = "${google_compute_instance_template.node.self_link}"
+  target_size        = "${var.node-count / 4}"
+}
+
+resource "google_compute_instance_group_manager" "nodes-b" {
+  name = "nodes-b"
+
+  base_instance_name = "node"
+  instance_template  = "${google_compute_instance_template.node.self_link}"
+  target_size        = "${var.node-count / 4}"
+}
+
+resource "google_compute_instance_group_manager" "nodes-c" {
+  name = "nodes-c"
+
+  base_instance_name = "node"
+  instance_template  = "${google_compute_instance_template.node.self_link}"
+  target_size        = "${var.node-count / 4}"
+}
+
+resource "google_compute_instance_group_manager" "nodes-d" {
+  name = "nodes-d"
+
+  base_instance_name = "node"
+  instance_template  = "${google_compute_instance_template.node.self_link}"
+  target_size        = "${var.node-count / 4}"
+}

tests/ansible/lib/inventory/gcloud.py

@@ -14,14 +14,14 @@ import googleapiclient.discovery
 def main():
     project = 'mitogen-load-testing'
     zone = 'europe-west1-d'
-    group_name = 'micro-debian9'
+    prefix = 'node-'
 
     client = googleapiclient.discovery.build('compute', 'v1')
     resp = client.instances().list(project=project, zone=zone).execute()
 
     ips = []
     for inst in resp['items']:
-        if inst['status'] == 'RUNNING' and inst['name'].startswith(group_name):
+        if inst['status'] == 'RUNNING' and inst['name'].startswith(prefix):
             ips.extend(
                 #bytes(config['natIP'])
                 bytes(interface['networkIP'])

tests/ansible/tests/affinity_test.py

@@ -64,32 +64,32 @@ class FixedPolicyTest(testlib.TestCase):
     def test_assign_muxprocess_1core(self):
         # Uniprocessor .
         policy = self.klass(cpu_count=1)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
 
     def test_assign_muxprocess_2core(self):
         # Small SMP gets dedicated core.
         policy = self.klass(cpu_count=2)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
 
     def test_assign_muxprocess_3core(self):
         # Small SMP gets a dedicated core.
         policy = self.klass(cpu_count=3)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
 
     def test_assign_muxprocess_4core(self):
         # Big SMP gets a dedicated core.
         policy = self.klass(cpu_count=4)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
-        policy.assign_muxprocess()
+        policy.assign_muxprocess(0)
         self.assertEquals(0x1, policy.mask)
 
     def test_assign_worker_1core(self):

tests/ansible/tests/connection_test.py

@@ -23,6 +23,8 @@ import testlib
 
 
 class MuxProcessMixin(object):
+    no_zombie_check = True
+
     @classmethod
     def setUpClass(cls):
         #mitogen.utils.log_to_file()
@@ -61,7 +63,23 @@ class ConnectionMixin(MuxProcessMixin):
         super(ConnectionMixin, self).tearDown()
 
 
-class OptionalIntTest(unittest2.TestCase):
+class MuxShutdownTest(ConnectionMixin, testlib.TestCase):
+    def test_connection_failure_raised(self):
+        # ensure if a WorkerProcess tries to connect to a MuxProcess that has
+        # already shut down, it fails with a graceful error.
+        path = self.model._muxes[0].path
+        os.rename(path, path + '.tmp')
+        try:
+            #e = self.assertRaises(ansible.errors.AnsibleError,
+                #lambda: self.conn._connect()
+            #)
+            e = 1
+            print(e)
+        finally:
+            os.rename(path + '.tmp', path)
+
+
+class OptionalIntTest(testlib.TestCase):
     func = staticmethod(ansible_mitogen.connection.optional_int)
 
     def test_already_int(self):
@@ -81,7 +99,7 @@ class OptionalIntTest(unittest2.TestCase):
         self.assertEquals(None, self.func({1:2}))
 
 
-class PutDataTest(ConnectionMixin, unittest2.TestCase):
+class PutDataTest(ConnectionMixin, testlib.TestCase):
     def test_out_path(self):
         path = tempfile.mktemp(prefix='mitotest')
         contents = mitogen.core.b('contents')
@@ -102,7 +120,7 @@ class PutDataTest(ConnectionMixin, unittest2.TestCase):
         os.unlink(path)
 
 
-class PutFileTest(ConnectionMixin, unittest2.TestCase):
+class PutFileTest(ConnectionMixin, testlib.TestCase):
     @classmethod
     def setUpClass(cls):
         super(PutFileTest, cls).setUpClass()

tests/testlib.py

@@ -343,7 +343,14 @@ class TestCase(unittest2.TestCase):
                 self, self._fd_count_before, get_fd_count(),
             )
 
+    # Some class fixtures (like Ansible MuxProcess) start persistent children
+    # for the duration of the class.
+    no_zombie_check = False
+
     def _teardown_check_zombies(self):
+        if self.no_zombie_check:
+            return
+
         try:
             pid, status = os.waitpid(0, os.WNOHANG)
         except OSError:
@@ -354,7 +361,7 @@ class TestCase(unittest2.TestCase):
                 self, pid, status
             )
 
-        print()
+        print('')
         print('Children of unit test process:')
         os.system('ps uww --ppid ' + str(os.getpid()))
         assert 0, "%s leaked still-running subprocesses." % (self,)

tests/unix_test.py

@@ -90,7 +90,7 @@ class ClientTest(testlib.TestCase):
         while True:
             try:
                 return mitogen.unix.connect(path)
-            except socket.error:
+            except mitogen.unix.ConnectError:
                 if time.time() > timeout:
                     raise
                 time.sleep(0.1)