dvc: bump to 2.7.3

Bumps [adlfs](https://github.com/dask/adlfs) from 2021.8.2 to 2021.9.1.
- [Release notes](https://github.com/dask/adlfs/releases)
- [Changelog](https://github.com/dask/adlfs/blob/master/CHANGELOG.md)
- [Commits](https://github.com/dask/adlfs/compare/2021.08.2...2021.09.1)

---
updated-dependencies:
- dependency-name: adlfs
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

MANIFEST.in

@@ -1,3 +1,3 @@
 include fastentrypoints.py
 include LICENSE
-include test_requirements.txt
+recursive-include requirements *.txt

dvc/command/experiments.py

@@ -1151,9 +1151,9 @@ def add_parser(subparsers, parent_parser):
         nargs="?",
         default=None,
         help=(
-            "Optional Git remote name or Git URL. If provided, experiments "
-            "from the specified Git repository will be listed instead of "
-            "local experiments."
+            "Optional Git remote name or Git URL. "
+            "If provided, experiments from the specified Git repository "
+            " will be listed instead of local ones."
         ),
         metavar="[<git_remote>]",
     )

dvc/fs/base.py

@@ -217,44 +217,50 @@ class BaseFileSystem:
             return False
         return hash_.endswith(cls.CHECKSUM_DIR_SUFFIX)
 
-    def upload(self, from_info, to_info, name=None, no_progress_bar=False):
-        if not hasattr(self, "_upload"):
-            raise RemoteActionNotImplemented("upload", self.scheme)
+    def upload(
+        self,
+        from_info,
+        to_info,
+        total=None,
+        desc=None,
+        no_progress_bar=False,
+        **pbar_args,
+    ):
+        is_file_obj = hasattr(from_info, "read")
+        method = "upload_fobj" if is_file_obj else "_upload"
+        if not hasattr(self, method):
+            raise RemoteActionNotImplemented(method, self.scheme)
 
         if to_info.scheme != self.scheme:
             raise NotImplementedError
 
+        logger.debug("Uploading '%s' to '%s'", from_info, to_info)
+        if is_file_obj:
+            with Tqdm.wrapattr(
+                from_info,
+                "read",
+                disable=no_progress_bar,
+                bytes=True,
+                total=total,
+                desc=desc,
+                **pbar_args,
+            ) as wrapped:
+                # `size` is used to provide hints to the WebdavFileSystem
+                # for legacy servers.
+                # pylint: disable=no-member
+                return self.upload_fobj(wrapped, to_info, size=total)
+
         if from_info.scheme != "local":
             raise NotImplementedError
 
-        logger.debug("Uploading '%s' to '%s'", from_info, to_info)
-
-        name = name or from_info.name
-
-        self._upload(  # noqa, pylint: disable=no-member
+        name = desc or from_info.name
+        return self._upload(  # noqa, pylint: disable=no-member
             from_info.fspath,
             to_info,
             name=name,
             no_progress_bar=no_progress_bar,
-        )
-
-    def upload_fobj(
-        self, fobj, to_info, no_progress_bar=False, size=None, **pbar_args
-    ):
-        if not hasattr(self, "_upload_fobj"):
-            raise RemoteActionNotImplemented("upload_fobj", self.scheme)
-
-        with Tqdm.wrapattr(
-            fobj,
-            "read",
-            disable=no_progress_bar,
-            bytes=True,
-            total=size,
             **pbar_args,
-        ) as wrapped:
-            self._upload_fobj(  # pylint: disable=no-member
-                wrapped, to_info, size=size
-            )
+        )
 
     def download(
         self,

dvc/fs/fsspec_wrapper.py

@@ -130,7 +130,7 @@ class FSSpecWrapper(BaseFileSystem):
             self._with_bucket(path_info), exist_ok=kwargs.pop("exist_ok", True)
         )
 
-    def _upload_fobj(self, fobj, to_info, size=None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         self.makedirs(to_info.parent)
         with self.open(to_info, "wb") as fdest:
             shutil.copyfileobj(fobj, fdest, length=fdest.blocksize)
@@ -141,13 +141,15 @@ class FSSpecWrapper(BaseFileSystem):
         self.makedirs(to_info.parent)
         size = os.path.getsize(from_file)
         with open(from_file, "rb") as fobj:
-            self.upload_fobj(
+            with Tqdm.wrapattr(
                 fobj,
-                to_info,
-                size=size,
+                "read",
+                disable=no_progress_bar,
+                bytes=True,
+                total=size,
                 desc=name,
-                no_progress_bar=no_progress_bar,
-            )
+            ) as wrapped:
+                self.upload_fobj(wrapped, to_info, size=size)
         self.fs.invalidate_cache(self._with_bucket(to_info.parent))
 
     def _download(

dvc/fs/gdrive.py

@@ -298,7 +298,7 @@ class GDriveFileSystem(
 
         return super()._with_bucket(path)
 
-    def _upload_fobj(self, fobj, to_info, size: int = None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         rpath = self._with_bucket(to_info)
         self.makedirs(os.path.dirname(rpath))
-        return self.fs.upload_fobj(fobj, rpath, size=size)
+        return self.fs.upload_fobj(fobj, rpath, **kwargs)

dvc/fs/hdfs.py

@@ -244,7 +244,7 @@ class HDFSFileSystem(BaseFileSystem):
             size=self.getsize(path_info),
         )
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with self.hdfs(to_info) as hdfs:
             with hdfs.open_output_stream(to_info.path) as fdest:
                 shutil.copyfileobj(fobj, fdest, self.BLOCK_SIZE)

dvc/fs/local.py

@@ -108,7 +108,7 @@ class LocalFileSystem(BaseFileSystem):
             self.remove(tmp_info)
             raise
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         self.makedirs(to_info.parent)
         tmp_info = to_info.parent / tmp_fname("")
         try:

dvc/fs/ssh.py

@@ -67,6 +67,7 @@ class SSHFileSystem(CallbackMixin, FSSpecWrapper):
         )
 
         login_info["password"] = config.get("password")
+        login_info["passphrase"] = config.get("password")
 
         raw_keys = []
         if config.get("keyfile"):
@@ -119,9 +120,9 @@ class SSHFileSystem(CallbackMixin, FSSpecWrapper):
     # Ensure that if an interrupt happens during the transfer, we don't
     # pollute the cache.
 
-    def _upload_fobj(self, fobj, to_info, *args, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with as_atomic(self, to_info) as tmp_file:
-            super()._upload_fobj(fobj, tmp_file, *args, **kwargs)
+            super().upload_fobj(fobj, tmp_file, **kwargs)
 
     def _upload(self, from_file, to_info, *args, **kwargs):
         with as_atomic(self, to_info) as tmp_file:

dvc/fs/utils.py

@@ -33,7 +33,8 @@ def transfer(
             return from_fs.download_file(from_info, to_info)
 
         with from_fs.open(from_info, mode="rb") as fobj:
-            return to_fs.upload_fobj(fobj, to_info)
+            size = from_fs.getsize(from_info)
+            return to_fs.upload(fobj, to_info, total=size)
     except OSError as exc:
         # If the target file already exists, we are going to simply
         # ignore the exception (#4992).

dvc/fs/webdav.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import threading
 from functools import lru_cache
 
@@ -71,15 +70,13 @@ class WebDAVFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
 
         return WebdavFileSystem(**self.fs_args)
 
-    def _upload_fobj(self, fobj, to_info, size: int = None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         rpath = self.translate_path_info(to_info)
-        self.makedirs(os.path.dirname(rpath))
+        size = kwargs.get("size")
         # using upload_fileobj to directly upload fileobj rather than buffering
         # and using overwrite=True to avoid check for an extra exists call,
         # as caller should ensure that the file does not exist beforehand.
-        return self.fs.client.upload_fileobj(
-            fobj, rpath, overwrite=True, size=size
-        )
+        return self.fs.upload_fileobj(fobj, rpath, overwrite=True, size=size)
 
     @lru_cache(512)
     def translate_path_info(self, path):

dvc/fs/webhdfs.py

@@ -141,7 +141,7 @@ class WebHDFSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         self.hdfs_client.makedirs(to_info.parent.path)
         self.hdfs_client.rename(from_info.path, to_info.path)
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with self.hdfs_client.write(to_info.path) as fdest:
             shutil.copyfileobj(fobj, fdest)
 

dvc/objects/db/reference.py

@@ -70,7 +70,7 @@ class ReferenceObjectDB(ObjectDB):
         ref_fobj = io.BytesIO(ref_file.to_bytes())
         ref_fobj.seek(0)
         try:
-            self.fs.upload_fobj(ref_fobj, to_info)
+            self.fs.upload(ref_fobj, to_info)
         except OSError as exc:
             if isinstance(exc, FileExistsError) or (
                 os.name == "nt"

dvc/objects/stage.py

@@ -34,9 +34,8 @@ def _upload_file(path_info, fs, odb, upload_odb):
     tmp_info = upload_odb.path_info / tmp_fname()
     with fs.open(path_info, mode="rb", chunk_size=fs.CHUNK_SIZE) as stream:
         stream = HashedStreamReader(stream)
-        upload_odb.fs.upload_fobj(
-            stream, tmp_info, desc=path_info.name, size=fs.getsize(path_info)
-        )
+        size = fs.getsize(path_info)
+        upload_odb.fs.upload(stream, tmp_info, desc=path_info.name, total=size)
 
     odb.add(tmp_info, upload_odb.fs, stream.hash_info)
     return path_info, odb.get(stream.hash_info)

dvc/objects/transfer.py

@@ -1,20 +1,19 @@
 import errno
-import itertools
 import logging
-from concurrent import futures
-from concurrent.futures import ThreadPoolExecutor
 from functools import partial, wraps
-from typing import TYPE_CHECKING, Callable, Iterable, Optional
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional
 
 from funcy import split
 
 from dvc.progress import Tqdm
+from dvc.utils.threadpool import ThreadPoolExecutor
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
 
     from .db.base import ObjectDB
     from .db.index import ObjectDBIndexBase
+    from .tree import Tree
 
 logger = logging.getLogger(__name__)
 
@@ -38,109 +37,48 @@ def _log_exceptions(func):
     return wrapper
 
 
-def _transfer(
-    src, dest, dir_ids, file_ids, missing_ids, jobs, verify, move, **kwargs
-):
-    func = _log_exceptions(dest.add)
-    total = len(dir_ids) + len(file_ids)
-    if total == 0:
-        return 0
-    with Tqdm(total=total, unit="file", desc="Transferring") as pbar:
-        func = pbar.wrap_fn(func)
-        with ThreadPoolExecutor(max_workers=jobs) as executor:
-            processor = partial(
-                _create_tasks,
-                executor,
-                jobs,
-                func,
-                src,
-                verify,
-                move,
-            )
-            processor.add_func = func
-            _do_transfer(
-                src,
-                dest,
-                dir_ids,
-                file_ids,
-                missing_ids,
-                processor,
-                verify=verify,
-                move=move,
-                **kwargs,
-            )
-    return total
+def find_tree_by_obj_id(
+    odbs: Iterable[Optional["ObjectDB"]], obj_id: "HashInfo"
+) -> Optional["Tree"]:
+    from .errors import ObjectFormatError
+    from .tree import Tree
 
-
-def _create_tasks(executor, jobs, func, src, verify, move, obj_ids):
-    fails = 0
-    hash_iter = iter(obj_ids)
-
-    def submit(hash_info):
-        obj = src.get(hash_info)
-        return executor.submit(
-            func,
-            obj.path_info,
-            obj.fs,
-            obj.hash_info,
-            verify=verify,
-            move=move,
-        )
-
-    def create_taskset(amount):
-        return {
-            submit(hash_info)
-            for hash_info in itertools.islice(hash_iter, amount)
-        }
-
-    tasks = create_taskset(jobs * 5)
-    while tasks:
-        done, tasks = futures.wait(tasks, return_when=futures.FIRST_COMPLETED)
-        fails += sum(task.result() for task in done)
-        tasks.update(create_taskset(len(done)))
-    return fails
+    for odb in odbs:
+        if odb is not None:
+            try:
+                return Tree.load(odb, obj_id)
+            except (FileNotFoundError, ObjectFormatError):
+                pass
+    return None
 
 
 def _do_transfer(
     src: "ObjectDB",
     dest: "ObjectDB",
-    dir_ids: Iterable["HashInfo"],
-    file_ids: Iterable["HashInfo"],
+    obj_ids: Iterable["HashInfo"],
     missing_ids: Iterable["HashInfo"],
     processor: Callable,
     src_index: Optional["ObjectDBIndexBase"] = None,
     dest_index: Optional["ObjectDBIndexBase"] = None,
     cache_odb: Optional["ObjectDB"] = None,
-    **kwargs,
+    **kwargs: Any,
 ):
     from dvc.exceptions import FileTransferError
-    from dvc.objects.errors import ObjectFormatError
 
+    dir_ids, file_ids = split(lambda hash_info: hash_info.isdir, obj_ids)
     total_fails = 0
     succeeded_dir_objs = []
     all_file_ids = set(file_ids)
 
     for dir_hash in dir_ids:
-        from .tree import Tree
-
-        bound_file_ids = set()
-        dir_obj: Optional["Tree"] = None
-        for odb in (cache_odb, src):
-            if odb is not None:
-                try:
-                    dir_obj = Tree.load(odb, dir_hash)
-                    break
-                except (FileNotFoundError, ObjectFormatError):
-                    pass
+        dir_obj = find_tree_by_obj_id([cache_odb, src], dir_hash)
         assert dir_obj
+
         entry_ids = {entry.hash_info for _, entry in dir_obj}
+        bound_file_ids = all_file_ids & entry_ids
+        all_file_ids -= entry_ids
 
-        for file_hash in all_file_ids.copy():
-            if file_hash in entry_ids:
-                bound_file_ids.add(file_hash)
-                all_file_ids.remove(file_hash)
-
-        dir_fails = processor(bound_file_ids)
+        dir_fails = sum(processor(bound_file_ids))
         if dir_fails:
             logger.debug(
                 "failed to upload full contents of '%s', "
@@ -164,19 +102,13 @@ def _do_transfer(
                 dir_obj.name,
             )
         else:
-            raw_obj = src.get(dir_obj.hash_info)
-            is_dir_failed = processor.add_func(  # type: ignore[attr-defined]
-                raw_obj.path_info,
-                raw_obj.fs,
-                raw_obj.hash_info,
-                **kwargs,
-            )
+            is_dir_failed = sum(processor([dir_obj.hash_info]))
             total_fails += is_dir_failed
             if not is_dir_failed:
                 succeeded_dir_objs.append(dir_obj)
 
     # insert the rest
-    total_fails += processor(all_file_ids)
+    total_fails += sum(processor(all_file_ids))
     if total_fails:
         if src_index:
             src_index.clear()
@@ -222,18 +154,19 @@ def transfer(
     if not status.new:
         return 0
 
-    dir_ids, file_ids = split(lambda hash_info: hash_info.isdir, status.new)
-    if jobs is None:
-        jobs = dest.fs.jobs
+    def func(hash_info: "HashInfo") -> None:
+        obj = src.get(hash_info)
+        return dest.add(
+            obj.path_info, obj.fs, obj.hash_info, verify=verify, move=move
+        )
 
-    return _transfer(
-        src,
-        dest,
-        set(dir_ids),
-        set(file_ids),
-        status.missing,
-        jobs,
-        verify,
-        move,
-        **kwargs,
-    )
+    total = len(status.new)
+    jobs = jobs or dest.fs.jobs
+    with Tqdm(total=total, unit="file", desc="Transferring") as pbar:
+        with ThreadPoolExecutor(max_workers=jobs) as executor:
+            wrapped_func = pbar.wrap_fn(_log_exceptions(func))
+            processor = partial(executor.imap_unordered, wrapped_func)
+            _do_transfer(
+                src, dest, status.new, status.missing, processor, **kwargs
+            )
+    return total

dvc/repo/diff.py

@@ -4,7 +4,6 @@ from collections import defaultdict
 from typing import Dict, List
 
 from dvc.exceptions import PathMissingError
-from dvc.objects.stage import get_file_hash
 from dvc.repo import locked
 from dvc.repo.experiments.utils import fix_exp_head
 
@@ -45,7 +44,7 @@ def diff(self, a_rev="HEAD", b_rev=None, targets=None):
                 repo_fs, targets
             )
 
-        results[rev] = _paths_checksums(self, repo_fs, targets_path_infos)
+        results[rev] = _paths_checksums(self, targets_path_infos)
 
     if targets is not None:
         # check for overlapping missing targets between a_rev and b_rev
@@ -98,7 +97,7 @@ def diff(self, a_rev="HEAD", b_rev=None, targets=None):
     return ret if any(ret.values()) else {}
 
 
-def _paths_checksums(repo, repo_fs, targets):
+def _paths_checksums(repo, targets):
     """
     A dictionary of checksums addressed by relpaths collected from
     the current fs outputs.
@@ -110,10 +109,10 @@ def _paths_checksums(repo, repo_fs, targets):
         file:      "data"
     """
 
-    return dict(_output_paths(repo, repo_fs, targets))
+    return dict(_output_paths(repo, targets))
 
 
-def _output_paths(repo, repo_fs, targets):
+def _output_paths(repo, targets):
     from dvc.fs.local import LocalFileSystem
     from dvc.objects.stage import stage as ostage
 
@@ -131,45 +130,47 @@ def _output_paths(repo, repo_fs, targets):
             else os.path.join(str(output), "")
         )
 
-    def _to_checksum(output):
-        if on_working_fs:
-            return ostage(
-                repo.odb.local,
-                output.path_info,
-                repo.odb.local.fs,
-                "md5",
-                dry_run=True,
-            )[1].hash_info.value
-        return output.hash_info.value
-
     for output in repo.index.outs:
         if _exists(output):
             yield_output = targets is None or any(
                 output.path_info.isin_or_eq(target) for target in targets
             )
 
+            if on_working_fs:
+                _, obj = ostage(
+                    repo.odb.local,
+                    output.path_info,
+                    repo.odb.local.fs,
+                    "md5",
+                    dry_run=True,
+                    dvcignore=output.dvcignore,
+                )
+                hash_info = obj.hash_info
+            else:
+                hash_info = output.hash_info
+                obj = output.get_obj()
+
             if yield_output:
-                yield _to_path(output), _to_checksum(output)
+                yield _to_path(output), hash_info.value
+
+            if not obj:
+                continue
 
             if output.is_dir_checksum and (
                 yield_output
                 or any(target.isin(output.path_info) for target in targets)
             ):
-                yield from _dir_output_paths(repo_fs, output, targets)
+                yield from _dir_output_paths(output.path_info, obj, targets)
 
 
-def _dir_output_paths(repo_fs, output, targets=None):
-    from dvc.config import NoRemoteError
-
-    try:
-        for fname in repo_fs.walk_files(output.path_info):
-            if targets is None or any(
-                fname.isin_or_eq(target) for target in targets
-            ):
-                # pylint: disable=no-member
-                yield str(fname), get_file_hash(fname, repo_fs, "md5").value
-    except NoRemoteError:
-        logger.warning("dir cache entry for '%s' is missing", output)
+def _dir_output_paths(path_info, obj, targets=None):
+    for key, entry_obj in obj:
+        fname = path_info.joinpath(*key)
+        if targets is None or any(
+            fname.isin_or_eq(target) for target in targets
+        ):
+            # pylint: disable=no-member
+            yield str(fname), entry_obj.hash_info.value
 
 
 def _filter_missing(repo_fs, paths):

dvc/scm/base.py

@@ -47,6 +47,15 @@ class InvalidRemoteSCMRepo(SCMError):
         super().__init__(msg)
 
 
+class GitAuthError(SCMError):
+    def __init__(self, url: str):
+        super().__init__(
+            f"HTTP Git authentication is not supported: '{url}'"
+            "\nSee https://dvc.org/doc//user-guide/"
+            "troubleshooting#git-auth"
+        )
+
+
 class Base:
     """Base class for source control management driver implementations."""
 

dvc/scm/git/backend/dulwich.py

@@ -20,7 +20,7 @@ from funcy import cached_property
 
 from dvc.path_info import PathInfo
 from dvc.progress import Tqdm
-from dvc.scm.base import InvalidRemoteSCMRepo, SCMError
+from dvc.scm.base import GitAuthError, InvalidRemoteSCMRepo, SCMError
 from dvc.utils import relpath
 
 from ..objects import GitObject
@@ -355,7 +355,7 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 yield os.fsdecode(key)
 
     def iter_remote_refs(self, url: str, base: Optional[str] = None):
-        from dulwich.client import get_transport_and_path
+        from dulwich.client import HTTPUnauthorized, get_transport_and_path
         from dulwich.errors import NotGitRepository
         from dulwich.porcelain import get_remote_repo
 
@@ -376,6 +376,8 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 yield from (os.fsdecode(ref) for ref in client.get_refs(path))
         except NotGitRepository as exc:
             raise InvalidRemoteSCMRepo(url) from exc
+        except HTTPUnauthorized:
+            raise GitAuthError(url)
 
     def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
         raise NotImplementedError
@@ -388,7 +390,7 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
         force: bool = False,
         on_diverged: Optional[Callable[[str, str], bool]] = None,
     ):
-        from dulwich.client import get_transport_and_path
+        from dulwich.client import HTTPUnauthorized, get_transport_and_path
         from dulwich.errors import NotGitRepository, SendPackError
         from dulwich.porcelain import (
             DivergedBranches,
@@ -447,6 +449,8 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 )
         except (NotGitRepository, SendPackError) as exc:
             raise SCMError("Git failed to push '{src}' to '{url}'") from exc
+        except HTTPUnauthorized:
+            raise GitAuthError(url)
 
     def _push_dest_refs(
         self, src: Optional[str], dest: str

dvc/state.py

@@ -53,7 +53,6 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         self.tmp_dir = tmp_dir
         self.root_dir = root_dir
         self.dvcignore = dvcignore
-        self.fs = LocalFileSystem()
 
         if not tmp_dir:
             return
@@ -85,7 +84,7 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         assert isinstance(hash_info, HashInfo)
         assert os.path.exists(path_info)
 
-        mtime, size = get_mtime_and_size(path_info, self.fs, self.dvcignore)
+        mtime, size = get_mtime_and_size(path_info, fs, self.dvcignore)
         inode = get_inode(path_info)
 
         logger.debug(
@@ -121,7 +120,7 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         if not os.path.exists(path):
             return None
 
-        mtime, size = get_mtime_and_size(path, self.fs, self.dvcignore)
+        mtime, size = get_mtime_and_size(path, fs, self.dvcignore)
         inode = get_inode(path)
 
         value = self.md5s.get(inode)
@@ -143,10 +142,10 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
 
         assert isinstance(path_info, str) or path_info.scheme == "local"
 
-        if not self.fs.exists(path_info):
+        if not fs.exists(path_info):
             return
 
-        mtime, _ = get_mtime_and_size(path_info, self.fs, self.dvcignore)
+        mtime, _ = get_mtime_and_size(path_info, fs, self.dvcignore)
         inode = get_inode(path_info)
         relative_path = relpath(path_info, self.root_dir)
 
@@ -168,11 +167,11 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
             for relative_path in ref:
                 path = os.path.join(self.root_dir, relative_path)
 
-                if path in used or not self.fs.exists(path):
+                if path in used or not fs.exists(path):
                     continue
 
                 inode = get_inode(path)
-                mtime, _ = get_mtime_and_size(path, self.fs, self.dvcignore)
+                mtime, _ = get_mtime_and_size(path, fs, self.dvcignore)
 
                 if ref[relative_path] == (inode, mtime):
                     logger.debug("Removing '%s' as unused link.", path)

dvc/utils/threadpool.py

@@ -0,0 +1,34 @@
+from concurrent import futures
+from itertools import islice
+from typing import Any, Callable, Iterable, Iterator, Set, TypeVar
+
+_T = TypeVar("_T")
+
+
+class ThreadPoolExecutor(futures.ThreadPoolExecutor):
+    _max_workers: int
+
+    @property
+    def max_workers(self) -> int:
+        return self._max_workers
+
+    def imap_unordered(
+        self, fn: Callable[..., _T], *iterables: Iterable[Any]
+    ) -> Iterator[_T]:
+        """Lazier version of map that does not preserve ordering of results.
+
+        It does not create all the futures at once to reduce memory usage.
+        """
+
+        def create_taskset(n: int) -> Set[futures.Future]:
+            return {self.submit(fn, *args) for args in islice(it, n)}
+
+        it = zip(*iterables)
+        tasks = create_taskset(self.max_workers * 5)
+        while tasks:
+            done, tasks = futures.wait(
+                tasks, return_when=futures.FIRST_COMPLETED
+            )
+            for fut in done:
+                yield fut.result()
+            tasks.update(create_taskset(len(done)))

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "2.7.2"
+_BASE_VERSION = "2.7.3"
 
 
 def _generate_version(base_version):

requirements/azure.txt

@@ -0,0 +1,3 @@
+adlfs==2021.9.1
+azure-identity>=1.4.0
+knack

requirements/default.txt

@@ -0,0 +1,45 @@
+# See https://github.com/pyinstaller/pyinstaller/issues/1945
+ply>=3.9
+colorama>=0.3.9
+configobj>=5.0.6
+gitpython>3
+dulwich>=0.20.23
+pygit2>=1.5.0
+setuptools>=34.0.0
+nanotime>=0.5.2
+pyasn1>=0.4.1
+voluptuous>=0.11.7
+requests>=2.22.0
+grandalf>=0.6
+distro>=1.3.0
+appdirs>=1.4.3
+ruamel.yaml>=0.17.11
+toml>=0.10.1
+funcy>=1.14
+pathspec>=0.6.0,<0.9.0
+shortuuid>=0.5.0
+tqdm>=4.45.0,<5
+packaging>=19.0
+zc.lockfile>=1.2.1
+flufl.lock>=3.2,<4
+win-unicode-console>=0.5; sys_platform == 'win32'
+pywin32>=225; sys_platform == 'win32' and python_version < '3.10'
+networkx>=2.5
+psutil>=5.8.0
+pydot>=1.2.4
+speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'
+dataclasses==0.7; python_version < '3.7'
+importlib-metadata>=1.4; python_version < '3.8'
+flatten_dict>=0.4.1,<1
+tabulate>=0.8.7
+pygtrie>=2.3.2
+dpath>=2.0.1,<3
+shtab>=1.3.4,<2
+rich>=10.0.0
+dictdiffer>=0.8.1
+python-benedict>=0.21.1
+pyparsing==2.4.7
+typing_extensions>=3.10.0.2
+fsspec[http]>=2021.8.1
+aiohttp-retry==2.4.5
+diskcache>=5.2.1

requirements/gdrive.txt

@@ -0,0 +1 @@
+pydrive2[fsspec]>=1.9.2

requirements/gs.txt

@@ -0,0 +1 @@
+gcsfs==2021.8.1

requirements/hdfs.txt

@@ -0,0 +1 @@
+pyarrow>=2.0.0; python_version < '3.10'

requirements/oss.txt

@@ -0,0 +1 @@
+ossfs==2021.8.0

requirements/s3.txt

@@ -0,0 +1,2 @@
+s3fs==2021.8.1
+aiobotocore[boto3]>1.0.1

requirements/ssh.txt

@@ -0,0 +1 @@
+sshfs>=2021.8.1

requirements/ssh_gssapi.txt

@@ -0,0 +1 @@
+sshfs[gssapi]>=2021.8.1

requirements/terraform.txt

@@ -0,0 +1 @@
+tpi[ssh]>=0.0.0

requirements/tests.txt

@@ -13,14 +13,13 @@ flaky==3.7.0
 mock==4.0.3
 rangehttpserver==1.2.0
 mock-ssh-server==0.9.0
-beautifulsoup4==4.9.3
 wget==3.2
 filelock==3.0.12
 wsgidav==3.1.1
 crc32c==2.2.post0
 xmltodict==0.12.0
 google-compute-engine==2.8.13
-google-cloud-storage==1.42.0
+google-cloud-storage==1.42.1
 urllib3==1.26.6
 dvclive>=0.2.1
 

requirements/webdav.txt

@@ -0,0 +1 @@
+webdav4>=0.9.1

requirements/webhdfs.txt

@@ -0,0 +1 @@
+hdfs==2.6.0

setup.py

@@ -46,80 +46,25 @@ class build_py(_build_py):
         _build_py.run(self)
 
 
-install_requires = [
-    "ply>=3.9",  # See https://github.com/pyinstaller/pyinstaller/issues/1945
-    "colorama>=0.3.9",
-    "configobj>=5.0.6",
-    "gitpython>3",
-    "dulwich>=0.20.23",
-    "pygit2>=1.5.0",
-    "setuptools>=34.0.0",
-    "nanotime>=0.5.2",
-    "pyasn1>=0.4.1",
-    "voluptuous>=0.11.7",
-    "requests>=2.22.0",
-    "grandalf==0.6",
-    "distro>=1.3.0",
-    "appdirs>=1.4.3",
-    "ruamel.yaml>=0.17.11",
-    "toml>=0.10.1",
-    "funcy>=1.14",
-    "pathspec>=0.6.0,<0.9.0",
-    "shortuuid>=0.5.0",
-    "tqdm>=4.45.0,<5",
-    "packaging>=19.0",
-    "zc.lockfile>=1.2.1",
-    "flufl.lock>=3.2,<4",
-    "win-unicode-console>=0.5; sys_platform == 'win32'",
-    "pywin32>=225; sys_platform == 'win32' and python_version < '3.10'",
-    "networkx>=2.5",
-    "psutil>=5.8.0",
-    "pydot>=1.2.4",
-    "speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'",
-    "dataclasses==0.7; python_version < '3.7'",
-    "importlib-metadata>=1.4; python_version < '3.8'",
-    "flatten_dict>=0.4.1,<1",
-    "tabulate>=0.8.7",
-    "pygtrie>=2.3.2",
-    "dpath>=2.0.1,<3",
-    "shtab>=1.3.4,<2",
-    "rich>=10.0.0",
-    "dictdiffer>=0.8.1",
-    "python-benedict>=0.21.1",
-    "pyparsing==2.4.7",
-    "typing_extensions>=3.7.4; python_version < '3.10'",
-    # until https://github.com/python/typing/issues/865 is fixed for python3.10
-    "typing_extensions==3.10.0.0; python_version >= '3.10'",
-    "fsspec[http]>=2021.8.1",
-    "aiohttp-retry==2.4.5",
-    "diskcache>=5.2.1",
-]
-
-
 # Extra dependencies for remote integrations
+requirements = {
+    path.stem: path.read_text().strip().splitlines()
+    for path in Path("requirements").glob("*.txt")
+}
 
-gs = ["gcsfs==2021.8.1"]
-gdrive = ["pydrive2[fsspec]>=1.9.2"]
-s3 = ["s3fs==2021.8.1", "aiobotocore[boto3]>1.0.1"]
-azure = ["adlfs==2021.8.2", "azure-identity>=1.4.0", "knack"]
-oss = ["ossfs==2021.8.0"]
-ssh = ["sshfs>=2021.8.1"]
-
-hdfs = ["pyarrow>=2.0.0; python_version < '3.10'"]
-webhdfs = ["hdfs==2.5.8"]
-webdav = ["webdav4>=0.9.1"]
 # gssapi should not be included in all_remotes, because it doesn't have wheels
 # for linux and mac, so it will fail to compile if user doesn't have all the
 # requirements, including kerberos itself. Once all the wheels are available,
 # we can start shipping it by default.
-ssh_gssapi = ["sshfs[gssapi]>=2021.8.1"]
-all_remotes = gs + s3 + azure + ssh + oss + gdrive + hdfs + webhdfs + webdav
 
-terraform = ["tpi[ssh]>=0.0.0"]
-
-tests_requirements = (
-    Path("test_requirements.txt").read_text().strip().splitlines()
-) + terraform
+install_requires = requirements.pop("default")
+requirements["all"] = [
+    requirements
+    for key, requirements in requirements.items()
+    if key not in ("tests", "ssh_gssapi", "terraform")
+]
+requirements["tests"] += requirements["terraform"]
+requirements["dev"] = requirements["all"] + requirements["tests"]
 
 setup(
     name="dvc",
@@ -133,21 +78,7 @@ setup(
     download_url="https://github.com/iterative/dvc",
     license="Apache License 2.0",
     install_requires=install_requires,
-    extras_require={
-        "all": all_remotes,
-        "gs": gs,
-        "gdrive": gdrive,
-        "s3": s3,
-        "azure": azure,
-        "oss": oss,
-        "ssh": ssh,
-        "ssh_gssapi": ssh_gssapi,
-        "hdfs": hdfs,
-        "webhdfs": webhdfs,
-        "webdav": webdav,
-        "terraform": terraform,
-        "tests": tests_requirements,
-    },
+    extras_require=requirements,
     keywords="data-science data-version-control machine-learning git"
     " developer-tools reproducibility collaboration ai",
     python_requires=">=3.6",

tests/func/experiments/conftest.py

@@ -88,3 +88,17 @@ def checkpoint_stage(tmp_dir, scm, dvc, mocker):
     scm.commit("init")
     stage.iterations = DEFAULT_ITERATIONS
     return stage
+
+
+@pytest.fixture
+def http_auth_patch(mocker):
+    from dulwich.client import HTTPUnauthorized
+
+    url = "https://0.0.0.0"
+    client = mocker.MagicMock()
+    client.get_refs.side_effect = HTTPUnauthorized("", url)
+    client.send_pack.side_effect = HTTPUnauthorized("", url)
+
+    patch = mocker.patch("dulwich.client.get_transport_and_path")
+    patch.return_value = (client, url)
+    return url

tests/func/experiments/test_remote.py

@@ -252,3 +252,37 @@ def test_push_pull_cache(
         path = os.path.join(dvc.odb.local.cache_dir, hash_[:2], hash_[2:])
         assert os.path.exists(path)
         assert open(path).read() == str(x)
+
+
+def test_auth_error_list(tmp_dir, scm, dvc, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.ls(git_remote=http_auth_patch)
+
+
+def test_auth_error_pull(tmp_dir, scm, dvc, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.pull(http_auth_patch, "foo")
+
+
+def test_auth_error_push(tmp_dir, scm, dvc, exp_stage, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    exp = first(results)
+    ref_info = first(exp_refs_by_rev(scm, exp))
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.push(http_auth_patch, ref_info.name)

tests/func/test_fs.py

@@ -405,7 +405,7 @@ def test_fs_makedirs_on_upload_and_copy(dvc, cloud):
     fs = cls(**config)
 
     with io.BytesIO(b"foo") as stream:
-        fs.upload_fobj(stream, cloud / "dir" / "foo")
+        fs.upload(stream, cloud / "dir" / "foo")
 
     assert fs.isdir(cloud / "dir")
     assert fs.exists(cloud / "dir" / "foo")

tests/unit/fs/test_ssh.py

@@ -34,10 +34,13 @@ def test_get_kwargs_from_urls():
 
 
 def test_init():
-    fs = SSHFileSystem(user="test", host="12.34.56.78", port="1234")
+    fs = SSHFileSystem(
+        user="test", host="12.34.56.78", port="1234", password="xxx"
+    )
     assert fs.fs_args["username"] == "test"
     assert fs.fs_args["host"] == "12.34.56.78"
     assert fs.fs_args["port"] == "1234"
+    assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "xxx"
 
 
 mock_ssh_config = """