dvc: introduce per-remote --jobs default

Fixes #1167

Signed-off-by: Ruslan Kuprieiev <[email protected]>

dvc/cli.py

@@ -3,7 +3,6 @@ from __future__ import print_function
 import os
 import sys
 import argparse
-from multiprocessing import cpu_count
 
 from dvc.command.init import CmdInit
 from dvc.command.destroy import CmdDestroy
@@ -257,7 +256,7 @@ def parse_args(argv=None):
                         '-j',
                         '--jobs',
                         type=int,
-                        default=4*cpu_count(),
+                        default=None,
                         help='Number of jobs to run simultaneously.')
     parent_cache_parser.add_argument(
                         '--show-checksums',

dvc/remote/base.py

@@ -1,6 +1,7 @@
 import os
 import re
 import errno
+from multiprocessing import cpu_count
 
 from dvc.config import Config
 from dvc.logger import Logger
@@ -30,6 +31,7 @@ class DataCloudError(DvcException):
 class RemoteBase(object):
     REGEX = None
     REQUIRES = {}
+    JOBS = 4 * cpu_count()
 
     def __init__(self, project, config):
         pass

dvc/remote/local.py

@@ -487,7 +487,7 @@ class RemoteLOCAL(RemoteBase):
 
         return removed
 
-    def status(self, checksum_infos, remote, jobs=1, show_checksums=False):
+    def status(self, checksum_infos, remote, jobs=None, show_checksums=False):
         Logger.info("Preparing to pull data from {}".format(remote.url))
         title = "Collecting information"
 
@@ -523,7 +523,7 @@ class RemoteLOCAL(RemoteBase):
     def _do_pull(self,
                  checksum_infos,
                  remote,
-                 jobs=1,
+                 jobs=None,
                  show_checksums=False):
         title = "Collecting information"
 
@@ -561,6 +561,9 @@ class RemoteLOCAL(RemoteBase):
 
         assert len(path_infos) == len(cache) == len(md5s) == len(names)
 
+        if jobs is None:
+            jobs = remote.JOBS
+
         chunks = list(zip(to_chunks(path_infos, jobs),
                           to_chunks(cache, jobs),
                           to_chunks(names, jobs)))
@@ -584,7 +587,7 @@ class RemoteLOCAL(RemoteBase):
         for f in futures:
             f.result()
 
-    def pull(self, checksum_infos, remote, jobs=1, show_checksums=False):
+    def pull(self, checksum_infos, remote, jobs=None, show_checksums=False):
         Logger.info("Preparing to pull data from {}".format(remote.url))
 
         # NOTE: try fetching missing dir info
@@ -601,7 +604,7 @@ class RemoteLOCAL(RemoteBase):
                       jobs,
                       show_checksums=show_checksums)
 
-    def push(self, checksum_infos, remote, jobs=1, show_checksums=False):
+    def push(self, checksum_infos, remote, jobs=None, show_checksums=False):
         Logger.info("Preparing to push data to {}".format(remote.url))
         title = "Collecting information"
 
@@ -646,6 +649,9 @@ class RemoteLOCAL(RemoteBase):
 
         progress.update_target(title, 90, 100)
 
+        if jobs is None:
+            jobs = remote.JOBS
+
         chunks = list(zip(to_chunks(path_infos, jobs),
                           to_chunks(cache, jobs),
                           to_chunks(names, jobs)))

dvc/remote/ssh.py

@@ -44,6 +44,9 @@ class RemoteSSH(RemoteBase):
     REGEX = r'^ssh://((?P<user>.*)@)?(?P<host>[^/]*):(?P<path>/.*)$'
 
     REQUIRES = {'paramiko': paramiko}
+
+    JOBS = 4
+
     PARAM_MD5 = 'md5'
 
     DEFAULT_PORT = 22