Merge remote-tracking branch 'origin/dmw'

* origin/dmw:
  docs: merge signals.rst into internals.rst
  os_fork: do not attempt to cork the active thread.
  parent: fix get_log_level() for split out loggers.
  issue #547: fix service_test failures.
  issue #547: update Changelog.
  issue #547: core/service: race/deadlock-free service pool init
  docs: update Changelog.
  select: make Select.add() handle multiple buffered items.
  core/select: add {Select,Latch,Receiver}.size(), deprecate empty()
  parent: docstring fixes
  core: remove dead Router.on_shutdown() and Router "shutdown" signal
  testlib: use lsof +E for much clearer leaked FD output
  [stream-refactor] stop leaking FD 100 for the life of the child
  core: split preserve_tty_fp() out into a function
  parent: zombie reaping v3
  issue #410: fix test failure due to obsolete parentfp/childfp
  issue #170: replace Timer.cancelled with Timer.active
  core: more descriptive graceful shutdown timeout error
  docs: update changelog
  core: fix Python2.4 crash due to missing Logger.getChild().
  issue #410: automatically work around SELinux braindamage.
  core: cache stream reference in DelimitedProtocol
  parent: docstring formatting
  docs: remove fakessh from home page, it's been broken forever
  docs: add changelog thanks
  Disable Azure pipelines build for docs-master too.
  docs: udpate Changelog.
  docs: tweak Changelog wording
  [linear2] merge fallout: re-enable _send_module_forwards().
  docs: another round of docstring cleanups.
  master: allow filtering forwarded logs using logging package functions.
  docs: many more internals.rst tidyups
  tests: fix error in affinity_test
  service: centralize fetching thread name, and tidy up logs
  [stream-refactor] get caught up on internals.rst updates
  Stop using mitogen root logger in more modules, remove unused loggers
  tests: stop dumping Docker help output in the log.
  parent: move subprocess creation to mux thread too
  Split out and make readable more log messages across both packages
  ansible: log affinity assignments
  ci: log failed command line, and try enabling stdout line buffering
  ansible: improve docstring
  [linear2] simplify _listener_for_name()
  ansible: stop relying on SIGTERM to shut down service pool
  tests: move tty_create_child tests together
  ansible: cleanup various docstrings
  parent: define Connection behaviour during Broker.shutdown()
  issue #549: ansible: reduce risk by capping RLIM_INFINITY

Author: dw

.ci/azure-pipelines.yml

@@ -3,6 +3,11 @@
 # Add steps that analyze code, save the dist with the build record, publish to a PyPI-compatible index, and more:
 # https://docs.microsoft.com/azure/devops/pipelines/languages/python
 
+trigger:
+  branches:
+    exclude:
+    - docs-master
+
 jobs:
 
 - job: Mac

.ci/ci_lib.py

@@ -57,8 +57,10 @@ def have_docker():
 
 # -----------------
 
-# Force stdout FD 1 to be a pipe, so tools like pip don't spam progress bars.
+# Force line buffering on stdout.
+sys.stdout = os.fdopen(1, 'w', 1)
 
+# Force stdout FD 1 to be a pipe, so tools like pip don't spam progress bars.
 if 'TRAVIS_HOME' in os.environ:
     proc = subprocess.Popen(
         args=['stdbuf', '-oL', 'cat'],
@@ -86,8 +88,13 @@ def _argv(s, *args):
 def run(s, *args, **kwargs):
     argv = ['/usr/bin/time', '--'] + _argv(s, *args)
     print('Running: %s' % (argv,))
-    ret = subprocess.check_call(argv, **kwargs)
-    print('Finished running: %s' % (argv,))
+    try:
+        ret = subprocess.check_call(argv, **kwargs)
+        print('Finished running: %s' % (argv,))
+    except Exception:
+        print('Exception occurred while running: %s' % (argv,))
+        raise
+
     return ret
 
 
@@ -217,6 +224,7 @@ def start_containers(containers):
             "docker rm -f %(name)s || true" % container,
             "docker run "
                 "--rm "
+                "--cpuset-cpus 0,1 "
                 "--detach "
                 "--privileged "
                 "--cap-add=SYS_PTRACE "

ansible_mitogen/affinity.py

@@ -73,7 +73,9 @@
 decisions.
 """
 
+from __future__ import absolute_import
 import ctypes
+import logging
 import mmap
 import multiprocessing
 import os
@@ -83,6 +85,9 @@
 import mitogen.parent
 
 
+LOG = logging.getLogger(__name__)
+
+
 try:
     _libc = ctypes.CDLL(None, use_errno=True)
     _strerror = _libc.strerror
@@ -207,40 +212,42 @@ def __init__(self, cpu_count=None):
             self._reserve_mask = 3
             self._reserve_shift = 2
 
-    def _set_affinity(self, mask):
+    def _set_affinity(self, descr, mask):
+        if descr:
+            LOG.debug('CPU mask for %s: %#08x', descr, mask)
         mitogen.parent._preexec_hook = self._clear
         self._set_cpu_mask(mask)
 
-    def _balance(self):
+    def _balance(self, descr):
         self.state.lock.acquire()
         try:
             n = self.state.counter
             self.state.counter += 1
         finally:
             self.state.lock.release()
 
-        self._set_cpu(self._reserve_shift + (
+        self._set_cpu(descr, self._reserve_shift + (
             (n % (self.cpu_count - self._reserve_shift))
         ))
 
-    def _set_cpu(self, cpu):
-        self._set_affinity(1 << (cpu % self.cpu_count))
+    def _set_cpu(self, descr, cpu):
+        self._set_affinity(descr, 1 << (cpu % self.cpu_count))
 
     def _clear(self):
         all_cpus = (1 << self.cpu_count) - 1
-        self._set_affinity(all_cpus & ~self._reserve_mask)
+        self._set_affinity(None, all_cpus & ~self._reserve_mask)
 
     def assign_controller(self):
         if self._reserve_controller:
-            self._set_cpu(1)
+            self._set_cpu('Ansible top-level process', 1)
         else:
-            self._balance()
+            self._balance('Ansible top-level process')
 
     def assign_muxprocess(self, index):
-        self._set_cpu(index)
+        self._set_cpu('MuxProcess %d' % (index,), index)
 
     def assign_worker(self):
-        self._balance()
+        self._balance('WorkerProcess')
 
     def assign_subprocess(self):
         self._clear()

ansible_mitogen/process.py

@@ -246,7 +246,14 @@ def increase_open_file_limit():
     limit is much higher.
     """
     soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
-    LOG.debug('inherited open file limits: soft=%d hard=%d', soft, hard)
+    if hard == resource.RLIM_INFINITY:
+        hard_s = '(infinity)'
+        # cap in case of O(RLIMIT_NOFILE) algorithm in some subprocess.
+        hard = 524288
+    else:
+        hard_s = str(hard)
+
+    LOG.debug('inherited open file limits: soft=%d hard=%s', soft, hard_s)
     if soft >= hard:
         LOG.debug('max open files already set to hard limit: %d', hard)
         return
@@ -268,13 +275,13 @@ def common_setup(enable_affinity=True, _init_logging=True):
     save_pid('controller')
     ansible_mitogen.logging.set_process_name('top')
 
+    if _init_logging:
+        ansible_mitogen.logging.setup()
+
     if enable_affinity:
         ansible_mitogen.affinity.policy.assign_controller()
 
     mitogen.utils.setup_gil()
-    if _init_logging:
-        ansible_mitogen.logging.setup()
-
     if faulthandler is not None:
         faulthandler.enable()
 
@@ -352,6 +359,11 @@ def close(self):
 
 
 class WorkerModel(object):
+    """
+    Interface used by StrategyMixin to manage various Mitogen services, by
+    default running in one or more connection multiplexer subprocesses spawned
+    off the top-level Ansible process.
+    """
     def on_strategy_start(self):
         """
         Called prior to strategy start in the top-level process. Responsible
@@ -368,6 +380,11 @@ def on_strategy_complete(self):
         raise NotImplementedError()
 
     def get_binding(self, inventory_name):
+        """
+        Return a :class:`Binding` to access Mitogen services for
+        `inventory_name`. Usually called from worker processes, but may also be
+        called from top-level process to handle "meta: reset_connection".
+        """
         raise NotImplementedError()
 
 
@@ -427,13 +444,10 @@ class ClassicWorkerModel(WorkerModel):
 
     def __init__(self, _init_logging=True):
         """
-        Arrange for classic model multiplexers to be started, if they are not
-        already running.
-
-        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.
+        Arrange for classic model multiplexers to be started. The parent choses
+        UNIX socket paths 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.
@@ -466,12 +480,10 @@ def _listener_for_name(self, 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
-
-        idx = abs(hash(name)) % len(self._muxes)
-        LOG.debug('Picked worker %d: %s', idx, self._muxes[idx].path)
-        return self._muxes[idx].path
+        mux = self._muxes[abs(hash(name)) % len(self._muxes)]
+        LOG.debug('will use multiplexer %d (%s) to connect to "%s"',
+                  mux.index, mux.path, name)
+        return mux.path
 
     def _reconnect(self, path):
         if self.router is not None:
@@ -498,9 +510,9 @@ def _on_process_exit(self):
         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 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.
+        every :class:`MuxProcess` 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
@@ -523,30 +535,19 @@ def _on_process_exit(self):
         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,
+            LOG.debug('multiplexer %d PID %d %s', mux.index, mux.pid,
                       mitogen.parent.returncode_to_str(status))
 
     def _test_reset(self):
         """
         Used to clean up in unit tests.
         """
-        # TODO: split this up a bit.
-        global _classic_worker_model
-        assert self.parent_sock is not None
-        self.parent_sock.close()
-        self.parent_sock = None
-        self.listener_path = None
-        self.router = None
-        self.parent = None
-
-        for mux in self._muxes:
-            pid, status = os.waitpid(mux.pid, 0)
-            status = mitogen.fork._convert_exit_status(status)
-            LOG.debug('mux PID %d %s', pid,
-                mitogen.parent.returncode_to_str(status))
+        self.on_binding_close()
+        self._on_process_exit()
+        set_worker_model(None)
 
+        global _classic_worker_model
         _classic_worker_model = None
-        set_worker_model(None)
 
     def on_strategy_start(self):
         """
@@ -579,6 +580,7 @@ def on_binding_close(self):
         self.broker.join()
         self.router = None
         self.broker = None
+        self.parent = None
         self.listener_path = None
 
         # #420: Ansible executes "meta" actions in the top-level process,
@@ -694,8 +696,8 @@ def _setup_master(self):
             max_message_size=4096 * 1048576,
         )
         _setup_responder(self.router.responder)
-        mitogen.core.listen(self.broker, 'shutdown', self.on_broker_shutdown)
-        mitogen.core.listen(self.broker, 'exit', self.on_broker_exit)
+        mitogen.core.listen(self.broker, 'shutdown', self._on_broker_shutdown)
+        mitogen.core.listen(self.broker, 'exit', self._on_broker_exit)
         self.listener = mitogen.unix.Listener.build_stream(
             router=self.router,
             path=self.path,
@@ -715,26 +717,20 @@ def _setup_services(self):
         )
         setup_pool(self.pool)
 
-    def on_broker_shutdown(self):
+    def _on_broker_shutdown(self):
         """
-        Respond to broker shutdown by beginning service pool shutdown. Do not
-        join on the pool yet, since that would block the broker thread which
-        then cannot clean up pending handlers, which is required for the
-        threads to exit gracefully.
+        Respond to broker shutdown by shutting down the pool. Do not join on it
+        yet, since that would block the broker thread which then cannot clean
+        up pending handlers and connections, which is required for the threads
+        to exit gracefully.
         """
-        # In normal operation we presently kill the process because there is
-        # not yet any way to cancel connect().
-        self.pool.stop(join=self.profiling)
+        self.pool.stop(join=False)
 
-    def on_broker_exit(self):
+    def _on_broker_exit(self):
         """
-        Respond to the broker thread about to exit by sending SIGTERM to
-        ourself. In future this should gracefully join the pool, but TERM is
-        fine for now.
+        Respond to the broker thread about to exit by finally joining on the
+        pool. This is safe since pools only block in connection attempts, and
+        connection attempts fail with CancelledError when broker shutdown
+        begins.
         """
-        if not os.environ.get('MITOGEN_PROFILING'):
-            # In normal operation we presently kill the process because there is
-            # not yet any way to cancel connect(). When profiling, threads
-            # including the broker must shut down gracefully, otherwise pstats
-            # won't be written.
-            os.kill(os.getpid(), signal.SIGTERM)
+        self.pool.join()

ansible_mitogen/services.py

@@ -180,7 +180,7 @@ def put(self, context):
         Return a reference, making it eligable for recycling once its reference
         count reaches zero.
         """
-        LOG.debug('%r.put(%r)', self, context)
+        LOG.debug('decrementing reference count for %r', context)
         self._lock.acquire()
         try:
             if self._refs_by_context.get(context, 0) == 0:
@@ -326,7 +326,6 @@ def _on_context_disconnect(self, context):
     )
 
     def _send_module_forwards(self, context):
-        return
         self.router.responder.forward_modules(context, self.ALWAYS_PRELOAD)
 
     _candidate_temp_dirs = None
@@ -383,7 +382,6 @@ def _connect(self, key, spec, via=None):
         mitogen.core.listen(context, 'disconnect',
             lambda: self._on_context_disconnect(context))
 
-        #self._send_module_forwards(context) TODO
         self._send_module_forwards(context)
         init_child_result = context.call(
             ansible_mitogen.target.init_child,