Repository: PyFilesystem/pyfilesystem2
Pull request #189 (2018-07-17 02:38:19)

Use `os.open()` in osfs in exclusive mode (#182)
`io.open()` takes a file descriptor, `os.open()` returns one and lets you pass O_EXCL to open the file.

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

Coverage remained the same at 99.976% when pulling **42f1c4d2f52bd16c42f767a83570e21217da9d17 on sqwishy:open-with-O_EXCL-182** into **e63f5af191fb477d42720ec19feec81147a2b889 on PyFilesystem: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/README.md b/README.md
index 2254bbf..4d542e7 100644
--- a/README.md
+++ b/README.md
@@ -82,6 +82,7 @@ def count_py_loc(path):
             if name.endswith('.py'):
                 with open(os.path.join(root, name), 'rt') as python_file:
                     count += sum(1 for line in python_file if line.strip())
+    return count
 ```
 
 This version is similar to the PyFilesystem code above, but would only
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/_fscompat.py b/fs/_fscompat.py
index 1cd6529..e77b878 100644
--- a/fs/_fscompat.py
+++ b/fs/_fscompat.py
@@ -5,7 +5,7 @@ import six
 try:
     from os import fsencode, fsdecode
 except ImportError:
-    from backports.os import fsencode, fsdecode
+    from backports.os import fsencode, fsdecode  # type: ignore
 
 try:
     from os import fspath
diff --git a/fs/_typing.py b/fs/_typing.py
index 12953e1..cf869ea 100644
--- a/fs/_typing.py
+++ b/fs/_typing.py
@@ -8,14 +8,13 @@ import six
 
 _PY = sys.version_info
 
+from typing import overload  # type: ignore
 
 if _PY.major == 3 and _PY.minor == 5 and _PY.micro in (0, 1):
     def overload(func):  # pragma: no cover
         return func
-else:
-    from typing import overload
 
 try:
     from typing import Text
 except ImportError:  # pragma: no cover
-    Text = six.text_type
+    Text = six.text_type  # type: ignore
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/compress.py b/fs/compress.py
index 92ae0ea..04b2d7b 100644
--- a/fs/compress.py
+++ b/fs/compress.py
@@ -90,7 +90,8 @@ def write_zip(src_fs,                            # type: FS
             zip_info = zipfile.ZipInfo(zip_name, zip_time)  # type: ignore
 
             try:
-                zip_info.external_attr = info.permissions.mode << 16
+                if info.permissions is not None:
+                    zip_info.external_attr = info.permissions.mode << 16
             except MissingInfoNamespace:
                 pass
 
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/ftpfs.py b/fs/ftpfs.py
index 4303a46..fbe6b60 100644
--- a/fs/ftpfs.py
+++ b/fs/ftpfs.py
@@ -16,6 +16,7 @@ from contextlib import contextmanager
 from ftplib import FTP
 from ftplib import error_perm
 from ftplib import error_temp
+from typing import cast
 
 from six import PY2
 from six import text_type
@@ -82,7 +83,7 @@ def ftp_errors(fs, path=None):
                 msg=message
             )
         elif code in ('501', '550'):
-            raise errors.ResourceNotFound(path=path)
+            raise errors.ResourceNotFound(path=cast(str, path))
         raise errors.PermissionDenied(
             msg=message
         )
diff --git a/fs/info.py b/fs/info.py
index 75057c1..668c376 100644
--- a/fs/info.py
+++ b/fs/info.py
@@ -6,6 +6,7 @@ from __future__ import print_function
 from __future__ import unicode_literals
 
 import typing
+from typing import cast
 from copy import deepcopy
 
 import six
@@ -15,12 +16,11 @@ from .enums import ResourceType
 from .errors import MissingInfoNamespace
 from .permissions import Permissions
 from .time import epoch_to_datetime
-from ._typing import overload
+from ._typing import overload, Text
 
 if False:  # typing.TYPE_CHECKING
     from datetime import datetime
-    from typing import Callable, List, Mapping, Optional
-    from ._typing import Text
+    from typing import Any, Callable, List, Mapping, Optional, Union
     RawInfo = Mapping[Text, Mapping[Text, object]]
     ToDatetime = Callable[[int], datetime]
     T = typing.TypeVar("T")
@@ -82,23 +82,18 @@ class Info(object):
         else:
             return None
 
-    @overload
-    def get(self, namespace, key, default=None):  # pragma: no cover
-        # type: (Text, Text, Optional[T]) -> Optional[T]
-        pass
-
     @overload
     def get(self, namespace, key):  # pragma: no cover
-        # type: (Text, Text) -> Optional[T]
+        # type: (Text, Text) -> Any
         pass
 
     @overload
     def get(self, namespace, key, default):  # pragma: no cover
-        # type: (Text, Text, T) -> T
+        # type: (Text, Text, T) -> Union[Any, T]
         pass
 
     def get(self, namespace, key, default=None):
-        # type: (Text, Text, Optional[T]) -> Optional[T]
+        # type: (Text, Text, Optional[Any]) -> Optional[Any]
         """Get a raw info value.
 
         Arguments:
@@ -187,7 +182,7 @@ class Info(object):
         # type: () -> Text
         """`str`: the resource name.
         """
-        return self.get('basic', 'name')
+        return cast(Text, self.get('basic', 'name'))
 
     @property
     def suffix(self):
@@ -245,14 +240,14 @@ class Info(object):
         # type: () -> bool
         """`bool`: `True` if the resource references a directory.
         """
-        return self.get('basic', 'is_dir')
+        return cast(bool, self.get('basic', 'is_dir'))
 
     @property
     def is_file(self):
         # type: () -> bool
         """`bool`: `True` if the resource references a file.
         """
-        return not self.get('basic', 'is_dir')
+        return not cast(bool, self.get('basic', 'is_dir'))
 
     @property
     def is_link(self):
@@ -381,7 +376,7 @@ class Info(object):
 
         """
         self._require_namespace('details')
-        return self.get('details', 'size')
+        return cast(int, self.get('details', 'size'))
 
     @property
     def user(self):
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..5da6b82 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):
-    # type: (Union[Text, FS], Union[Text, FS]) -> None
+def move_fs(src_fs, dst_fs, workers=0):
+    # type: (Union[Text, FS], Union[Text, FS], int) -> 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)
diff --git a/fs/osfs.py b/fs/osfs.py
index e53e5e2..9b2c330 100644
--- a/fs/osfs.py
+++ b/fs/osfs.py
@@ -422,8 +422,8 @@ class OSFS(FS):
         _path = self.validatepath(path)
         sys_path = self._to_sys_path(_path)
         with convert_os_errors('open', path):
-            if six.PY2 and _mode.exclusive and self.exists(path):
-                raise FileExists(path)
+            if six.PY2 and _mode.exclusive:
+                sys_path = os.open(sys_path, os.O_CREAT | os.O_EXCL)
             _encoding = encoding or 'utf-8'
             return io.open(
                 sys_path,
diff --git a/fs/tarfs.py b/fs/tarfs.py
index fc7ae65..325bd0c 100644
--- a/fs/tarfs.py
+++ b/fs/tarfs.py
@@ -9,6 +9,7 @@ import os
 import tarfile
 import typing
 from collections import OrderedDict
+from typing import cast, IO
 
 import six
 
@@ -23,6 +24,7 @@ from .path import dirname, relpath, basename, isbase, parts, frombase
 from .wrapfs import WrapFS
 from .permissions import Permissions
 
+
 if False:  # typing.TYPE_CHECKING
     from tarfile import TarInfo
     from typing import (
@@ -401,7 +403,7 @@ class ReadTarFS(FS):
         if not member.isfile():
             raise errors.FileExpected(path)
 
-        rw = RawWrapper(self._tar.extractfile(member))
+        rw = RawWrapper(cast(IO, self._tar.extractfile(member)))
 
         if six.PY2: # Patch nonexistent file.flush in Python2
             def _flush():


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