remote: use configured jobs number (#5747)

Author: pared

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 @@ def jobs(self):
         return (
             self.config.get("jobs")
             or (self.repo and self.repo.config["core"].get("jobs"))
-            or self.JOBS
+            or self._JOBS
         )
 
     @cached_property

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

dvc/objects/db/base.py

@@ -229,7 +229,7 @@ def list_hashes_traverse(
         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 @@ def list_with_update(prefix):
                 )
 
             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 @@ def exists_with_progress(path_info):
                 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)

dvc/objects/db/ssh.py

@@ -64,10 +64,10 @@ def exists_with_progress(chunks):
                 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))

dvc/remote/base.py

@@ -318,7 +318,7 @@ def _process(
             desc = "Uploading"
 
         if jobs is None:
-            jobs = self.fs.JOBS
+            jobs = self.fs.jobs
 
         dir_status, file_status, dir_contents = self._status(
             cache,

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(