From eccc320606ffbcf05aa0d3fec0e9918f7c9b9551 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pawe=C5=82=20Redzy=C5=84ski?= <pawelredzynski@gmail.com>
Date: Thu, 8 Apr 2021 16:52:42 +0200
Subject: [PATCH] remote: use configured jobs number (#5747)

---
 dvc/fs/base.py                 | 4 ++--
 dvc/fs/ssh/__init__.py         | 2 +-
 dvc/objects/db/base.py         | 6 +++---
 dvc/objects/db/ssh.py          | 4 ++--
 dvc/remote/base.py             | 2 +-
 tests/unit/remote/test_base.py | 4 ++--
 6 files changed, 11 insertions(+), 11 deletions(-)

diff --git a/dvc/fs/base.py b/dvc/fs/base.py
index 03bb442a4..604bed544 100644
--- a/dvc/fs/base.py
+++ b/dvc/fs/base.py
@@ -38,7 +38,7 @@ class BaseFileSystem:
     scheme = "base"
     REQUIRES: ClassVar[Dict[str, str]] = {}
     PATH_CLS = URLInfo  # type: Any
-    JOBS = 4 * cpu_count()
+    _JOBS = 4 * cpu_count()
 
     CHECKSUM_DIR_SUFFIX = ".dir"
     HASH_JOBS = max(1, min(4, cpu_count() // 2))
@@ -67,7 +67,7 @@ class BaseFileSystem:
         return (
             self.config.get("jobs")
             or (self.repo and self.repo.config["core"].get("jobs"))
-            or self.JOBS
+            or self._JOBS
         )
 
     @cached_property
diff --git a/dvc/fs/ssh/__init__.py b/dvc/fs/ssh/__init__.py
index 54a182368..4ad1fca36 100644
--- a/dvc/fs/ssh/__init__.py
+++ b/dvc/fs/ssh/__init__.py
@@ -33,7 +33,7 @@ def ask_password(host, user, port):
 class SSHFileSystem(BaseFileSystem):
     scheme = Schemes.SSH
     REQUIRES = {"paramiko": "paramiko"}
-    JOBS = 4
+    _JOBS = 4
 
     PARAM_CHECKSUM = "md5"
     # At any given time some of the connections will go over network and
diff --git a/dvc/objects/db/base.py b/dvc/objects/db/base.py
index 4288f0c7c..885f214cd 100644
--- a/dvc/objects/db/base.py
+++ b/dvc/objects/db/base.py
@@ -229,7 +229,7 @@ class ObjectDB:
         keeping all of it in memory.
         """
         num_pages = remote_size / self.fs.LIST_OBJECT_PAGE_SIZE
-        if num_pages < 256 / self.fs.JOBS:
+        if num_pages < 256 / self.fs.jobs:
             # Fetching prefixes in parallel requires at least 255 more
             # requests, for small enough remotes it will be faster to fetch
             # entire cache without splitting it into prefixes.
@@ -263,7 +263,7 @@ class ObjectDB:
                 )
 
             with ThreadPoolExecutor(
-                max_workers=jobs or self.fs.JOBS
+                max_workers=jobs or self.fs.jobs
             ) as executor:
                 in_remote = executor.map(list_with_update, traverse_prefixes,)
                 yield from itertools.chain.from_iterable(in_remote)
@@ -331,7 +331,7 @@ class ObjectDB:
                 return ret
 
             with ThreadPoolExecutor(
-                max_workers=jobs or self.fs.JOBS
+                max_workers=jobs or self.fs.jobs
             ) as executor:
                 path_infos = map(self.hash_to_path_info, hashes)
                 in_remote = executor.map(exists_with_progress, path_infos)
diff --git a/dvc/objects/db/ssh.py b/dvc/objects/db/ssh.py
index e2b0198c9..5be814871 100644
--- a/dvc/objects/db/ssh.py
+++ b/dvc/objects/db/ssh.py
@@ -64,10 +64,10 @@ class SSHObjectDB(ObjectDB):
                 return self.batch_exists(chunks, callback=pbar.update_msg)
 
             with ThreadPoolExecutor(
-                max_workers=jobs or self.fs.JOBS
+                max_workers=jobs or self.fs.jobs
             ) as executor:
                 path_infos = [self.hash_to_path_info(x) for x in hashes]
-                chunks = to_chunks(path_infos, num_chunks=self.fs.JOBS)
+                chunks = to_chunks(path_infos, num_chunks=self.fs.jobs)
                 results = executor.map(exists_with_progress, chunks)
                 in_remote = itertools.chain.from_iterable(results)
                 ret = list(itertools.compress(hashes, in_remote))
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 1671d1bb7..583883980 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -318,7 +318,7 @@ class Remote:
             desc = "Uploading"
 
         if jobs is None:
-            jobs = self.fs.JOBS
+            jobs = self.fs.jobs
 
         dir_status, file_status, dir_contents = self._status(
             cache,
diff --git a/tests/unit/remote/test_base.py b/tests/unit/remote/test_base.py
index 322ec5e83..eb5a833ee 100644
--- a/tests/unit/remote/test_base.py
+++ b/tests/unit/remote/test_base.py
@@ -70,7 +70,7 @@ def test_hashes_exist(object_exists, traverse, dvc):
     # large remote, large local
     object_exists.reset_mock()
     traverse.reset_mock()
-    odb.fs.JOBS = 16
+    odb.fs._JOBS = 16
     with mock.patch.object(odb, "list_hashes", return_value=list(range(256))):
         hashes = list(range(1000000))
         odb.hashes_exist(hashes)
@@ -94,7 +94,7 @@ def test_list_hashes_traverse(_path_to_hash, list_hashes, dvc):
     odb.fs.path_info = PathInfo("foo")
 
     # parallel traverse
-    size = 256 / odb.fs.JOBS * odb.fs.LIST_OBJECT_PAGE_SIZE
+    size = 256 / odb.fs._JOBS * odb.fs.LIST_OBJECT_PAGE_SIZE
     list(odb.list_hashes_traverse(size, {0}))
     for i in range(1, 16):
         list_hashes.assert_any_call(