progress: do not block on async filesystems

dvc/fs/base.py

@@ -6,7 +6,6 @@
 from multiprocessing import cpu_count
 from typing import Any, ClassVar, Dict, FrozenSet, Optional
 
-from funcy import cached_property
 from tqdm.utils import CallbackIOWrapper
 
 from dvc.exceptions import DvcException
@@ -223,12 +222,6 @@ def is_dir_hash(cls, hash_):
             return False
         return hash_.endswith(cls.CHECKSUM_DIR_SUFFIX)
 
-    @cached_property
-    def _local_fs(self):
-        from dvc.fs import LocalFileSystem
-
-        return LocalFileSystem()
-
     def upload(
         self,
         from_info,
@@ -260,7 +253,7 @@ def upload(
                 **pbar_args,
             )
             stack.enter_context(pbar)
-            callback = pbar.as_callback(self._local_fs, from_info)
+            callback = pbar.as_callback()
             if total:
                 callback.set_size(total)
 
@@ -321,7 +314,7 @@ def download(
                 **pbar_kwargs,
             )
             stack.enter_context(pbar)
-            callback = pbar.as_callback(self, from_info)
+            callback = pbar.as_callback()
 
         with stack:
             if download_dir:

dvc/fs/fsspec_wrapper.py

@@ -8,7 +8,6 @@
 from dvc.progress import DEFAULT_CALLBACK
 
 from .base import BaseFileSystem
-from .local import LocalFileSystem
 
 
 # pylint: disable=no-member
@@ -238,9 +237,6 @@ def ls(self, *args, **kwargs):
         raise NotImplementedError
 
 
-_LOCAL_FS = LocalFileSystem()
-
-
 class CallbackMixin:
     """Provides callback support for the filesystem that don't support yet."""
 

dvc/progress.py

@@ -136,8 +136,8 @@ def wrapped(*args, **kwargs):
 
         return wrapped
 
-    def as_callback(self, fs, path_info):
-        return FsspecCallback(fs, path_info, self)
+    def as_callback(self):
+        return FsspecCallback(self)
 
     def close(self):
         self.postfix["info"] = ""
@@ -167,21 +167,15 @@ def format_dict(self):
 
 
 class FsspecCallback(fsspec.Callback):
-    def __init__(self, fs, path_info, progress_bar):
-        self.fs = fs
-        self.path_info = path_info
+    def __init__(self, progress_bar):
         self.progress_bar = progress_bar
         super().__init__()
 
     def set_size(self, size):
-        """``set_size()`` is an API that might be called with ``None`` if the
-        information is not already present on the caller. In that case, we'll
-        retrieve the size from an ``info()`` call."""
-        if size is None:
-            size = self.fs.info(self.path_info)["size"]
-        self.progress_bar.total = size
-        self.progress_bar.refresh()
-        super().set_size(size)
+        if size is not None:
+            self.progress_bar.total = size
+            self.progress_bar.refresh()
+            super().set_size(size)
 
     def relative_update(self, inc=1):
         self.progress_bar.update(inc)