Changes proposed by code review.

Author: atollk

fs/_bulk.py

@@ -11,13 +11,14 @@
 
 from six.moves.queue import Queue
 
-from .copy import copy_file_internal
+from .copy import copy_file_internal, copy_modified_time
 from .errors import BulkCopyFailed
+from .tools import copy_file_data
 
 if typing.TYPE_CHECKING:
     from .base import FS
     from types import TracebackType
-    from typing import List, Optional, Text, Type
+    from typing import List, Optional, Text, Type, IO, Tuple
 
 
 class _Worker(threading.Thread):
@@ -55,40 +56,32 @@ def __call__(self):
 class _CopyTask(_Task):
     """A callable that copies from one file another."""
 
-    def __init__(
-        self,
-        src_fs,  # type: FS
-        src_path,  # type: Text
-        dst_fs,  # type: FS
-        dst_path,  # type: Text
-        preserve_time,  # type: bool
-    ):
-        # type: (...) -> None
-        self.src_fs = src_fs
-        self.src_path = src_path
-        self.dst_fs = dst_fs
-        self.dst_path = dst_path
-        self.preserve_time = preserve_time
+    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
-        copy_file_internal(
-            self.src_fs,
-            self.src_path,
-            self.dst_fs,
-            self.dst_path,
-            preserve_time=self.preserve_time,
-        )
+        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
+    def __init__(self, num_workers=4, preserve_time=False):
+        # type: (int, bool) -> None
         if num_workers < 0:
             raise ValueError("num_workers must be >= 0")
         self.num_workers = num_workers
+        self.preserve_time = preserve_time
+        self.all_tasks = []  # type: List[Tuple[FS, Text, FS, Text]]
         self.queue = None  # type: Optional[Queue[_Task]]
         self.workers = []  # type: List[_Worker]
         self.errors = []  # type: List[Exception]
@@ -97,7 +90,7 @@ def __init__(self, num_workers=4):
     def start(self):
         """Start the workers."""
         if self.num_workers:
-            self.queue = Queue()
+            self.queue = Queue(maxsize=self.num_workers)
             self.workers = [_Worker(self) for _ in range(self.num_workers)]
             for worker in self.workers:
                 worker.start()
@@ -106,10 +99,18 @@ def start(self):
     def stop(self):
         """Stop the workers (will block until they are finished)."""
         if self.running and self.num_workers:
+            # Notify the workers that all tasks have arrived
+            # and wait for them to finish.
             for _worker in self.workers:
                 self.queue.put(None)
             for worker in self.workers:
                 worker.join()
+
+            # If the "last modified" time is to be preserved, do it now.
+            if self.preserve_time:
+                for args in self.all_tasks:
+                    copy_modified_time(*args)
+
             # Free up references held by workers
             del self.workers[:]
             self.queue.join()
@@ -139,8 +140,15 @@ def copy(self, src_fs, src_path, dst_fs, dst_path, preserve_time=False):
         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, preserve_time=preserve_time
+                src_fs, src_path, dst_fs, dst_path, preserve_time=self.preserve_time
             )
         else:
-            task = _CopyTask(src_fs, src_path, dst_fs, dst_path, preserve_time)
+            self.all_tasks.append((src_fs, src_path, dst_fs, dst_path))
+            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)

fs/base.py

@@ -22,7 +22,7 @@
 import six
 
 from . import copy, errors, fsencode, iotools, move, tools, walk, wildcard
-from .copy import copy_mtime
+from .copy import copy_modified_time
 from .glob import BoundGlobber
 from .mode import validate_open_mode
 from .path import abspath, join, normpath
@@ -426,7 +426,8 @@ def copy(
             with closing(self.open(src_path, "rb")) as read_file:
                 # FIXME(@althonos): typing complains because open return IO
                 self.upload(dst_path, read_file)  # type: ignore
-            copy_mtime(self, src_path, self, dst_path)
+            if preserve_time:
+                copy_modified_time(self, src_path, self, dst_path)
 
     def copydir(
         self,
@@ -1150,12 +1151,15 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
                 except OSError:
                     pass
                 else:
+                    if preserve_time:
+                        copy_modified_time(self, src_path, self, dst_path)
                     return
         with self._lock:
             with self.open(src_path, "rb") as read_file:
                 # FIXME(@althonos): typing complains because open return IO
                 self.upload(dst_path, read_file)  # type: ignore
-            copy_mtime(self, src_path, self, dst_path)
+            if preserve_time:
+                copy_modified_time(self, src_path, self, dst_path)
             self.remove(src_path)
 
     def open(

fs/copy.py

@@ -166,7 +166,7 @@ def copy_file(
                     else:
                         with _src_fs.openbin(src_path) as read_file:
                             _dst_fs.upload(dst_path, read_file)
-                    copy_mtime(_src_fs, src_path, _dst_fs, dst_path)
+                    copy_modified_time(_src_fs, src_path, _dst_fs, dst_path)
 
 
 def copy_file_internal(
@@ -207,7 +207,7 @@ def copy_file_internal(
             dst_fs.upload(dst_path, read_file)
 
     if preserve_time:
-        copy_mtime(src_fs, src_path, dst_fs, dst_path)
+        copy_modified_time(src_fs, src_path, dst_fs, dst_path)
 
 
 def copy_file_if_newer(
@@ -334,10 +334,11 @@ def dst():
     from ._bulk import Copier
 
     with src() as _src_fs, dst() as _dst_fs:
-        _thread_safe = is_thread_safe(_src_fs, _dst_fs)
-        copier = Copier(num_workers=workers if _thread_safe else 0)
-        with copier:
-            with _src_fs.lock(), _dst_fs.lock():
+        with _src_fs.lock(), _dst_fs.lock():
+            _thread_safe = is_thread_safe(_src_fs, _dst_fs)
+            with Copier(
+                num_workers=workers if _thread_safe else 0, preserve_time=preserve_time
+            ) 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))
@@ -351,7 +352,6 @@ def dst():
                             src_path,
                             _dst_fs,
                             dst_path,
-                            preserve_time=preserve_time,
                         )
                         on_copy(_src_fs, src_path, _dst_fs, dst_path)
             pass
@@ -408,7 +408,9 @@ def dst():
     with src() as _src_fs, dst() as _dst_fs:
         with _src_fs.lock(), _dst_fs.lock():
             _thread_safe = is_thread_safe(_src_fs, _dst_fs)
-            with Copier(num_workers=workers if _thread_safe else 0) as copier:
+            with Copier(
+                num_workers=workers if _thread_safe else 0, preserve_time=preserve_time
+            ) as copier:
                 _dst_fs.makedir(_dst_path, recreate=True)
                 namespace = ("details", "modified")
                 dst_state = {
@@ -445,12 +447,11 @@ def dst():
                                 dir_path,
                                 _dst_fs,
                                 copy_path,
-                                preserve_time=preserve_time,
                             )
                             on_copy(_src_fs, dir_path, _dst_fs, copy_path)
 
 
-def copy_mtime(
+def copy_modified_time(
     src_fs,  # type: Union[FS, Text]
     src_path,  # type: Text
     dst_fs,  # type: Union[FS, Text]

fs/memoryfs.py

@@ -15,6 +15,7 @@
 
 from . import errors
 from .base import FS
+from .copy import copy_modified_time
 from .enums import ResourceType, Seek
 from .info import Info
 from .mode import Mode
@@ -465,6 +466,9 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
             dst_dir_entry.set_entry(dst_name, src_entry)
             src_dir_entry.remove_entry(src_name)
 
+            if preserve_time:
+                copy_modified_time(self, src_path, self, dst_path)
+
     def movedir(self, src_path, dst_path, create=False, preserve_time=False):
         src_dir, src_name = split(self.validatepath(src_path))
         dst_dir, dst_name = split(self.validatepath(dst_path))
@@ -484,6 +488,9 @@ def movedir(self, src_path, dst_path, create=False, preserve_time=False):
             dst_dir_entry.set_entry(dst_name, src_entry)
             src_dir_entry.remove_entry(src_name)
 
+            if preserve_time:
+                copy_modified_time(self, src_path, self, dst_path)
+
     def openbin(self, path, mode="r", buffering=-1, **options):
         # type: (Text, Text, int, **Any) -> BinaryIO
         _mode = Mode(mode)

fs/mirror.py

@@ -87,7 +87,9 @@ def dst():
 
     with src() as _src_fs, dst() as _dst_fs:
         _thread_safe = is_thread_safe(_src_fs, _dst_fs)
-        with Copier(num_workers=workers if _thread_safe else 0) as copier:
+        with Copier(
+            num_workers=workers if _thread_safe else 0, preserve_time=preserve_time
+        ) as copier:
             with _src_fs.lock(), _dst_fs.lock():
                 _mirror(
                     _src_fs,

fs/osfs.py

@@ -49,7 +49,7 @@
 from .mode import Mode, validate_open_mode
 from .errors import FileExpected, NoURL
 from ._url_tools import url_quote
-from .copy import copy_mtime
+from .copy import copy_modified_time
 
 if typing.TYPE_CHECKING:
     from typing import (
@@ -454,7 +454,7 @@ def copy(self, src_path, dst_path, overwrite=False, preserve_time=False):
                                 sent = sendfile(fd_dst, fd_src, offset, maxsize)
                                 offset += sent
                     if preserve_time:
-                        copy_mtime(self, src_path, self, dst_path)
+                        copy_modified_time(self, src_path, self, dst_path)
                 except OSError as e:
                     # the error is not a simple "sendfile not supported" error
                     if e.errno not in self._sendfile_error_codes:

fs/wrapfs.py

@@ -186,7 +186,7 @@ def movedir(self, src_path, dst_path, create=False, preserve_time=False):
                 raise errors.ResourceNotFound(dst_path)
             if not src_fs.getinfo(_src_path).is_dir:
                 raise errors.DirectoryExpected(src_path)
-            move_dir(src_fs, _src_path, dst_fs, _dst_path)
+            move_dir(src_fs, _src_path, dst_fs, _dst_path, preserve_time=preserve_time)
 
     def openbin(self, path, mode="r", buffering=-1, **options):
         # type: (Text, Text, int, **Any) -> BinaryIO

tests/test_move.py

@@ -21,6 +21,9 @@ def test_move_fs(self):
         src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
 
         dst_fs = open_fs("mem://")
+        dst_fs.create("test.txt")
+        dst_fs.setinfo("test.txt", {"details": {"modified": 1000000}})
+
         fs.move.move_fs(src_fs, dst_fs, preserve_time=self.preserve_time)
 
         self.assertTrue(dst_fs.isdir("foo/bar"))
@@ -44,6 +47,9 @@ def test_move_dir(self):
         src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
 
         dst_fs = open_fs("mem://")
+        dst_fs.create("test.txt")
+        dst_fs.setinfo("test.txt", {"details": {"modified": 1000000}})
+
         fs.move.move_dir(src_fs, "/foo", dst_fs, "/", preserve_time=self.preserve_time)
 
         self.assertTrue(dst_fs.isdir("bar"))