diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 1fef1a63a..9fb7c43b6 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -68,5 +68,5 @@ class DependencyBase(object): def dumpd(self): return {self.PARAM_PATH: self.path} - def download(self, path): - self.remote.download([self.path_info], [path]) + def download(self, to_info): + self.remote.download([self.path_info], [to_info]) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index aa2b43500..474cfbf5a 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -55,7 +55,7 @@ class RemoteGS(RemoteBase): return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} - def _copy(from_info, to_info, gs=None): + def _copy(self, from_info, to_info, gs=None): gs = gs if gs else self.gs blob = gs.bucket(from_info['bucket']).get_blob(from_info['key']) @@ -160,7 +160,7 @@ class RemoteGS(RemoteBase): if from_info['scheme'] != 'gs': raise NotImplementedError - if from_info['scheme'] == 'gs': + if to_info['scheme'] == 'gs': self._copy(from_info, to_info, gs=gs) continue diff --git a/dvc/remote/local.py b/dvc/remote/local.py index a351181b8..5bde700fd 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -369,12 +369,18 @@ class RemoteLOCAL(RemoteBase): if len(chunks) == 0: return + futures = [] with ThreadPoolExecutor(max_workers=len(chunks)) as executor: for from_infos, to_infos, md5s in chunks: - executor.submit(remote.download, from_infos, - to_infos, - names=md5s, - no_progress_bar=no_progress_bar) + res = executor.submit(remote.download, + from_infos, + to_infos, + names=md5s, + no_progress_bar=no_progress_bar) + futures.append(res) + + for f in futures: + f.result() def pull(self, checksum_infos, remote, jobs=1): # NOTE: try fetching missing dir info @@ -410,6 +416,11 @@ class RemoteLOCAL(RemoteBase): if len(chunks) == 0: return + futures = [] with ThreadPoolExecutor(max_workers=len(chunks)) as executor: for to_infos, from_infos, md5s in chunks: - executor.submit(remote.upload, from_infos, to_infos, names=md5s) + res = executor.submit(remote.upload, from_infos, to_infos, names=md5s) + futures.append(res) + + for f in futures: + f.result() diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index a6ea8a863..9901011ce 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -167,7 +167,7 @@ class RemoteS3(RemoteBase): return ret - def upload(self, from_infos, path_infos, names=None): + def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) s3 = self.s3_session_client