diff --git a/dvc/cli.py b/dvc/cli.py index a1be92d47..60153446c 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -123,15 +123,6 @@ def parse_args(argv=None): help='Command or command file to execute') run_parser.set_defaults(func=CmdRun) - # Parent parser used in pull/push/status - parent_cache_parser = argparse.ArgumentParser( - add_help=False, - parents=[parent_parser]) - parent_cache_parser.add_argument('-j', - '--jobs', - type=int, - default=cpu_count(), - help='Number of jobs to run simultaneously.') # Cache cache_parser = subparsers.add_parser( 'cache', @@ -144,21 +135,21 @@ def parse_args(argv=None): # Pull pull_parser = cache_subparsers.add_parser( 'pull', - parents=[parent_cache_parser], + parents=[parent_parser], help='Pull data files from the cloud') pull_parser.set_defaults(func=CmdDataPull) # Push push_parser = cache_subparsers.add_parser( 'push', - parents=[parent_cache_parser], + parents=[parent_parser], help='Push data files to the cloud') push_parser.set_defaults(func=CmdDataPush) # Status status_parser = cache_subparsers.add_parser( 'status', - parents=[parent_cache_parser], + parents=[parent_parser], help='Show mismatches between local cache and cloud cache') status_parser.set_defaults(func=CmdDataStatus) diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py index 5298855ea..2e01b7726 100644 --- a/dvc/cloud/data_cloud.py +++ b/dvc/cloud/data_cloud.py @@ -2,7 +2,7 @@ from dvc.cloud.instance_manager import CloudSettings from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.config import ConfigError -from dvc.utils import map_progress +from dvc.progress import progress from dvc.cloud.aws import DataCloudAWS from dvc.cloud.gcp import DataCloudGCP @@ -79,26 +79,36 @@ class DataCloud(object): self._cloud.sanity_check() - def _map_targets(self, func, targets, jobs): + def _process_targets(self, func, targets): """ - Process targets as data items in parallel. + Process targets with a progress bar. """ - return map_progress(func, targets, jobs) + ret = [] + progress.set_n_total(len(targets)) + try: + for target in targets: + ret += [func(target)] + except Exception as exc: + raise + finally: + progress.finish() - def push(self, targets, jobs=1): + return list(zip(targets, ret)) + + def push(self, targets): """ Push data items in a cloud-agnostic way. """ - return self._map_targets(self._cloud.push, targets, jobs) + return self._process_targets(self._cloud.push, targets) - def pull(self, targets, jobs=1): + def pull(self, targets): """ Pull data items in a cloud-agnostic way. """ - return self._map_targets(self._cloud.pull, targets, jobs) + return self._process_targets(self._cloud.pull, targets) - def status(self, targets, jobs=1): + def status(self, targets): """ Check status of data items in a cloud-agnostic way. """ - return self._map_targets(self._cloud.status, targets, jobs) + return self._process_targets(self._cloud.status, targets) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index aca9803d7..fec28915b 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -7,7 +7,7 @@ from dvc.exceptions import DvcException class CmdDataPull(CmdBase): def run(self): try: - self.project.pull(self.args.jobs) + self.project.pull() except DvcException as exc: self.project.logger.error('Failed to pull data from the cloud', exc) return 1 @@ -17,7 +17,7 @@ class CmdDataPull(CmdBase): class CmdDataPush(CmdBase): def run(self): try: - self.project.push(self.args.jobs) + self.project.push() except DvcException as exc: self.project.logger.error('Failed to push data to the cloud', exc) return 1 @@ -42,7 +42,7 @@ class CmdDataStatus(CmdBase): def run(self): try: - status = self.project.status(self.args.jobs) + status = self.project.status() self._show(status) except DvcException as exc: self.project.logger.error('Failed to obtain data status', exc) diff --git a/dvc/project.py b/dvc/project.py index 8050cc30d..1e2b8ecfe 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -195,15 +195,15 @@ class Project(object): self._remove_cache(cache) self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache))) - def push(self, jobs=1): - self.cloud.push(self._used_cache(), jobs) + def push(self): + self.cloud.push(self._used_cache()) - def pull(self, jobs=1): - self.cloud.pull(self._used_cache(), jobs) + def pull(self): + self.cloud.pull(self._used_cache()) self.checkout() - def status(self, jobs=1): - return self.cloud.status(self._used_cache(), jobs) + def status(self): + return self.cloud.status(self._used_cache()) def graph(self): G = nx.DiGraph() diff --git a/dvc/utils.py b/dvc/utils.py index 55e4c037a..24e540443 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -72,24 +72,6 @@ def copyfile(src, dest): fdest.close() -def map_progress(func, targets, n_threads): - """ - Process targets in multi-threaded mode with progress bar - """ - progress.set_n_total(len(targets)) - pool = ThreadPool(processes=n_threads) - ret = [] - - try: - ret = pool.map(func, targets) - except Exception as exc: - raise - finally: - progress.finish() - - return list(zip(targets, ret)) - - def parse_target_metric_file(file_name): with open(file_name, 'r') as fd: try: diff --git a/tests/test_utils.py b/tests/test_utils.py index b1344d89c..e377a941d 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -26,14 +26,3 @@ class TestUtils(TestDvc): shutil.rmtree(dest_dir) os.remove(src) os.remove(dest) - - def test_map_progress(self): - def f(target): - with open(target, 'w+') as o: - o.write(target) - - targets = ['map{}'.format(i) for i in range(1, 10)] - n_threads = [1, 10, 20] - - for n in n_threads: - utils.map_progress(f, targets, n)