Repository: PyFilesystem/pyfilesystem2
Pull request #188 (2018-07-15 08:01:05)

Parallel copy
Adds parallel copying to fs.mirror, fs.copy, and fs.move.

A `workers` parameter has been added to these methods. If 0 (the default) then the copy will be done as it always has. If 1 or greater, it will use that many threads to copy the data.

This is probably only useful for copying to network filesystems, where the available bandwidth can be maximised. 

This may not be the last work in concurrent uploads, it assumed that its the data that is the bottleneck. The scanning directories etc. is still done in the main thread. 

NB. Also ran 'black' on the new code. 

coveralls: 
[![Coverage Status](https://coveralls.io/builds/17990956/badge)](https://coveralls.io/builds/17990956)

Coverage increased (+0.02%) to 100.0% when pulling **1b0fb020e4061d33aab36053315a2d3f6491a23d on parallel-copy** into **e63f5af191fb477d42720ec19feec81147a2b889 on master**.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 86775b0..5fed0d7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,9 +4,18 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## [2.0.25] - 2018-07-20
+
+### Added
+
+- workers parameter to fs.copy, fs.move, and fs.mirror for concurrent
+    copies
+
 ## [2.0.24] - 2018-06-28
 
-### Added timeout to FTP opener
+### Added
+
+- timeout to FTP opener
 
 ## [2.0.23] - 2018-05-02
 
@@ -210,5 +219,3 @@ No changes, pushed wrong branch to PyPi.
 ## [2.0.0] - 2016-12-07
 
 New version of the PyFilesystem API.
-
-
diff --git a/fs/_bulk.py b/fs/_bulk.py
new file mode 100644
index 0000000..8b53438
--- /dev/null
+++ b/fs/_bulk.py
@@ -0,0 +1,140 @@
+"""
+
+Implements a thread pool for parallel copying of files.
+
+"""
+
+from __future__ import unicode_literals
+
+import threading
+
+from six.moves.queue import Queue
+
+from .copy import copy_file_internal
+from .errors import BulkCopyFailed
+from .tools import copy_file_data
+
+if False:  # typing.TYPE_CHECKING
+    from .base import FS
+    from types import TracebackType
+    from typing import IO, Iterator, List, Optional, Mapping, Text, Type, Union
+
+
+class _Worker(threading.Thread):
+    """Worker thread that pulls tasks from a queue."""
+
+    def __init__(self, copier):
+        # type (Copier) -> None
+        self.copier = copier
+        super(_Worker, self).__init__()
+        self.daemon = True
+
+    def run(self):
+        # type () -> None
+        queue = self.copier.queue
+        while True:
+            task = queue.get(block=True)
+            try:
+                if task is None:
+                    break  # Sentinel to exit thread
+                task()
+            except Exception as error:
+                self.copier.add_error(error)
+            finally:
+                queue.task_done()
+
+
+class _Task(object):
+    """Base class for a task."""
+
+    def __call__(self):
+        # type: () -> None
+        """Task implementation."""
+
+
+class _CopyTask(_Task):
+    """A callable that copies from one file another."""
+
+    def __init__(self, src_file, dst_file):
+        # type: (IO, IO) -> None
+        self.src_file = src_file
+        self.dst_file = dst_file
+
+    def __call__(self):
+        # type: () -> None
+        try:
+            copy_file_data(self.src_file, self.dst_file, chunk_size=1024 * 1024)
+        finally:
+            try:
+                self.src_file.close()
+            finally:
+                self.dst_file.close()
+
+
+class Copier(object):
+    """Copy files in worker threads."""
+
+    def __init__(self, num_workers=4):
+        # type: (int) -> None
+        if num_workers < 0:
+            raise ValueError('num_workers must be >= 0')
+        self.num_workers = num_workers
+        self.queue = None  # type: Optional[Queue[_Task]]
+        self.workers = []  # type: List[_Worker]
+        self.errors = []  # type: List[Exception]
+        self.running = False
+
+    def start(self):
+        """Start the workers."""
+        if self.num_workers:
+            self.queue = Queue(maxsize=self.num_workers)
+            self.workers = [_Worker(self) for _ in range(self.num_workers)]
+            for worker in self.workers:
+                worker.start()
+        self.running = True
+
+    def stop(self):
+        """Stop the workers (will block until they are finished)."""
+        if self.running and self.num_workers:
+            for worker in self.workers:
+                self.queue.put(None)
+            for worker in self.workers:
+                worker.join()
+            # Free up references help by workers
+            del self.workers[:]
+            self.queue.join()
+        self.running = False
+
+    def add_error(self, error):
+        """Add an exception raised by a task."""
+        self.errors.append(error)
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(
+        self,
+        exc_type,  # type: Optional[Type[BaseException]]
+        exc_value,  # type: Optional[BaseException]
+        traceback,  # type: Optional[TracebackType]
+    ):
+        self.stop()
+        if traceback is None and self.errors:
+            raise BulkCopyFailed(self.errors)
+
+    def copy(self, src_fs, src_path, dst_fs, dst_path):
+        # type: (FS, Text, FS, Text) -> None
+        """Copy a file from one fs to another."""
+        if self.queue is None:
+            # This should be the most performant for a single-thread
+            copy_file_internal(src_fs, src_path, dst_fs, dst_path)
+        else:
+            src_file = src_fs.openbin(src_path, "r")
+            try:
+                dst_file = dst_fs.openbin(dst_path, "w")
+            except Exception:
+                src_file.close()
+                raise
+            task = _CopyTask(src_file, dst_file)
+            self.queue.put(task)
diff --git a/fs/_version.py b/fs/_version.py
index 55c2db1..4fb3d7b 100644
--- a/fs/_version.py
+++ b/fs/_version.py
@@ -1,3 +1,3 @@
 """Version, used in module and setup.py.
 """
-__version__ = "2.0.24"
+__version__ = "2.0.25"
diff --git a/fs/base.py b/fs/base.py
index c9c4835..7186ad0 100644
--- a/fs/base.py
+++ b/fs/base.py
@@ -1368,7 +1368,8 @@ class FS(object):
         """Create or replace a file with text.
 
         Arguments:
-            contents (str): A path on the filesystem.
+            path (str): Destination path on the filesystem.
+            contents (str): Text to be written.
             encoding (str, optional): Encoding of destination file
                 (defaults to ``'ut-8'``).
             errors (str, optional): How encoding errors should be treated
diff --git a/fs/copy.py b/fs/copy.py
index 8eddcd4..624449c 100644
--- a/fs/copy.py
+++ b/fs/copy.py
@@ -1,31 +1,30 @@
 """Functions for copying resources *between* filesystem.
 """
 
-from __future__ import print_function
-from __future__ import unicode_literals
+from __future__ import print_function, unicode_literals
 
 import typing
 
-from .walk import Walker
-from .opener import manage_fs
-from .path import abspath
-from .path import combine
-from .path import frombase
-from .path import normpath
 from .errors import FSError
+from .opener import manage_fs
+from .path import abspath, combine, frombase, normpath
+from .walk import Walker
 
 if False:  # typing.TYPE_CHECKING
     from typing import Callable, Optional, Text, Union
     from .base import FS
     from .walk import Walker
+
     _OnCopy = Callable[[FS, Text, FS, Text], object]
 
 
-def copy_fs(src_fs,       # type: Union[FS, Text]
-            dst_fs,       # type: Union[FS, Text]
-            walker=None,  # type: Optional[Walker]
-            on_copy=None  # type: Optional[_OnCopy]
-            ):
+def copy_fs(
+    src_fs,  # type: Union[FS, Text]
+    dst_fs,  # type: Union[FS, Text]
+    walker=None,  # type: Optional[Walker]
+    on_copy=None,  # type: Optional[_OnCopy]
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Copy the contents of one filesystem to another.
 
@@ -38,17 +37,22 @@ def copy_fs(src_fs,       # type: Union[FS, Text]
         on_copy (callable):A function callback called after a single file copy
             is executed. Expected signature is ``(src_fs, src_path, dst_fs,
             dst_path)``.
+        workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
-    return copy_dir(src_fs, '/', dst_fs, '/',
-                    walker=walker, on_copy=on_copy)
-
-
-def copy_fs_if_newer(src_fs,        # type: Union[FS, Text]
-                     dst_fs,        # type: Union[FS, Text]
-                     walker=None,   # type: Optional[Walker]
-                     on_copy=None   # type: Optional[_OnCopy]
-                     ):
+    return copy_dir(
+        src_fs, "/", dst_fs, "/", walker=walker, on_copy=on_copy, workers=workers
+    )
+
+
+def copy_fs_if_newer(
+    src_fs,  # type: Union[FS, Text]
+    dst_fs,  # type: Union[FS, Text]
+    walker=None,  # type: Optional[Walker]
+    on_copy=None,  # type: Optional[_OnCopy]
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Copy the contents of one filesystem to another, checking times.
 
@@ -66,10 +70,13 @@ def copy_fs_if_newer(src_fs,        # type: Union[FS, Text]
         on_copy (callable):A function callback called after a single file copy
             is executed. Expected signature is ``(src_fs, src_path, dst_fs,
             dst_path)``.
+         workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
-    return copy_dir_if_newer(src_fs, '/', dst_fs, '/',
-                             walker=walker, on_copy=on_copy)
+    return copy_dir_if_newer(
+        src_fs, "/", dst_fs, "/", walker=walker, on_copy=on_copy, workers=workers
+    )
 
 
 def _source_is_newer(src_fs, src_path, dst_fs, dst_path):
@@ -90,22 +97,23 @@ def _source_is_newer(src_fs, src_path, dst_fs, dst_path):
     """
     try:
         if dst_fs.exists(dst_path):
-            namespace = ('details', 'modified')
+            namespace = ("details", "modified")
             src_modified = src_fs.getinfo(src_path, namespace).modified
             if src_modified is not None:
                 dst_modified = dst_fs.getinfo(dst_path, namespace).modified
                 return dst_modified is None or src_modified > dst_modified
         return True
     except FSError:  # pragma: no cover
-        #todo: should log something here
+        # todo: should log something here
         return True
 
 
-def copy_file(src_fs,    # type: Union[FS, Text]
-              src_path,  # type: Text
-              dst_fs,    # type: Union[FS, Text]
-              dst_path   # type: Text
-              ):
+def copy_file(
+    src_fs,  # type: Union[FS, Text]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[FS, Text]
+    dst_path,  # type: Text
+):
     # type: (...) -> None
     """Copy a file from one filesystem to another.
 
@@ -118,7 +126,7 @@ def copy_file(src_fs,    # type: Union[FS, Text]
         dst_path (str): Path to a file on the destination filesystem.
 
     """
-    with manage_fs(src_fs) as _src_fs:
+    with manage_fs(src_fs, writeable=False) as _src_fs:
         with manage_fs(dst_fs, create=True) as _dst_fs:
             if _src_fs is _dst_fs:
                 # Same filesystem, so we can do a potentially optimized
@@ -128,18 +136,19 @@ def copy_file(src_fs,    # type: Union[FS, Text]
                 # Standard copy
                 with _src_fs.lock(), _dst_fs.lock():
                     if _dst_fs.hassyspath(dst_path):
-                        with _dst_fs.openbin(dst_path, 'w') as write_file:
+                        with _dst_fs.openbin(dst_path, "w") as write_file:
                             _src_fs.getfile(src_path, write_file)
                     else:
                         with _src_fs.openbin(src_path) as read_file:
                             _dst_fs.setbinfile(dst_path, read_file)
 
 
-def copy_file_internal(src_fs,      # type: FS
-                       src_path,    # type: Text
-                       dst_fs,      # type: FS
-                       dst_path     # type: Text
-                       ):
+def copy_file_internal(
+    src_fs,  # type: FS
+    src_path,  # type: Text
+    dst_fs,  # type: FS
+    dst_path,  # type: Text
+):
     # type: (...) -> None
     """Low level copy, that doesn't call manage_fs or lock.
 
@@ -160,18 +169,19 @@ def copy_file_internal(src_fs,      # type: FS
         # copy
         src_fs.copy(src_path, dst_path, overwrite=True)
     elif dst_fs.hassyspath(dst_path):
-        with dst_fs.openbin(dst_path, 'w') as write_file:
+        with dst_fs.openbin(dst_path, "w") as write_file:
             src_fs.getfile(src_path, write_file)
     else:
         with src_fs.openbin(src_path) as read_file:
             dst_fs.setbinfile(dst_path, read_file)
 
 
-def copy_file_if_newer(src_fs,      # type: Union[FS, Text]
-                       src_path,    # type: Text
-                       dst_fs,      # type: Union[FS, Text]
-                       dst_path     # type: Text
-                       ):
+def copy_file_if_newer(
+    src_fs,  # type: Union[FS, Text]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[FS, Text]
+    dst_path,  # type: Text
+):
     # type: (...) -> bool
     """Copy a file from one filesystem to another, checking times.
 
@@ -191,7 +201,7 @@ def copy_file_if_newer(src_fs,      # type: Union[FS, Text]
         bool: `True` if the file copy was executed, `False` otherwise.
 
     """
-    with manage_fs(src_fs) as _src_fs:
+    with manage_fs(src_fs, writeable=False) as _src_fs:
         with manage_fs(dst_fs, create=True) as _dst_fs:
             if _src_fs is _dst_fs:
                 # Same filesystem, so we can do a potentially optimized
@@ -211,10 +221,11 @@ def copy_file_if_newer(src_fs,      # type: Union[FS, Text]
                         return False
 
 
-def copy_structure(src_fs,      # type: Union[FS, Text]
-                   dst_fs,      # type: Union[FS, Text]
-                   walker=None  # type: Optional[Walker]
-                   ):
+def copy_structure(
+    src_fs,  # type: Union[FS, Text]
+    dst_fs,  # type: Union[FS, Text]
+    walker=None,  # type: Optional[Walker]
+):
     # type: (...) -> None
     """Copy directories (but not files) from ``src_fs`` to ``dst_fs``.
 
@@ -234,13 +245,15 @@ def copy_structure(src_fs,      # type: Union[FS, Text]
                     _dst_fs.makedir(dir_path, recreate=True)
 
 
-def copy_dir(src_fs,        # type: Union[FS, Text]
-             src_path,      # type: Text
-             dst_fs,        # type: Union[FS, Text]
-             dst_path,      # type: Text
-             walker=None,   # type: Optional[Walker]
-             on_copy=None   # type: Optional[_OnCopy]
-             ):
+def copy_dir(
+    src_fs,  # type: Union[FS, Text]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[FS, Text]
+    dst_path,  # type: Text
+    walker=None,  # type: Optional[Walker]
+    on_copy=None,  # type: Optional[_OnCopy]
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Copy a directory from one filesystem to another.
 
@@ -255,45 +268,47 @@ def copy_dir(src_fs,        # type: Union[FS, Text]
         on_copy (callable, optional):  A function callback called after
             a single file copy is executed. Expected signature is
             ``(src_fs, src_path, dst_fs, dst_path)``.
+        workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
     on_copy = on_copy or (lambda *args: None)
     walker = walker or Walker()
     _src_path = abspath(normpath(src_path))
     _dst_path = abspath(normpath(dst_path))
-    with manage_fs(src_fs) as _src_fs:
-        with manage_fs(dst_fs, create=True) as _dst_fs:
-            with _src_fs.lock(), _dst_fs.lock():
+
+    def src():
+        return manage_fs(src_fs, writeable=False)
+
+    def dst():
+        return manage_fs(dst_fs, create=True)
+
+    from ._bulk import Copier
+
+    with src() as _src_fs, dst() as _dst_fs:
+        with _src_fs.lock(), _dst_fs.lock():
+            with Copier(num_workers=workers) as copier:
                 _dst_fs.makedir(_dst_path, recreate=True)
                 for dir_path, dirs, files in walker.walk(_src_fs, _src_path):
-                    copy_path = combine(
-                        _dst_path,
-                        frombase(_src_path, dir_path)
-                    )
+                    copy_path = combine(_dst_path, frombase(_src_path, dir_path))
                     for info in dirs:
-                        _dst_fs.makedir(
-                            info.make_path(copy_path),
-                            recreate=True
-                        )
+                        _dst_fs.makedir(info.make_path(copy_path), recreate=True)
                     for info in files:
                         src_path = info.make_path(dir_path)
                         dst_path = info.make_path(copy_path)
-                        copy_file_internal(
-                            _src_fs,
-                            src_path,
-                            _dst_fs,
-                            dst_path
-                        )
+                        copier.copy(_src_fs, src_path, _dst_fs, dst_path)
                         on_copy(_src_fs, src_path, _dst_fs, dst_path)
 
 
-def copy_dir_if_newer(src_fs,       # type: Union[FS, Text]
-                      src_path,     # type: Text
-                      dst_fs,       # type: Union[FS, Text]
-                      dst_path,     # type: Text
-                      walker=None,  # type: Optional[Walker]
-                      on_copy=None  # type: Optional[_OnCopy]
-                      ):
+def copy_dir_if_newer(
+    src_fs,  # type: Union[FS, Text]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[FS, Text]
+    dst_path,  # type: Text
+    walker=None,  # type: Optional[Walker]
+    on_copy=None,  # type: Optional[_OnCopy]
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Copy a directory from one filesystem to another, checking times.
 
@@ -313,17 +328,28 @@ def copy_dir_if_newer(src_fs,       # type: Union[FS, Text]
         on_copy (callable, optional):  A function callback called after
             a single file copy is executed. Expected signature is
             ``(src_fs, src_path, dst_fs, dst_path)``.
+        workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
     on_copy = on_copy or (lambda *args: None)
     walker = walker or Walker()
     _src_path = abspath(normpath(src_path))
     _dst_path = abspath(normpath(dst_path))
-    with manage_fs(src_fs) as _src_fs:
-        with manage_fs(dst_fs, create=True) as _dst_fs:
-            with _src_fs.lock(), _dst_fs.lock():
+
+    def src():
+        return manage_fs(src_fs, writeable=False)
+
+    def dst():
+        return manage_fs(dst_fs, create=True)
+
+    from ._bulk import Copier
+
+    with src() as _src_fs, dst() as _dst_fs:
+        with _src_fs.lock(), _dst_fs.lock():
+            with Copier(num_workers=workers) as copier:
                 _dst_fs.makedir(_dst_path, recreate=True)
-                namespace = ('details', 'modified')
+                namespace = ("details", "modified")
                 dst_state = {
                     path: info
                     for path, info in walker.info(_dst_fs, _dst_path, namespace)
@@ -334,10 +360,7 @@ def copy_dir_if_newer(src_fs,       # type: Union[FS, Text]
                     for path, info in walker.info(_src_fs, _src_path, namespace)
                 ]
                 for dir_path, copy_info in src_state:
-                    copy_path = combine(
-                        _dst_path,
-                        frombase(_src_path, dir_path)
-                    )
+                    copy_path = combine(_dst_path, frombase(_src_path, dir_path))
                     if copy_info.is_dir:
                         _dst_fs.makedir(copy_path, recreate=True)
                     elif copy_info.is_file:
@@ -350,11 +373,11 @@ def copy_dir_if_newer(src_fs,       # type: Union[FS, Text]
                             do_copy = True
                         else:
                             do_copy = (
-                                src_modified is None or
-                                dst_modified is None or
-                                src_modified > dst_modified
+                                src_modified is None
+                                or dst_modified is None
+                                or src_modified > dst_modified
                             )
 
                         if do_copy:
-                            copy_file_internal(_src_fs, dir_path, _dst_fs, copy_path)
+                            copier.copy(_src_fs, dir_path, _dst_fs, copy_path)
                             on_copy(_src_fs, dir_path, _dst_fs, copy_path)
diff --git a/fs/errors.py b/fs/errors.py
index cfc3abc..bff976d 100644
--- a/fs/errors.py
+++ b/fs/errors.py
@@ -22,34 +22,35 @@ if False:  # typing.TYPE_CHECKING
 
 
 __all__ = [
-    'CreateFailed',
-    'DestinationExists',
-    'DirectoryExists',
-    'DirectoryExpected',
-    'DirectoryNotEmpty',
-    'FileExists',
-    'FileExpected',
-    'FilesystemClosed',
-    'FSError',
-    'IllegalBackReference',
-    'InsufficientStorage',
-    'InvalidCharsInPath',
-    'InvalidPath',
-    'MissingInfoNamespace',
-    'NoSysPath',
-    'NoURL',
-    'OperationFailed',
-    'OperationTimeout',
-    'PathError',
-    'PermissionDenied',
-    'RemoteConnectionError',
-    'RemoveRootError',
-    'ResourceError',
-    'ResourceInvalid',
-    'ResourceLocked',
-    'ResourceNotFound',
-    'ResourceReadOnly',
-    'Unsupported',
+    "BulkCopyFailed",
+    "CreateFailed",
+    "DestinationExists",
+    "DirectoryExists",
+    "DirectoryExpected",
+    "DirectoryNotEmpty",
+    "FileExists",
+    "FileExpected",
+    "FilesystemClosed",
+    "FSError",
+    "IllegalBackReference",
+    "InsufficientStorage",
+    "InvalidCharsInPath",
+    "InvalidPath",
+    "MissingInfoNamespace",
+    "NoSysPath",
+    "NoURL",
+    "OperationFailed",
+    "OperationTimeout",
+    "PathError",
+    "PermissionDenied",
+    "RemoteConnectionError",
+    "RemoveRootError",
+    "ResourceError",
+    "ResourceInvalid",
+    "ResourceLocked",
+    "ResourceNotFound",
+    "ResourceReadOnly",
+    "Unsupported",
 ]
 
 
@@ -60,9 +61,7 @@ class MissingInfoNamespace(AttributeError):
     def __init__(self, namespace):
         # type: (Text) -> None
         msg = "namespace '{}' is required for this attribute"
-        super(MissingInfoNamespace, self).__init__(
-            msg.format(namespace)
-        )
+        super(MissingInfoNamespace, self).__init__(msg.format(namespace))
 
 
 @six.python_2_unicode_compatible
@@ -97,6 +96,18 @@ class FilesystemClosed(FSError):
     default_message = "attempt to use closed filesystem"
 
 
+class BulkCopyFailed(FSError):
+    """A copy operation failed in worker threads."""
+
+    default_message = (
+        "One or more copy operations failed (see errors attribute)"
+    )
+
+    def __init__(self, errors):
+        self.errors = errors
+        super(BulkCopyFailed, self).__init__()
+
+
 class CreateFailed(FSError):
     """Filesystem could not be created.
     """
@@ -106,7 +117,7 @@ class CreateFailed(FSError):
     def __init__(self, msg=None, exc=None):
         # type: (Optional[Text], Optional[Exception]) -> None
         self._msg = msg or self.default_message
-        self.details = '' if exc is None else text_type(exc)
+        self.details = "" if exc is None else text_type(exc)
         self.exc = exc
 
     @classmethod
@@ -119,8 +130,10 @@ class CreateFailed(FSError):
                 raise
             except Exception as e:
                 raise cls(exc=e)
+
         return new_func  # type: ignore
 
+
 class PathError(FSError):
     """Base exception for errors to do with a path string.
     """
@@ -172,15 +185,16 @@ class OperationFailed(FSError):
 
     default_message = "operation failed, {details}"
 
-    def __init__(self,
-                 path=None,     # type: Optional[Text]
-                 exc=None,      # type: Optional[Exception]
-                 msg=None       # type: Optional[Text]
-                 ):
+    def __init__(
+        self,
+        path=None,  # type: Optional[Text]
+        exc=None,  # type: Optional[Exception]
+        msg=None,  # type: Optional[Text]
+    ):
         # type: (...) -> None
         self.path = path
         self.exc = exc
-        self.details = '' if exc is None else text_type(exc)
+        self.details = "" if exc is None else text_type(exc)
         self.errno = getattr(exc, "errno", None)
         super(OperationFailed, self).__init__(msg=msg)
 
@@ -325,7 +339,6 @@ class IllegalBackReference(ValueError):
 
     def __init__(self, path):
         # type: (Text) -> None
-        _msg = \
-            "path '{path}' contains back-references outside of filesystem"
+        _msg = "path '{path}' contains back-references outside of filesystem"
         _msg = _msg.format(path=path)
         super(IllegalBackReference, self).__init__(_msg)
diff --git a/fs/mirror.py b/fs/mirror.py
index 6bf8ad1..88a8aa3 100644
--- a/fs/mirror.py
+++ b/fs/mirror.py
@@ -19,15 +19,17 @@ the expense of potentially copying extra files.
 from __future__ import print_function
 from __future__ import unicode_literals
 
+from contextlib import contextmanager
 import typing
 
+from ._bulk import Copier
 from .copy import copy_file_internal
 from .errors import ResourceNotFound
 from .walk import Walker
 from .opener import manage_fs
 
 if False:  # typing.TYPE_CHECKING
-    from typing import Optional, Text, Union
+    from typing import Callable, Optional, Text, Union
     from .base import FS
     from .info import Info
 
@@ -49,11 +51,13 @@ def _compare(info1, info2):
     return date1 is None or date2 is None or date1 > date2
 
 
-def mirror(src_fs,              # type: Union[FS, Text]
-           dst_fs,              # type: Union[FS, Text]
-           walker=None,         # type: Optional[Walker]
-           copy_if_newer=True   # type: bool
-           ):
+def mirror(
+    src_fs,  # type: Union[FS, Text]
+    dst_fs,  # type: Union[FS, Text]
+    walker=None,  # type: Optional[Walker]
+    copy_if_newer=True,  # type: bool
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Mirror files / directories from one filesystem to another.
 
@@ -66,28 +70,44 @@ def mirror(src_fs,              # type: Union[FS, Text]
         dst_fs (FS or str): Destination filesystem (URL or instance).
         walker (~fs.walk.Walker, optional): An optional walker instance.
         copy_if_newer (bool): Only copy newer files (the default).
-
+        workers (int): Number of worker threads used
+            (0 for single threaded). Set to a relatively low number
+            for network filesystems, 4 would be a good start.
     """
-    with manage_fs(src_fs, writeable=False) as _src_fs:
-        with manage_fs(dst_fs, create=True) as _dst_fs:
-            with _src_fs.lock(), _dst_fs.lock():
-                return _mirror(
+
+    def src():
+        return manage_fs(src_fs, writeable=False)
+
+    def dst():
+        return manage_fs(dst_fs, create=True)
+
+    with src() as _src_fs, dst() as _dst_fs:
+        with _src_fs.lock(), _dst_fs.lock():
+            with Copier(num_workers=workers) as copier:
+                _mirror(
                     _src_fs,
                     _dst_fs,
                     walker=walker,
-                    copy_if_newer=copy_if_newer
+                    copy_if_newer=copy_if_newer,
+                    copy_file=copier.copy,
                 )
 
 
-def _mirror(src_fs, dst_fs, walker=None, copy_if_newer=True):
-    # type: (FS, FS, Optional[Walker], bool) -> None
+def _mirror(
+    src_fs,
+    dst_fs,
+    walker=None,
+    copy_if_newer=True,
+    copy_file=copy_file_internal,
+):
+    # type: (FS, FS, Optional[Walker], bool, Callable[[FS, str, FS, str], None]) -> None
     walker = walker or Walker()
-    walk = walker.walk(src_fs, namespaces=['details'])
+    walk = walker.walk(src_fs, namespaces=["details"])
     for path, dirs, files in walk:
         try:
             dst = {
                 info.name: info
-                for info in dst_fs.scandir(path, namespaces=['details'])
+                for info in dst_fs.scandir(path, namespaces=["details"])
             }
         except ResourceNotFound:
             dst_fs.makedir(path)
@@ -105,7 +125,7 @@ def _mirror(src_fs, dst_fs, walker=None, copy_if_newer=True):
                     # Compare file info
                     if copy_if_newer and not _compare(_file, dst_file):
                         continue
-            copy_file_internal(src_fs, _path, dst_fs, _path)
+            copy_file(src_fs, _path, dst_fs, _path)
 
         # Make directories
         for _dir in dirs:
diff --git a/fs/move.py b/fs/move.py
index a1d9161..07a29ee 100644
--- a/fs/move.py
+++ b/fs/move.py
@@ -15,23 +15,26 @@ if False:  # typing.TYPE_CHECKING
     from typing import Text, Union
 
 
-def move_fs(src_fs, dst_fs):
+def move_fs(src_fs, dst_fs, workers=0):
     # type: (Union[Text, FS], Union[Text, FS]) -> None
     """Move the contents of a filesystem to another filesystem.
 
     Arguments:
         src_fs (FS or str): Source filesystem (instance or URL).
         dst_fs (FS or str): Destination filesystem (instance or URL).
+        workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
-    move_dir(src_fs, '/', dst_fs, '/')
+    move_dir(src_fs, "/", dst_fs, "/", workers=workers)
 
 
-def move_file(src_fs,       # type: Union[Text, FS]
-              src_path,     # type: Text
-              dst_fs,       # type: Union[Text, FS]
-              dst_path      # type: Text
-              ):
+def move_file(
+    src_fs,  # type: Union[Text, FS]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[Text, FS]
+    dst_path,  # type: Text
+):
     # type: (...) -> None
     """Move a file from one filesystem to another.
 
@@ -54,11 +57,13 @@ def move_file(src_fs,       # type: Union[Text, FS]
                     _src_fs.remove(src_path)
 
 
-def move_dir(src_fs,        # type: Union[Text, FS]
-             src_path,      # type: Text
-             dst_fs,        # type: Union[Text, FS]
-             dst_path       # type: Text
-             ):
+def move_dir(
+    src_fs,  # type: Union[Text, FS]
+    src_path,  # type: Text
+    dst_fs,  # type: Union[Text, FS]
+    dst_path,  # type: Text
+    workers=0,  # type: int
+):
     # type: (...) -> None
     """Move a directory from one filesystem to another.
 
@@ -66,17 +71,20 @@ def move_dir(src_fs,        # type: Union[Text, FS]
         src_fs (FS or str): Source filesystem (instance or URL).
         src_path (str): Path to a directory on ``src_fs``
         dst_fs (FS or str): Destination filesystem (instance or URL).
-        dst_path (str): Path to a directory on ``dst_fs``
+        dst_path (str): Path to a directory on ``dst_fs``.
+        workers (int): Use `worker` threads to copy data, or ``0`` (default) for
+            a single-threaded copy.
 
     """
-    with manage_fs(src_fs) as _src_fs:
-        with manage_fs(dst_fs, create=True) as _dst_fs:
-            with _src_fs.lock(), _dst_fs.lock():
-                _dst_fs.makedir(dst_path, recreate=True)
-                copy_dir(
-                    src_fs,
-                    src_path,
-                    dst_fs,
-                    dst_path
-                )
-                _src_fs.removetree(src_path)
+
+    def src():
+        return manage_fs(src_fs, writeable=False)
+
+    def dst():
+        return manage_fs(dst_fs, create=True)
+
+    with src() as _src_fs, dst() as _dst_fs:
+        with _src_fs.lock(), _dst_fs.lock():
+            _dst_fs.makedir(dst_path, recreate=True)
+            copy_dir(src_fs, src_path, dst_fs, dst_path, workers=workers)
+            _src_fs.removetree(src_path)


Test patch:
diff --git a/fs/test.py b/fs/test.py
index 653b866..2e3c081 100644
--- a/fs/test.py
+++ b/fs/test.py
@@ -1193,6 +1193,67 @@ class FSTestCases(object):
         with self.assertRaises(errors.FileExpected):
             self.fs.copy('dir', 'folder')
 
+    def _test_upload(self, workers):
+        """Test fs.copy with varying number of worker threads."""
+        data1 = b'foo' * 256 * 1024
+        data2 = b'bar' * 2 * 256 * 1024
+        data3 = b'baz' * 3 * 256 * 1024
+        data4 = b'egg' * 7 * 256 * 1024
+
+        with open_fs('temp://') as src_fs:
+            src_fs.setbytes('foo', data1)
+            src_fs.setbytes('bar', data2)
+            src_fs.makedir('dir1').setbytes('baz', data3)
+            src_fs.makedirs('dir2/dir3').setbytes('egg', data4)
+            dst_fs = self.fs
+            fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
+            self.assertEqual(dst_fs.getbytes('foo'), data1)
+            self.assertEqual(dst_fs.getbytes('bar'), data2)
+            self.assertEqual(dst_fs.getbytes('dir1/baz'), data3)
+            self.assertEqual(dst_fs.getbytes('dir2/dir3/egg'), data4)
+
+    def test_upload_0(self):
+        self._test_upload(0)
+
+    def test_upload_1(self):
+        self._test_upload(1)
+
+    def test_upload_2(self):
+        self._test_upload(2)
+
+    def test_upload_4(self):
+        self._test_upload(4)
+
+    def _test_download(self, workers):
+        """Test fs.copy with varying number of worker threads."""
+        data1 = b'foo' * 256 * 1024
+        data2 = b'bar' * 2 * 256 * 1024
+        data3 = b'baz' * 3 * 256 * 1024
+        data4 = b'egg' * 7 * 256 * 1024
+        src_fs = self.fs
+        with open_fs('temp://') as dst_fs:
+            src_fs.setbytes('foo', data1)
+            src_fs.setbytes('bar', data2)
+            src_fs.makedir('dir1').setbytes('baz', data3)
+            src_fs.makedirs('dir2/dir3').setbytes('egg', data4)
+            fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
+            self.assertEqual(dst_fs.getbytes('foo'), data1)
+            self.assertEqual(dst_fs.getbytes('bar'), data2)
+            self.assertEqual(dst_fs.getbytes('dir1/baz'), data3)
+            self.assertEqual(dst_fs.getbytes('dir2/dir3/egg'), data4)
+
+    def test_download_0(self):
+        self._test_download(0)
+
+    def test_download_1(self):
+        self._test_download(1)
+
+    def test_download_2(self):
+        self._test_download(2)
+
+    def test_download_4(self):
+        self._test_download(4)
+
     def test_create(self):
         # Test create new file
         self.assertFalse(self.fs.exists('foo'))
diff --git a/tests/test_bulk.py b/tests/test_bulk.py
new file mode 100644
index 0000000..5ff4c24
--- /dev/null
+++ b/tests/test_bulk.py
@@ -0,0 +1,19 @@
+from __future__ import unicode_literals
+
+import unittest
+
+from fs._bulk import Copier, _Task
+from fs.errors import BulkCopyFailed
+
+
+class BrokenTask(_Task):
+    def __call__(self):
+        1/0
+
+
+class TestBulk(unittest.TestCase):
+
+    def test_worker_error(self):
+        with self.assertRaises(BulkCopyFailed):
+            with Copier(num_workers=2) as copier:
+                copier.queue.put(BrokenTask())
diff --git a/tests/test_copy.py b/tests/test_copy.py
index c88720b..45d1d53 100644
--- a/tests/test_copy.py
+++ b/tests/test_copy.py
@@ -16,18 +16,19 @@ from fs import open_fs
 class TestCopy(unittest.TestCase):
 
     def test_copy_fs(self):
-        src_fs = open_fs('mem://')
-        src_fs.makedirs('foo/bar')
-        src_fs.makedirs('foo/empty')
-        src_fs.touch('test.txt')
-        src_fs.touch('foo/bar/baz.txt')
+        for workers in (0, 1, 2, 4):
+            src_fs = open_fs('mem://')
+            src_fs.makedirs('foo/bar')
+            src_fs.makedirs('foo/empty')
+            src_fs.touch('test.txt')
+            src_fs.touch('foo/bar/baz.txt')
 
-        dst_fs = open_fs('mem://')
-        fs.copy.copy_fs(src_fs, dst_fs)
+            dst_fs = open_fs('mem://')
+            fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
 
-        self.assertTrue(dst_fs.isdir('foo/empty'))
-        self.assertTrue(dst_fs.isdir('foo/bar'))
-        self.assertTrue(dst_fs.isfile('test.txt'))
+            self.assertTrue(dst_fs.isdir('foo/empty'))
+            self.assertTrue(dst_fs.isdir('foo/bar'))
+            self.assertTrue(dst_fs.isfile('test.txt'))
 
     def test_copy_dir(self):
         src_fs = open_fs('mem://')
@@ -35,13 +36,30 @@ class TestCopy(unittest.TestCase):
         src_fs.makedirs('foo/empty')
         src_fs.touch('test.txt')
         src_fs.touch('foo/bar/baz.txt')
-
-        dst_fs = open_fs('mem://')
-        fs.copy.copy_dir(src_fs, '/foo', dst_fs, '/')
-
-        self.assertTrue(dst_fs.isdir('bar'))
-        self.assertTrue(dst_fs.isdir('empty'))
-        self.assertTrue(dst_fs.isfile('bar/baz.txt'))
+        for workers in (0, 1, 2, 4):
+            with open_fs('mem://') as dst_fs:
+                fs.copy.copy_dir(src_fs, '/foo', dst_fs, '/', workers=workers)
+                self.assertTrue(dst_fs.isdir('bar'))
+                self.assertTrue(dst_fs.isdir('empty'))
+                self.assertTrue(dst_fs.isfile('bar/baz.txt'))
+
+    def test_copy_large(self):
+        data1 = b'foo' * 512 * 1024
+        data2 = b'bar' * 2 * 512 * 1024
+        data3 = b'baz' * 3 * 512 * 1024
+        data4 = b'egg' * 7 * 512 * 1024
+        with open_fs('temp://') as src_fs:
+            src_fs.setbytes('foo', data1)
+            src_fs.setbytes('bar', data2)
+            src_fs.makedir('dir1').setbytes('baz', data3)
+            src_fs.makedirs('dir2/dir3').setbytes('egg', data4)
+            for workers in (0, 1, 2, 4):
+                with open_fs('temp://') as dst_fs:
+                    fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
+                    self.assertEqual(dst_fs.getbytes('foo'), data1)
+                    self.assertEqual(dst_fs.getbytes('bar'), data2)
+                    self.assertEqual(dst_fs.getbytes('dir1/baz'), data3)
+                    self.assertEqual(dst_fs.getbytes('dir2/dir3/egg'), data4)
 
     def test_copy_dir_on_copy(self):
         src_fs = open_fs('mem://')
@@ -53,7 +71,6 @@ class TestCopy(unittest.TestCase):
 
         dst_fs = open_fs('mem://')
         fs.copy.copy_dir(src_fs, '/', dst_fs, '/', on_copy=on_copy)
-        print(on_copy_calls)
         self.assertEqual(
             on_copy_calls,
             [(src_fs, '/baz.txt', dst_fs, '/baz.txt')]
@@ -179,7 +196,6 @@ class TestCopy(unittest.TestCase):
             shutil.rmtree(src_dir)
             shutil.rmtree(dst_dir)
 
-
     def test_copy_fs_if_newer_dst_older(self):
         try:
             #create first dst ==> dst is older the src ==> file should be copied
@@ -338,7 +354,6 @@ class TestCopy(unittest.TestCase):
             shutil.rmtree(src_dir)
 
     def test_copy_dir_if_newer_multiple_files(self):
-
         try:
             src_dir = self._create_sandbox_dir()
             src_fs = open_fs('osfs://' + src_dir)
diff --git a/tests/test_mirror.py b/tests/test_mirror.py
index 662093f..8e71dd7 100644
--- a/tests/test_mirror.py
+++ b/tests/test_mirror.py
@@ -7,7 +7,7 @@ from fs import open_fs
 
 
 class TestMirror(unittest.TestCase):
-
+    WORKERS = 0  # Single threaded
 
     def _contents(self, fs):
         """Extract an FS in to a simple data structure."""
@@ -31,14 +31,14 @@ class TestMirror(unittest.TestCase):
     def test_empty_mirror(self):
         m1 = open_fs('mem://')
         m2 = open_fs('mem://')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_one_file(self):
         m1 = open_fs('mem://')
         m1.settext('foo', 'hello')
         m2 = open_fs('mem://')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_one_file_one_dir(self):
@@ -46,7 +46,7 @@ class TestMirror(unittest.TestCase):
         m1.settext('foo', 'hello')
         m1.makedir('bar')
         m2 = open_fs('mem://')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_delete_replace(self):
@@ -54,13 +54,13 @@ class TestMirror(unittest.TestCase):
         m1.settext('foo', 'hello')
         m1.makedir('bar')
         m2 = open_fs('mem://')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
         m2.remove('foo')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
         m2.removedir('bar')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_extra_dir(self):
@@ -69,7 +69,7 @@ class TestMirror(unittest.TestCase):
         m1.makedir('bar')
         m2 = open_fs('mem://')
         m2.makedir('baz')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_extra_file(self):
@@ -79,7 +79,7 @@ class TestMirror(unittest.TestCase):
         m2 = open_fs('mem://')
         m2.makedir('baz')
         m2.touch('egg')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_wrong_type(self):
@@ -89,7 +89,7 @@ class TestMirror(unittest.TestCase):
         m2 = open_fs('mem://')
         m2.makedir('foo')
         m2.touch('bar')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_update(self):
@@ -97,8 +97,20 @@ class TestMirror(unittest.TestCase):
         m1.settext('foo', 'hello')
         m1.makedir('bar')
         m2 = open_fs('mem://')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
         m2.appendtext('foo', ' world!')
-        mirror(m1, m2)
+        mirror(m1, m2, workers=self.WORKERS)
         self.assert_compare_fs(m1, m2)
+
+
+class TestMirrorWorkers1(TestMirror):
+    WORKERS = 1
+
+
+class TestMirrorWorkers2(TestMirror):
+    WORKERS = 2
+
+
+class TestMirrorWorkers4(TestMirror):
+    WORKERS = 4