Adding upstream version 2.14.3.upstream/2.14.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

5 files changed, 1993 insertions, 0 deletions

diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
new file mode 100644
index 0000000..5cc05ee
--- /dev/null
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -0,0 +1,1202 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import cmd
+import functools
+import os
+import pprint
+import queue
+import sys
+import threading
+import time
+
+from collections import deque
+from multiprocessing import Lock
+
+from jinja2.exceptions import UndefinedError
+
+from ansible import constants as C
+from ansible import context
+from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleUndefinedVariable, AnsibleParserError
+from ansible.executor import action_write_locks
+from ansible.executor.play_iterator import IteratingStates
+from ansible.executor.process.worker import WorkerProcess
+from ansible.executor.task_result import TaskResult
+from ansible.executor.task_queue_manager import CallbackSend, DisplaySend
+from ansible.module_utils.six import string_types
+from ansible.module_utils._text import to_text
+from ansible.module_utils.connection import Connection, ConnectionError
+from ansible.playbook.conditional import Conditional
+from ansible.playbook.handler import Handler
+from ansible.playbook.helpers import load_list_of_blocks
+from ansible.playbook.task import Task
+from ansible.playbook.task_include import TaskInclude
+from ansible.plugins import loader as plugin_loader
+from ansible.template import Templar
+from ansible.utils.display import Display
+from ansible.utils.fqcn import add_internal_fqcns
+from ansible.utils.unsafe_proxy import wrap_var
+from ansible.utils.vars import combine_vars, isidentifier
+from ansible.vars.clean import strip_internal_keys, module_response_deepcopy
+
+display = Display()
+
+__all__ = ['StrategyBase']
+
+# This list can be an exact match, or start of string bound
+# does not accept regex
+ALWAYS_DELEGATE_FACT_PREFIXES = frozenset((
+    'discovered_interpreter_',
+))
+
+
+class StrategySentinel:
+    pass
+
+
+_sentinel = StrategySentinel()
+
+
+def post_process_whens(result, task, templar, task_vars):
+    cond = None
+    if task.changed_when:
+        with templar.set_temporary_context(available_variables=task_vars):
+            cond = Conditional(loader=templar._loader)
+            cond.when = task.changed_when
+            result['changed'] = cond.evaluate_conditional(templar, templar.available_variables)
+
+    if task.failed_when:
+        with templar.set_temporary_context(available_variables=task_vars):
+            if cond is None:
+                cond = Conditional(loader=templar._loader)
+            cond.when = task.failed_when
+            failed_when_result = cond.evaluate_conditional(templar, templar.available_variables)
+            result['failed_when_result'] = result['failed'] = failed_when_result
+
+
+def _get_item_vars(result, task):
+    item_vars = {}
+    if task.loop or task.loop_with:
+        loop_var = result.get('ansible_loop_var', 'item')
+        index_var = result.get('ansible_index_var')
+        if loop_var in result:
+            item_vars[loop_var] = result[loop_var]
+        if index_var and index_var in result:
+            item_vars[index_var] = result[index_var]
+        if '_ansible_item_label' in result:
+            item_vars['_ansible_item_label'] = result['_ansible_item_label']
+        if 'ansible_loop' in result:
+            item_vars['ansible_loop'] = result['ansible_loop']
+    return item_vars
+
+
+def results_thread_main(strategy):
+    while True:
+        try:
+            result = strategy._final_q.get()
+            if isinstance(result, StrategySentinel):
+                break
+            elif isinstance(result, DisplaySend):
+                display.display(*result.args, **result.kwargs)
+            elif isinstance(result, CallbackSend):
+                for arg in result.args:
+                    if isinstance(arg, TaskResult):
+                        strategy.normalize_task_result(arg)
+                        break
+                strategy._tqm.send_callback(result.method_name, *result.args, **result.kwargs)
+            elif isinstance(result, TaskResult):
+                strategy.normalize_task_result(result)
+                with strategy._results_lock:
+                    strategy._results.append(result)
+            else:
+                display.warning('Received an invalid object (%s) in the result queue: %r' % (type(result), result))
+        except (IOError, EOFError):
+            break
+        except queue.Empty:
+            pass
+
+
+def debug_closure(func):
+    """Closure to wrap ``StrategyBase._process_pending_results`` and invoke the task debugger"""
+    @functools.wraps(func)
+    def inner(self, iterator, one_pass=False, max_passes=None):
+        status_to_stats_map = (
+            ('is_failed', 'failures'),
+            ('is_unreachable', 'dark'),
+            ('is_changed', 'changed'),
+            ('is_skipped', 'skipped'),
+        )
+
+        # We don't know the host yet, copy the previous states, for lookup after we process new results
+        prev_host_states = iterator.host_states.copy()
+
+        results = func(self, iterator, one_pass=one_pass, max_passes=max_passes)
+        _processed_results = []
+
+        for result in results:
+            task = result._task
+            host = result._host
+            _queued_task_args = self._queued_task_cache.pop((host.name, task._uuid), None)
+            task_vars = _queued_task_args['task_vars']
+            play_context = _queued_task_args['play_context']
+            # Try to grab the previous host state, if it doesn't exist use get_host_state to generate an empty state
+            try:
+                prev_host_state = prev_host_states[host.name]
+            except KeyError:
+                prev_host_state = iterator.get_host_state(host)
+
+            while result.needs_debugger(globally_enabled=self.debugger_active):
+                next_action = NextAction()
+                dbg = Debugger(task, host, task_vars, play_context, result, next_action)
+                dbg.cmdloop()
+
+                if next_action.result == NextAction.REDO:
+                    # rollback host state
+                    self._tqm.clear_failed_hosts()
+                    if task.run_once and iterator._play.strategy in add_internal_fqcns(('linear',)) and result.is_failed():
+                        for host_name, state in prev_host_states.items():
+                            if host_name == host.name:
+                                continue
+                            iterator.set_state_for_host(host_name, state)
+                            iterator._play._removed_hosts.remove(host_name)
+                    iterator.set_state_for_host(host.name, prev_host_state)
+                    for method, what in status_to_stats_map:
+                        if getattr(result, method)():
+                            self._tqm._stats.decrement(what, host.name)
+                    self._tqm._stats.decrement('ok', host.name)
+
+                    # redo
+                    self._queue_task(host, task, task_vars, play_context)
+
+                    _processed_results.extend(debug_closure(func)(self, iterator, one_pass))
+                    break
+                elif next_action.result == NextAction.CONTINUE:
+                    _processed_results.append(result)
+                    break
+                elif next_action.result == NextAction.EXIT:
+                    # Matches KeyboardInterrupt from bin/ansible
+                    sys.exit(99)
+            else:
+                _processed_results.append(result)
+
+        return _processed_results
+    return inner
+
+
+class StrategyBase:
+
+    '''
+    This is the base class for strategy plugins, which contains some common
+    code useful to all strategies like running handlers, cleanup actions, etc.
+    '''
+
+    # by default, strategies should support throttling but we allow individual
+    # strategies to disable this and either forego supporting it or managing
+    # the throttling internally (as `free` does)
+    ALLOW_BASE_THROTTLING = True
+
+    def __init__(self, tqm):
+        self._tqm = tqm
+        self._inventory = tqm.get_inventory()
+        self._workers = tqm._workers
+        self._variable_manager = tqm.get_variable_manager()
+        self._loader = tqm.get_loader()
+        self._final_q = tqm._final_q
+        self._step = context.CLIARGS.get('step', False)
+        self._diff = context.CLIARGS.get('diff', False)
+
+        # the task cache is a dictionary of tuples of (host.name, task._uuid)
+        # used to find the original task object of in-flight tasks and to store
+        # the task args/vars and play context info used to queue the task.
+        self._queued_task_cache = {}
+
+        # Backwards compat: self._display isn't really needed, just import the global display and use that.
+        self._display = display
+
+        # internal counters
+        self._pending_results = 0
+        self._cur_worker = 0
+
+        # this dictionary is used to keep track of hosts that have
+        # outstanding tasks still in queue
+        self._blocked_hosts = dict()
+
+        self._results = deque()
+        self._results_lock = threading.Condition(threading.Lock())
+
+        # create the result processing thread for reading results in the background
+        self._results_thread = threading.Thread(target=results_thread_main, args=(self,))
+        self._results_thread.daemon = True
+        self._results_thread.start()
+
+        # holds the list of active (persistent) connections to be shutdown at
+        # play completion
+        self._active_connections = dict()
+
+        # Caches for get_host calls, to avoid calling excessively
+        # These values should be set at the top of the ``run`` method of each
+        # strategy plugin. Use ``_set_hosts_cache`` to set these values
+        self._hosts_cache = []
+        self._hosts_cache_all = []
+
+        self.debugger_active = C.ENABLE_TASK_DEBUGGER
+
+    def _set_hosts_cache(self, play, refresh=True):
+        """Responsible for setting _hosts_cache and _hosts_cache_all
+
+        See comment in ``__init__`` for the purpose of these caches
+        """
+        if not refresh and all((self._hosts_cache, self._hosts_cache_all)):
+            return
+
+        if not play.finalized and Templar(None).is_template(play.hosts):
+            _pattern = 'all'
+        else:
+            _pattern = play.hosts or 'all'
+        self._hosts_cache_all = [h.name for h in self._inventory.get_hosts(pattern=_pattern, ignore_restrictions=True)]
+        self._hosts_cache = [h.name for h in self._inventory.get_hosts(play.hosts, order=play.order)]
+
+    def cleanup(self):
+        # close active persistent connections
+        for sock in self._active_connections.values():
+            try:
+                conn = Connection(sock)
+                conn.reset()
+            except ConnectionError as e:
+                # most likely socket is already closed
+                display.debug("got an error while closing persistent connection: %s" % e)
+        self._final_q.put(_sentinel)
+        self._results_thread.join()
+
+    def run(self, iterator, play_context, result=0):
+        # execute one more pass through the iterator without peeking, to
+        # make sure that all of the hosts are advanced to their final task.
+        # This should be safe, as everything should be IteratingStates.COMPLETE by
+        # this point, though the strategy may not advance the hosts itself.
+
+        for host in self._hosts_cache:
+            if host not in self._tqm._unreachable_hosts:
+                try:
+                    iterator.get_next_task_for_host(self._inventory.hosts[host])
+                except KeyError:
+                    iterator.get_next_task_for_host(self._inventory.get_host(host))
+
+        # return the appropriate code, depending on the status hosts after the run
+        if not isinstance(result, bool) and result != self._tqm.RUN_OK:
+            return result
+        elif len(self._tqm._unreachable_hosts.keys()) > 0:
+            return self._tqm.RUN_UNREACHABLE_HOSTS
+        elif len(iterator.get_failed_hosts()) > 0:
+            return self._tqm.RUN_FAILED_HOSTS
+        else:
+            return self._tqm.RUN_OK
+
+    def get_hosts_remaining(self, play):
+        self._set_hosts_cache(play, refresh=False)
+        ignore = set(self._tqm._failed_hosts).union(self._tqm._unreachable_hosts)
+        return [host for host in self._hosts_cache if host not in ignore]
+
+    def get_failed_hosts(self, play):
+        self._set_hosts_cache(play, refresh=False)
+        return [host for host in self._hosts_cache if host in self._tqm._failed_hosts]
+
+    def add_tqm_variables(self, vars, play):
+        '''
+        Base class method to add extra variables/information to the list of task
+        vars sent through the executor engine regarding the task queue manager state.
+        '''
+        vars['ansible_current_hosts'] = self.get_hosts_remaining(play)
+        vars['ansible_failed_hosts'] = self.get_failed_hosts(play)
+
+    def _queue_task(self, host, task, task_vars, play_context):
+        ''' handles queueing the task up to be sent to a worker '''
+
+        display.debug("entering _queue_task() for %s/%s" % (host.name, task.action))
+
+        # Add a write lock for tasks.
+        # Maybe this should be added somewhere further up the call stack but
+        # this is the earliest in the code where we have task (1) extracted
+        # into its own variable and (2) there's only a single code path
+        # leading to the module being run.  This is called by two
+        # functions: linear.py::run(), and
+        # free.py::run() so we'd have to add to both to do it there.
+        # The next common higher level is __init__.py::run() and that has
+        # tasks inside of play_iterator so we'd have to extract them to do it
+        # there.
+
+        if task.action not in action_write_locks.action_write_locks:
+            display.debug('Creating lock for %s' % task.action)
+            action_write_locks.action_write_locks[task.action] = Lock()
+
+        # create a templar and template things we need later for the queuing process
+        templar = Templar(loader=self._loader, variables=task_vars)
+
+        try:
+            throttle = int(templar.template(task.throttle))
+        except Exception as e:
+            raise AnsibleError("Failed to convert the throttle value to an integer.", obj=task._ds, orig_exc=e)
+
+        # and then queue the new task
+        try:
+            # Determine the "rewind point" of the worker list. This means we start
+            # iterating over the list of workers until the end of the list is found.
+            # Normally, that is simply the length of the workers list (as determined
+            # by the forks or serial setting), however a task/block/play may "throttle"
+            # that limit down.
+            rewind_point = len(self._workers)
+            if throttle > 0 and self.ALLOW_BASE_THROTTLING:
+                if task.run_once:
+                    display.debug("Ignoring 'throttle' as 'run_once' is also set for '%s'" % task.get_name())
+                else:
+                    if throttle <= rewind_point:
+                        display.debug("task: %s, throttle: %d" % (task.get_name(), throttle))
+                        rewind_point = throttle
+
+            queued = False
+            starting_worker = self._cur_worker
+            while True:
+                if self._cur_worker >= rewind_point:
+                    self._cur_worker = 0
+
+                worker_prc = self._workers[self._cur_worker]
+                if worker_prc is None or not worker_prc.is_alive():
+                    self._queued_task_cache[(host.name, task._uuid)] = {
+                        'host': host,
+                        'task': task,
+                        'task_vars': task_vars,
+                        'play_context': play_context
+                    }
+
+                    worker_prc = WorkerProcess(self._final_q, task_vars, host, task, play_context, self._loader, self._variable_manager, plugin_loader)
+                    self._workers[self._cur_worker] = worker_prc
+                    self._tqm.send_callback('v2_runner_on_start', host, task)
+                    worker_prc.start()
+                    display.debug("worker is %d (out of %d available)" % (self._cur_worker + 1, len(self._workers)))
+                    queued = True
+
+                self._cur_worker += 1
+
+                if self._cur_worker >= rewind_point:
+                    self._cur_worker = 0
+
+                if queued:
+                    break
+                elif self._cur_worker == starting_worker:
+                    time.sleep(0.0001)
+
+            self._pending_results += 1
+        except (EOFError, IOError, AssertionError) as e:
+            # most likely an abort
+            display.debug("got an error while queuing: %s" % e)
+            return
+        display.debug("exiting _queue_task() for %s/%s" % (host.name, task.action))
+
+    def get_task_hosts(self, iterator, task_host, task):
+        if task.run_once:
+            host_list = [host for host in self._hosts_cache if host not in self._tqm._unreachable_hosts]
+        else:
+            host_list = [task_host.name]
+        return host_list
+
+    def get_delegated_hosts(self, result, task):
+        host_name = result.get('_ansible_delegated_vars', {}).get('ansible_delegated_host', None)
+        return [host_name or task.delegate_to]
+
+    def _set_always_delegated_facts(self, result, task):
+        """Sets host facts for ``delegate_to`` hosts for facts that should
+        always be delegated
+
+        This operation mutates ``result`` to remove the always delegated facts
+
+        See ``ALWAYS_DELEGATE_FACT_PREFIXES``
+        """
+        if task.delegate_to is None:
+            return
+
+        facts = result['ansible_facts']
+        always_keys = set()
+        _add = always_keys.add
+        for fact_key in facts:
+            for always_key in ALWAYS_DELEGATE_FACT_PREFIXES:
+                if fact_key.startswith(always_key):
+                    _add(fact_key)
+        if always_keys:
+            _pop = facts.pop
+            always_facts = {
+                'ansible_facts': dict((k, _pop(k)) for k in list(facts) if k in always_keys)
+            }
+            host_list = self.get_delegated_hosts(result, task)
+            _set_host_facts = self._variable_manager.set_host_facts
+            for target_host in host_list:
+                _set_host_facts(target_host, always_facts)
+
+    def normalize_task_result(self, task_result):
+        """Normalize a TaskResult to reference actual Host and Task objects
+        when only given the ``Host.name``, or the ``Task._uuid``
+
+        Only the ``Host.name`` and ``Task._uuid`` are commonly sent back from
+        the ``TaskExecutor`` or ``WorkerProcess`` due to performance concerns
+
+        Mutates the original object
+        """
+
+        if isinstance(task_result._host, string_types):
+            # If the value is a string, it is ``Host.name``
+            task_result._host = self._inventory.get_host(to_text(task_result._host))
+
+        if isinstance(task_result._task, string_types):
+            # If the value is a string, it is ``Task._uuid``
+            queue_cache_entry = (task_result._host.name, task_result._task)
+            try:
+                found_task = self._queued_task_cache[queue_cache_entry]['task']
+            except KeyError:
+                # This should only happen due to an implicit task created by the
+                # TaskExecutor, restrict this behavior to the explicit use case
+                # of an implicit async_status task
+                if task_result._task_fields.get('action') != 'async_status':
+                    raise
+                original_task = Task()
+            else:
+                original_task = found_task.copy(exclude_parent=True, exclude_tasks=True)
+                original_task._parent = found_task._parent
+            original_task.from_attrs(task_result._task_fields)
+            task_result._task = original_task
+
+        return task_result
+
+    @debug_closure
+    def _process_pending_results(self, iterator, one_pass=False, max_passes=None):
+        '''
+        Reads results off the final queue and takes appropriate action
+        based on the result (executing callbacks, updating state, etc.).
+        '''
+
+        ret_results = []
+        handler_templar = Templar(self._loader)
+
+        def search_handler_blocks_by_name(handler_name, handler_blocks):
+            # iterate in reversed order since last handler loaded with the same name wins
+            for handler_block in reversed(handler_blocks):
+                for handler_task in handler_block.block:
+                    if handler_task.name:
+                        try:
+                            if not handler_task.cached_name:
+                                if handler_templar.is_template(handler_task.name):
+                                    handler_templar.available_variables = self._variable_manager.get_vars(play=iterator._play,
+                                                                                                          task=handler_task,
+                                                                                                          _hosts=self._hosts_cache,
+                                                                                                          _hosts_all=self._hosts_cache_all)
+                                    handler_task.name = handler_templar.template(handler_task.name)
+                                handler_task.cached_name = True
+
+                            # first we check with the full result of get_name(), which may
+                            # include the role name (if the handler is from a role). If that
+                            # is not found, we resort to the simple name field, which doesn't
+                            # have anything extra added to it.
+                            candidates = (
+                                handler_task.name,
+                                handler_task.get_name(include_role_fqcn=False),
+                                handler_task.get_name(include_role_fqcn=True),
+                            )
+
+                            if handler_name in candidates:
+                                return handler_task
+                        except (UndefinedError, AnsibleUndefinedVariable) as e:
+                            # We skip this handler due to the fact that it may be using
+                            # a variable in the name that was conditionally included via
+                            # set_fact or some other method, and we don't want to error
+                            # out unnecessarily
+                            if not handler_task.listen:
+                                display.warning(
+                                    "Handler '%s' is unusable because it has no listen topics and "
+                                    "the name could not be templated (host-specific variables are "
+                                    "not supported in handler names). The error: %s" % (handler_task.name, to_text(e))
+                                )
+                            continue
+
+        cur_pass = 0
+        while True:
+            try:
+                self._results_lock.acquire()
+                task_result = self._results.popleft()
+            except IndexError:
+                break
+            finally:
+                self._results_lock.release()
+
+            original_host = task_result._host
+            original_task = task_result._task
+
+            # all host status messages contain 2 entries: (msg, task_result)
+            role_ran = False
+            if task_result.is_failed():
+                role_ran = True
+                ignore_errors = original_task.ignore_errors
+                if not ignore_errors:
+                    # save the current state before failing it for later inspection
+                    state_when_failed = iterator.get_state_for_host(original_host.name)
+                    display.debug("marking %s as failed" % original_host.name)
+                    if original_task.run_once:
+                        # if we're using run_once, we have to fail every host here
+                        for h in self._inventory.get_hosts(iterator._play.hosts):
+                            if h.name not in self._tqm._unreachable_hosts:
+                                iterator.mark_host_failed(h)
+                    else:
+                        iterator.mark_host_failed(original_host)
+
+                    state, _ = iterator.get_next_task_for_host(original_host, peek=True)
+
+                    if iterator.is_failed(original_host) and state and state.run_state == IteratingStates.COMPLETE:
+                        self._tqm._failed_hosts[original_host.name] = True
+
+                    # if we're iterating on the rescue portion of a block then
+                    # we save the failed task in a special var for use
+                    # within the rescue/always
+                    if iterator.is_any_block_rescuing(state_when_failed):
+                        self._tqm._stats.increment('rescued', original_host.name)
+                        iterator._play._removed_hosts.remove(original_host.name)
+                        self._variable_manager.set_nonpersistent_facts(
+                            original_host.name,
+                            dict(
+                                ansible_failed_task=wrap_var(original_task.serialize()),
+                                ansible_failed_result=task_result._result,
+                            ),
+                        )
+                    else:
+                        self._tqm._stats.increment('failures', original_host.name)
+                else:
+                    self._tqm._stats.increment('ok', original_host.name)
+                    self._tqm._stats.increment('ignored', original_host.name)
+                    if 'changed' in task_result._result and task_result._result['changed']:
+                        self._tqm._stats.increment('changed', original_host.name)
+                self._tqm.send_callback('v2_runner_on_failed', task_result, ignore_errors=ignore_errors)
+            elif task_result.is_unreachable():
+                ignore_unreachable = original_task.ignore_unreachable
+                if not ignore_unreachable:
+                    self._tqm._unreachable_hosts[original_host.name] = True
+                    iterator._play._removed_hosts.append(original_host.name)
+                    self._tqm._stats.increment('dark', original_host.name)
+                else:
+                    self._tqm._stats.increment('ok', original_host.name)
+                    self._tqm._stats.increment('ignored', original_host.name)
+                self._tqm.send_callback('v2_runner_on_unreachable', task_result)
+            elif task_result.is_skipped():
+                self._tqm._stats.increment('skipped', original_host.name)
+                self._tqm.send_callback('v2_runner_on_skipped', task_result)
+            else:
+                role_ran = True
+
+                if original_task.loop:
+                    # this task had a loop, and has more than one result, so
+                    # loop over all of them instead of a single result
+                    result_items = task_result._result.get('results', [])
+                else:
+                    result_items = [task_result._result]
+
+                for result_item in result_items:
+                    if '_ansible_notify' in result_item:
+                        if task_result.is_changed():
+                            # The shared dictionary for notified handlers is a proxy, which
+                            # does not detect when sub-objects within the proxy are modified.
+                            # So, per the docs, we reassign the list so the proxy picks up and
+                            # notifies all other threads
+                            for handler_name in result_item['_ansible_notify']:
+                                found = False
+                                # Find the handler using the above helper.  First we look up the
+                                # dependency chain of the current task (if it's from a role), otherwise
+                                # we just look through the list of handlers in the current play/all
+                                # roles and use the first one that matches the notify name
+                                target_handler = search_handler_blocks_by_name(handler_name, iterator._play.handlers)
+                                if target_handler is not None:
+                                    found = True
+                                    if target_handler.notify_host(original_host):
+                                        self._tqm.send_callback('v2_playbook_on_notify', target_handler, original_host)
+
+                                for listening_handler_block in iterator._play.handlers:
+                                    for listening_handler in listening_handler_block.block:
+                                        listeners = getattr(listening_handler, 'listen', []) or []
+                                        if not listeners:
+                                            continue
+
+                                        listeners = listening_handler.get_validated_value(
+                                            'listen', listening_handler.fattributes.get('listen'), listeners, handler_templar
+                                        )
+                                        if handler_name not in listeners:
+                                            continue
+                                        else:
+                                            found = True
+
+                                        if listening_handler.notify_host(original_host):
+                                            self._tqm.send_callback('v2_playbook_on_notify', listening_handler, original_host)
+
+                                # and if none were found, then we raise an error
+                                if not found:
+                                    msg = ("The requested handler '%s' was not found in either the main handlers list nor in the listening "
+                                           "handlers list" % handler_name)
+                                    if C.ERROR_ON_MISSING_HANDLER:
+                                        raise AnsibleError(msg)
+                                    else:
+                                        display.warning(msg)
+
+                    if 'add_host' in result_item:
+                        # this task added a new host (add_host module)
+                        new_host_info = result_item.get('add_host', dict())
+                        self._inventory.add_dynamic_host(new_host_info, result_item)
+                        # ensure host is available for subsequent plays
+                        if result_item.get('changed') and new_host_info['host_name'] not in self._hosts_cache_all:
+                            self._hosts_cache_all.append(new_host_info['host_name'])
+
+                    elif 'add_group' in result_item:
+                        # this task added a new group (group_by module)
+                        self._inventory.add_dynamic_group(original_host, result_item)
+
+                    if 'add_host' in result_item or 'add_group' in result_item:
+                        item_vars = _get_item_vars(result_item, original_task)
+                        found_task_vars = self._queued_task_cache.get((original_host.name, task_result._task._uuid))['task_vars']
+                        if item_vars:
+                            all_task_vars = combine_vars(found_task_vars, item_vars)
+                        else:
+                            all_task_vars = found_task_vars
+                        all_task_vars[original_task.register] = wrap_var(result_item)
+                        post_process_whens(result_item, original_task, handler_templar, all_task_vars)
+                        if original_task.loop or original_task.loop_with:
+                            new_item_result = TaskResult(
+                                task_result._host,
+                                task_result._task,
+                                result_item,
+                                task_result._task_fields,
+                            )
+                            self._tqm.send_callback('v2_runner_item_on_ok', new_item_result)
+                            if result_item.get('changed', False):
+                                task_result._result['changed'] = True
+                            if result_item.get('failed', False):
+                                task_result._result['failed'] = True
+
+                    if 'ansible_facts' in result_item and original_task.action not in C._ACTION_DEBUG:
+                        # if delegated fact and we are delegating facts, we need to change target host for them
+                        if original_task.delegate_to is not None and original_task.delegate_facts:
+                            host_list = self.get_delegated_hosts(result_item, original_task)
+                        else:
+                            # Set facts that should always be on the delegated hosts
+                            self._set_always_delegated_facts(result_item, original_task)
+
+                            host_list = self.get_task_hosts(iterator, original_host, original_task)
+
+                        if original_task.action in C._ACTION_INCLUDE_VARS:
+                            for (var_name, var_value) in result_item['ansible_facts'].items():
+                                # find the host we're actually referring too here, which may
+                                # be a host that is not really in inventory at all
+                                for target_host in host_list:
+                                    self._variable_manager.set_host_variable(target_host, var_name, var_value)
+                        else:
+                            cacheable = result_item.pop('_ansible_facts_cacheable', False)
+                            for target_host in host_list:
+                                # so set_fact is a misnomer but 'cacheable = true' was meant to create an 'actual fact'
+                                # to avoid issues with precedence and confusion with set_fact normal operation,
+                                # we set BOTH fact and nonpersistent_facts (aka hostvar)
+                                # when fact is retrieved from cache in subsequent operations it will have the lower precedence,
+                                # but for playbook setting it the 'higher' precedence is kept
+                                is_set_fact = original_task.action in C._ACTION_SET_FACT
+                                if not is_set_fact or cacheable:
+                                    self._variable_manager.set_host_facts(target_host, result_item['ansible_facts'].copy())
+                                if is_set_fact:
+                                    self._variable_manager.set_nonpersistent_facts(target_host, result_item['ansible_facts'].copy())
+
+                    if 'ansible_stats' in result_item and 'data' in result_item['ansible_stats'] and result_item['ansible_stats']['data']:
+
+                        if 'per_host' not in result_item['ansible_stats'] or result_item['ansible_stats']['per_host']:
+                            host_list = self.get_task_hosts(iterator, original_host, original_task)
+                        else:
+                            host_list = [None]
+
+                        data = result_item['ansible_stats']['data']
+                        aggregate = 'aggregate' in result_item['ansible_stats'] and result_item['ansible_stats']['aggregate']
+                        for myhost in host_list:
+                            for k in data.keys():
+                                if aggregate:
+                                    self._tqm._stats.update_custom_stats(k, data[k], myhost)
+                                else:
+                                    self._tqm._stats.set_custom_stats(k, data[k], myhost)
+
+                if 'diff' in task_result._result:
+                    if self._diff or getattr(original_task, 'diff', False):
+                        self._tqm.send_callback('v2_on_file_diff', task_result)
+
+                if not isinstance(original_task, TaskInclude):
+                    self._tqm._stats.increment('ok', original_host.name)
+                    if 'changed' in task_result._result and task_result._result['changed']:
+                        self._tqm._stats.increment('changed', original_host.name)
+
+                # finally, send the ok for this task
+                self._tqm.send_callback('v2_runner_on_ok', task_result)
+
+            # register final results
+            if original_task.register:
+
+                if not isidentifier(original_task.register):
+                    raise AnsibleError("Invalid variable name in 'register' specified: '%s'" % original_task.register)
+
+                host_list = self.get_task_hosts(iterator, original_host, original_task)
+
+                clean_copy = strip_internal_keys(module_response_deepcopy(task_result._result))
+                if 'invocation' in clean_copy:
+                    del clean_copy['invocation']
+
+                for target_host in host_list:
+                    self._variable_manager.set_nonpersistent_facts(target_host, {original_task.register: clean_copy})
+
+            self._pending_results -= 1
+            if original_host.name in self._blocked_hosts:
+                del self._blocked_hosts[original_host.name]
+
+            # If this is a role task, mark the parent role as being run (if
+            # the task was ok or failed, but not skipped or unreachable)
+            if original_task._role is not None and role_ran:  # TODO:  and original_task.action not in C._ACTION_INCLUDE_ROLE:?
+                # lookup the role in the ROLE_CACHE to make sure we're dealing
+                # with the correct object and mark it as executed
+                for (entry, role_obj) in iterator._play.ROLE_CACHE[original_task._role.get_name()].items():
+                    if role_obj._uuid == original_task._role._uuid:
+                        role_obj._had_task_run[original_host.name] = True
+
+            ret_results.append(task_result)
+
+            if isinstance(original_task, Handler):
+                for handler in (h for b in iterator._play.handlers for h in b.block if h._uuid == original_task._uuid):
+                    handler.remove_host(original_host)
+
+            if one_pass or max_passes is not None and (cur_pass + 1) >= max_passes:
+                break
+
+            cur_pass += 1
+
+        return ret_results
+
+    def _wait_on_pending_results(self, iterator):
+        '''
+        Wait for the shared counter to drop to zero, using a short sleep
+        between checks to ensure we don't spin lock
+        '''
+
+        ret_results = []
+
+        display.debug("waiting for pending results...")
+        while self._pending_results > 0 and not self._tqm._terminated:
+
+            if self._tqm.has_dead_workers():
+                raise AnsibleError("A worker was found in a dead state")
+
+            results = self._process_pending_results(iterator)
+            ret_results.extend(results)
+            if self._pending_results > 0:
+                time.sleep(C.DEFAULT_INTERNAL_POLL_INTERVAL)
+
+        display.debug("no more pending results, returning what we have")
+
+        return ret_results
+
+    def _copy_included_file(self, included_file):
+        '''
+        A proven safe and performant way to create a copy of an included file
+        '''
+        ti_copy = included_file._task.copy(exclude_parent=True)
+        ti_copy._parent = included_file._task._parent
+
+        temp_vars = ti_copy.vars | included_file._vars
+
+        ti_copy.vars = temp_vars
+
+        return ti_copy
+
+    def _load_included_file(self, included_file, iterator, is_handler=False):
+        '''
+        Loads an included YAML file of tasks, applying the optional set of variables.
+
+        Raises AnsibleError exception in case of a failure during including a file,
+        in such case the caller is responsible for marking the host(s) as failed
+        using PlayIterator.mark_host_failed().
+        '''
+        display.debug("loading included file: %s" % included_file._filename)
+        try:
+            data = self._loader.load_from_file(included_file._filename)
+            if data is None:
+                return []
+            elif not isinstance(data, list):
+                raise AnsibleError("included task files must contain a list of tasks")
+
+            ti_copy = self._copy_included_file(included_file)
+
+            block_list = load_list_of_blocks(
+                data,
+                play=iterator._play,
+                parent_block=ti_copy.build_parent_block(),
+                role=included_file._task._role,
+                use_handlers=is_handler,
+                loader=self._loader,
+                variable_manager=self._variable_manager,
+            )
+
+            # since we skip incrementing the stats when the task result is
+            # first processed, we do so now for each host in the list
+            for host in included_file._hosts:
+                self._tqm._stats.increment('ok', host.name)
+        except AnsibleParserError:
+            raise
+        except AnsibleError as e:
+            if isinstance(e, AnsibleFileNotFound):
+                reason = "Could not find or access '%s' on the Ansible Controller." % to_text(e.file_name)
+            else:
+                reason = to_text(e)
+
+            for r in included_file._results:
+                r._result['failed'] = True
+
+            for host in included_file._hosts:
+                tr = TaskResult(host=host, task=included_file._task, return_data=dict(failed=True, reason=reason))
+                self._tqm._stats.increment('failures', host.name)
+                self._tqm.send_callback('v2_runner_on_failed', tr)
+            raise AnsibleError(reason) from e
+
+        # finally, send the callback and return the list of blocks loaded
+        self._tqm.send_callback('v2_playbook_on_include', included_file)
+        display.debug("done processing included file")
+        return block_list
+
+    def _take_step(self, task, host=None):
+
+        ret = False
+        msg = u'Perform task: %s ' % task
+        if host:
+            msg += u'on %s ' % host
+        msg += u'(N)o/(y)es/(c)ontinue: '
+        resp = display.prompt(msg)
+
+        if resp.lower() in ['y', 'yes']:
+            display.debug("User ran task")
+            ret = True
+        elif resp.lower() in ['c', 'continue']:
+            display.debug("User ran task and canceled step mode")
+            self._step = False
+            ret = True
+        else:
+            display.debug("User skipped task")
+
+        display.banner(msg)
+
+        return ret
+
+    def _cond_not_supported_warn(self, task_name):
+        display.warning("%s task does not support when conditional" % task_name)
+
+    def _execute_meta(self, task, play_context, iterator, target_host):
+
+        # meta tasks store their args in the _raw_params field of args,
+        # since they do not use k=v pairs, so get that
+        meta_action = task.args.get('_raw_params')
+
+        def _evaluate_conditional(h):
+            all_vars = self._variable_manager.get_vars(play=iterator._play, host=h, task=task,
+                                                       _hosts=self._hosts_cache, _hosts_all=self._hosts_cache_all)
+            templar = Templar(loader=self._loader, variables=all_vars)
+            return task.evaluate_conditional(templar, all_vars)
+
+        skipped = False
+        msg = meta_action
+        skip_reason = '%s conditional evaluated to False' % meta_action
+        if isinstance(task, Handler):
+            self._tqm.send_callback('v2_playbook_on_handler_task_start', task)
+        else:
+            self._tqm.send_callback('v2_playbook_on_task_start', task, is_conditional=False)
+
+        # These don't support "when" conditionals
+        if meta_action in ('noop', 'refresh_inventory', 'reset_connection') and task.when:
+            self._cond_not_supported_warn(meta_action)
+
+        if meta_action == 'noop':
+            msg = "noop"
+        elif meta_action == 'flush_handlers':
+            if _evaluate_conditional(target_host):
+                host_state = iterator.get_state_for_host(target_host.name)
+                if host_state.run_state == IteratingStates.HANDLERS:
+                    raise AnsibleError('flush_handlers cannot be used as a handler')
+                if target_host.name not in self._tqm._unreachable_hosts:
+                    host_state.pre_flushing_run_state = host_state.run_state
+                    host_state.run_state = IteratingStates.HANDLERS
+                msg = "triggered running handlers for %s" % target_host.name
+            else:
+                skipped = True
+                skip_reason += ', not running handlers for %s' % target_host.name
+        elif meta_action == 'refresh_inventory':
+            self._inventory.refresh_inventory()
+            self._set_hosts_cache(iterator._play)
+            msg = "inventory successfully refreshed"
+        elif meta_action == 'clear_facts':
+            if _evaluate_conditional(target_host):
+                for host in self._inventory.get_hosts(iterator._play.hosts):
+                    hostname = host.get_name()
+                    self._variable_manager.clear_facts(hostname)
+                msg = "facts cleared"
+            else:
+                skipped = True
+                skip_reason += ', not clearing facts and fact cache for %s' % target_host.name
+        elif meta_action == 'clear_host_errors':
+            if _evaluate_conditional(target_host):
+                for host in self._inventory.get_hosts(iterator._play.hosts):
+                    self._tqm._failed_hosts.pop(host.name, False)
+                    self._tqm._unreachable_hosts.pop(host.name, False)
+                    iterator.clear_host_errors(host)
+                msg = "cleared host errors"
+            else:
+                skipped = True
+                skip_reason += ', not clearing host error state for %s' % target_host.name
+        elif meta_action == 'end_batch':
+            if _evaluate_conditional(target_host):
+                for host in self._inventory.get_hosts(iterator._play.hosts):
+                    if host.name not in self._tqm._unreachable_hosts:
+                        iterator.set_run_state_for_host(host.name, IteratingStates.COMPLETE)
+                msg = "ending batch"
+            else:
+                skipped = True
+                skip_reason += ', continuing current batch'
+        elif meta_action == 'end_play':
+            if _evaluate_conditional(target_host):
+                for host in self._inventory.get_hosts(iterator._play.hosts):
+                    if host.name not in self._tqm._unreachable_hosts:
+                        iterator.set_run_state_for_host(host.name, IteratingStates.COMPLETE)
+                        # end_play is used in PlaybookExecutor/TQM to indicate that
+                        # the whole play is supposed to be ended as opposed to just a batch
+                        iterator.end_play = True
+                msg = "ending play"
+            else:
+                skipped = True
+                skip_reason += ', continuing play'
+        elif meta_action == 'end_host':
+            if _evaluate_conditional(target_host):
+                iterator.set_run_state_for_host(target_host.name, IteratingStates.COMPLETE)
+                iterator._play._removed_hosts.append(target_host.name)
+                msg = "ending play for %s" % target_host.name
+            else:
+                skipped = True
+                skip_reason += ", continuing execution for %s" % target_host.name
+                # TODO: Nix msg here? Left for historical reasons, but skip_reason exists now.
+                msg = "end_host conditional evaluated to false, continuing execution for %s" % target_host.name
+        elif meta_action == 'role_complete':
+            # Allow users to use this in a play as reported in https://github.com/ansible/ansible/issues/22286?
+            # How would this work with allow_duplicates??
+            if task.implicit:
+                if target_host.name in task._role._had_task_run:
+                    task._role._completed[target_host.name] = True
+                    msg = 'role_complete for %s' % target_host.name
+        elif meta_action == 'reset_connection':
+            all_vars = self._variable_manager.get_vars(play=iterator._play, host=target_host, task=task,
+                                                       _hosts=self._hosts_cache, _hosts_all=self._hosts_cache_all)
+            templar = Templar(loader=self._loader, variables=all_vars)
+
+            # apply the given task's information to the connection info,
+            # which may override some fields already set by the play or
+            # the options specified on the command line
+            play_context = play_context.set_task_and_variable_override(task=task, variables=all_vars, templar=templar)
+
+            # fields set from the play/task may be based on variables, so we have to
+            # do the same kind of post validation step on it here before we use it.
+            play_context.post_validate(templar=templar)
+
+            # now that the play context is finalized, if the remote_addr is not set
+            # default to using the host's address field as the remote address
+            if not play_context.remote_addr:
+                play_context.remote_addr = target_host.address
+
+            # We also add "magic" variables back into the variables dict to make sure
+            # a certain subset of variables exist. This 'mostly' works here cause meta
+            # disregards the loop, but should not really use play_context at all
+            play_context.update_vars(all_vars)
+
+            if target_host in self._active_connections:
+                connection = Connection(self._active_connections[target_host])
+                del self._active_connections[target_host]
+            else:
+                connection = plugin_loader.connection_loader.get(play_context.connection, play_context, os.devnull)
+                connection.set_options(task_keys=task.dump_attrs(), var_options=all_vars)
+                play_context.set_attributes_from_plugin(connection)
+
+            if connection:
+                try:
+                    connection.reset()
+                    msg = 'reset connection'
+                except ConnectionError as e:
+                    # most likely socket is already closed
+                    display.debug("got an error while closing persistent connection: %s" % e)
+            else:
+                msg = 'no connection, nothing to reset'
+        else:
+            raise AnsibleError("invalid meta action requested: %s" % meta_action, obj=task._ds)
+
+        result = {'msg': msg}
+        if skipped:
+            result['skipped'] = True
+            result['skip_reason'] = skip_reason
+        else:
+            result['changed'] = False
+
+        if not task.implicit:
+            header = skip_reason if skipped else msg
+            display.vv(f"META: {header}")
+
+        if isinstance(task, Handler):
+            task.remove_host(target_host)
+
+        res = TaskResult(target_host, task, result)
+        if skipped:
+            self._tqm.send_callback('v2_runner_on_skipped', res)
+        return [res]
+
+    def get_hosts_left(self, iterator):
+        ''' returns list of available hosts for this iterator by filtering out unreachables '''
+
+        hosts_left = []
+        for host in self._hosts_cache:
+            if host not in self._tqm._unreachable_hosts:
+                try:
+                    hosts_left.append(self._inventory.hosts[host])
+                except KeyError:
+                    hosts_left.append(self._inventory.get_host(host))
+        return hosts_left
+
+    def update_active_connections(self, results):
+        ''' updates the current active persistent connections '''
+        for r in results:
+            if 'args' in r._task_fields:
+                socket_path = r._task_fields['args'].get('_ansible_socket')
+                if socket_path:
+                    if r._host not in self._active_connections:
+                        self._active_connections[r._host] = socket_path
+
+
+class NextAction(object):
+    """ The next action after an interpreter's exit. """
+    REDO = 1
+    CONTINUE = 2
+    EXIT = 3
+
+    def __init__(self, result=EXIT):
+        self.result = result
+
+
+class Debugger(cmd.Cmd):
+    prompt_continuous = '> '  # multiple lines
+
+    def __init__(self, task, host, task_vars, play_context, result, next_action):
+        # cmd.Cmd is old-style class
+        cmd.Cmd.__init__(self)
+
+        self.prompt = '[%s] %s (debug)> ' % (host, task)
+        self.intro = None
+        self.scope = {}
+        self.scope['task'] = task
+        self.scope['task_vars'] = task_vars
+        self.scope['host'] = host
+        self.scope['play_context'] = play_context
+        self.scope['result'] = result
+        self.next_action = next_action
+
+    def cmdloop(self):
+        try:
+            cmd.Cmd.cmdloop(self)
+        except KeyboardInterrupt:
+            pass
+
+    do_h = cmd.Cmd.do_help
+
+    def do_EOF(self, args):
+        """Quit"""
+        return self.do_quit(args)
+
+    def do_quit(self, args):
+        """Quit"""
+        display.display('User interrupted execution')
+        self.next_action.result = NextAction.EXIT
+        return True
+
+    do_q = do_quit
+
+    def do_continue(self, args):
+        """Continue to next result"""
+        self.next_action.result = NextAction.CONTINUE
+        return True
+
+    do_c = do_continue
+
+    def do_redo(self, args):
+        """Schedule task for re-execution. The re-execution may not be the next result"""
+        self.next_action.result = NextAction.REDO
+        return True
+
+    do_r = do_redo
+
+    def do_update_task(self, args):
+        """Recreate the task from ``task._ds``, and template with updated ``task_vars``"""
+        templar = Templar(None, variables=self.scope['task_vars'])
+        task = self.scope['task']
+        task = task.load_data(task._ds)
+        task.post_validate(templar)
+        self.scope['task'] = task
+
+    do_u = do_update_task
+
+    def evaluate(self, args):
+        try:
+            return eval(args, globals(), self.scope)
+        except Exception:
+            t, v = sys.exc_info()[:2]
+            if isinstance(t, str):
+                exc_type_name = t
+            else:
+                exc_type_name = t.__name__
+            display.display('***%s:%s' % (exc_type_name, repr(v)))
+            raise
+
+    def do_pprint(self, args):
+        """Pretty Print"""
+        try:
+            result = self.evaluate(args)
+            display.display(pprint.pformat(result))
+        except Exception:
+            pass
+
+    do_p = do_pprint
+
+    def execute(self, args):
+        try:
+            code = compile(args + '\n', '<stdin>', 'single')
+            exec(code, globals(), self.scope)
+        except Exception:
+            t, v = sys.exc_info()[:2]
+            if isinstance(t, str):
+                exc_type_name = t
+            else:
+                exc_type_name = t.__name__
+            display.display('***%s:%s' % (exc_type_name, repr(v)))
+            raise
+
+    def default(self, line):
+        try:
+            self.execute(line)
+        except Exception:
+            pass
diff --git a/lib/ansible/plugins/strategy/debug.py b/lib/ansible/plugins/strategy/debug.py
new file mode 100644
index 0000000..f808bcf
--- /dev/null
+++ b/lib/ansible/plugins/strategy/debug.py
@@ -0,0 +1,37 @@
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    name: debug
+    short_description: Executes tasks in interactive debug session.
+    description:
+        - Task execution is 'linear' but controlled by an interactive debug session.
+    version_added: "2.1"
+    author: Kishin Yagami (!UNKNOWN)
+'''
+
+import cmd
+import pprint
+import sys
+
+from ansible.plugins.strategy.linear import StrategyModule as LinearStrategyModule
+
+
+class StrategyModule(LinearStrategyModule):
+    def __init__(self, tqm):
+        super(StrategyModule, self).__init__(tqm)
+        self.debugger_active = True
diff --git a/lib/ansible/plugins/strategy/free.py b/lib/ansible/plugins/strategy/free.py
new file mode 100644
index 0000000..6f45114
--- /dev/null
+++ b/lib/ansible/plugins/strategy/free.py
@@ -0,0 +1,303 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    name: free
+    short_description: Executes tasks without waiting for all hosts
+    description:
+        - Task execution is as fast as possible per batch as defined by C(serial) (default all).
+          Ansible will not wait for other hosts to finish the current task before queuing more tasks for other hosts.
+          All hosts are still attempted for the current task, but it prevents blocking new tasks for hosts that have already finished.
+        - With the free strategy, unlike the default linear strategy, a host that is slow or stuck on a specific task
+          won't hold up the rest of the hosts and tasks.
+    version_added: "2.0"
+    author: Ansible Core Team
+'''
+
+import time
+
+from ansible import constants as C
+from ansible.errors import AnsibleError, AnsibleParserError
+from ansible.playbook.handler import Handler
+from ansible.playbook.included_file import IncludedFile
+from ansible.plugins.loader import action_loader
+from ansible.plugins.strategy import StrategyBase
+from ansible.template import Templar
+from ansible.module_utils._text import to_text
+from ansible.utils.display import Display
+
+display = Display()
+
+
+class StrategyModule(StrategyBase):
+
+    # This strategy manages throttling on its own, so we don't want it done in queue_task
+    ALLOW_BASE_THROTTLING = False
+
+    def __init__(self, tqm):
+        super(StrategyModule, self).__init__(tqm)
+        self._host_pinned = False
+
+    def run(self, iterator, play_context):
+        '''
+        The "free" strategy is a bit more complex, in that it allows tasks to
+        be sent to hosts as quickly as they can be processed. This means that
+        some hosts may finish very quickly if run tasks result in little or no
+        work being done versus other systems.
+
+        The algorithm used here also tries to be more "fair" when iterating
+        through hosts by remembering the last host in the list to be given a task
+        and starting the search from there as opposed to the top of the hosts
+        list again, which would end up favoring hosts near the beginning of the
+        list.
+        '''
+
+        # the last host to be given a task
+        last_host = 0
+
+        result = self._tqm.RUN_OK
+
+        # start with all workers being counted as being free
+        workers_free = len(self._workers)
+
+        self._set_hosts_cache(iterator._play)
+
+        if iterator._play.max_fail_percentage is not None:
+            display.warning("Using max_fail_percentage with the free strategy is not supported, as tasks are executed independently on each host")
+
+        work_to_do = True
+        while work_to_do and not self._tqm._terminated:
+
+            hosts_left = self.get_hosts_left(iterator)
+
+            if len(hosts_left) == 0:
+                self._tqm.send_callback('v2_playbook_on_no_hosts_remaining')
+                result = False
+                break
+
+            work_to_do = False        # assume we have no more work to do
+            starting_host = last_host  # save current position so we know when we've looped back around and need to break
+
+            # try and find an unblocked host with a task to run
+            host_results = []
+            while True:
+                host = hosts_left[last_host]
+                display.debug("next free host: %s" % host)
+                host_name = host.get_name()
+
+                # peek at the next task for the host, to see if there's
+                # anything to do do for this host
+                (state, task) = iterator.get_next_task_for_host(host, peek=True)
+                display.debug("free host state: %s" % state, host=host_name)
+                display.debug("free host task: %s" % task, host=host_name)
+
+                # check if there is work to do, either there is a task or the host is still blocked which could
+                # mean that it is processing an include task and after its result is processed there might be
+                # more tasks to run
+                if (task or self._blocked_hosts.get(host_name, False)) and not self._tqm._unreachable_hosts.get(host_name, False):
+                    display.debug("this host has work to do", host=host_name)
+                    # set the flag so the outer loop knows we've still found
+                    # some work which needs to be done
+                    work_to_do = True
+
+                if not self._tqm._unreachable_hosts.get(host_name, False) and task:
+                    # check to see if this host is blocked (still executing a previous task)
+                    if not self._blocked_hosts.get(host_name, False):
+                        display.debug("getting variables", host=host_name)
+                        task_vars = self._variable_manager.get_vars(play=iterator._play, host=host, task=task,
+                                                                    _hosts=self._hosts_cache,
+                                                                    _hosts_all=self._hosts_cache_all)
+                        self.add_tqm_variables(task_vars, play=iterator._play)
+                        templar = Templar(loader=self._loader, variables=task_vars)
+                        display.debug("done getting variables", host=host_name)
+
+                        try:
+                            throttle = int(templar.template(task.throttle))
+                        except Exception as e:
+                            raise AnsibleError("Failed to convert the throttle value to an integer.", obj=task._ds, orig_exc=e)
+
+                        if throttle > 0:
+                            same_tasks = 0
+                            for worker in self._workers:
+                                if worker and worker.is_alive() and worker._task._uuid == task._uuid:
+                                    same_tasks += 1
+
+                            display.debug("task: %s, same_tasks: %d" % (task.get_name(), same_tasks))
+                            if same_tasks >= throttle:
+                                break
+
+                        # advance the host, mark the host blocked, and queue it
+                        self._blocked_hosts[host_name] = True
+                        iterator.set_state_for_host(host.name, state)
+
+                        try:
+                            action = action_loader.get(task.action, class_only=True, collection_list=task.collections)
+                        except KeyError:
+                            # we don't care here, because the action may simply not have a
+                            # corresponding action plugin
+                            action = None
+
+                        try:
+                            task.name = to_text(templar.template(task.name, fail_on_undefined=False), nonstring='empty')
+                            display.debug("done templating", host=host_name)
+                        except Exception:
+                            # just ignore any errors during task name templating,
+                            # we don't care if it just shows the raw name
+                            display.debug("templating failed for some reason", host=host_name)
+
+                        run_once = templar.template(task.run_once) or action and getattr(action, 'BYPASS_HOST_LOOP', False)
+                        if run_once:
+                            if action and getattr(action, 'BYPASS_HOST_LOOP', False):
+                                raise AnsibleError("The '%s' module bypasses the host loop, which is currently not supported in the free strategy "
+                                                   "and would instead execute for every host in the inventory list." % task.action, obj=task._ds)
+                            else:
+                                display.warning("Using run_once with the free strategy is not currently supported. This task will still be "
+                                                "executed for every host in the inventory list.")
+
+                        # check to see if this task should be skipped, due to it being a member of a
+                        # role which has already run (and whether that role allows duplicate execution)
+                        if not isinstance(task, Handler) and task._role and task._role.has_run(host):
+                            # If there is no metadata, the default behavior is to not allow duplicates,
+                            # if there is metadata, check to see if the allow_duplicates flag was set to true
+                            if task._role._metadata is None or task._role._metadata and not task._role._metadata.allow_duplicates:
+                                display.debug("'%s' skipped because role has already run" % task, host=host_name)
+                                del self._blocked_hosts[host_name]
+                                continue
+
+                        if task.action in C._ACTION_META:
+                            self._execute_meta(task, play_context, iterator, target_host=host)
+                            self._blocked_hosts[host_name] = False
+                        else:
+                            # handle step if needed, skip meta actions as they are used internally
+                            if not self._step or self._take_step(task, host_name):
+                                if task.any_errors_fatal:
+                                    display.warning("Using any_errors_fatal with the free strategy is not supported, "
+                                                    "as tasks are executed independently on each host")
+                                if isinstance(task, Handler):
+                                    self._tqm.send_callback('v2_playbook_on_handler_task_start', task)
+                                else:
+                                    self._tqm.send_callback('v2_playbook_on_task_start', task, is_conditional=False)
+                                self._queue_task(host, task, task_vars, play_context)
+                                # each task is counted as a worker being busy
+                                workers_free -= 1
+                                del task_vars
+                    else:
+                        display.debug("%s is blocked, skipping for now" % host_name)
+
+                # all workers have tasks to do (and the current host isn't done with the play).
+                # loop back to starting host and break out
+                if self._host_pinned and workers_free == 0 and work_to_do:
+                    last_host = starting_host
+                    break
+
+                # move on to the next host and make sure we
+                # haven't gone past the end of our hosts list
+                last_host += 1
+                if last_host > len(hosts_left) - 1:
+                    last_host = 0
+
+                # if we've looped around back to the start, break out
+                if last_host == starting_host:
+                    break
+
+            results = self._process_pending_results(iterator)
+            host_results.extend(results)
+
+            # each result is counted as a worker being free again
+            workers_free += len(results)
+
+            self.update_active_connections(results)
+
+            included_files = IncludedFile.process_include_results(
+                host_results,
+                iterator=iterator,
+                loader=self._loader,
+                variable_manager=self._variable_manager
+            )
+
+            if len(included_files) > 0:
+                all_blocks = dict((host, []) for host in hosts_left)
+                failed_includes_hosts = set()
+                for included_file in included_files:
+                    display.debug("collecting new blocks for %s" % included_file)
+                    is_handler = False
+                    try:
+                        if included_file._is_role:
+                            new_ir = self._copy_included_file(included_file)
+
+                            new_blocks, handler_blocks = new_ir.get_block_list(
+                                play=iterator._play,
+                                variable_manager=self._variable_manager,
+                                loader=self._loader,
+                            )
+                        else:
+                            is_handler = isinstance(included_file._task, Handler)
+                            new_blocks = self._load_included_file(included_file, iterator=iterator, is_handler=is_handler)
+
+                        # let PlayIterator know about any new handlers included via include_role or
+                        # import_role within include_role/include_taks
+                        iterator.handlers = [h for b in iterator._play.handlers for h in b.block]
+                    except AnsibleParserError:
+                        raise
+                    except AnsibleError as e:
+                        if included_file._is_role:
+                            # include_role does not have on_include callback so display the error
+                            display.error(to_text(e), wrap_text=False)
+                        for r in included_file._results:
+                            r._result['failed'] = True
+                            failed_includes_hosts.add(r._host)
+                        continue
+
+                    for new_block in new_blocks:
+                        if is_handler:
+                            for task in new_block.block:
+                                task.notified_hosts = included_file._hosts[:]
+                            final_block = new_block
+                        else:
+                            task_vars = self._variable_manager.get_vars(
+                                play=iterator._play,
+                                task=new_block.get_first_parent_include(),
+                                _hosts=self._hosts_cache,
+                                _hosts_all=self._hosts_cache_all,
+                            )
+                            final_block = new_block.filter_tagged_tasks(task_vars)
+                        for host in hosts_left:
+                            if host in included_file._hosts:
+                                all_blocks[host].append(final_block)
+                    display.debug("done collecting new blocks for %s" % included_file)
+
+                for host in failed_includes_hosts:
+                    self._tqm._failed_hosts[host.name] = True
+                    iterator.mark_host_failed(host)
+
+                display.debug("adding all collected blocks from %d included file(s) to iterator" % len(included_files))
+                for host in hosts_left:
+                    iterator.add_tasks(host, all_blocks[host])
+                display.debug("done adding collected blocks to iterator")
+
+            # pause briefly so we don't spin lock
+            time.sleep(C.DEFAULT_INTERNAL_POLL_INTERVAL)
+
+        # collect all the final results
+        results = self._wait_on_pending_results(iterator)
+
+        # run the base class run() method, which executes the cleanup function
+        # and runs any outstanding handlers which have been triggered
+        return super(StrategyModule, self).run(iterator, play_context, result)
diff --git a/lib/ansible/plugins/strategy/host_pinned.py b/lib/ansible/plugins/strategy/host_pinned.py
new file mode 100644
index 0000000..70f22eb
--- /dev/null
+++ b/lib/ansible/plugins/strategy/host_pinned.py
@@ -0,0 +1,45 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    name: host_pinned
+    short_description: Executes tasks on each host without interruption
+    description:
+        - Task execution is as fast as possible per host in batch as defined by C(serial) (default all).
+          Ansible will not start a play for a host unless the play can be finished without interruption by tasks for another host,
+          i.e. the number of hosts with an active play does not exceed the number of forks.
+          Ansible will not wait for other hosts to finish the current task before queuing the next task for a host that has finished.
+          Once a host is done with the play, it opens it's slot to a new host that was waiting to start.
+          Other than that, it behaves just like the "free" strategy.
+    version_added: "2.7"
+    author: Ansible Core Team
+'''
+
+from ansible.plugins.strategy.free import StrategyModule as FreeStrategyModule
+from ansible.utils.display import Display
+
+display = Display()
+
+
+class StrategyModule(FreeStrategyModule):
+
+    def __init__(self, tqm):
+        super(StrategyModule, self).__init__(tqm)
+        self._host_pinned = True
diff --git a/lib/ansible/plugins/strategy/linear.py b/lib/ansible/plugins/strategy/linear.py
new file mode 100644
index 0000000..a3c91c2
--- /dev/null
+++ b/lib/ansible/plugins/strategy/linear.py
@@ -0,0 +1,406 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    name: linear
+    short_description: Executes tasks in a linear fashion
+    description:
+        - Task execution is in lockstep per host batch as defined by C(serial) (default all).
+          Up to the fork limit of hosts will execute each task at the same time and then
+          the next series of hosts until the batch is done, before going on to the next task.
+    version_added: "2.0"
+    notes:
+     - This was the default Ansible behaviour before 'strategy plugins' were introduced in 2.0.
+    author: Ansible Core Team
+'''
+
+from ansible import constants as C
+from ansible.errors import AnsibleError, AnsibleAssertionError, AnsibleParserError
+from ansible.executor.play_iterator import IteratingStates, FailedStates
+from ansible.module_utils._text import to_text
+from ansible.playbook.handler import Handler
+from ansible.playbook.included_file import IncludedFile
+from ansible.playbook.task import Task
+from ansible.plugins.loader import action_loader
+from ansible.plugins.strategy import StrategyBase
+from ansible.template import Templar
+from ansible.utils.display import Display
+
+display = Display()
+
+
+class StrategyModule(StrategyBase):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        # used for the lockstep to indicate to run handlers
+        self._in_handlers = False
+
+    def _get_next_task_lockstep(self, hosts, iterator):
+        '''
+        Returns a list of (host, task) tuples, where the task may
+        be a noop task to keep the iterator in lock step across
+        all hosts.
+        '''
+        noop_task = Task()
+        noop_task.action = 'meta'
+        noop_task.args['_raw_params'] = 'noop'
+        noop_task.implicit = True
+        noop_task.set_loader(iterator._play._loader)
+
+        state_task_per_host = {}
+        for host in hosts:
+            state, task = iterator.get_next_task_for_host(host, peek=True)
+            if task is not None:
+                state_task_per_host[host] = state, task
+
+        if not state_task_per_host:
+            return [(h, None) for h in hosts]
+
+        if self._in_handlers and not any(filter(
+            lambda rs: rs == IteratingStates.HANDLERS,
+            (s.run_state for s, _ in state_task_per_host.values()))
+        ):
+            self._in_handlers = False
+
+        if self._in_handlers:
+            lowest_cur_handler = min(
+                s.cur_handlers_task for s, t in state_task_per_host.values()
+                if s.run_state == IteratingStates.HANDLERS
+            )
+        else:
+            task_uuids = [t._uuid for s, t in state_task_per_host.values()]
+            _loop_cnt = 0
+            while _loop_cnt <= 1:
+                try:
+                    cur_task = iterator.all_tasks[iterator.cur_task]
+                except IndexError:
+                    # pick up any tasks left after clear_host_errors
+                    iterator.cur_task = 0
+                    _loop_cnt += 1
+                else:
+                    iterator.cur_task += 1
+                    if cur_task._uuid in task_uuids:
+                        break
+            else:
+                # prevent infinite loop
+                raise AnsibleAssertionError(
+                    'BUG: There seems to be a mismatch between tasks in PlayIterator and HostStates.'
+                )
+
+        host_tasks = []
+        for host, (state, task) in state_task_per_host.items():
+            if ((self._in_handlers and lowest_cur_handler == state.cur_handlers_task) or
+                    (not self._in_handlers and cur_task._uuid == task._uuid)):
+                iterator.set_state_for_host(host.name, state)
+                host_tasks.append((host, task))
+            else:
+                host_tasks.append((host, noop_task))
+
+        # once hosts synchronize on 'flush_handlers' lockstep enters
+        # '_in_handlers' phase where handlers are run instead of tasks
+        # until at least one host is in IteratingStates.HANDLERS
+        if (not self._in_handlers and cur_task.action in C._ACTION_META and
+                cur_task.args.get('_raw_params') == 'flush_handlers'):
+            self._in_handlers = True
+
+        return host_tasks
+
+    def run(self, iterator, play_context):
+        '''
+        The linear strategy is simple - get the next task and queue
+        it for all hosts, then wait for the queue to drain before
+        moving on to the next task
+        '''
+
+        # iterate over each task, while there is one left to run
+        result = self._tqm.RUN_OK
+        work_to_do = True
+
+        self._set_hosts_cache(iterator._play)
+
+        while work_to_do and not self._tqm._terminated:
+
+            try:
+                display.debug("getting the remaining hosts for this loop")
+                hosts_left = self.get_hosts_left(iterator)
+                display.debug("done getting the remaining hosts for this loop")
+
+                # queue up this task for each host in the inventory
+                callback_sent = False
+                work_to_do = False
+
+                host_tasks = self._get_next_task_lockstep(hosts_left, iterator)
+
+                # skip control
+                skip_rest = False
+                choose_step = True
+
+                # flag set if task is set to any_errors_fatal
+                any_errors_fatal = False
+
+                results = []
+                for (host, task) in host_tasks:
+                    if not task:
+                        continue
+
+                    if self._tqm._terminated:
+                        break
+
+                    run_once = False
+                    work_to_do = True
+
+                    # check to see if this task should be skipped, due to it being a member of a
+                    # role which has already run (and whether that role allows duplicate execution)
+                    if not isinstance(task, Handler) and task._role and task._role.has_run(host):
+                        # If there is no metadata, the default behavior is to not allow duplicates,
+                        # if there is metadata, check to see if the allow_duplicates flag was set to true
+                        if task._role._metadata is None or task._role._metadata and not task._role._metadata.allow_duplicates:
+                            display.debug("'%s' skipped because role has already run" % task)
+                            continue
+
+                    display.debug("getting variables")
+                    task_vars = self._variable_manager.get_vars(play=iterator._play, host=host, task=task,
+                                                                _hosts=self._hosts_cache, _hosts_all=self._hosts_cache_all)
+                    self.add_tqm_variables(task_vars, play=iterator._play)
+                    templar = Templar(loader=self._loader, variables=task_vars)
+                    display.debug("done getting variables")
+
+                    # test to see if the task across all hosts points to an action plugin which
+                    # sets BYPASS_HOST_LOOP to true, or if it has run_once enabled. If so, we
+                    # will only send this task to the first host in the list.
+
+                    task_action = templar.template(task.action)
+
+                    try:
+                        action = action_loader.get(task_action, class_only=True, collection_list=task.collections)
+                    except KeyError:
+                        # we don't care here, because the action may simply not have a
+                        # corresponding action plugin
+                        action = None
+
+                    if task_action in C._ACTION_META:
+                        # for the linear strategy, we run meta tasks just once and for
+                        # all hosts currently being iterated over rather than one host
+                        results.extend(self._execute_meta(task, play_context, iterator, host))
+                        if task.args.get('_raw_params', None) not in ('noop', 'reset_connection', 'end_host', 'role_complete', 'flush_handlers'):
+                            run_once = True
+                        if (task.any_errors_fatal or run_once) and not task.ignore_errors:
+                            any_errors_fatal = True
+                    else:
+                        # handle step if needed, skip meta actions as they are used internally
+                        if self._step and choose_step:
+                            if self._take_step(task):
+                                choose_step = False
+                            else:
+                                skip_rest = True
+                                break
+
+                        run_once = templar.template(task.run_once) or action and getattr(action, 'BYPASS_HOST_LOOP', False)
+
+                        if (task.any_errors_fatal or run_once) and not task.ignore_errors:
+                            any_errors_fatal = True
+
+                        if not callback_sent:
+                            display.debug("sending task start callback, copying the task so we can template it temporarily")
+                            saved_name = task.name
+                            display.debug("done copying, going to template now")
+                            try:
+                                task.name = to_text(templar.template(task.name, fail_on_undefined=False), nonstring='empty')
+                                display.debug("done templating")
+                            except Exception:
+                                # just ignore any errors during task name templating,
+                                # we don't care if it just shows the raw name
+                                display.debug("templating failed for some reason")
+                            display.debug("here goes the callback...")
+                            if isinstance(task, Handler):
+                                self._tqm.send_callback('v2_playbook_on_handler_task_start', task)
+                            else:
+                                self._tqm.send_callback('v2_playbook_on_task_start', task, is_conditional=False)
+                            task.name = saved_name
+                            callback_sent = True
+                            display.debug("sending task start callback")
+
+                        self._blocked_hosts[host.get_name()] = True
+                        self._queue_task(host, task, task_vars, play_context)
+                        del task_vars
+
+                    # if we're bypassing the host loop, break out now
+                    if run_once:
+                        break
+
+                    results.extend(self._process_pending_results(iterator, max_passes=max(1, int(len(self._tqm._workers) * 0.1))))
+
+                # go to next host/task group
+                if skip_rest:
+                    continue
+
+                display.debug("done queuing things up, now waiting for results queue to drain")
+                if self._pending_results > 0:
+                    results.extend(self._wait_on_pending_results(iterator))
+
+                self.update_active_connections(results)
+
+                included_files = IncludedFile.process_include_results(
+                    results,
+                    iterator=iterator,
+                    loader=self._loader,
+                    variable_manager=self._variable_manager
+                )
+
+                if len(included_files) > 0:
+                    display.debug("we have included files to process")
+
+                    display.debug("generating all_blocks data")
+                    all_blocks = dict((host, []) for host in hosts_left)
+                    display.debug("done generating all_blocks data")
+                    included_tasks = []
+                    failed_includes_hosts = set()
+                    for included_file in included_files:
+                        display.debug("processing included file: %s" % included_file._filename)
+                        is_handler = False
+                        try:
+                            if included_file._is_role:
+                                new_ir = self._copy_included_file(included_file)
+
+                                new_blocks, handler_blocks = new_ir.get_block_list(
+                                    play=iterator._play,
+                                    variable_manager=self._variable_manager,
+                                    loader=self._loader,
+                                )
+                            else:
+                                is_handler = isinstance(included_file._task, Handler)
+                                new_blocks = self._load_included_file(included_file, iterator=iterator, is_handler=is_handler)
+
+                            # let PlayIterator know about any new handlers included via include_role or
+                            # import_role within include_role/include_taks
+                            iterator.handlers = [h for b in iterator._play.handlers for h in b.block]
+
+                            display.debug("iterating over new_blocks loaded from include file")
+                            for new_block in new_blocks:
+                                if is_handler:
+                                    for task in new_block.block:
+                                        task.notified_hosts = included_file._hosts[:]
+                                    final_block = new_block
+                                else:
+                                    task_vars = self._variable_manager.get_vars(
+                                        play=iterator._play,
+                                        task=new_block.get_first_parent_include(),
+                                        _hosts=self._hosts_cache,
+                                        _hosts_all=self._hosts_cache_all,
+                                    )
+                                    display.debug("filtering new block on tags")
+                                    final_block = new_block.filter_tagged_tasks(task_vars)
+                                    display.debug("done filtering new block on tags")
+
+                                    included_tasks.extend(final_block.get_tasks())
+
+                                for host in hosts_left:
+                                    if host in included_file._hosts:
+                                        all_blocks[host].append(final_block)
+
+                            display.debug("done iterating over new_blocks loaded from include file")
+                        except AnsibleParserError:
+                            raise
+                        except AnsibleError as e:
+                            if included_file._is_role:
+                                # include_role does not have on_include callback so display the error
+                                display.error(to_text(e), wrap_text=False)
+                            for r in included_file._results:
+                                r._result['failed'] = True
+                                failed_includes_hosts.add(r._host)
+                            continue
+
+                    for host in failed_includes_hosts:
+                        self._tqm._failed_hosts[host.name] = True
+                        iterator.mark_host_failed(host)
+
+                    # finally go through all of the hosts and append the
+                    # accumulated blocks to their list of tasks
+                    display.debug("extending task lists for all hosts with included blocks")
+
+                    for host in hosts_left:
+                        iterator.add_tasks(host, all_blocks[host])
+
+                    iterator.all_tasks[iterator.cur_task:iterator.cur_task] = included_tasks
+
+                    display.debug("done extending task lists")
+                    display.debug("done processing included files")
+
+                display.debug("results queue empty")
+
+                display.debug("checking for any_errors_fatal")
+                failed_hosts = []
+                unreachable_hosts = []
+                for res in results:
+                    # execute_meta() does not set 'failed' in the TaskResult
+                    # so we skip checking it with the meta tasks and look just at the iterator
+                    if (res.is_failed() or res._task.action in C._ACTION_META) and iterator.is_failed(res._host):
+                        failed_hosts.append(res._host.name)
+                    elif res.is_unreachable():
+                        unreachable_hosts.append(res._host.name)
+
+                # if any_errors_fatal and we had an error, mark all hosts as failed
+                if any_errors_fatal and (len(failed_hosts) > 0 or len(unreachable_hosts) > 0):
+                    dont_fail_states = frozenset([IteratingStates.RESCUE, IteratingStates.ALWAYS])
+                    for host in hosts_left:
+                        (s, _) = iterator.get_next_task_for_host(host, peek=True)
+                        # the state may actually be in a child state, use the get_active_state()
+                        # method in the iterator to figure out the true active state
+                        s = iterator.get_active_state(s)
+                        if s.run_state not in dont_fail_states or \
+                           s.run_state == IteratingStates.RESCUE and s.fail_state & FailedStates.RESCUE != 0:
+                            self._tqm._failed_hosts[host.name] = True
+                            result |= self._tqm.RUN_FAILED_BREAK_PLAY
+                display.debug("done checking for any_errors_fatal")
+
+                display.debug("checking for max_fail_percentage")
+                if iterator._play.max_fail_percentage is not None and len(results) > 0:
+                    percentage = iterator._play.max_fail_percentage / 100.0
+
+                    if (len(self._tqm._failed_hosts) / iterator.batch_size) > percentage:
+                        for host in hosts_left:
+                            # don't double-mark hosts, or the iterator will potentially
+                            # fail them out of the rescue/always states
+                            if host.name not in failed_hosts:
+                                self._tqm._failed_hosts[host.name] = True
+                                iterator.mark_host_failed(host)
+                        self._tqm.send_callback('v2_playbook_on_no_hosts_remaining')
+                        result |= self._tqm.RUN_FAILED_BREAK_PLAY
+                    display.debug('(%s failed / %s total )> %s max fail' % (len(self._tqm._failed_hosts), iterator.batch_size, percentage))
+                display.debug("done checking for max_fail_percentage")
+
+                display.debug("checking to see if all hosts have failed and the running result is not ok")
+                if result != self._tqm.RUN_OK and len(self._tqm._failed_hosts) >= len(hosts_left):
+                    display.debug("^ not ok, so returning result now")
+                    self._tqm.send_callback('v2_playbook_on_no_hosts_remaining')
+                    return result
+                display.debug("done checking to see if all hosts have failed")
+
+            except (IOError, EOFError) as e:
+                display.debug("got IOError/EOFError in task loop: %s" % e)
+                # most likely an abort, return failed
+                return self._tqm.RUN_UNKNOWN_ERROR
+
+        # run the base class run() method, which executes the cleanup function
+        # and runs any outstanding handlers which have been triggered
+
+        return super(StrategyModule, self).run(iterator, play_context, result)