diff --git a/lib/ansible/runner/action_plugins/copy.py b/lib/ansible/runner/action_plugins/copy.py
index 9d3973a831..00f12a8bc1 100644
--- a/lib/ansible/runner/action_plugins/copy.py
+++ b/lib/ansible/runner/action_plugins/copy.py
@@ -59,6 +59,10 @@ class ActionModule(object):
             result=dict(failed=True, msg="src and content are mutually exclusive")
             return ReturnData(conn=conn, result=result)
 
+        source_trailing_slash = False
+        if source:
+            source_trailing_slash = source.endswith("/")
+
         # if we have first_available_file in our vars
         # look up the files and use the first one we find as src
         if 'first_available_file' in inject:
@@ -95,88 +99,144 @@ class ActionModule(object):
                 source = utils.path_dwim(self.runner.basedir, source)
 
 
-        local_md5 = utils.md5(source)
-        if local_md5 is None:
-            result=dict(failed=True, msg="could not find src=%s" % source)
-            return ReturnData(conn=conn, result=result)
-
-        if dest.endswith("/"):
-            base = os.path.basename(source)
-            dest = os.path.join(dest, base)
-
-        remote_md5 = self.runner._remote_md5(conn, tmp, dest)
-        if remote_md5 == '3':
-            # Destination is a directory
-            if content is not None:
-                os.remove(tmp_content)
-                result = dict(failed=True, msg="can not use content with a dir as dest")
-                return ReturnData(conn=conn, result=result)
-            dest = os.path.join(dest, os.path.basename(source))
-            remote_md5 = self.runner._remote_md5(conn, tmp, dest)
-
-        # remote_md5 == '1' would mean that the file does not exist.
-        if remote_md5 != '1' and not force:
-            return ReturnData(conn=conn, result=dict(changed=False))
-
-        exec_rc = None
-        if local_md5 != remote_md5:
-
-            if self.runner.diff and not raw:
-                diff = self._get_diff_data(conn, tmp, inject, dest, source)
+        source_files = []
+        if os.path.isdir(source):
+            # Implement rsync-like behavior: if source is "dir/" , only
+            # inside its contents will be copied to destination. Otherwise
+            # if it's "dir", dir itself will be copied to destination.
+            if source_trailing_slash:
+                sz = len(source) + 1
             else:
-                diff = {}
+                sz = len(source.rsplit('/', 1)[0]) + 1
+            for base_path, sub_folders, files in os.walk(source):
+                for file in files:
+                    full_path = os.path.join(base_path, file)
+                    rel_path = full_path[sz:]
+                    source_files.append((full_path, rel_path))
+            # If it's recursive copy, destination is always a dir,
+            # explictly mark it so (note - copy module relies on this).
+            if not dest.endswith("/"):
+                dest += "/"
+        else:
+            source_files.append((source, os.path.basename(source)))
 
-            if self.runner.noop_on_check(inject):
+        changed = False
+        diffs = []
+        module_result = {"changed": False}
+        for source_full, source_rel in source_files:
+            # We need to get a new tmp path for each file, otherwise the copy module deletes the folder.
+            tmp = self.runner._make_tmp_path(conn)
+            local_md5 = utils.md5(source_full)
+
+            if local_md5 is None:
+                result=dict(failed=True, msg="could not find src=%s" % source_full)
+                return ReturnData(conn=conn, result=result)
+
+            # This is kind of optimization - if user told us destination is
+            # dir, do path manipulation right away, otherwise we still check
+            # for dest being a dir via remote call below.
+            if dest.endswith("/"):
+                dest_file = os.path.join(dest, source_rel)
+            else:
+                dest_file = dest
+
+            remote_md5 = self.runner._remote_md5(conn, tmp, dest_file)
+            if remote_md5 == '3':
+                # Destination is a directory
                 if content is not None:
                     os.remove(tmp_content)
-                return ReturnData(conn=conn, result=dict(changed=True), diff=diff)
+                    result = dict(failed=True, msg="can not use content with a dir as dest")
+                    return ReturnData(conn=conn, result=result)
+                dest_file = os.path.join(dest, source_rel)
+                remote_md5 = self.runner._remote_md5(conn, tmp, dest_file)
+
+            # remote_md5 == '1' would mean that the file does not exist.
+            if remote_md5 != '1' and not force:
+                continue
+
+            exec_rc = None
+            if local_md5 != remote_md5:
+                # Assume we either really change file or error out
+                changed = True
+
+                if self.runner.diff and not raw:
+                    diff = self._get_diff_data(conn, tmp, inject, dest_file, source_full)
+                else:
+                    diff = {}
+
+                if self.runner.noop_on_check(inject):
+                    if content is not None:
+                        os.remove(tmp_content)
+                    diffs.append(diff)
+                    changed = True
+                    module_result = dict(changed=True)
+                    continue
 
 
-            # transfer the file to a remote tmp location
-            tmp_src = tmp + 'source'
+                # transfer the file to a remote tmp location
+                tmp_src = tmp + 'source'
+
+                if not raw:
+                    conn.put_file(source_full, tmp_src)
+                else:
+                    conn.put_file(source_full, dest_file)
+
+                if content is not None:
+                    os.remove(tmp_content)
+
+                # fix file permissions when the copy is done as a different user
+                if self.runner.sudo and self.runner.sudo_user != 'root' and not raw:
+                    self.runner._low_level_exec_command(conn, "chmod a+r %s" % tmp_src, tmp)
+
+                if raw:
+                    continue
+
+                # run the copy module
+                if raw:
+                    # don't send down raw=no
+                    module_args.pop('raw')
+
+                # src and dest here come after original and override them
+                # we pass dest only to make sure it includes trailing slash
+                # in case of recursive copy
+                module_args_tmp = "%s src=%s dest=%s original_basename=%s" % (module_args,
+                                  pipes.quote(tmp_src), pipes.quote(dest), pipes.quote(source_rel))
+                module_return = self.runner._execute_module(conn, tmp, 'copy', module_args_tmp, inject=inject, complex_args=complex_args)
 
-            if not raw:
-                conn.put_file(source, tmp_src)
             else:
-                conn.put_file(source, dest)
+                # no need to transfer the file, already correct md5, but still need to call
+                # the file module in case we want to change attributes
+                if content is not None:
+                    os.remove(tmp_content)
 
-            if content is not None:
-                os.remove(tmp_content)
+                if raw:
+                    continue
 
-            # fix file permissions when the copy is done as a different user
-            if self.runner.sudo and self.runner.sudo_user != 'root' and not raw:
-                self.runner._low_level_exec_command(conn, "chmod a+r %s" % tmp_src, tmp)
+                tmp_src = tmp + source_rel
+                if raw:
+                    # don't send down raw=no
+                    module_args.pop('raw')
+                module_args_tmp = "%s src=%s original_basename=%s" % (module_args,
+                                  pipes.quote(tmp_src), pipes.quote(source_rel))
+                if self.runner.noop_on_check(inject):
+                    module_args_tmp = "%s CHECKMODE=True" % module_args_tmp
+                module_return = self.runner._execute_module(conn, tmp, 'file', module_args_tmp, inject=inject, complex_args=complex_args)
 
-            if raw:
-                return ReturnData(conn=conn, result=dict(dest=dest, changed=True))
-
-            # run the copy module
-            if raw:
-                # don't send down raw=no
-                module_args.pop('raw')
-            module_args = "%s src=%s original_basename=%s" % (module_args, pipes.quote(tmp_src), pipes.quote(os.path.basename(source)))
-            return self.runner._execute_module(conn, tmp, 'copy', module_args, inject=inject, complex_args=complex_args)
+            module_result = module_return.result
+            if module_result.get('failed') == True:
+                return module_return
+            if module_result.get('changed') == True:
+                changed = True
 
+        # TODO: Support detailed status/diff for multiple files
+        if len(source_files) == 1:
+            result = module_result
         else:
-            # no need to transfer the file, already correct md5, but still need to call
-            # the file module in case we want to change attributes
-
-            if content is not None:
-                os.remove(tmp_content)
-
-            if raw:
-                return ReturnData(conn=conn, result=dict(dest=dest, changed=False))
-
-            tmp_src = tmp + os.path.basename(source)
-            if raw:
-                # don't send down raw=no
-                module_args.pop('raw')
-            module_args = "%s src=%s" % (module_args, pipes.quote(tmp_src))
-            if self.runner.noop_on_check(inject):
-                module_args = "%s CHECKMODE=True" % module_args
-            results = self.runner._execute_module(conn, tmp, 'file', module_args, inject=inject, complex_args=complex_args)
-            results = self._result_key_merge(options, results)
-            return results
+            result = dict(dest=dest, src=source, changed=changed)
+        if len(diffs) == 1:
+            return ReturnData(conn=conn, result=result, diff=diffs[0])
+        else:
+            return ReturnData(conn=conn, result=result)
 
     def _get_diff_data(self, conn, tmp, inject, destination, source):
         peek_result = self.runner._execute_module(conn, tmp, 'file', "path=%s diff_peek=1" % destination, inject=inject, persist_files=True)
diff --git a/library/files/copy b/library/files/copy
index 8f2843b314..99c3391083 100644
--- a/library/files/copy
+++ b/library/files/copy
@@ -31,6 +31,10 @@ options:
   src:
     description:
       - Local path to a file to copy to the remote server; can be absolute or relative.
+        If path is a directory, it is copied recursively. In this case, if path ends
+        with "/", only inside contents of that directory are copied to destination.
+        Otherwise, if it does not end with "/", the directory itself with all contents
+        is copied. This behavior is similar to Rsync.
     required: false
     default: null
     aliases: []
@@ -42,7 +46,8 @@ options:
     default: null
   dest:
     description:
-      - Remote absolute path where the file should be copied to.
+      - Remote absolute path where the file should be copied to. If src is a directory,
+        this must be a directory too.
     required: true
     default: null
   backup:
@@ -76,8 +81,8 @@ options:
     required: false
 author: Michael DeHaan
 notes:
-   - The "copy" module can't be used to recursively copy directory structures to the target machine. Please see the
-     "Delegation" section of the Advanced Playbooks documentation for a better approach to recursive copies.
+   - The "copy" module recursively copy facility does not scale to lots (>hundreds) of files.
+     For alternative, see "Delegation" section of the Advanced Playbooks documentation.
 '''
 
 EXAMPLES = '''
@@ -122,6 +127,13 @@ def main():
     md5sum_src = module.md5(src)
     md5sum_dest = None
 
+    # Special handling for recursive copy - create intermediate dirs
+    if original_basename and dest.endswith("/"):
+        dest = os.path.join(dest, original_basename)
+        dirname = os.path.dirname(dest)
+        if not os.path.exists(dirname):
+            os.makedirs(dirname)
+
     if os.path.exists(dest):
         if not force:
             module.exit_json(msg="file already exists", src=src, dest=dest, changed=False)
diff --git a/library/files/file b/library/files/file
index 01772a7e71..cb81bf1ce1 100644
--- a/library/files/file
+++ b/library/files/file
@@ -145,6 +145,7 @@ def main():
         argument_spec = dict(
             state = dict(choices=['file','directory','link','hard','touch','absent'], default=None),
             path  = dict(aliases=['dest', 'name'], required=True),
+            original_basename = dict(required=False), # Internal use only, for recursive ops
             recurse  = dict(default='no', type='bool'),
             force = dict(required=False,default=False,type='bool'),
             diff_peek = dict(default=None),
@@ -200,7 +201,11 @@ def main():
         src = os.path.expanduser(src)
 
     if src is not None and os.path.isdir(path) and state not in ["link", "absent"]:
-        params['path'] = path = os.path.join(path, os.path.basename(src))
+        if params['original_basename']:
+            basename = params['original_basename']
+        else:
+            basename = os.path.basename(src)
+        params['path'] = path = os.path.join(path, basename)
 
     file_args = module.load_file_common_arguments(params)
 
diff --git a/test/TestPlayBook.py b/test/TestPlayBook.py
index 8a4ceb9584..fcee39db4f 100644
--- a/test/TestPlayBook.py
+++ b/test/TestPlayBook.py
@@ -393,6 +393,23 @@ class TestPlaybook(unittest.TestCase):
       assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
 
 
+   def test_recursive_copy(self):
+       pb = 'test/playbook-recursive-copy.yml'
+       actual = self._run(pb)
+
+       expected =  {
+            "localhost": {
+                "changed": 65,
+                "failures": 0,
+                "ok": 73,
+                "skipped": 0,
+                "unreachable": 0
+            }
+       }
+
+       assert utils.jsonify(expected, format=True) == utils.jsonify(actual,format=True)
+
+
    def _compare_file_output(self, filename, expected_lines):
       actual_lines = []
       with open(filename) as f:
diff --git a/test/playbook-recursive-copy.yml b/test/playbook-recursive-copy.yml
new file mode 100644
index 0000000000..a4c6b948d8
--- /dev/null
+++ b/test/playbook-recursive-copy.yml
@@ -0,0 +1,133 @@
+---
+# To run me manually, use: -i "localhost,"
+- hosts: localhost
+  connection: local
+  gather_facts: no
+  vars:
+    - testdir: /tmp/ansible-rcopy
+    - filesdir: test_recursive_copy/files
+  tasks:
+
+    #
+    # First, regression tests for single-file behavior
+    #
+
+    - name: "src single file, dest file"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}/file1
+      register: res
+    - command: test -f {{testdir}}/file1
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}/file1
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    - name: "src single file, dest dir w/trailing slash"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}/
+      register: res
+    - command: test -f {{testdir}}/test1
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}/
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    - name: "src single file, dest dir wo/trailing slash - doesn't behave in sane way"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}
+      register: res
+    - shell: test -f {{testdir}}/test1
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir/subdir2/subdir3/test1 dest={{testdir}}
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    #
+    # Now, test recursive behavior
+    #
+
+    - name: "src dir w/trailing slash, dest w/trailing slash"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir/ dest={{testdir}}/
+      register: res
+    - command: test -d {{testdir}}/subdir2
+    - command: test -d {{testdir}}/subdir2/subdir3
+    - command: test -d {{testdir}}/subdir2/subdir3
+    - command: test -f {{testdir}}/subdir2/subdir3/test1
+    - command: test -f {{testdir}}/subdir2/subdir3/test2
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir/ dest={{testdir}}/
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    # Expecting the same behavior
+    - name: "src dir w/trailing slash, dest wo/trailing slash"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir/ dest={{testdir}}
+      register: res
+    - command: test -d {{testdir}}/subdir2
+    - command: test -d {{testdir}}/subdir2/subdir3
+    - command: test -d {{testdir}}/subdir2/subdir3
+    - command: test -f {{testdir}}/subdir2/subdir3/test1
+    - command: test -f {{testdir}}/subdir2/subdir3/test2
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir/ dest={{testdir}}
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    - name: "src dir wo/trailing slash, dest w/trailing slash"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir dest={{testdir}}/
+      register: res
+    - command: test -d {{testdir}}/subdir/subdir2
+    - command: test -d {{testdir}}/subdir/subdir2/subdir3
+    - command: test -d {{testdir}}/subdir/subdir2/subdir3
+    - command: test -f {{testdir}}/subdir/subdir2/subdir3/test1
+    - command: test -f {{testdir}}/subdir/subdir2/subdir3/test2
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir dest={{testdir}}/
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+    # Expecting the same behavior
+    - name: "src dir wo/trailing slash, dest wo/trailing slash"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir dest={{testdir}}
+      register: res
+    - command: test -d {{testdir}}/subdir/subdir2
+    - command: test -d {{testdir}}/subdir/subdir2/subdir3
+    - command: test -d {{testdir}}/subdir/subdir2/subdir3
+    - command: test -f {{testdir}}/subdir/subdir2/subdir3/test1
+    - command: test -f {{testdir}}/subdir/subdir2/subdir3/test2
+    - command: test "{{res.changed}}" == "True"
+    - copy: src={{filesdir}}/subdir dest={{testdir}}
+      register: res
+    - command: test "{{res.changed}}" == "False"
+
+
+    - name: "Verifying notify handling for recursive files"
+      command: rm -rf {{testdir}}
+    - file: state=directory dest={{testdir}}
+    - copy: src={{filesdir}}/subdir dest={{testdir}}
+      notify:
+        - files changed
+    - meta: flush_handlers
+    - command: test -f {{testdir}}/notify_fired
+
+    - command: rm {{testdir}}/notify_fired
+    - copy: src={{filesdir}}/subdir dest={{testdir}}
+      notify:
+        - files changed
+    - meta: flush_handlers
+    - command: test ! -f {{testdir}}/notify_fired
+
+  handlers:
+    - name: files changed
+      command: touch {{testdir}}/notify_fired
diff --git a/test/test_recursive_copy/files/subdir/subdir2/subdir3/test1 b/test/test_recursive_copy/files/subdir/subdir2/subdir3/test1
new file mode 100644
index 0000000000..9f71d140ff
--- /dev/null
+++ b/test/test_recursive_copy/files/subdir/subdir2/subdir3/test1
@@ -0,0 +1,2 @@
+test1
+
diff --git a/test/test_recursive_copy/files/subdir/subdir2/subdir3/test2 b/test/test_recursive_copy/files/subdir/subdir2/subdir3/test2
new file mode 100644
index 0000000000..4a19b05426
--- /dev/null
+++ b/test/test_recursive_copy/files/subdir/subdir2/subdir3/test2
@@ -0,0 +1,2 @@
+test2
+