diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 3499dc1a48e60d..82cf533fb62715 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -583,6 +583,16 @@ def _execute(self, variables=None):
             include_file = templar.template(include_file)
             return dict(include=include_file, include_args=include_args)
 
+        # if this task is a TaskDo, we just return now with a success code so the
+        # main thread can expand the task list for the given host
+        if self._task.action == 'do':
+            do_args = self._task.args.copy()
+            do_block = do_args.get('_raw_params', None)
+            if not do_block:
+                return dict(failed=True, msg="Do block was specified without a body.")
+
+            return dict(do=do_block, do_args=do_args)
+
         # if this task is a IncludeRole, we just return now with a success code so the main thread can expand the task list for the given host
         elif self._task.action == 'include_role':
             include_args = self._task.args.copy()
diff --git a/lib/ansible/modules/do.py b/lib/ansible/modules/do.py
new file mode 100644
index 00000000000000..d1cb59b19cc57c
--- /dev/null
+++ b/lib/ansible/modules/do.py
@@ -0,0 +1,84 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright:  Ansible Project
+# Copyright:  Estelle Poulin <dev@inspiredby.es>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = r'''
+---
+author: Estelle Poulin (dev@inspiredby.es)
+module: do
+short_description: Dynamically include a task block
+description:
+  - Includes a block with a list of tasks to be executed in the current playbook.
+version_added: '2.10'
+options:
+  apply:
+    description:
+      - Accepts a hash of task keywords (e.g. C(tags), C(become)) that will be applied to the tasks within the include.
+    type: str
+    version_added: '2.7'
+  free-form:
+    description:
+      - |
+        Accepts a list of tasks specificed in the same manner as C(block).
+notes:
+  - This is a core feature of the Ansible, rather than a module, and cannot be overridden like a module.
+seealso:
+- module: ansible.builtin.include
+- module: ansible.builtin.include_tasks
+- module: ansible.builtin.include_role
+- ref: playbooks_reuse_includes
+  description: More information related to including and importing playbooks, roles and tasks.
+'''
+
+EXAMPLES = r'''
+- hosts: all
+  tasks:
+    - debug:
+        msg: task1
+
+    - name: Run a task list within a play.
+      do:
+        - debug:
+            msg: stuff
+
+    - debug:
+        msg: task10
+
+- hosts: all
+  tasks:
+    - debug:
+        msg: task1
+
+    - name: Run the task list only if the condition is true.
+      do:
+        - debug:
+            msg: stuff
+      when: hostvar is defined
+
+- name: Apply tags to tasks within included file
+  do:
+    - debug:
+        msg: stuff
+  args:
+    apply:
+      tags: [install]
+  tags: [always]
+
+- name: Loop over a block of tasks.
+  do:
+    - debug:
+       var: item
+  loop: [1, 2, 3]
+
+'''
+
+RETURN = r'''
+# This module does not return anything except tasks to execute.
+'''
diff --git a/lib/ansible/parsing/mod_args.py b/lib/ansible/parsing/mod_args.py
index d23f2f2b9d1596..e4c5986b763f6a 100644
--- a/lib/ansible/parsing/mod_args.py
+++ b/lib/ansible/parsing/mod_args.py
@@ -23,6 +23,7 @@
 from ansible.module_utils.six import iteritems, string_types
 from ansible.module_utils._text import to_text
 from ansible.parsing.splitter import parse_kv, split_args
+from ansible.parsing.yaml.objects import AnsibleSequence
 from ansible.plugins.loader import module_loader, action_loader
 from ansible.template import Templar
 from ansible.utils.sentinel import Sentinel
@@ -44,6 +45,7 @@
     'include_tasks',
     'include_role',
     'import_tasks',
+    'do',
     'import_role',
     'add_host',
     'group_by',
@@ -208,6 +210,8 @@ def _normalize_new_style_args(self, thing, action):
         elif thing is None:
             # this can happen with modules which take no params, like ping:
             args = None
+        elif isinstance(thing, AnsibleSequence):
+            args = { u'_raw_params': thing }
         else:
             raise AnsibleParserError("unexpected parameter type in action: %s" % type(thing), obj=self._task_ds)
         return args
diff --git a/lib/ansible/playbook/do_block.py b/lib/ansible/playbook/do_block.py
new file mode 100644
index 00000000000000..d43b0411d3a9d3
--- /dev/null
+++ b/lib/ansible/playbook/do_block.py
@@ -0,0 +1,134 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+# (c) 2020, Estelle Poulin <dev@inspiredby.es>
+#
+# 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 os
+
+from ansible.errors import AnsibleError
+from ansible.module_utils._text import to_text
+from ansible.playbook.task_include import TaskInclude
+from ansible.template import Templar
+from ansible.utils.display import Display
+
+display = Display()
+
+
+class DoBlock:
+
+    def __init__(self, block, args, vars, task):
+        self._block = block
+        self._args = args
+        self._vars = vars
+        self._task = task
+        self._hosts = []
+
+    def add_host(self, host):
+        if host not in self._hosts:
+            self._hosts.append(host)
+            return
+        raise ValueError()
+
+    def __eq__(self, other):
+        return (other._args == self._args and
+                other._vars == self._vars and
+                other._task._uuid == self._task._uuid and
+                other._task._parent._uuid == self._task._parent._uuid)
+
+    def __repr__(self):
+        return "do_block (args=%s vars=%s): %s" % (self._args, self._vars, self._hosts)
+
+    @staticmethod
+    def process_do_results(results, iterator, loader, variable_manager):
+        do_blocks = []
+        task_vars_cache = {}
+
+        for res in results:
+
+            original_host = res._host
+            original_task = res._task
+
+            if original_task.action in ('do'):
+                if original_task.loop:
+                    if 'results' not in res._result:
+                        continue
+                    do_results = res._result['results']
+                else:
+                    do_results = [res._result]
+
+                for do_result in do_results:
+                    # if the task result was skipped or failed, continue
+                    if 'skipped' in do_result and do_result['skipped'] or 'failed' in do_result and do_result['failed']:
+                        continue
+
+                    cache_key = (iterator._play, original_host, original_task)
+                    try:
+                        task_vars = task_vars_cache[cache_key]
+                    except KeyError:
+                        task_vars = task_vars_cache[cache_key] = variable_manager.get_vars(play=iterator._play, host=original_host, task=original_task)
+
+                    do_args = do_result.get('do_args', dict())
+                    special_vars = {}
+                    loop_var = do_result.get('ansible_loop_var', 'item')
+                    index_var = do_result.get('ansible_index_var')
+                    if loop_var in do_result:
+                        task_vars[loop_var] = special_vars[loop_var] = do_result[loop_var]
+                    if index_var and index_var in do_result:
+                        task_vars[index_var] = special_vars[index_var] = do_result[index_var]
+                    if '_ansible_item_label' in do_result:
+                        task_vars['_ansible_item_label'] = special_vars['_ansible_item_label'] = do_result['_ansible_item_label']
+                    if 'ansible_loop' in do_result:
+                        task_vars['ansible_loop'] = special_vars['ansible_loop'] = do_result['ansible_loop']
+                    if original_task.no_log and '_ansible_no_log' not in do_args:
+                        task_vars['_ansible_no_log'] = special_vars['_ansible_no_log'] = original_task.no_log
+
+                    # get search path for this task to pass to lookup plugins that may be used in pathing to
+                    # the do block
+                    task_vars['ansible_search_path'] = original_task.get_search_path()
+
+                    # ensure basedir is always in (dwim already searches here but we need to display it)
+                    if loader.get_basedir() not in task_vars['ansible_search_path']:
+                        task_vars['ansible_search_path'].append(loader.get_basedir())
+
+                    do_block = original_task
+
+                    do_blk = DoBlock(do_block, do_args, special_vars, original_task)
+
+                    idx = 0
+                    orig_do_blk = do_blk
+                    while 1:
+                        try:
+                            pos = do_blocks[idx:].index(orig_do_blk)
+                            # pos is relative to idx since we are slicing
+                            # use idx + pos due to relative indexing
+                            do_blk = do_blocks[idx + pos]
+                        except ValueError:
+                            do_blocks.append(orig_do_blk)
+                            do_blk = orig_do_blk
+
+                        try:
+                            do_blk.add_host(original_host)
+                        except ValueError:
+                            # The host already exists for this do block, advance forward, this is a new do block
+                            idx += pos + 1
+                        else:
+                            break
+
+        return do_blocks
diff --git a/lib/ansible/playbook/task_include.py b/lib/ansible/playbook/task_include.py
index 365ce30bb65068..f5451c7fc29a51 100644
--- a/lib/ansible/playbook/task_include.py
+++ b/lib/ansible/playbook/task_include.py
@@ -104,7 +104,7 @@ def get_vars(self):
         we need to include the args of the include into the vars as
         they are params to the included tasks. But ONLY for 'include'
         '''
-        if self.action != 'include':
+        if self.action not in ('include', 'do') :
             all_vars = super(TaskInclude, self).get_vars()
         else:
             all_vars = dict()
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index df7db3685f8dce..0e3bd807776468 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -778,6 +778,20 @@ def _copy_included_file(self, included_file):
 
         return ti_copy
 
+    def _copy_do(self, do_block):
+        '''
+        A proven safe and performant way to create a copy of do block.
+        '''
+        ti_copy = do_block._task.copy(exclude_parent=True)
+        ti_copy._parent = do_block._task._parent
+
+        temp_vars = ti_copy.vars.copy()
+        temp_vars.update(do_block._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.
@@ -843,6 +857,65 @@ def _load_included_file(self, included_file, iterator, is_handler=False):
         display.debug("done processing included file")
         return block_list
 
+    def _load_do_block(self, do_block, iterator, is_handler=False):
+        '''
+        Loads a list of tasks from the args of a do block, applying the optional set of variables.
+        '''
+
+        try:
+            data = do_block._args['_raw_params']
+            if data is None:
+                return []
+            elif not isinstance(data, list):
+                raise AnsibleError("do blocks must contain a list of tasks")
+
+            ti_copy = self._copy_do(do_block)
+            # pop tags out of the do args, if they were specified there, and assign
+            # them to the block. If the do already had tags specified, we raise an
+            # error so that users know not to specify them both ways
+            tags = do_block._task.vars.pop('tags', [])
+            if isinstance(tags, string_types):
+                tags = tags.split(',')
+            if len(tags) > 0:
+                if len(do_block._task.tags) > 0:
+                    raise AnsibleParserError("Do blocks should not specify tags in more than one way (both via args and directly on the task). "
+                                             "Mixing tag specify styles is prohibited for whole import hierarchy, not only for single import statement",
+                                             obj=do_block._task._ds)
+                display.deprecated("You should not specify tags in the do parameters. All tags should be specified using the task-level option",
+                                   version='2.12', collection_name='ansible.builtin')
+                do_block._task.tags = tags
+
+            block_list = load_list_of_blocks(
+                data,
+                play=iterator._play,
+                parent_block=ti_copy,
+                role=do_block._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 do_block._hosts:
+                self._tqm._stats.increment('ok', host.name)
+
+        except AnsibleError as e:
+            reason = to_text(e)
+
+            # mark all of the hosts including this file as failed, send callbacks,
+            # and increment the stats for this host
+            for host in do_block._hosts:
+                tr = TaskResult(host=host, task=do_block._task, return_data=dict(failed=True, reason=reason))
+                iterator.mark_host_failed(host)
+                self._tqm._failed_hosts[host.name] = True
+                self._tqm._stats.increment('failures', host.name)
+                self._tqm.send_callback('v2_runner_on_failed', tr)
+            return []
+
+        display.debug("done processing do blocks")
+        return block_list
+
     def run_handlers(self, iterator, play_context):
         '''
         Runs handlers on those hosts which have been notified.
diff --git a/lib/ansible/plugins/strategy/free.py b/lib/ansible/plugins/strategy/free.py
index 4e779eb9a1c5db..2fe145f5145aa8 100644
--- a/lib/ansible/plugins/strategy/free.py
+++ b/lib/ansible/plugins/strategy/free.py
@@ -36,6 +36,7 @@
 from ansible import constants as C
 from ansible.errors import AnsibleError
 from ansible.playbook.included_file import IncludedFile
+from ansible.playbook.do_block import DoBlock
 from ansible.plugins.loader import action_loader
 from ansible.plugins.strategy import StrategyBase
 from ansible.template import Templar
@@ -209,6 +210,13 @@ def run(self, iterator, play_context):
                 variable_manager=self._variable_manager
             )
 
+            do_blocks = DoBlock.process_do_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)
                 for included_file in included_files:
@@ -243,6 +251,32 @@ def run(self, iterator, play_context):
                     iterator.add_tasks(host, all_blocks[host])
                 display.debug("done adding collected blocks to iterator")
 
+            if len(do_blocks) > 0:
+                all_blocks = dict((host, []) for host in hosts_left)
+                for do_block in do_blocks:
+                    display.debug("collecting new blocks")
+                    try:
+                        new_blocks = self._load_do_block(do_block, iterator=iterator)
+                    except AnsibleError as e:
+                        for host in do_block._hosts:
+                            iterator.mark_host_failed(host)
+                        display.warning(to_text(e))
+                        continue
+
+                    for new_block in new_blocks:
+                        task_vars = self._variable_manager.get_vars(play=iterator._play, task=new_block._parent)
+                        final_block = new_block.filter_tagged_tasks(task_vars)
+                        for host in hosts_left:
+                            if host in do_block._hosts:
+                                all_blocks[host].append(final_block)
+                    display.debug("done collecting new blocks")
+
+                display.debug("adding all collected blocks from %d do blocks to iterator" % len(do_blocks))
+                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)
 
diff --git a/lib/ansible/plugins/strategy/linear.py b/lib/ansible/plugins/strategy/linear.py
index 34960323299134..d00b1bca90d312 100644
--- a/lib/ansible/plugins/strategy/linear.py
+++ b/lib/ansible/plugins/strategy/linear.py
@@ -37,6 +37,7 @@
 from ansible.module_utils._text import to_text
 from ansible.playbook.block import Block
 from ansible.playbook.included_file import IncludedFile
+from ansible.playbook.do_block import DoBlock
 from ansible.playbook.task import Task
 from ansible.plugins.loader import action_loader
 from ansible.plugins.strategy import StrategyBase
@@ -331,6 +332,13 @@ def run(self, iterator, play_context):
                     variable_manager=self._variable_manager
                 )
 
+                do_blocks = DoBlock.process_do_results(
+                    host_results,
+                    iterator=iterator,
+                    loader=self._loader,
+                    variable_manager=self._variable_manager
+                )
+
                 include_failure = False
                 if len(included_files) > 0:
                     display.debug("we have included files to process")
@@ -391,6 +399,57 @@ def run(self, iterator, play_context):
                     display.debug("done extending task lists")
                     display.debug("done processing included files")
 
+                if len(do_blocks) > 0:
+                    display.debug("we have included do blocks to process")
+
+                    display.debug("generating all_blocks data")
+                    all_blocks = dict((host, []) for host in hosts_left)
+                    display.debug("done generating all_blocks data")
+                    for do_block in do_blocks:
+                        display.debug("processing do_block")
+                        # included hosts get the task list while those excluded get an equal-length
+                        # list of noop tasks, to make sure that they continue running in lock-step
+                        try:
+                            new_blocks = self._load_do_block(do_block, iterator=iterator)
+
+                            display.debug("iterating over new_blocks loaded from do blocks")
+                            for new_block in new_blocks:
+                                task_vars = self._variable_manager.get_vars(
+                                    play=iterator._play,
+                                    task=new_block._parent
+                                )
+                                display.debug("filtering new block on tags")
+                                final_block = new_block.filter_tagged_tasks(task_vars)
+                                display.debug("done filtering new block on tags")
+
+                                noop_block = self._prepare_and_create_noop_block_from(final_block, task._parent, iterator)
+
+                                for host in hosts_left:
+                                    # TODO
+                                    if host in do_block._hosts:
+                                        all_blocks[host].append(final_block)
+                                    else:
+                                        all_blocks[host].append(noop_block)
+                            display.debug("done iterating over new_blocks loaded from do block")
+
+                        except AnsibleError as e:
+                            for host in do_block._hosts:
+                                self._tqm._failed_hosts[host.name] = True
+                                iterator.mark_host_failed(host)
+                            display.error(to_text(e), wrap_text=False)
+                            include_failure = True
+                            continue
+
+                    # 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 do blocks")
+
+                    for host in hosts_left:
+                        iterator.add_tasks(host, all_blocks[host])
+
+                    display.debug("done extending task lists")
+                    display.debug("done processing do blocks")
+
                 display.debug("results queue empty")
 
                 display.debug("checking for any_errors_fatal")