fs: transfer: try reflink (#6853)

If supported, reflink, will provide the fastest transfer possible.
That being said, right now we use `move` in most of the places,
which still takes presidence.

Pre-requisite for #6387

dvc/fs/utils.py

@@ -1,7 +1,9 @@
+import errno
 import logging
 import os
 from typing import TYPE_CHECKING
 
+from .base import RemoteActionNotImplemented
 from .local import LocalFileSystem
 
 if TYPE_CHECKING:
@@ -19,11 +21,21 @@ def transfer(
     to_info: "DvcPath",
     move: bool = False,
 ) -> None:
-    use_move = isinstance(from_fs, type(to_fs)) and move
+    same_fs = isinstance(from_fs, type(to_fs))
+    use_move = same_fs and move
     try:
         if use_move:
             return to_fs.move(from_info, to_info)
 
+        if same_fs:
+            try:
+                return from_fs.reflink(from_info, to_info)
+            except RemoteActionNotImplemented:
+                pass
+            except OSError as exc:
+                if exc.errno not in [errno.EXDEV, errno.ENOTSUP]:
+                    raise
+
         if isinstance(from_fs, LocalFileSystem):
             if not isinstance(from_info, from_fs.PATH_CLS):
                 from_info = from_fs.PATH_CLS(from_info)

tests/func/objects/db/test_index.py

@@ -66,7 +66,7 @@ def test_clear_on_download_err(tmp_dir, dvc, index, mocker):
 
     assert list(index.hashes())
 
-    mocker.patch("dvc.fs.local.LocalFileSystem.upload", side_effect=Exception)
+    mocker.patch("dvc.fs.utils.transfer", side_effect=Exception)
     with pytest.raises(DownloadError):
         dvc.pull()
     assert not list(index.hashes())
@@ -83,7 +83,7 @@ def test_partial_upload(tmp_dir, dvc, index, mocker):
             raise Exception("stop baz")
         return original(self, from_file, to_info, name, **kwargs)
 
-    mocker.patch.object(LocalFileSystem, "upload", unreliable_upload)
+    mocker.patch("dvc.fs.utils.transfer", unreliable_upload)
     with pytest.raises(UploadError):
         dvc.push()
     assert not list(index.hashes())

tests/func/test_external_repo.py

@@ -2,7 +2,6 @@ import os
 from unittest.mock import ANY, patch
 
 from dvc.external_repo import CLONES, external_repo
-from dvc.fs.local import LocalFileSystem
 from dvc.objects.stage import stage
 from dvc.objects.transfer import transfer
 from dvc.path_info import PathInfo
@@ -48,12 +47,14 @@ def test_source_change(erepo_dir):
 
 
 def test_cache_reused(erepo_dir, mocker, local_cloud):
+    import dvc.fs.utils
+
     erepo_dir.add_remote(config=local_cloud.config)
     with erepo_dir.chdir():
         erepo_dir.dvc_gen("file", "text", commit="add file")
     erepo_dir.dvc.push()
 
-    download_spy = mocker.spy(LocalFileSystem, "upload")
+    download_spy = mocker.spy(dvc.fs.utils, "transfer")
 
     # Use URL to prevent any fishy optimizations
     url = f"file://{erepo_dir}"

tests/func/test_import.py

@@ -308,9 +308,9 @@ def test_download_error_pulling_imported_stage(tmp_dir, dvc, erepo_dir):
     remove("foo_imported")
     remove(dst_cache)
 
-    with patch(
-        "dvc.fs.local.LocalFileSystem.upload", side_effect=Exception
-    ), pytest.raises(DownloadError):
+    with patch("dvc.fs.utils.transfer", side_effect=Exception), pytest.raises(
+        DownloadError
+    ):
         dvc.pull(["foo_imported.dvc"])
 
 

tests/func/test_remote.py

@@ -8,7 +8,6 @@ import pytest
 
 from dvc.config import Config
 from dvc.exceptions import DownloadError, RemoteCacheRequiredError, UploadError
-from dvc.fs.local import LocalFileSystem
 from dvc.main import main
 from dvc.path_info import PathInfo
 from dvc.utils.fs import remove
@@ -172,22 +171,24 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
 
 
 def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
+    import dvc.fs.utils as fs_utils
+
     foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
     bar = tmp_dir.dvc_gen({"bar": "bar content"})[0].outs[0]
     baz = tmp_dir.dvc_gen({"baz": {"foo": "foo content"}})[0].outs[0]
 
     # Faulty upload version, failing on foo
-    original = LocalFileSystem.upload
+    original = fs_utils.transfer
     odb = dvc.cloud.get_remote_odb("upstream")
 
-    def unreliable_upload(self, fobj, to_info, **kwargs):
+    def unreliable_upload(from_fs, from_info, to_fs, to_info, **kwargs):
         if os.path.abspath(to_info) == os.path.abspath(
             odb.get(foo.hash_info).path_info
         ):
             raise Exception("stop foo")
-        return original(self, fobj, to_info, **kwargs)
+        return original(from_fs, from_info, to_fs, to_info, **kwargs)
 
-    with patch.object(LocalFileSystem, "upload", unreliable_upload):
+    with patch.object(fs_utils, "transfer", unreliable_upload):
         with pytest.raises(UploadError) as upload_error_info:
             dvc.push()
         assert upload_error_info.value.amount == 2
@@ -201,7 +202,7 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
     remove(dvc.odb.local.cache_dir)
 
     baz._collect_used_dir_cache()
-    with patch.object(LocalFileSystem, "upload", side_effect=Exception):
+    with patch.object(fs_utils, "transfer", side_effect=Exception):
         with pytest.raises(DownloadError) as download_error_info:
             dvc.pull()
         # error count should be len(.dir + standalone file checksums)
@@ -214,9 +215,8 @@ def test_raise_on_too_many_open_files(
 ):
     tmp_dir.dvc_gen({"file": "file content"})
 
-    mocker.patch.object(
-        LocalFileSystem,
-        "upload",
+    mocker.patch(
+        "dvc.fs.utils.transfer",
         side_effect=OSError(errno.EMFILE, "Too many open files"),
     )
 
@@ -263,19 +263,19 @@ def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
 
 
 def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
+    import dvc.fs.utils as fs_utils
+
     foo = tmp_dir.dvc_gen({"foo": {"bar": "bar content"}})[0].outs[0]
     tmp_dir.dvc_gen({"baz": "baz content"})
 
-    mocked_upload = mocker.patch.object(
-        LocalFileSystem, "upload", return_value=0
-    )
+    mocked_upload = mocker.spy(fs_utils, "transfer")
     dvc.push()
 
     # foo .dir file should be uploaded after bar
     odb = dvc.cloud.get_remote_odb("upstream")
     foo_path = odb.hash_to_path_info(foo.hash_info.value)
     bar_path = odb.hash_to_path_info(foo.obj.trie[("bar",)][1].value)
-    paths = [args[1] for args, _ in mocked_upload.call_args_list]
+    paths = [args[3] for args, _ in mocked_upload.call_args_list]
     assert paths.index(foo_path) > paths.index(bar_path)