Merge remote-tracking branch 'origin/linear2'

* origin/linear2:
  [linear2] fix another test relying on Connection.parent
  [linear2] more merge fallout, fix Connection._mitogen_reset(mode=)
  [linear2] update mitogen_get_stack for new _build_stack() return value
  [linear2] fix mitogen_shutdown_all service context access
  docs: changelog concision / additions
  add 363 to changelog
  docs: update Changelog
  docs: note fd usage has halved
  docs: more stream-refactor work
  docs: update Changelog for stream-refactor.
  docs: Add lineinfile bug to changelog.
  [linear2] fix MuxProcess test fixture and some merge fallout
  service: avoid taking another lock in the usual case
  service: don't acquire lock when pool already initialized
  profiler: marginal improvements
  core: ensure 'exit' signal fires even on Broker crash.
  core: wake Waker outside of lock.
  core: wake Latch outside of lock.
  core: remove old blocking call guard, it's in the wrong place
  Make setting affinity optional.
  ansible: abstract worker process model.
  [stream-refactor] parent: fix crash on graceful shutdown
  parent: tidy up create_socketpair()
  core: more concise Side.repr.

Author: dw

ansible_mitogen/connection.py

@@ -38,6 +38,7 @@
 import time
 
 import jinja2.runtime
+from ansible.module_utils import six
 import ansible.constants as C
 import ansible.errors
 import ansible.plugins.connection
@@ -459,15 +460,10 @@ def call(self, func, *args, **kwargs):
 
 
 class Connection(ansible.plugins.connection.ConnectionBase):
-    #: mitogen.master.Broker for this worker.
-    broker = None
-
-    #: mitogen.master.Router for this worker.
-    router = None
-
-    #: mitogen.parent.Context representing the parent Context, which is
-    #: presently always the connection multiplexer process.
-    parent = None
+    #: The :class:`ansible_mitogen.process.Binding` representing the connection
+    #: multiplexer this connection's target is assigned to. :data:`None` when
+    #: disconnected.
+    binding = None
 
     #: mitogen.parent.Context for the target account on the target, possibly
     #: reached via become.
@@ -518,13 +514,6 @@ class Connection(ansible.plugins.connection.ConnectionBase):
     #: matching vanilla Ansible behaviour.
     loader_basedir = None
 
-    def __init__(self, play_context, new_stdin, **kwargs):
-        assert ansible_mitogen.process.MuxProcess.unix_listener_path, (
-            'Mitogen connection types may only be instantiated '
-            'while the "mitogen" strategy is active.'
-        )
-        super(Connection, self).__init__(play_context, new_stdin)
-
     def __del__(self):
         """
         Ansible cannot be trusted to always call close() e.g. the synchronize
@@ -585,6 +574,15 @@ def homedir(self):
         self._connect()
         return self.init_child_result['home_dir']
 
+    def get_binding(self):
+        """
+        Return the :class:`ansible_mitogen.process.Binding` representing the
+        process that hosts the physical connection and services (context
+        establishment, file transfer, ..) for our desired target.
+        """
+        assert self.binding is not None
+        return self.binding
+
     @property
     def connected(self):
         return self.context is not None
@@ -672,33 +670,21 @@ def _stack_from_spec(self, spec, stack=(), seen_names=()):
 
         return stack
 
-    def _connect_broker(self):
-        """
-        Establish a reference to the Broker, Router and parent context used for
-        connections.
-        """
-        if not self.broker:
-            self.broker = mitogen.master.Broker()
-            self.router, self.parent = mitogen.unix.connect(
-                path=ansible_mitogen.process.MuxProcess.unix_listener_path,
-                broker=self.broker,
-            )
-
     def _build_stack(self):
         """
         Construct a list of dictionaries representing the connection
         configuration between the controller and the target. This is
         additionally used by the integration tests "mitogen_get_stack" action
         to fetch the would-be connection configuration.
         """
-        return self._stack_from_spec(
-            ansible_mitogen.transport_config.PlayContextSpec(
-                connection=self,
-                play_context=self._play_context,
-                transport=self.transport,
-                inventory_name=self.inventory_hostname,
-            )
+        spec = ansible_mitogen.transport_config.PlayContextSpec(
+            connection=self,
+            play_context=self._play_context,
+            transport=self.transport,
+            inventory_name=self.inventory_hostname,
         )
+        stack = self._stack_from_spec(spec)
+        return spec.inventory_name(), stack
 
     def _connect_stack(self, stack):
         """
@@ -711,7 +697,8 @@ def _connect_stack(self, stack):
         description of the returned dictionary.
         """
         try:
-            dct = self.parent.call_service(
+            dct = mitogen.service.call(
+                call_context=self.binding.get_service_context(),
                 service_name='ansible_mitogen.services.ContextService',
                 method_name='get',
                 stack=mitogen.utils.cast(list(stack)),
@@ -758,8 +745,9 @@ def _connect(self):
         if self.connected:
             return
 
-        self._connect_broker()
-        stack = self._build_stack()
+        inventory_name, stack = self._build_stack()
+        worker_model = ansible_mitogen.process.get_worker_model()
+        self.binding = worker_model.get_binding(inventory_name)
         self._connect_stack(stack)
 
     def _mitogen_reset(self, mode):
@@ -776,7 +764,8 @@ def _mitogen_reset(self, mode):
             return
 
         self.chain.reset()
-        self.parent.call_service(
+        mitogen.service.call(
+            call_context=self.binding.get_service_context(),
             service_name='ansible_mitogen.services.ContextService',
             method_name=mode,
             context=self.context
@@ -787,35 +776,16 @@ def _mitogen_reset(self, mode):
         self.init_child_result = None
         self.chain = None
 
-    def _shutdown_broker(self):
-        """
-        Shutdown the broker thread during :meth:`close` or :meth:`reset`.
-        """
-        if self.broker:
-            self.broker.shutdown()
-            self.broker.join()
-            self.broker = None
-            self.router = None
-
-            # #420: Ansible executes "meta" actions in the top-level process,
-            # meaning "reset_connection" will cause :class:`mitogen.core.Latch`
-            # FDs to be cached and erroneously shared by children on subsequent
-            # WorkerProcess forks. To handle that, call on_fork() to ensure any
-            # shared state is discarded.
-            # #490: only attempt to clean up when it's known that some
-            # resources exist to cleanup, otherwise later __del__ double-call
-            # to close() due to GC at random moment may obliterate an unrelated
-            # Connection's resources.
-            mitogen.fork.on_fork()
-
     def close(self):
         """
         Arrange for the mitogen.master.Router running in the worker to
         gracefully shut down, and wait for shutdown to complete. Safe to call
         multiple times.
         """
         self._mitogen_reset(mode='put')
-        self._shutdown_broker()
+        if self.binding:
+            self.binding.close()
+            self.binding = None
 
     def _reset_find_task_vars(self):
         """
@@ -853,7 +823,8 @@ def reset(self):
 
         self._connect()
         self._mitogen_reset(mode='reset')
-        self._shutdown_broker()
+        self.binding.close()
+        self.binding = None
 
     # Compatibility with Ansible 2.4 wait_for_connection plug-in.
     _reset = reset
@@ -1024,7 +995,8 @@ def put_file(self, in_path, out_path):
                                      utimes=(st.st_atime, st.st_mtime))
 
         self._connect()
-        self.parent.call_service(
+        mitogen.service.call(
+            call_context=self.binding.get_service_context(),
             service_name='mitogen.service.FileService',
             method_name='register',
             path=mitogen.utils.cast(in_path)
@@ -1036,7 +1008,7 @@ def put_file(self, in_path, out_path):
         # file alive, but that requires more work.
         self.get_chain().call(
             ansible_mitogen.target.transfer_file,
-            context=self.parent,
+            context=self.binding.get_child_service_context(),
             in_path=in_path,
             out_path=out_path
         )

ansible_mitogen/planner.py

@@ -148,14 +148,16 @@ def get_kwargs(self, **kwargs):
                 # named by `runner_name`.
             }
         """
+        binding = self._inv.connection.get_binding()
+
         new = dict((mitogen.core.UnicodeType(k), kwargs[k])
                    for k in kwargs)
         new.setdefault('good_temp_dir',
             self._inv.connection.get_good_temp_dir())
         new.setdefault('cwd', self._inv.connection.get_default_cwd())
         new.setdefault('extra_env', self._inv.connection.get_default_env())
         new.setdefault('emulate_tty', True)
-        new.setdefault('service_context', self._inv.connection.parent)
+        new.setdefault('service_context', binding.get_child_service_context())
         return new
 
     def __repr__(self):
@@ -328,7 +330,9 @@ def get_search_path(self):
 
     def get_module_map(self):
         if self._module_map is None:
-            self._module_map = self._inv.connection.parent.call_service(
+            binding = self._inv.connection.get_binding()
+            self._module_map = mitogen.service.call(
+                call_context=binding.get_service_context(),
                 service_name='ansible_mitogen.services.ModuleDepService',
                 method_name='scan',
 
@@ -405,9 +409,12 @@ def get_module_data(name):
 
 
 def _propagate_deps(invocation, planner, context):
-    invocation.connection.parent.call_service(
+    binding = invocation.connection.get_binding()
+    mitogen.service.call(
+        call_context=binding.get_service_context(),
         service_name='mitogen.service.PushFileService',
         method_name='propagate_paths_and_modules',
+
         context=context,
         paths=planner.get_push_files(),
         modules=planner.get_module_deps(),

ansible_mitogen/plugins/action/mitogen_get_stack.py

@@ -47,8 +47,9 @@ def run(self, tmp=None, task_vars=None):
                 'skipped': True,
             }
 
+        _, stack = self._connection._build_stack()
         return {
             'changed': True,
-            'result': self._connection._build_stack(),
+            'result': stack,
             '_ansible_verbose_always': True,
         }

ansible_mitogen/plugins/connection/mitogen_local.py

@@ -81,6 +81,6 @@ def get_default_env(self):
         from WorkerProcess, we must emulate that.
         """
         return dict_diff(
-            old=ansible_mitogen.process.MuxProcess.original_env,
+            old=ansible_mitogen.process.MuxProcess.cls_original_env,
             new=os.environ,
         )