diff --git a/dvc/ignore.py b/dvc/ignore.py index d11c43053..c9257e520 100644 --- a/dvc/ignore.py +++ b/dvc/ignore.py @@ -244,3 +244,7 @@ class CleanTree(BaseTree): if self.exists(path): return self.tree.stat(path) raise FileNotFoundError + + @property + def hash_jobs(self): + return self.tree.hash_jobs diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 8affd571a..5d26c1f25 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -323,7 +323,8 @@ class BaseRemoteTree: self.PARAM_CHECKSUM: self.get_hash(path_info, tree=tree, **kwargs) } - def _calculate_hashes(self, file_infos, tree): + @staticmethod + def _calculate_hashes(file_infos, tree): file_infos = list(file_infos) with Tqdm( total=len(file_infos), @@ -331,7 +332,7 @@ class BaseRemoteTree: desc="Computing file/dir hashes (only done once)", ) as pbar: worker = pbar.wrap_fn(tree.get_file_hash) - with ThreadPoolExecutor(max_workers=self.hash_jobs) as executor: + with ThreadPoolExecutor(max_workers=tree.hash_jobs) as executor: tasks = executor.map(worker, file_infos) hashes = dict(zip(file_infos, tasks)) return hashes diff --git a/dvc/repo/tree.py b/dvc/repo/tree.py index ff7d4e707..175d80f95 100644 --- a/dvc/repo/tree.py +++ b/dvc/repo/tree.py @@ -421,3 +421,7 @@ class RepoTree(BaseTree): src = root / fname with self.open(src, mode="rb") as fobj: copy_fobj_to_file(fobj, dest / fname) + + @property + def hash_jobs(self): + return self.repo.tree.hash_jobs diff --git a/dvc/scm/git/tree.py b/dvc/scm/git/tree.py index 99d22a00e..2ea999c66 100644 --- a/dvc/scm/git/tree.py +++ b/dvc/scm/git/tree.py @@ -183,3 +183,9 @@ class GitTree(BaseTree): to_ctime(entry.ctime), ) ) + + @property + def hash_jobs(self): + # NOTE: gitpython is not threadsafe. See + # https://github.com/iterative/dvc/issues/4079 + return 1 diff --git a/dvc/scm/tree.py b/dvc/scm/tree.py index 5061c5fbf..c923cd6be 100644 --- a/dvc/scm/tree.py +++ b/dvc/scm/tree.py @@ -1,5 +1,8 @@ import os import stat +from multiprocessing import cpu_count + +from funcy import cached_property class BaseTree: @@ -83,6 +86,10 @@ class WorkingTree(BaseTree): def stat(path): return os.stat(path) + @cached_property + def hash_jobs(self): + return max(1, min(4, cpu_count() // 2)) + def is_working_tree(tree): return isinstance(tree, WorkingTree) or isinstance(