From 4fa8618734814361b8674faf8533fd802c373979 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 22:08:47 +0200 Subject: [PATCH 01/11] git: update .gitignore Signed-off-by: Ruslan Kuprieiev --- .dvc/.gitignore | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/.dvc/.gitignore b/.dvc/.gitignore index a53edd2f3..5b2f952a6 100644 --- a/.dvc/.gitignore +++ b/.dvc/.gitignore @@ -1,10 +1,8 @@ -link.state -link.state.lock -lock -config.local -updater -cache -state -state.lock -state-journal -state-wal +/state +/lock +/config.local +/updater +/updater.lock +/state-journal +/state-wal +/cache From 7600ccf8bef204527f7fb0e6c0c20d0f8da3c86a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Fri, 14 Dec 2018 21:31:27 +0200 Subject: [PATCH 02/11] remote: introduce cache_exists() method Making this method separate from exists() makes it easier to optimize requests for large amounts of cache, since we can re-use the same prefix for retrieving all the necessary information in a one single request. Related #1430 Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/base.py | 2 +- dvc/remote/azure.py | 54 ++++++++++++++++++++++------------ dvc/remote/base.py | 3 ++ dvc/remote/gs.py | 67 +++++++++++++++++++++++------------------- dvc/remote/hdfs.py | 43 +++++++++++++++++---------- dvc/remote/http.py | 18 ++++++++---- dvc/remote/local.py | 26 ++++++++-------- dvc/remote/s3.py | 64 +++++++++++++++++++++------------------- dvc/remote/ssh.py | 44 ++++++++++++++------------- 9 files changed, 188 insertions(+), 133 deletions(-) diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 62637cb14..4700e8fc8 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -60,7 +60,7 @@ class DependencyBase(object): @property def exists(self): - return self.remote.exists([self.path_info])[0] + return self.remote.exists(self.path_info) def changed(self): raise NotImplementedError diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 228c68551..4b5ee2a68 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -49,7 +49,7 @@ class RemoteAzure(RemoteBase): or match.group('container_name') # backward compatibility or os.getenv('AZURE_STORAGE_CONTAINER_NAME')) - self.prefix = urlparse(self.url).path.lstrip('/') or '/' + self.prefix = urlparse(self.url).path.lstrip('/') self.connection_string = ( config.get(Config.SECTION_AZURE_CONNECTION_STRING) @@ -145,29 +145,39 @@ class RemoteAzure(RemoteBase): return [{ 'scheme': self.scheme, 'bucket': self.bucket, - 'key': '{}/{}/{}'.format(self.prefix, md5[0:2], md5[2:]) + 'key': posixpath.join(self.prefix, md5[0:2], md5[2:]) } for md5 in md5s] - def _all_keys(self, bucket, prefix): - return {blob.name - for blob in self.blob_service.list_blobs(bucket, - prefix=prefix)} + def _list_keys(self, bucket, prefix): + blob_service = self.blob_service + next_marker = None + while True: + blobs = blob_service.list_blobs(bucket, + prefix=prefix, + marker=next_marker) - def exists(self, path_infos): - ret = [] + for blob in blobs: + yield blob.name - if len(path_infos) == 0: - return ret + if not blobs.next_marker: + break - bucket = path_infos[0]['bucket'] - key = path_infos[0]['key'] + next_marker = blobs.next_marker - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - assert path_info['bucket'] == bucket + def cache_exists(self, md5s): + assert isinstance(md5s, list) - key = path_info['key'] - ret.append(key in self._all_keys(bucket, key)) + if len(md5s) == 0: + return [] + + ret = len(md5s) * [False] + keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] + for key in self._list_keys(self.bucket, self.prefix): + for i, k in enumerate(keys): + if k == key: + ret[i] = True + + assert len(ret) == len(keys) == len(md5s) return ret @@ -246,8 +256,14 @@ class RemoteAzure(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all(self): - keys = self._all_keys(self.bucket, self.prefix) - return [self._path_to_etag(key) for key in keys] + # NOTE: The list might be way too big(e.g. 100M entries, md5 for each + # is 32 bytes, so ~3200Mb list) and we don't really need all of it at + # the same time, so it makes sense to use a generator to gradually + # iterate over it, without keeping all of it in memory. + return ( + self._path_to_etag(key) + for key in self._list_keys(self.bucket, self.prefix) + ) def gc(self, cinfos): from dvc.remote.local import RemoteLOCAL diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 02ab6b588..1d7f82128 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -131,6 +131,9 @@ class RemoteBase(object): def md5s_to_path_infos(self, md5s): raise NotImplementedError + def cache_exists(self, md5s): + raise NotImplementedError + def exists(self, path_infos): raise NotImplementedError diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 23830071f..78a9f75eb 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -33,6 +33,10 @@ class RemoteGS(RemoteBase): self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) + parsed = urlparse(self.url) + self.bucket = parsed.netloc + self.prefix = parsed.path.lstrip('/') + @staticmethod def compat_config(config): ret = config.copy() @@ -40,14 +44,6 @@ class RemoteGS(RemoteBase): ret[Config.SECTION_REMOTE_URL] = url return ret - @property - def bucket(self): - return urlparse(self.url).netloc - - @property - def prefix(self): - return urlparse(self.url).path.lstrip('/') - @property def gs(self): return storage.Client() @@ -82,7 +78,7 @@ class RemoteGS(RemoteBase): cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} if {self.PARAM_MD5: md5} != self.save_info(cache): - if self.exists([cache])[0]: + if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) self.remove(cache) @@ -91,7 +87,7 @@ class RemoteGS(RemoteBase): return False def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True md5 = checksum_info.get(self.PARAM_MD5, None) @@ -146,7 +142,7 @@ class RemoteGS(RemoteBase): logger.warn(msg.format(md5, self.to_string(path_info))) return - if self.exists([path_info])[0]: + if self.exists(path_info): msg = "Data '{}' exists. Removing before checkout." logger.warn(msg.format(self.to_string(path_info))) self.remove(path_info) @@ -179,26 +175,32 @@ class RemoteGS(RemoteBase): 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s] - def exists(self, path_infos): - ret = [] + def _list_keys(self, bucket, prefix): + for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): + yield blob.name - if len(path_infos) == 0: - return ret + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'gs' - gs = self.gs - bucket_name = path_infos[0]['bucket'] - bucket = gs.bucket(bucket_name) + keys = self._list_keys(path_info['bucket'], path_info['key']) + return any(path_info['key'] == key for key in keys) - for path_info in path_infos: - assert self.scheme == path_info['scheme'] - assert bucket_name == path_info['bucket'] + def cache_exists(self, md5s): + assert isinstance(md5s, list) + + if len(md5s) == 0: + return [] + + ret = len(md5s) * [False] + keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] + for key in self._list_keys(self.bucket, self.prefix): + for i, k in enumerate(keys): + if k == key: + ret[i] = True + + assert len(ret) == len(keys) == len(md5s) - exists = False - key = path_info['key'] - keys = [blob.name for blob in bucket.list_blobs(prefix=key)] - if key in keys: - exists = True - ret.append(exists) return ret def upload(self, from_infos, to_infos, names=None): @@ -292,9 +294,14 @@ class RemoteGS(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all_md5s(self): - blobs = self.gs.bucket(self.bucket).list_blobs(prefix=self.prefix) - blobs = list(blobs) - return [self._path_to_md5(blob.name) for blob in blobs] + # NOTE: The list might be way too big(e.g. 100M entries, md5 for each + # is 32 bytes, so ~3200Mb list) and we don't really need all of it at + # the same time, so it makes sense to use a generator to gradually + # iterate over it, without keeping all of it in memory. + return ( + self._path_to_md5(key) + for key in self._list_keys(self.bucket, self.prefix) + ) def gc(self, cinfos): used = [info[self.PARAM_MD5] for info in cinfos['gs']] diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index dfc1f331c..3b7158b2d 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -93,7 +93,7 @@ class RemoteHDFS(RemoteBase): cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): - if self.exists([cache])[0]: + if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) self.remove(cache) @@ -102,7 +102,7 @@ class RemoteHDFS(RemoteBase): return False def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True checksum = checksum_info.get(self.PARAM_CHECKSUM, None) @@ -148,7 +148,7 @@ class RemoteHDFS(RemoteBase): logger.warn(msg.format(checksum, self.to_string(path_info))) return - if self.exists([path_info])[0]: + if self.exists(path_info): msg = "Data '{}' exists. Removing before checkout." logger.warn(msg.format(self.to_string(path_info))) self.remove(path_info) @@ -178,20 +178,28 @@ class RemoteHDFS(RemoteBase): 'url': posixpath.join(self.url, md5[0:2], md5[2:])} for md5 in md5s] - def exists(self, path_infos): - ret = [] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'hdfs' - if len(path_infos) == 0: - return ret + try: + self.hadoop_fs('test -e {}'.format(path_info['url'])) + return True + except RemoteHDFSCmdError: + return False - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - try: - self.hadoop_fs('test -e {}'.format(path_info['url'])) - exists = True - except RemoteHDFSCmdError: - exists = False - ret.append(exists) + def cache_exists(self, md5s): + assert isinstance(md5s, list) + + if len(md5s) == 0: + return [] + + existing_md5s = self._all_checksums() + ret = len(md5s) * [False] + for existing_md5 in existing_md5s: + for i, md5 in enumerate(md5s): + if md5 == existing_md5: + ret[i] = True return ret @@ -242,6 +250,11 @@ class RemoteHDFS(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all_checksums(self): + try: + self.hadoop_fs('test -e {}'.format(self.url)) + except RemoteHDFSCmdError: + return [] + stdout = self.hadoop_fs('ls -R {}'.format(self.url)) lines = stdout.split('\n') flist = [] diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 50bf1910c..4eb6c82a9 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -82,11 +82,19 @@ class RemoteHTTP(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def exists(self, path_infos): - return [ - bool(self._request('HEAD', path_info.get('url'))) - for path_info in path_infos - ] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] in ['http', 'https'] + return bool(self._request('HEAD', path_info.get('url'))) + + def cache_exists(self, md5s): + assert isinstance(md5s, list) + ret = [] + for md5 in md5s: + url = posixpath.join(self.prefix, md5[0:2], md5[2:]) + exists = bool(self._request('HEAD', url)) + ret.append(exists) + return ret def save_info(self, path_info): if path_info['scheme'] not in ['http', 'https']: diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 369820fab..d6a01599f 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -101,6 +101,11 @@ class RemoteLOCAL(RemoteBase): return True return False + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'local' + return os.path.exists(path_info['path']) + def changed_cache(self, md5): cache = self.get(md5) clist = [(cache, md5)] @@ -465,7 +470,7 @@ class RemoteLOCAL(RemoteBase): - The checkusm stored in the State is different from the given one - There's no file in the cache """ - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True md5 = checksum_info.get(self.PARAM_MD5, None) @@ -494,11 +499,12 @@ class RemoteLOCAL(RemoteBase): 'path': os.path.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s] - def exists(self, path_infos): + def cache_exists(self, md5s): + assert isinstance(md5s, list) ret = [] - for path_info in path_infos: - assert path_info['scheme'] == 'local' - ret.append(os.path.exists(path_info['path'])) + for md5 in md5s: + path = os.path.join(self.prefix, md5[0:2], md5[2:]) + ret.append(os.path.exists(path)) return ret def upload(self, from_infos, to_infos, names=None): @@ -610,13 +616,9 @@ class RemoteLOCAL(RemoteBase): md5s, names = self._group(checksum_infos, show_checksums=show_checksums) - progress.update_target(title, 20, 100) - - path_infos = remote.md5s_to_path_infos(md5s) - progress.update_target(title, 30, 100) - remote_exists = remote.exists(path_infos) + remote_exists = remote.cache_exists(md5s) progress.update_target(title, 90, 100) @@ -663,7 +665,7 @@ class RemoteLOCAL(RemoteBase): # NOTE: dummy call to try to establish a connection # to see if we need to ask user for a password. - remote.exists(remote.md5s_to_path_infos(['000'])) + remote.cache_exists(['000']) progress.update_target(title, 70, 100) @@ -711,7 +713,7 @@ class RemoteLOCAL(RemoteBase): # NOTE: filter files that are already uploaded md5s = [i[self.PARAM_MD5] for i in checksum_infos] - exists = remote.exists(remote.md5s_to_path_infos(md5s)) + exists = remote.cache_exists(md5s) progress.update_target(title, 30, 100) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 3242f42a9..633568fb4 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -66,6 +66,10 @@ class RemoteS3(RemoteBase): if shared_creds: os.environ.setdefault('AWS_SHARED_CREDENTIALS_FILE', shared_creds) + parsed = urlparse(self.url) + self.bucket = parsed.netloc + self.prefix = parsed.path.lstrip('/') + @staticmethod def compat_config(config): ret = config.copy() @@ -73,14 +77,6 @@ class RemoteS3(RemoteBase): ret[Config.SECTION_REMOTE_URL] = url return ret - @property - def bucket(self): - return urlparse(self.url).netloc - - @property - def prefix(self): - return urlparse(self.url).path.lstrip('/') - @property def s3(self): session = boto3.session.Session(profile_name=self.profile, @@ -106,7 +102,7 @@ class RemoteS3(RemoteBase): path_info['key'])} def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True etag = checksum_info.get(self.PARAM_ETAG, None) @@ -146,7 +142,7 @@ class RemoteS3(RemoteBase): cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key} if {self.PARAM_ETAG: etag} != self.save_info(cache): - if self.exists([cache])[0]: + if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) self.remove(cache) @@ -172,7 +168,7 @@ class RemoteS3(RemoteBase): logger.warn(msg.format(etag, self.to_string(path_info))) return - if self.exists([path_info])[0]: + if self.exists(path_info): msg = "Data '{}' exists. Removing before checkout." logger.warn(msg.format(self.to_string(path_info))) self.remove(path_info) @@ -202,10 +198,9 @@ class RemoteS3(RemoteBase): 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s] - def _all_keys(self, bucket, prefix): + def _list_keys(self, bucket, prefix): s3 = self.s3 - keys = [] kwargs = {'Bucket': bucket, 'Prefix': prefix} while True: @@ -219,7 +214,7 @@ class RemoteS3(RemoteBase): break for obj in contents: - keys.append(obj['Key']) + yield obj['Key'] token = resp.get('NextContinuationToken', None) if not token: @@ -227,24 +222,27 @@ class RemoteS3(RemoteBase): kwargs['ContinuationToken'] = token - return keys + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 's3' - def exists(self, path_infos): - ret = [] + keys = self._list_keys(path_info['bucket'], path_info['key']) + return any(path_info['key'] == key for key in keys) - if len(path_infos) == 0: - return ret + def cache_exists(self, md5s): + assert isinstance(md5s, list) - bucket = path_infos[0]['bucket'] + if len(md5s) == 0: + return [] - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - assert path_info['bucket'] == bucket - exists = False - key = path_info['key'] - if key in self._all_keys(bucket, key): - exists = True - ret.append(exists) + ret = len(md5s) * [False] + keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] + for key in self._list_keys(self.bucket, self.prefix): + for i, k in enumerate(keys): + if k == key: + ret[i] = True + + assert len(keys) == len(ret) == len(md5s) return ret @@ -342,8 +340,14 @@ class RemoteS3(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all(self): - keys = self._all_keys(self.bucket, self.prefix) - return [self._path_to_etag(key) for key in keys] + # NOTE: The list might be way too big(e.g. 100M entries, md5 for each + # is 32 bytes, so ~3200Mb list) and we don't really need all of it at + # the same time, so it makes sense to use a generator to gradually + # iterate over it, without keeping all of it in memory. + return ( + self._path_to_etag(key) + for key in self._list_keys(self.bucket, self.prefix) + ) def gc(self, cinfos): used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']] diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 755468728..922be2205 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -105,31 +105,33 @@ class RemoteSSH(RemoteBase): return ssh - def exists(self, path_infos): - ret = [] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'ssh' - if len(path_infos) == 0: - return ret - - host = path_infos[0]['host'] - user = path_infos[0]['user'] - port = path_infos[0]['port'] - ssh = self.ssh(host=host, - user=user, - port=port) - - for path_info in path_infos: - assert host == path_info['host'] - assert user == path_info['user'] - assert port == path_info['port'] + with self.ssh(path_info['host'], + path_info['user'], + path_info['port']) as ssh: try: self._exec(ssh, 'test -e {}'.format(path_info['path'])) exists = True except RemoteSSHCmdError: exists = False - ret.append(exists) - ssh.close() + return exists + + def cache_exists(self, md5s): + assert isinstance(md5s, list) + + if len(md5s) == 0: + return [] + + existing_md5s = self._all_md5s() + ret = len(md5s) * [False] + for existing_md5 in existing_md5s: + for i, md5 in enumerate(md5s): + if md5 == existing_md5: + ret[i] = True return ret @@ -245,7 +247,7 @@ class RemoteSSH(RemoteBase): cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) if {self.PARAM_MD5: md5} != self.save_info(cache): - if self.exists([cache])[0]: + if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) self.remove(cache) @@ -254,7 +256,7 @@ class RemoteSSH(RemoteBase): return False def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True md5 = checksum_info.get(self.PARAM_MD5, None) @@ -296,7 +298,7 @@ class RemoteSSH(RemoteBase): logger.warn(msg.format(md5, self.to_string(path_info))) return - if self.exists([path_info])[0]: + if self.exists(path_info): msg = "Data '{}' exists. Removing before checkout." logger.warn(msg.format(self.to_string(path_info))) self.remove(path_info) From b39625113086fcf6ba47ff88f0e4344791f90b96 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 19 Dec 2018 20:20:41 +0200 Subject: [PATCH 03/11] azure: remove unused code Signed-off-by: Ruslan Kuprieiev --- dvc/remote/azure.py | 60 --------------------------------------------- 1 file changed, 60 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 4b5ee2a68..f781dcfc2 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -72,66 +72,6 @@ class RemoteAzure(RemoteBase): self.__blob_service.create_container(self.bucket) return self.__blob_service -# FIXME: temporarily disabled because of the lack of test for external azure -# dependencies/outputs/cache. -# -# def _get_etag(self, bucket, key): -# try: -# blob = self.blob_service.get_blob_properties(bucket, key) -# return blob.properties.etag -# except Exception: -# return None -# -# def save_info(self, path_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# return {self.PARAM_ETAG: self._get_etag( -# path_info['bucket'], path_info['key'])} -# -# def save(self, path_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# etag = self._get_etag(path_info['bucket'], path_info['key']) -# dest_key = '{}/{}'.format(etag[0:2], etag[2:]) -# -# self._copy_blob(to_bucket=self.bucket, -# to_key=dest_key, -# from_bucket=path_info['bucket'], -# from_key=path_info['key']) -# -# return {self.PARAM_ETAG: etag} -# -# def changed(self, path_info, checksum_info): -# return checksum_info != self.save_info(path_info) -# -# def _copy_blob(self, to_bucket, to_key, from_bucket, from_key): -# source = self.blob_service.make_blob_url(from_bucket, from_key) -# -# copy = self.blob_service.copy_blob(to_bucket, to_key, source) -# -# if self.COPY_POLL_SECONDS <= 0: -# return -# -# while copy.status != 'success': -# time.sleep(self.COPY_POLL_SECONDS) -# copy = self.blob_service.get_blob_properties( -# to_bucket, to_key).properties.copy -# -# def checkout(self, path_info, checksum_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# etag = checksum_info.get(self.PARAM_ETAG, None) -# if not etag: -# return -# -# self._copy_blob(to_bucket=path_info['bucket'], -# to_key=path_info['key'], -# from_bucket=self.bucket, -# from_key='{}/{}'.format(etag[0:2], etag[2:])) -# def remove(self, path_info): if path_info['scheme'] != self.scheme: raise NotImplementedError From 933d524ebe57cedd5a25f6b6e814e69f8b9f4fef Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 13:32:12 +0200 Subject: [PATCH 04/11] remote: RFC: use 'path' instead of 'key' Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/gs.py | 6 ++-- dvc/dependency/s3.py | 6 ++-- dvc/remote/azure.py | 44 +++++++++++++------------- dvc/remote/gs.py | 75 ++++++++++++++++++++++---------------------- dvc/remote/s3.py | 75 +++++++++++++++++++++++--------------------- 5 files changed, 106 insertions(+), 100 deletions(-) diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index 2c0be604e..6f31f3d9e 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -17,9 +17,9 @@ class DependencyGS(DependencyS3): self.info = info self.remote = remote if remote else RemoteGS(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 'gs', 'bucket': bucket, - 'key': key} + 'path': path} diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 9eb2a48d0..d2f0c4eef 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -18,12 +18,12 @@ class DependencyS3(DependencyBase): self.remote = remote or RemoteS3(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 's3', 'bucket': bucket, - 'key': key} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index f781dcfc2..c5eac284f 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -77,18 +77,18 @@ class RemoteAzure(RemoteBase): raise NotImplementedError logger.debug('Removing azure://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - self.blob_service.delete_blob(path_info['bucket'], path_info['key']) + self.blob_service.delete_blob(path_info['bucket'], path_info['path']) def md5s_to_path_infos(self, md5s): return [{ 'scheme': self.scheme, 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, md5[0:2], md5[2:]) + 'path': posixpath.join(self.prefix, md5[0:2], md5[2:]) } for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): blob_service = self.blob_service next_marker = None while True: @@ -111,13 +111,15 @@ class RemoteAzure(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], + md5[2:]) for md5 in md5s] + for path in self._list_keys(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(ret) == len(keys) == len(md5s) + assert len(ret) == len(paths) == len(md5s) return ret @@ -132,10 +134,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = to_info['bucket'] - key = to_info['key'] + path = to_info['path'] logger.debug("Uploading '{}' to '{}/{}'".format( - from_info['path'], bucket, key)) + from_info['path'], bucket, path)) if not name: name = os.path.basename(from_info['path']) @@ -144,7 +146,7 @@ class RemoteAzure(RemoteBase): try: self.blob_service.create_blob_from_path( - bucket, key, from_info['path'], progress_callback=cb) + bucket, path, from_info['path'], progress_callback=cb) except Exception as ex: msg = "Failed to upload '{}'".format(from_info['path']) logger.warn(msg, ex) @@ -166,10 +168,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = from_info['bucket'] - key = from_info['key'] + path = from_info['path'] logger.debug("Downloading '{}/{}' to '{}'".format( - bucket, key, to_info['path'])) + bucket, path, to_info['path'])) tmp_file = self.tmp_file(to_info['path']) if not name: @@ -181,9 +183,9 @@ class RemoteAzure(RemoteBase): try: self.blob_service.get_blob_to_path( - bucket, key, tmp_file, progress_callback=cb) + bucket, path, tmp_file, progress_callback=cb) except Exception as exc: - msg = "Failed to download '{}/{}'".format(bucket, key) + msg = "Failed to download '{}/{}'".format(bucket, path) logger.warn(msg, exc) else: os.rename(tmp_file, to_info['path']) @@ -201,8 +203,8 @@ class RemoteAzure(RemoteBase): # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_etag(path) + for path in self._list_keys(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -216,9 +218,9 @@ class RemoteAzure(RemoteBase): if etag in used: continue path_info = {'scheme': self.scheme, - 'key': posixpath.join(self.prefix, - etag[0:2], - etag[2:]), + 'path': posixpath.join(self.prefix, + etag[0:2], + etag[2:]), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 78a9f75eb..f1a3fa7e8 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -48,11 +48,11 @@ class RemoteGS(RemoteBase): def gs(self): return storage.Client() - def get_md5(self, bucket, key): + def get_md5(self, bucket, path): import base64 import codecs - blob = self.gs.bucket(bucket).get_blob(key) + blob = self.gs.bucket(bucket).get_blob(path) if not blob: return None @@ -65,17 +65,17 @@ class RemoteGS(RemoteBase): raise NotImplementedError return {self.PARAM_MD5: self.get_md5(path_info['bucket'], - path_info['key'])} + path_info['path'])} @staticmethod def to_string(path_info): return "{}://{}/{}".format(path_info['scheme'], path_info['bucket'], - path_info['key']) + path_info['path']) def changed_cache(self, md5): - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} if {self.PARAM_MD5: md5} != self.save_info(cache): if self.exists(cache): @@ -102,23 +102,23 @@ class RemoteGS(RemoteBase): 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']) + blob = gs.bucket(from_info['bucket']).get_blob(from_info['path']) if not blob: msg = '{} doesn\'t exist in the cloud' - raise DvcException(msg.format(from_info['key'])) + raise DvcException(msg.format(from_info['path'])) bucket = self.gs.bucket(to_info['bucket']) bucket.copy_blob(blob, self.gs.bucket(to_info['bucket']), - new_name=to_info['key']) + new_name=to_info['path']) def save(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError - md5 = self.get_md5(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - to_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + md5 = self.get_md5(path_info['bucket'], path_info['path']) + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -151,8 +151,8 @@ class RemoteGS(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), md5)) - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -161,9 +161,9 @@ class RemoteGS(RemoteBase): raise NotImplementedError logger.debug("Removing gs://{}/{}".format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key']) + blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['path']) if not blob: return @@ -172,10 +172,10 @@ class RemoteGS(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': 'gs', 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): yield blob.name @@ -183,8 +183,8 @@ class RemoteGS(RemoteBase): assert not isinstance(path_info, list) assert path_info['scheme'] == 'gs' - keys = self._list_keys(path_info['bucket'], path_info['key']) - return any(path_info['key'] == key for key in keys) + paths = self._list_paths(path_info['bucket'], path_info['path']) + return any(path_info['path'] == path for path in paths) def cache_exists(self, md5s): assert isinstance(md5s, list) @@ -193,13 +193,14 @@ class RemoteGS(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], md5[2:]) for md5 in md5s] + for path in self._list_paths(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(ret) == len(keys) == len(md5s) + assert len(ret) == len(paths) == len(md5s) return ret @@ -217,7 +218,7 @@ class RemoteGS(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -226,13 +227,13 @@ class RemoteGS(RemoteBase): try: bucket = gs.bucket(to_info['bucket']) - blob = bucket.blob(to_info['key']) + blob = bucket.blob(to_info['path']) blob.upload_from_filename(from_info['path']) except Exception as exc: msg = "Failed to upload '{}' to '{}/{}'" logger.warn(msg.format(from_info['path'], to_info['bucket'], - to_info['key']), exc) + to_info['path']), exc) continue progress.finish_target(name) @@ -258,7 +259,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -268,19 +269,19 @@ class RemoteGS(RemoteBase): if not no_progress_bar: # percent_cb is not available for download_to_filename, so - # lets at least update progress at keypoints(start, finish) + # lets at least update progress at pathpoints(start, finish) progress.update_target(name, 0, None) self._makedirs(to_info['path']) try: bucket = gs.bucket(from_info['bucket']) - blob = bucket.get_blob(from_info['key']) + blob = bucket.get_blob(from_info['path']) blob.download_to_filename(tmp_file) except Exception as exc: msg = "Failed to download '{}/{}' to '{}'" logger.warn(msg.format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']), exc) continue @@ -299,8 +300,8 @@ class RemoteGS(RemoteBase): # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_md5(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_md5(path) + for path in self._list_paths(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -313,8 +314,8 @@ class RemoteGS(RemoteBase): continue path_info = {'scheme': 'gs', 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} self.remove(path_info) removed = True diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 633568fb4..0fa9869a8 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -86,11 +86,12 @@ class RemoteS3(RemoteBase): endpoint_url=self.endpoint_url, use_ssl=self.use_ssl) - def get_etag(self, bucket, key): + def get_etag(self, bucket, path): try: - obj = self.s3.head_object(Bucket=bucket, Key=key) + obj = self.s3.head_object(Bucket=bucket, Key=path) except Exception: - raise DvcException('s3://{}/{} does not exist'.format(bucket, key)) + raise DvcException('s3://{}/{} does not exist'.format(bucket, + path)) return obj['ETag'].strip('"') @@ -99,7 +100,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], - path_info['key'])} + path_info['path'])} def changed(self, path_info, checksum_info): if not self.exists(path_info): @@ -118,16 +119,16 @@ class RemoteS3(RemoteBase): s3 = s3 if s3 else self.s3 source = {'Bucket': from_info['bucket'], - 'Key': from_info['key']} - self.s3.copy(source, to_info['bucket'], to_info['key']) + 'Key': from_info['path']} + self.s3.copy(source, to_info['bucket'], to_info['path']) def save(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError - etag = self.get_etag(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + etag = self.get_etag(path_info['bucket'], path_info['path']) + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -135,11 +136,11 @@ class RemoteS3(RemoteBase): @staticmethod def to_string(path_info): - return "s3://{}/{}".format(path_info['bucket'], path_info['key']) + return "s3://{}/{}".format(path_info['bucket'], path_info['path']) def changed_cache(self, etag): - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} if {self.PARAM_ETAG: etag} != self.save_info(cache): if self.exists(cache): @@ -177,8 +178,8 @@ class RemoteS3(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), etag)) - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -187,18 +188,18 @@ class RemoteS3(RemoteBase): raise NotImplementedError logger.debug('Removing s3://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) self.s3.delete_object(Bucket=path_info['bucket'], - Key=path_info['key']) + Key=path_info['path']) def md5s_to_path_infos(self, md5s): return [{'scheme': self.scheme, 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): s3 = self.s3 kwargs = {'Bucket': bucket, @@ -226,8 +227,8 @@ class RemoteS3(RemoteBase): assert not isinstance(path_info, list) assert path_info['scheme'] == 's3' - keys = self._list_keys(path_info['bucket'], path_info['key']) - return any(path_info['key'] == key for key in keys) + paths = self._list_paths(path_info['bucket'], path_info['path']) + return any(path_info['path'] == path for path in paths) def cache_exists(self, md5s): assert isinstance(md5s, list) @@ -236,13 +237,15 @@ class RemoteS3(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], + md5[2:]) for md5 in md5s] + for path in self._list_paths(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(keys) == len(ret) == len(md5s) + assert len(paths) == len(ret) == len(md5s) return ret @@ -260,7 +263,7 @@ class RemoteS3(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -271,7 +274,7 @@ class RemoteS3(RemoteBase): try: s3.upload_file(from_info['path'], to_info['bucket'], - to_info['key'], + to_info['path'], Callback=cb) except Exception as exc: msg = "Failed to upload '{}'".format(from_info['path']) @@ -301,7 +304,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -317,16 +320,16 @@ class RemoteS3(RemoteBase): else: total = s3.head_object( Bucket=from_info['bucket'], - Key=from_info['key'])['ContentLength'] + Key=from_info['path'])['ContentLength'] cb = Callback(name, total) s3.download_file(from_info['bucket'], - from_info['key'], + from_info['path'], tmp_file, Callback=cb) except Exception as exc: msg = "Failed to download '{}/{}'".format(from_info['bucket'], - from_info['key']) + from_info['path']) logger.warn(msg, exc) continue @@ -345,8 +348,8 @@ class RemoteS3(RemoteBase): # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_etag(path) + for path in self._list_paths(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -358,8 +361,8 @@ class RemoteS3(RemoteBase): if etag in used_etags: continue path_info = {'scheme': 's3', - 'key': posixpath.join(self.prefix, - etag[0:2], etag[2:]), + 'path': posixpath.join(self.prefix, + etag[0:2], etag[2:]), 'bucket': self.bucket} self.remove(path_info) removed = True From 2a46247cb22459b88b56f5c6b40372d245cdf3ea Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 17:20:11 +0200 Subject: [PATCH 05/11] remote: RFC: unify checksum <-> path convertions Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/hdfs.py | 2 +- dvc/dependency/http.py | 2 +- dvc/dependency/local.py | 2 +- dvc/remote/azure.py | 17 +++---------- dvc/remote/base.py | 12 +++++++++ dvc/remote/gs.py | 22 ++++++---------- dvc/remote/hdfs.py | 56 +++++++++++++++++++---------------------- dvc/remote/http.py | 17 ++++++------- dvc/remote/local.py | 16 ++++++------ dvc/remote/s3.py | 23 ++++++----------- dvc/remote/ssh.py | 18 +++++-------- tests/test_cache.py | 4 +-- 12 files changed, 84 insertions(+), 107 deletions(-) diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index 4ea959309..81506ba97 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -21,7 +21,7 @@ class DependencyHDFS(DependencyBase): user = remote.user if remote else self.group('user') self.path_info = {'scheme': 'hdfs', 'user': user, - 'url': path} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index f05bcbed4..a2cccf454 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -21,7 +21,7 @@ class DependencyHTTP(DependencyBase): self.path_info = { 'scheme': urlparse(path).scheme, - 'url': path, + 'path': path, } def save(self): diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index b5671cbee..a9ffb6efc 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -30,7 +30,7 @@ class DependencyLOCAL(DependencyBase): path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/')) if not os.path.isabs(path): - path = self.remote.ospath(path) + path = self.remote.to_ospath(path) path = os.path.join(stage.cwd, path) self.path = os.path.abspath(os.path.normpath(path)) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index c5eac284f..515054e9b 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,7 +1,6 @@ from __future__ import absolute_import import os import re -import posixpath try: from azure.storage.blob import BlockBlobService @@ -85,7 +84,7 @@ class RemoteAzure(RemoteBase): return [{ 'scheme': self.scheme, 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, md5[0:2], md5[2:]) + 'path': self.checksum_to_path(md5[0:2]), } for md5 in md5s] def _list_paths(self, bucket, prefix): @@ -111,9 +110,7 @@ class RemoteAzure(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], - md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_keys(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -193,17 +190,13 @@ class RemoteAzure(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(path) + self.path_to_checksum(path) for path in self._list_keys(self.bucket, self.prefix) ) @@ -218,9 +211,7 @@ class RemoteAzure(RemoteBase): if etag in used: continue path_info = {'scheme': self.scheme, - 'path': posixpath.join(self.prefix, - etag[0:2], - etag[2:]), + 'path': self.checksum_to_path(etag), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 1d7f82128..46dcd5591 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,6 +1,7 @@ import os import re import errno +import posixpath from multiprocessing import cpu_count from dvc.config import Config @@ -150,3 +151,14 @@ class RemoteBase(object): assert len(names) == len(to_infos) return names + + @property + def ospath(self): + return posixpath + + def checksum_to_path(self, checksum): + return self.ospath.join(self.prefix, checksum[0:2], checksum[2:]) + + def path_to_checksum(self, path): + relpath = self.ospath.relpath(path, self.prefix) + return self.ospath.dirname(relpath) + self.ospath.basename(relpath) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index f1a3fa7e8..0c8687d88 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,5 +1,4 @@ import os -import posixpath try: from google.cloud import storage @@ -74,7 +73,7 @@ class RemoteGS(RemoteBase): path_info['path']) def changed_cache(self, md5): - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} if {self.PARAM_MD5: md5} != self.save_info(cache): @@ -117,7 +116,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError md5 = self.get_md5(path_info['bucket'], path_info['path']) - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -151,7 +150,7 @@ class RemoteGS(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), md5)) - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -172,8 +171,7 @@ class RemoteGS(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': 'gs', 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def _list_paths(self, bucket, prefix): for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): @@ -193,8 +191,7 @@ class RemoteGS(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_paths(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -290,17 +287,13 @@ class RemoteGS(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_md5s(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_md5(path) + self.path_to_checksum(path) for path in self._list_paths(self.bucket, self.prefix) ) @@ -314,8 +307,7 @@ class RemoteGS(RemoteBase): continue path_info = {'scheme': 'gs', 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': self.checksum_to_path(md5)} self.remove(path_info) removed = True diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 3b7158b2d..3db6402b8 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -22,7 +22,7 @@ class RemoteHDFS(RemoteBase): def __init__(self, project, config): self.project = project - self.url = config.get(Config.SECTION_REMOTE_URL, '/') + self.prefix = config.get(Config.SECTION_REMOTE_URL, '/') self.user = self.group('user') if not self.user: self.user = config.get(Config.SECTION_REMOTE_USER, @@ -59,38 +59,39 @@ class RemoteHDFS(RemoteBase): def checksum(self, path_info): regex = r'.*\t.*\t(?P.*)' - stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), + stdout = self.hadoop_fs('checksum {}'.format(path_info['path']), user=path_info['user']) return self._group(regex, stdout, 'checksum') def cp(self, from_info, to_info): - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), - user=to_info['user']) - self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']), + dname = posixpath.dirname(to_info['path']) + self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user']) + self.hadoop_fs('cp -f {} {}'.format(from_info['path'], + to_info['path']), user=to_info['user']) def rm(self, path_info): - self.hadoop_fs('rm {}'.format(path_info['url']), + self.hadoop_fs('rm {}'.format(path_info['path']), user=path_info['user']) def save_info(self, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') return {self.PARAM_CHECKSUM: self.checksum(path_info)} @staticmethod def to_string(path_info): return "{}://{}".format(path_info['scheme'], - path_info['url']) + path_info['path']) def changed_cache(self, checksum): cache = {} cache['scheme'] = 'hdfs' cache['user'] = self.user - cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + cache['path'] = self.checksum_to_path(checksum) if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): if self.exists(cache): @@ -118,11 +119,11 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') checksum = self.checksum(path_info) dest = path_info.copy() - dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + dest['path'] = self.checksum_to_path(checksum) self.cp(path_info, dest) @@ -132,7 +133,7 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') checksum = checksum_info.get(self.PARAM_CHECKSUM, None) if not checksum: @@ -158,7 +159,7 @@ class RemoteHDFS(RemoteBase): logger.info(msg.format(self.to_string(path_info), checksum)) src = path_info.copy() - src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + src['path'] = self.checksum_to_path(checksum) self.cp(src, path_info) @@ -166,24 +167,23 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') - logger.debug('Removing {}'.format(path_info['url'])) + logger.debug('Removing {}'.format(path_info['path'])) self.rm(path_info) def md5s_to_path_infos(self, md5s): return [{'scheme': 'hdfs', 'user': self.user, - 'url': posixpath.join(self.url, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] == 'hdfs' try: - self.hadoop_fs('test -e {}'.format(path_info['url'])) + self.hadoop_fs('test -e {}'.format(path_info['path'])) return True except RemoteHDFSCmdError: return False @@ -213,11 +213,11 @@ class RemoteHDFS(RemoteBase): if from_info['scheme'] != 'local': raise NotImplementedError - cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url'])) + cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['path'])) self.hadoop_fs(cmd, user=to_info['user']) cmd = 'copyFromLocal {} {}'.format(from_info['path'], - to_info['url']) + to_info['path']) self.hadoop_fs(cmd, user=to_info['user']) def download(self, @@ -242,27 +242,24 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path']) + cmd = 'copyToLocal {} {}'.format(from_info['path'], + to_info['path']) self.hadoop_fs(cmd, user=from_info['user']) - def _path_to_checksum(self, path): - relpath = posixpath.relpath(path, self.url) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_checksums(self): try: - self.hadoop_fs('test -e {}'.format(self.url)) + self.hadoop_fs('test -e {}'.format(self.prefix)) except RemoteHDFSCmdError: return [] - stdout = self.hadoop_fs('ls -R {}'.format(self.url)) + stdout = self.hadoop_fs('ls -R {}'.format(self.prefix)) lines = stdout.split('\n') flist = [] for line in lines: if not line.startswith('-'): continue flist.append(line.split()[-1]) - return [self._path_to_checksum(path) for path in flist] + return [self.path_to_checksum(path) for path in flist] def gc(self, cinfos): used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']] @@ -274,8 +271,7 @@ class RemoteHDFS(RemoteBase): continue path_info = {'scheme': 'hdfs', 'user': self.user, - 'url': posixpath.join(self.url, - checksum[0:2], checksum[2:])} + 'path': self.checksum_to_path(checksum)} self.remove(path_info) removed = True diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 4eb6c82a9..7b2cdc380 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -1,7 +1,6 @@ import os import threading import requests -import posixpath from dvc.logger import logger from dvc.progress import progress @@ -53,7 +52,7 @@ class RemoteHTTP(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - msg = "Downloading '{}' to '{}'".format(from_info['url'], + msg = "Downloading '{}' to '{}'".format(from_info['path'], to_info['path']) logger.debug(msg) @@ -63,7 +62,7 @@ class RemoteHTTP(RemoteBase): self._makedirs(to_info['path']) - total = self._content_length(from_info['url']) + total = self._content_length(from_info['path']) if no_progress_bar or not total: cb = None @@ -71,9 +70,9 @@ class RemoteHTTP(RemoteBase): cb = ProgressBarCallback(name, total) try: - self._download_to(from_info['url'], tmp_file, callback=cb) + self._download_to(from_info['path'], tmp_file, callback=cb) except Exception as exc: - msg = "Failed to download '{}'".format(from_info['url']) + msg = "Failed to download '{}'".format(from_info['path']) logger.warn(msg, exc) continue @@ -85,13 +84,13 @@ class RemoteHTTP(RemoteBase): def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] in ['http', 'https'] - return bool(self._request('HEAD', path_info.get('url'))) + return bool(self._request('HEAD', path_info.get('path'))) def cache_exists(self, md5s): assert isinstance(md5s, list) ret = [] for md5 in md5s: - url = posixpath.join(self.prefix, md5[0:2], md5[2:]) + url = self.checksum_to_path(md5) exists = bool(self._request('HEAD', url)) ret.append(exists) return ret @@ -100,13 +99,13 @@ class RemoteHTTP(RemoteBase): if path_info['scheme'] not in ['http', 'https']: raise NotImplementedError - return {self.PARAM_ETAG: self._etag(path_info['url'])} + return {self.PARAM_ETAG: self._etag(path_info['path'])} def md5s_to_path_infos(self, md5s): return [ { 'scheme': 'http', - 'url': posixpath.join(self.prefix, md5[0:2], md5[2:]), + 'path': self.checksum_to_path(md5), } for md5 in md5s ] diff --git a/dvc/remote/local.py b/dvc/remote/local.py index d6a01599f..089c01e5f 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -77,7 +77,7 @@ class RemoteLOCAL(RemoteBase): for cache in os.listdir(subdir): path = os.path.join(subdir, cache) - clist.append(self.path_to_md5(path)) + clist.append(self.path_to_checksum(path)) return clist @@ -87,10 +87,6 @@ class RemoteLOCAL(RemoteBase): return os.path.join(self.cache_dir, md5[0:2], md5[2:]) - def path_to_md5(self, path): - relpath = os.path.relpath(path, self.cache_dir) - return os.path.dirname(relpath) + os.path.basename(relpath) - def changed_cache_file(self, md5): cache = self.get(md5) if self.state.changed(cache, md5=md5): @@ -165,8 +161,14 @@ class RemoteLOCAL(RemoteBase): raise DvcException('No possible cache types left to try out.') + @property + def ospath(self): + if os.name == 'nt': + return ntpath + return posixpath + @classmethod - def ospath(cls, path): + def to_ospath(cls, path): if os.name == 'nt': return cls.ntpath(path) return cls.unixpath(path) @@ -242,7 +244,7 @@ class RemoteLOCAL(RemoteBase): return [] for info in d: - info['relpath'] = self.ospath(info['relpath']) + info['relpath'] = self.to_ospath(info['relpath']) return d diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 0fa9869a8..afe5ae7ee 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -1,6 +1,5 @@ import os import threading -import posixpath try: import boto3 @@ -127,7 +126,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError etag = self.get_etag(path_info['bucket'], path_info['path']) - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -139,7 +138,7 @@ class RemoteS3(RemoteBase): return "s3://{}/{}".format(path_info['bucket'], path_info['path']) def changed_cache(self, etag): - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} if {self.PARAM_ETAG: etag} != self.save_info(cache): @@ -178,7 +177,7 @@ class RemoteS3(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), etag)) - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -196,8 +195,7 @@ class RemoteS3(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': self.scheme, 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def _list_paths(self, bucket, prefix): s3 = self.s3 @@ -237,9 +235,7 @@ class RemoteS3(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], - md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_paths(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -338,17 +334,13 @@ class RemoteS3(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(path) + self.path_to_checksum(path) for path in self._list_paths(self.bucket, self.prefix) ) @@ -361,8 +353,7 @@ class RemoteS3(RemoteBase): if etag in used_etags: continue path_info = {'scheme': 's3', - 'path': posixpath.join(self.prefix, - etag[0:2], etag[2:]), + 'path': self.checksum_to_path(etag), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 922be2205..40b680393 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -78,8 +78,7 @@ class RemoteSSH(RemoteBase): 'host': self.host, 'user': self.user, 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def ssh(self, host=None, user=None, port=None): msg = "Establishing ssh connection with '{}' " \ @@ -244,7 +243,7 @@ class RemoteSSH(RemoteBase): cache['host'] = self.host cache['port'] = self.port cache['user'] = self.user - cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + cache['path'] = self.checksum_to_path(md5) if {self.PARAM_MD5: md5} != self.save_info(cache): if self.exists(cache): @@ -274,7 +273,7 @@ class RemoteSSH(RemoteBase): md5 = self.md5(path_info) dest = path_info.copy() - dest['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + dest['path'] = self.checksum_to_path(md5) self.cp(path_info, dest) @@ -308,7 +307,7 @@ class RemoteSSH(RemoteBase): logger.info(msg.format(self.to_string(path_info), md5)) src = path_info.copy() - src['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + src['path'] = self.checksum_to_path(md5) self.cp(src, path_info) @@ -418,10 +417,6 @@ class RemoteSSH(RemoteBase): sftp.close() ssh.close() - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_md5s(self): ssh = self.ssh(host=self.host, user=self.user, @@ -431,7 +426,7 @@ class RemoteSSH(RemoteBase): flist = stdout.split() ssh.close() - return [self._path_to_md5(path) for path in flist] + return [self.path_to_checksum(path) for path in flist] def gc(self, cinfos): used = [info[self.PARAM_MD5] for info in cinfos['ssh']] @@ -445,8 +440,7 @@ class RemoteSSH(RemoteBase): 'user': self.user, 'host': self.host, 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': self.checksum_to_path(md5)} self.remove(path_info) removed = True diff --git a/tests/test_cache.py b/tests/test_cache.py index 7a9dd3a53..fcad3717d 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -24,8 +24,8 @@ class TestCache(TestDvc): def test_all(self): md5_list = Cache(self.dvc).local.all() self.assertEquals(len(md5_list), 2) - self.assertTrue(self.cache1_md5 in md5_list) - self.assertTrue(self.cache2_md5 in md5_list) + self.assertIn(self.cache1_md5, md5_list) + self.assertIn(self.cache2_md5, md5_list) def test_get(self): cache = Cache(self.dvc).local.get(self.cache1_md5) From 141de1443c9a23333685983c567e7c9ab4831621 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 18:21:01 +0200 Subject: [PATCH 06/11] remote: RFC: unify gc code Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/__init__.py | 6 ++--- dvc/output/local.py | 2 +- dvc/project.py | 4 +-- dvc/remote/azure.py | 40 ++++------------------------ dvc/remote/base.py | 44 ++++++++++++++++++++++++++++--- dvc/remote/gs.py | 51 +++++++++--------------------------- dvc/remote/hdfs.py | 28 +++----------------- dvc/remote/http.py | 13 +++------- dvc/remote/local.py | 53 +++++++++++++------------------------- dvc/remote/s3.py | 51 +++++++++--------------------------- dvc/remote/ssh.py | 48 ++++++++++------------------------ dvc/stage.py | 4 +-- tests/test_cache.py | 2 +- tests/test_checkout.py | 6 ++--- tests/test_repro.py | 4 +-- tests/test_stage.py | 6 +++-- 16 files changed, 127 insertions(+), 235 deletions(-) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 7377f94f0..52c16bf30 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -39,11 +39,11 @@ DEP_MAP = { 'https': DependencyHTTP, } -# We are skipping RemoteHTTP.PARAM_ETAG because is the same as RemoteS3 +# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3 SCHEMA = { DependencyBase.PARAM_PATH: str, - schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None), - schema.Optional(RemoteS3.PARAM_ETAG): schema.Or(str, None), + schema.Optional(RemoteLOCAL.PARAM_CHECKSUM): schema.Or(str, None), + schema.Optional(RemoteS3.PARAM_CHECKSUM): schema.Or(str, None), schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None), } diff --git a/dvc/output/local.py b/dvc/output/local.py index d52bbaea5..6c1cf6541 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -34,7 +34,7 @@ class OutputLOCAL(DependencyLOCAL): @property def md5(self): - return self.info.get(self.project.cache.local.PARAM_MD5, None) + return self.info.get(self.project.cache.local.PARAM_CHECKSUM, None) @property def cache(self): diff --git a/dvc/project.py b/dvc/project.py index a48c32a26..35f4afe91 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -602,7 +602,7 @@ class Project(object): info = out.dumpd() ret = [info] r = out.remote - md5 = info[r.PARAM_MD5] + md5 = info[r.PARAM_CHECKSUM] if self.cache.local.changed_cache_file(md5): try: @@ -649,7 +649,7 @@ class Project(object): if out.path_info['scheme'] != 'local': return ret - md5 = info[out.remote.PARAM_MD5] + md5 = info[out.remote.PARAM_CHECKSUM] cache = self.cache.local.get(md5) if not out.remote.is_dir_cache(cache): return ret diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 515054e9b..6b1c91459 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -63,6 +63,8 @@ class RemoteAzure(RemoteBase): self.__blob_service = None + self.path_info = {'scheme': self.scheme, 'bucket': self.bucket} + @property def blob_service(self): if self.__blob_service is None: @@ -80,13 +82,6 @@ class RemoteAzure(RemoteBase): self.blob_service.delete_blob(path_info['bucket'], path_info['path']) - def md5s_to_path_infos(self, md5s): - return [{ - 'scheme': self.scheme, - 'bucket': self.bucket, - 'path': self.checksum_to_path(md5[0:2]), - } for md5 in md5s] - def _list_paths(self, bucket, prefix): blob_service = self.blob_service next_marker = None @@ -103,6 +98,9 @@ class RemoteAzure(RemoteBase): next_marker = blobs.next_marker + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) + def cache_exists(self, md5s): assert isinstance(md5s, list) @@ -189,31 +187,3 @@ class RemoteAzure(RemoteBase): if not no_progress_bar: progress.finish_target(name) - - def _all(self): - # NOTE: The list might be way too big(e.g. 100M entries, md5 for each - # is 32 bytes, so ~3200Mb list) and we don't really need all of it at - # the same time, so it makes sense to use a generator to gradually - # iterate over it, without keeping all of it in memory. - return ( - self.path_to_checksum(path) - for path in self._list_keys(self.bucket, self.prefix) - ) - - def gc(self, cinfos): - from dvc.remote.local import RemoteLOCAL - - used = [info[self.PARAM_ETAG] for info in cinfos['azure']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for etag in self._all(): - if etag in used: - continue - path_info = {'scheme': self.scheme, - 'path': self.checksum_to_path(etag), - 'bucket': self.bucket} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 46dcd5591..81c9ed10c 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -129,9 +129,6 @@ class RemoteBase(object): if e.errno != errno.EEXIST: raise - def md5s_to_path_infos(self, md5s): - raise NotImplementedError - def cache_exists(self, md5s): raise NotImplementedError @@ -162,3 +159,44 @@ class RemoteBase(object): def path_to_checksum(self, path): relpath = self.ospath.relpath(path, self.prefix) return self.ospath.dirname(relpath) + self.ospath.basename(relpath) + + def checksum_to_path_info(self, checksum): + path_info = self.path_info.copy() + path_info['path'] = self.checksum_to_path(checksum) + return path_info + + def md5s_to_path_infos(self, md5s): + return [ + self.checksum_to_path_info(md5) + for md5 in md5s + ] + + def list_cache_paths(self): + raise NotImplementedError + + def all(self): + # NOTE: The list might be way too big(e.g. 100M entries, md5 for each + # is 32 bytes, so ~3200Mb list) and we don't really need all of it at + # the same time, so it makes sense to use a generator to gradually + # iterate over it, without keeping all of it in memory. + return ( + self.path_to_checksum(path) + for path in self.list_cache_paths() + ) + + def gc(self, cinfos): + from dvc.remote.local import RemoteLOCAL + + used = [info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos['local']] + + if self.scheme != '': + used += [info[self.PARAM_CHECKSUM] for info in cinfos[self.scheme]] + + removed = False + for checksum in self.all(): + if checksum in used: + continue + path_info = self.checksum_to_path_info(checksum) + self.remove(path_info) + removed = True + return removed diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 0c8687d88..23a4d145e 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -12,7 +12,6 @@ except ImportError: from dvc.logger import logger from dvc.remote.base import RemoteBase -from dvc.remote.local import RemoteLOCAL from dvc.config import Config from dvc.progress import progress from dvc.exceptions import DvcException @@ -22,7 +21,7 @@ class RemoteGS(RemoteBase): scheme = 'gs' REGEX = r'^gs://(?P.*)$' REQUIRES = {'google.cloud.storage': storage} - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' def __init__(self, project, config): self.project = project @@ -36,6 +35,8 @@ class RemoteGS(RemoteBase): self.bucket = parsed.netloc self.prefix = parsed.path.lstrip('/') + self.path_info = {'scheme': 'gs', 'bucket': self.bucket} + @staticmethod def compat_config(config): ret = config.copy() @@ -63,8 +64,8 @@ class RemoteGS(RemoteBase): if path_info['scheme'] != 'gs': raise NotImplementedError - return {self.PARAM_MD5: self.get_md5(path_info['bucket'], - path_info['path'])} + return {self.PARAM_CHECKSUM: self.get_md5(path_info['bucket'], + path_info['path'])} @staticmethod def to_string(path_info): @@ -76,7 +77,7 @@ class RemoteGS(RemoteBase): path = self.checksum_to_path(md5) cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - if {self.PARAM_MD5: md5} != self.save_info(cache): + if {self.PARAM_CHECKSUM: md5} != self.save_info(cache): if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) @@ -89,7 +90,7 @@ class RemoteGS(RemoteBase): if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -121,13 +122,13 @@ class RemoteGS(RemoteBase): self._copy(path_info, to_info) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def checkout(self, path_info, checksum_info): if path_info['scheme'] != 'gs': raise NotImplementedError - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if not md5: return @@ -168,15 +169,13 @@ class RemoteGS(RemoteBase): blob.delete() - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'gs', - 'bucket': self.bucket, - 'path': self.checksum_to_path(md5)} for md5 in md5s] - def _list_paths(self, bucket, prefix): for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): yield blob.name + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) + def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] == 'gs' @@ -286,29 +285,3 @@ class RemoteGS(RemoteBase): if not no_progress_bar: progress.finish_target(name) - - def _all_md5s(self): - # NOTE: The list might be way too big(e.g. 100M entries, md5 for each - # is 32 bytes, so ~3200Mb list) and we don't really need all of it at - # the same time, so it makes sense to use a generator to gradually - # iterate over it, without keeping all of it in memory. - return ( - self.path_to_checksum(path) - for path in self._list_paths(self.bucket, self.prefix) - ) - - def gc(self, cinfos): - used = [info[self.PARAM_MD5] for info in cinfos['gs']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for md5 in self._all_md5s(): - if md5 in used: - continue - path_info = {'scheme': 'gs', - 'bucket': self.bucket, - 'path': self.checksum_to_path(md5)} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 3db6402b8..513303259 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -6,7 +6,6 @@ from subprocess import Popen, PIPE from dvc.config import Config from dvc.remote.base import RemoteBase, RemoteBaseCmdError -from dvc.remote.local import RemoteLOCAL from dvc.logger import logger from dvc.utils import fix_env @@ -28,6 +27,8 @@ class RemoteHDFS(RemoteBase): self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) + self.path_info = {'scheme': 'hdfs', 'user': self.user} + def hadoop_fs(self, cmd, user=None): cmd = 'hadoop fs -' + cmd if user: @@ -173,11 +174,6 @@ class RemoteHDFS(RemoteBase): self.rm(path_info) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'hdfs', - 'user': self.user, - 'path': self.checksum_to_path(md5)} for md5 in md5s] - def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] == 'hdfs' @@ -246,7 +242,7 @@ class RemoteHDFS(RemoteBase): to_info['path']) self.hadoop_fs(cmd, user=from_info['user']) - def _all_checksums(self): + def list_cache_paths(self): try: self.hadoop_fs('test -e {}'.format(self.prefix)) except RemoteHDFSCmdError: @@ -259,20 +255,4 @@ class RemoteHDFS(RemoteBase): if not line.startswith('-'): continue flist.append(line.split()[-1]) - return [self.path_to_checksum(path) for path in flist] - - def gc(self, cinfos): - used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for checksum in self._all_checksums(): - if checksum in used: - continue - path_info = {'scheme': 'hdfs', - 'user': self.user, - 'path': self.checksum_to_path(checksum)} - self.remove(path_info) - removed = True - - return removed + return flist diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 7b2cdc380..7147b9dd1 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -33,6 +33,7 @@ class RemoteHTTP(RemoteBase): self.project = project self.cache_dir = config.get(Config.SECTION_REMOTE_URL) self.url = self.cache_dir + self.path_info = {'scheme': 'http'} @property def prefix(self): @@ -101,15 +102,6 @@ class RemoteHTTP(RemoteBase): return {self.PARAM_ETAG: self._etag(path_info['path'])} - def md5s_to_path_infos(self, md5s): - return [ - { - 'scheme': 'http', - 'path': self.checksum_to_path(md5), - } - for md5 in md5s - ] - def _content_length(self, url): return self._request('HEAD', url).headers.get('Content-Length') @@ -148,3 +140,6 @@ class RemoteHTTP(RemoteBase): return requests.request(method, url, **kwargs) except requests.exceptions.RequestException: raise DvcException('Could not perform a {} request'.format(method)) + + def gc(self): + raise NotImplementedError diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 089c01e5f..ab9604352 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -22,7 +22,7 @@ from concurrent.futures import ThreadPoolExecutor class RemoteLOCAL(RemoteBase): scheme = '' REGEX = r'^(?P(/+|.:\\+).*)$' - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' PARAM_PATH = 'path' PARAM_RELPATH = 'relpath' MD5_DIR_SUFFIX = '.dir' @@ -53,6 +53,8 @@ class RemoteLOCAL(RemoteBase): if self.cache_dir is not None and not os.path.exists(self.cache_dir): os.mkdir(self.cache_dir) + self.path_info = {'scheme': 'local'} + @staticmethod def compat_config(config): ret = config.copy() @@ -68,7 +70,7 @@ class RemoteLOCAL(RemoteBase): def prefix(self): return self.cache_dir - def all(self): + def list_cache_paths(self): clist = [] for entry in os.listdir(self.cache_dir): subdir = os.path.join(self.cache_dir, entry) @@ -76,8 +78,7 @@ class RemoteLOCAL(RemoteBase): continue for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(self.path_to_checksum(path)) + clist.append(os.path.join(subdir, cache)) return clist @@ -118,7 +119,7 @@ class RemoteLOCAL(RemoteBase): continue for entry in self.load_dir_cache(md5): - md5 = entry[self.PARAM_MD5] + md5 = entry[self.PARAM_CHECKSUM] cache = self.get(md5) clist.append((cache, md5)) @@ -211,7 +212,7 @@ class RemoteLOCAL(RemoteBase): md5 = self.state.update(path) dir_info.append({self.PARAM_RELPATH: relpath, - self.PARAM_MD5: md5}) + self.PARAM_CHECKSUM: md5}) if bar: progress.finish_target(title) @@ -271,7 +272,7 @@ class RemoteLOCAL(RemoteBase): def checkout(self, path_info, checksum_info, force=False): path = path_info['path'] - md5 = checksum_info.get(self.PARAM_MD5) + md5 = checksum_info.get(self.PARAM_CHECKSUM) cache = self.get(md5) if not cache: @@ -318,13 +319,13 @@ class RemoteLOCAL(RemoteBase): for processed, entry in enumerate(dir_info): relpath = entry[self.PARAM_RELPATH] - m = entry[self.PARAM_MD5] + m = entry[self.PARAM_CHECKSUM] p = os.path.join(path, relpath) c = self.get(m) entry_info = {'scheme': path_info['scheme'], self.PARAM_PATH: p} - entry_checksum_info = {self.PARAM_MD5: m} + entry_checksum_info = {self.PARAM_CHECKSUM: m} if self.changed(entry_info, entry_checksum_info): if os.path.exists(p): @@ -409,7 +410,7 @@ class RemoteLOCAL(RemoteBase): self.link(cache, path) self.state.update_link(path) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def _save_dir(self, path_info): path = path_info['path'] @@ -422,7 +423,7 @@ class RemoteLOCAL(RemoteBase): for processed, entry in enumerate(dir_info): relpath = entry[self.PARAM_RELPATH] - m = entry[self.PARAM_MD5] + m = entry[self.PARAM_CHECKSUM] p = os.path.join(path, relpath) c = self.get(m) @@ -441,7 +442,7 @@ class RemoteLOCAL(RemoteBase): if bar: progress.finish_target(dir_relpath) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def save(self, path_info): if path_info['scheme'] != 'local': @@ -462,7 +463,7 @@ class RemoteLOCAL(RemoteBase): if path_info['scheme'] != 'local': raise NotImplementedError - return {self.PARAM_MD5: self.state.update(path_info['path'])} + return {self.PARAM_CHECKSUM: self.state.update(path_info['path'])} def changed(self, path_info, checksum_info): """ @@ -475,7 +476,7 @@ class RemoteLOCAL(RemoteBase): if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -496,11 +497,6 @@ class RemoteLOCAL(RemoteBase): move(from_info['path'], to_info['path']) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'local', - 'path': os.path.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] - def cache_exists(self, md5s): assert isinstance(md5s, list) ret = [] @@ -573,7 +569,7 @@ class RemoteLOCAL(RemoteBase): by_md5 = {} for info in checksum_infos: - md5 = info[self.PARAM_MD5] + md5 = info[self.PARAM_CHECKSUM] if show_checksums: by_md5[md5] = md5 @@ -593,19 +589,6 @@ class RemoteLOCAL(RemoteBase): return list(by_md5.keys()), list(by_md5.values()) - def gc(self, checksum_infos): - checksum_infos = checksum_infos['local'] - used_md5s = [info[self.PARAM_MD5] for info in checksum_infos] - - removed = False - for md5 in self.all(): - if md5 in used_md5s: - continue - remove(self.get(md5)) - removed = True - - return removed - def status(self, checksum_infos, remote, jobs=None, show_checksums=False): logger.info("Preparing to pull data from {}".format(remote.url)) title = "Collecting information" @@ -708,13 +691,13 @@ class RemoteLOCAL(RemoteBase): # NOTE: verifying that our cache is not corrupted def func(info): - return not self.changed_cache_file(info[self.PARAM_MD5]) + return not self.changed_cache_file(info[self.PARAM_CHECKSUM]) checksum_infos = list(filter(func, checksum_infos)) progress.update_target(title, 20, 100) # NOTE: filter files that are already uploaded - md5s = [i[self.PARAM_MD5] for i in checksum_infos] + md5s = [i[self.PARAM_CHECKSUM] for i in checksum_infos] exists = remote.cache_exists(md5s) progress.update_target(title, 30, 100) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index afe5ae7ee..8851fb072 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -15,7 +15,6 @@ from dvc.logger import logger from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBase -from dvc.remote.local import RemoteLOCAL from dvc.exceptions import DvcException @@ -36,7 +35,7 @@ class RemoteS3(RemoteBase): scheme = 's3' REGEX = r'^s3://(?P.*)$' REQUIRES = {'boto3': boto3} - PARAM_ETAG = 'etag' + PARAM_CHECKSUM = 'etag' def __init__(self, project, config): self.project = project @@ -69,6 +68,8 @@ class RemoteS3(RemoteBase): self.bucket = parsed.netloc self.prefix = parsed.path.lstrip('/') + self.path_info = {'scheme': self.scheme, 'bucket': self.bucket} + @staticmethod def compat_config(config): ret = config.copy() @@ -98,14 +99,14 @@ class RemoteS3(RemoteBase): if path_info['scheme'] != 's3': raise NotImplementedError - return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], - path_info['path'])} + return {self.PARAM_CHECKSUM: self.get_etag(path_info['bucket'], + path_info['path'])} def changed(self, path_info, checksum_info): if not self.exists(path_info): return True - etag = checksum_info.get(self.PARAM_ETAG, None) + etag = checksum_info.get(self.PARAM_CHECKSUM, None) if etag is None: return True @@ -131,7 +132,7 @@ class RemoteS3(RemoteBase): self._copy(path_info, to_info) - return {self.PARAM_ETAG: etag} + return {self.PARAM_CHECKSUM: etag} @staticmethod def to_string(path_info): @@ -141,7 +142,7 @@ class RemoteS3(RemoteBase): path = self.checksum_to_path(etag) cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - if {self.PARAM_ETAG: etag} != self.save_info(cache): + if {self.PARAM_CHECKSUM: etag} != self.save_info(cache): if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) @@ -154,7 +155,7 @@ class RemoteS3(RemoteBase): if path_info['scheme'] != 's3': raise NotImplementedError - etag = checksum_info.get(self.PARAM_ETAG, None) + etag = checksum_info.get(self.PARAM_CHECKSUM, None) if not etag: return @@ -192,11 +193,6 @@ class RemoteS3(RemoteBase): self.s3.delete_object(Bucket=path_info['bucket'], Key=path_info['path']) - def md5s_to_path_infos(self, md5s): - return [{'scheme': self.scheme, - 'bucket': self.bucket, - 'path': self.checksum_to_path(md5)} for md5 in md5s] - def _list_paths(self, bucket, prefix): s3 = self.s3 @@ -221,6 +217,9 @@ class RemoteS3(RemoteBase): kwargs['ContinuationToken'] = token + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) + def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] == 's3' @@ -333,29 +332,3 @@ class RemoteS3(RemoteBase): if not no_progress_bar: progress.finish_target(name) - - def _all(self): - # NOTE: The list might be way too big(e.g. 100M entries, md5 for each - # is 32 bytes, so ~3200Mb list) and we don't really need all of it at - # the same time, so it makes sense to use a generator to gradually - # iterate over it, without keeping all of it in memory. - return ( - self.path_to_checksum(path) - for path in self._list_paths(self.bucket, self.prefix) - ) - - def gc(self, cinfos): - used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']] - used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for etag in self._all(): - if etag in used_etags: - continue - path_info = {'scheme': 's3', - 'path': self.checksum_to_path(etag), - 'bucket': self.bucket} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 40b680393..1e04d9dc6 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -11,7 +11,6 @@ import dvc.prompt as prompt from dvc.logger import logger from dvc.progress import progress from dvc.remote.base import RemoteBase, RemoteBaseCmdError -from dvc.remote.local import RemoteLOCAL from dvc.config import Config from dvc.exceptions import DvcException @@ -52,7 +51,7 @@ class RemoteSSH(RemoteBase): JOBS = 4 - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' DEFAULT_PORT = 22 TIMEOUT = 1800 @@ -73,12 +72,10 @@ class RemoteSSH(RemoteBase): self.ask_password = config.get(Config.SECTION_REMOTE_ASK_PASSWORD, False) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'ssh', - 'host': self.host, - 'user': self.user, - 'port': self.port, - 'path': self.checksum_to_path(md5)} for md5 in md5s] + self.path_info = {'scheme': 'ssh', + 'host': self.host, + 'user': self.user, + 'port': self.port} def ssh(self, host=None, user=None, port=None): msg = "Establishing ssh connection with '{}' " \ @@ -125,7 +122,7 @@ class RemoteSSH(RemoteBase): if len(md5s) == 0: return [] - existing_md5s = self._all_md5s() + existing_md5s = self.all() ret = len(md5s) * [False] for existing_md5 in existing_md5s: for i, md5 in enumerate(md5s): @@ -228,7 +225,7 @@ class RemoteSSH(RemoteBase): if path_info['scheme'] != 'ssh': raise NotImplementedError - return {self.PARAM_MD5: self.md5(path_info)} + return {self.PARAM_CHECKSUM: self.md5(path_info)} @staticmethod def to_string(path_info): @@ -245,7 +242,7 @@ class RemoteSSH(RemoteBase): cache['user'] = self.user cache['path'] = self.checksum_to_path(md5) - if {self.PARAM_MD5: md5} != self.save_info(cache): + if {self.PARAM_CHECKSUM: md5} != self.save_info(cache): if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(self.to_string(cache))) @@ -258,7 +255,7 @@ class RemoteSSH(RemoteBase): if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -277,13 +274,13 @@ class RemoteSSH(RemoteBase): self.cp(path_info, dest) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def checkout(self, path_info, checksum_info): if path_info['scheme'] != 'ssh': raise NotImplementedError - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if not md5: return @@ -417,7 +414,7 @@ class RemoteSSH(RemoteBase): sftp.close() ssh.close() - def _all_md5s(self): + def list_cache_paths(self): ssh = self.ssh(host=self.host, user=self.user, port=self.port) @@ -425,23 +422,4 @@ class RemoteSSH(RemoteBase): stdout = self._exec(ssh, cmd) flist = stdout.split() ssh.close() - - return [self.path_to_checksum(path) for path in flist] - - def gc(self, cinfos): - used = [info[self.PARAM_MD5] for info in cinfos['ssh']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for md5 in self._all_md5s(): - if md5 in used: - continue - path_info = {'scheme': 'ssh', - 'user': self.user, - 'host': self.host, - 'port': self.port, - 'path': self.checksum_to_path(md5)} - self.remove(path_info) - removed = True - - return removed + return flist diff --git a/dvc/stage.py b/dvc/stage.py index e9a85341b..761378600 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -348,8 +348,8 @@ class Stage(object): new_d.pop(self.PARAM_MD5, None) outs = old_d.get(self.PARAM_OUTS, []) for out in outs: - out.pop(RemoteLOCAL.PARAM_MD5, None) - out.pop(RemoteS3.PARAM_ETAG, None) + out.pop(RemoteLOCAL.PARAM_CHECKSUM, None) + out.pop(RemoteS3.PARAM_CHECKSUM, None) return old_d == new_d diff --git a/tests/test_cache.py b/tests/test_cache.py index fcad3717d..17fb37569 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -22,7 +22,7 @@ class TestCache(TestDvc): self.create(self.cache2, '2') def test_all(self): - md5_list = Cache(self.dvc).local.all() + md5_list = list(Cache(self.dvc).local.all()) self.assertEquals(len(md5_list), 2) self.assertIn(self.cache1_md5, md5_list) self.assertIn(self.cache2_md5, md5_list) diff --git a/tests/test_checkout.py b/tests/test_checkout.py index da6d496c4..706b05236 100644 --- a/tests/test_checkout.py +++ b/tests/test_checkout.py @@ -81,7 +81,7 @@ class TestCheckoutCorruptedCacheDir(TestDvc): # NOTE: modifying cache file for one of the files inside the directory # to check if dvc will detect that the cache is corrupted. entry = self.dvc.cache.local.load_dir_cache(out.md5)[0] - md5 = entry[self.dvc.cache.local.PARAM_MD5] + md5 = entry[self.dvc.cache.local.PARAM_CHECKSUM] cache = self.dvc.cache.local.get(md5) with open(cache, 'w+') as fobj: @@ -281,8 +281,8 @@ class TestCheckoutMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_repro.py b/tests/test_repro.py index f6f60e637..460863ae0 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -544,8 +544,8 @@ class TestReproMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_stage.py b/tests/test_stage.py index f439641c1..ba227c5d0 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -52,8 +52,10 @@ class TestSchemaDepsOuts(TestSchema): Stage.validate(d) def test_list(self): - lst = [{OutputLOCAL.PARAM_PATH: 'foo', RemoteLOCAL.PARAM_MD5: '123'}, - {OutputLOCAL.PARAM_PATH: 'bar', RemoteLOCAL.PARAM_MD5: None}, + lst = [{OutputLOCAL.PARAM_PATH: 'foo', + RemoteLOCAL.PARAM_CHECKSUM: '123'}, + {OutputLOCAL.PARAM_PATH: 'bar', + RemoteLOCAL.PARAM_CHECKSUM: None}, {OutputLOCAL.PARAM_PATH: 'baz'}] d = {Stage.PARAM_DEPS: lst} Stage.validate(d) From 7758e53b35eb21606a11324f098c6504b751de56 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 18:29:43 +0200 Subject: [PATCH 07/11] remote: RFC: unify changed_cache Signed-off-by: Ruslan Kuprieiev --- dvc/remote/base.py | 12 ++++++++++++ dvc/remote/gs.py | 13 ------------- dvc/remote/hdfs.py | 15 --------------- dvc/remote/s3.py | 13 ------------- dvc/remote/ssh.py | 17 ----------------- 5 files changed, 12 insertions(+), 58 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 81c9ed10c..2b63eb9e4 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -200,3 +200,15 @@ class RemoteBase(object): self.remove(path_info) removed = True return removed + + def changed_cache(self, checksum): + cache = self.checksum_to_path_info(checksum) + + if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): + if self.exists(cache): + msg = 'Corrupted cache file {}' + logger.warn(msg.format(str(cache))) + self.remove(cache) + return True + + return False diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 23a4d145e..291ae5b8b 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -73,19 +73,6 @@ class RemoteGS(RemoteBase): path_info['bucket'], path_info['path']) - def changed_cache(self, md5): - path = self.checksum_to_path(md5) - cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - - if {self.PARAM_CHECKSUM: md5} != self.save_info(cache): - if self.exists(cache): - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 513303259..2cdf64a59 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -88,21 +88,6 @@ class RemoteHDFS(RemoteBase): return "{}://{}".format(path_info['scheme'], path_info['path']) - def changed_cache(self, checksum): - cache = {} - cache['scheme'] = 'hdfs' - cache['user'] = self.user - cache['path'] = self.checksum_to_path(checksum) - - if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): - if self.exists(cache): - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 8851fb072..849b76a04 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -138,19 +138,6 @@ class RemoteS3(RemoteBase): def to_string(path_info): return "s3://{}/{}".format(path_info['bucket'], path_info['path']) - def changed_cache(self, etag): - path = self.checksum_to_path(etag) - cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - - if {self.PARAM_CHECKSUM: etag} != self.save_info(cache): - if self.exists(cache): - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - def checkout(self, path_info, checksum_info): if path_info['scheme'] != 's3': raise NotImplementedError diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 1e04d9dc6..439c6912d 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -234,23 +234,6 @@ class RemoteSSH(RemoteBase): path_info['host'], path_info['path']) - def changed_cache(self, md5): - cache = {} - cache['scheme'] = 'ssh' - cache['host'] = self.host - cache['port'] = self.port - cache['user'] = self.user - cache['path'] = self.checksum_to_path(md5) - - if {self.PARAM_CHECKSUM: md5} != self.save_info(cache): - if self.exists(cache): - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True From ecb80d98dd9148d997ac829c33f1b692d426e86b Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 19:05:02 +0200 Subject: [PATCH 08/11] remote: RFC: unify cache_exists() Signed-off-by: Ruslan Kuprieiev --- dvc/remote/azure.py | 17 ----------------- dvc/remote/base.py | 21 ++++++++++++++++++--- dvc/remote/gs.py | 17 ----------------- dvc/remote/hdfs.py | 18 ++---------------- dvc/remote/s3.py | 17 ----------------- dvc/remote/ssh.py | 15 --------------- 6 files changed, 20 insertions(+), 85 deletions(-) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 6b1c91459..482c254dd 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -101,23 +101,6 @@ class RemoteAzure(RemoteBase): def list_cache_paths(self): return self._list_paths(self.bucket, self.prefix) - def cache_exists(self, md5s): - assert isinstance(md5s, list) - - if len(md5s) == 0: - return [] - - ret = len(md5s) * [False] - paths = [self.checksum_to_path(md5) for md5 in md5s] - for path in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(paths): - if k == path: - ret[i] = True - - assert len(ret) == len(paths) == len(md5s) - - return ret - def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 2b63eb9e4..9d67ce099 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -129,9 +129,6 @@ class RemoteBase(object): if e.errno != errno.EEXIST: raise - def cache_exists(self, md5s): - raise NotImplementedError - def exists(self, path_infos): raise NotImplementedError @@ -212,3 +209,21 @@ class RemoteBase(object): return True return False + + def cache_exists(self, checksums): + # NOTE: The reason for such an odd logic is that most of the remotes + # take much shorter time to just retrieve everything they have under + # a certain prefix(e.g. s3, gs, ssh, hdfs). Other remotes that can + # check if particular file exists much quicker, use their own + # implementation of cache_exists(see http, local). + # + # Result of all() might be way too big, so we should walk through + # it in one pass. Also currently, cache_exists() should return + # a list of True/False that matches order in checksums list, + # so we need to use such an ugly logic. + ret = len(checksums) * [False] + for existing in self.all(): + for i, checksum in enumerate(checksums): + if checksum == existing: + ret[i] = True + return ret diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 291ae5b8b..387221773 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -170,23 +170,6 @@ class RemoteGS(RemoteBase): paths = self._list_paths(path_info['bucket'], path_info['path']) return any(path_info['path'] == path for path in paths) - def cache_exists(self, md5s): - assert isinstance(md5s, list) - - if len(md5s) == 0: - return [] - - ret = len(md5s) * [False] - paths = [self.checksum_to_path(md5) for md5 in md5s] - for path in self._list_paths(self.bucket, self.prefix): - for i, k in enumerate(paths): - if k == path: - ret[i] = True - - assert len(ret) == len(paths) == len(md5s) - - return ret - def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 2cdf64a59..5eca0ba1c 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -21,7 +21,8 @@ class RemoteHDFS(RemoteBase): def __init__(self, project, config): self.project = project - self.prefix = config.get(Config.SECTION_REMOTE_URL, '/') + self.url = config.get(Config.SECTION_REMOTE_URL, '/') + self.prefix = self.url self.user = self.group('user') if not self.user: self.user = config.get(Config.SECTION_REMOTE_USER, @@ -169,21 +170,6 @@ class RemoteHDFS(RemoteBase): except RemoteHDFSCmdError: return False - def cache_exists(self, md5s): - assert isinstance(md5s, list) - - if len(md5s) == 0: - return [] - - existing_md5s = self._all_checksums() - ret = len(md5s) * [False] - for existing_md5 in existing_md5s: - for i, md5 in enumerate(md5s): - if md5 == existing_md5: - ret[i] = True - - return ret - def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 849b76a04..3fbacb4ca 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -214,23 +214,6 @@ class RemoteS3(RemoteBase): paths = self._list_paths(path_info['bucket'], path_info['path']) return any(path_info['path'] == path for path in paths) - def cache_exists(self, md5s): - assert isinstance(md5s, list) - - if len(md5s) == 0: - return [] - - ret = len(md5s) * [False] - paths = [self.checksum_to_path(md5) for md5 in md5s] - for path in self._list_paths(self.bucket, self.prefix): - for i, k in enumerate(paths): - if k == path: - ret[i] = True - - assert len(paths) == len(ret) == len(md5s) - - return ret - def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 439c6912d..b7da48978 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -116,21 +116,6 @@ class RemoteSSH(RemoteBase): return exists - def cache_exists(self, md5s): - assert isinstance(md5s, list) - - if len(md5s) == 0: - return [] - - existing_md5s = self.all() - ret = len(md5s) * [False] - for existing_md5 in existing_md5s: - for i, md5 in enumerate(md5s): - if md5 == existing_md5: - ret[i] = True - - return ret - def _exec(self, ssh, cmd): stdin, stdout, stderr = ssh.exec_command(cmd) channel = stdout.channel From ef766fb4d394e1f1fdc518678dabb3eb57202c16 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 22:00:04 +0200 Subject: [PATCH 09/11] remote: RFC: unify checkout() Signed-off-by: Ruslan Kuprieiev --- dvc/remote/base.py | 76 ++++++++++++++++++++++++++++++++++++++++++--- dvc/remote/gs.py | 44 ++------------------------ dvc/remote/hdfs.py | 45 ++------------------------- dvc/remote/local.py | 59 ++++++----------------------------- dvc/remote/s3.py | 42 ++----------------------- dvc/remote/ssh.py | 45 ++------------------------- 6 files changed, 93 insertions(+), 218 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 9d67ce099..8eb18829e 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -4,6 +4,7 @@ import errno import posixpath from multiprocessing import cpu_count +import dvc.prompt as prompt from dvc.config import Config from dvc.logger import logger from dvc.exceptions import DvcException @@ -105,9 +106,6 @@ class RemoteBase(object): def save(self, path_info): raise NotImplementedError - def checkout(self, path_info, checksum_info, force=False): - raise NotImplementedError - def download(self, from_infos, to_infos, no_progress_bar=False, name=None): raise NotImplementedError @@ -120,6 +118,9 @@ class RemoteBase(object): def move(self, path_info): raise NotImplementedError + def copy(self, from_info, to_info): + raise NotImplementedError + def _makedirs(self, fname): dname = os.path.dirname(fname) @@ -200,8 +201,13 @@ class RemoteBase(object): def changed_cache(self, checksum): cache = self.checksum_to_path_info(checksum) + expected = {self.PARAM_CHECKSUM: checksum} + actual = self.save_info(cache) - if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): + logger.debug("Cache '{}' actual '{}'.".format(str(expected), + str(actual))) + + if expected != actual: if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(str(cache))) @@ -227,3 +233,65 @@ class RemoteBase(object): if checksum == existing: ret[i] = True return ret + + def already_cached(self, path_info): + current = self.save_info(path_info)[self.PARAM_CHECKSUM] + + if not current: + return False + + return not self.changed_cache(current) + + def safe_remove(self, path_info, force=False): + if not self.exists(path_info): + return + + if not force and not self.already_cached(path_info): + msg = ( + 'File "{}" is going to be removed. ' + 'Are you sure you want to proceed?' + .format(str(path_info)) + ) + + if not prompt.confirm(msg): + raise DvcException("Unable to remove {} without a confirmation" + " from the user. Use '-f' to force." + .format(str(path_info))) + + self.remove(path_info) + + def do_checkout(self, path_info, checksum, force=False): + if self.exists(path_info): + msg = "Data '{}' exists. Removing before checkout." + logger.warn(msg.format(str(path_info))) + self.safe_remove(path_info, force=force) + + from_info = self.checksum_to_path_info(checksum) + self.copy(from_info, path_info) + + def checkout(self, path_info, checksum_info, force=False): + scheme = path_info['scheme'] + if scheme not in ['', 'local'] and scheme != self.scheme: + raise NotImplementedError + + checksum = checksum_info.get(self.PARAM_CHECKSUM) + if not checksum: + msg = "No checksum info for '{}'." + logger.info(msg.format(str(path_info))) + return + + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + logger.info(msg.format(str(path_info))) + return + + if self.changed_cache(checksum): + msg = "Cache '{}' not found. File '{}' won't be created." + logger.warn(msg.format(checksum, str(path_info))) + self.safe_remove(path_info, force=force) + return + + msg = "Checking out '{}' with cache '{}'." + logger.info(msg.format(str(path_info), checksum)) + + self.do_checkout(path_info, checksum, force=force) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 387221773..531e4e7e8 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -67,12 +67,6 @@ class RemoteGS(RemoteBase): return {self.PARAM_CHECKSUM: self.get_md5(path_info['bucket'], path_info['path'])} - @staticmethod - def to_string(path_info): - return "{}://{}/{}".format(path_info['scheme'], - path_info['bucket'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -86,7 +80,7 @@ class RemoteGS(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, 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['path']) @@ -107,42 +101,10 @@ class RemoteGS(RemoteBase): path = self.checksum_to_path(md5) to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) return {self.PARAM_CHECKSUM: md5} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'gs': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_CHECKSUM, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - path = self.checksum_to_path(md5) - from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - - self._copy(from_info, path_info) - def remove(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError @@ -218,7 +180,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'gs': - self._copy(from_info, to_info, gs=gs) + self.copy(from_info, to_info, gs=gs) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 5eca0ba1c..dc114f894 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -65,7 +65,7 @@ class RemoteHDFS(RemoteBase): user=path_info['user']) return self._group(regex, stdout, 'checksum') - def cp(self, from_info, to_info): + def copy(self, from_info, to_info): dname = posixpath.dirname(to_info['path']) self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user']) self.hadoop_fs('cp -f {} {}'.format(from_info['path'], @@ -84,11 +84,6 @@ class RemoteHDFS(RemoteBase): return {self.PARAM_CHECKSUM: self.checksum(path_info)} - @staticmethod - def to_string(path_info): - return "{}://{}".format(path_info['scheme'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -112,44 +107,10 @@ class RemoteHDFS(RemoteBase): dest = path_info.copy() dest['path'] = self.checksum_to_path(checksum) - self.cp(path_info, dest) + self.copy(path_info, dest) return {self.PARAM_CHECKSUM: checksum} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'hdfs': - raise NotImplementedError - - assert path_info.get('path') - - checksum = checksum_info.get(self.PARAM_CHECKSUM, None) - if not checksum: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(checksum): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(checksum, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), checksum)) - - src = path_info.copy() - src['path'] = self.checksum_to_path(checksum) - - self.cp(src, path_info) - def remove(self, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError @@ -199,7 +160,7 @@ class RemoteHDFS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'hdfs': - self.cp(from_info, to_info) + self.copy(from_info, to_info) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/local.py b/dvc/remote/local.py index ab9604352..5bce37e1a 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -7,7 +7,6 @@ import shutil import posixpath from operator import itemgetter -import dvc.prompt as prompt from dvc.system import System from dvc.remote.base import RemoteBase, STATUS_MAP from dvc.logger import logger @@ -270,36 +269,14 @@ class RemoteLOCAL(RemoteBase): def is_dir_cache(cls, cache): return cache.endswith(cls.MD5_DIR_SUFFIX) - def checkout(self, path_info, checksum_info, force=False): + def do_checkout(self, path_info, checksum, force=False): path = path_info['path'] - md5 = checksum_info.get(self.PARAM_CHECKSUM) + md5 = checksum cache = self.get(md5) - if not cache: - msg = 'No cache info for \'{}\'. Skipping checkout.' - logger.warn(msg.format(os.path.relpath(path))) - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(os.path.relpath(path))) - return - - if self.changed_cache(md5): - msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.' - logger.warn(msg.format(md5, os.path.relpath(path))) - remove(path) - return - - msg = u'Checking out \'{}\' with cache \'{}\'.' - logger.info(msg.format(os.path.relpath(path), md5)) - if not self.is_dir_cache(cache): if os.path.exists(path): - if force or self._already_cached(path): - remove(path) - else: - self._safe_remove(path) + self.safe_remove(path_info, force=force) self.link(cache, path) self.state.update_link(path) @@ -329,10 +306,7 @@ class RemoteLOCAL(RemoteBase): if self.changed(entry_info, entry_checksum_info): if os.path.exists(p): - if force or self._already_cached(p): - remove(p) - else: - self._safe_remove(p) + self.safe_remove(entry_info, force=force) self.link(c, p) @@ -346,8 +320,10 @@ class RemoteLOCAL(RemoteBase): if bar: progress.finish_target(dir_relpath) - def _already_cached(self, path): - current_md5 = self.state.update(path) + def already_cached(self, path_info): + assert path_info['scheme'] in ['', 'local'] + + current_md5 = self.state.update(path_info['path']) if not current_md5: return False @@ -369,24 +345,7 @@ class RemoteLOCAL(RemoteBase): delta = working_dir_files - cached_files for file in delta: - if force or self._already_cached(file): - remove(file) - else: - self._safe_remove(file) - - def _safe_remove(self, file): - msg = ( - 'File "{}" is going to be removed. ' - 'Are you sure you want to proceed?' - .format(file) - ) - - if not prompt.confirm(msg): - raise DvcException('Unable to remove {} without a confirmation' - " from the user. Use '-f' to force." - .format(file)) - - remove(file) + self.safe_remove({'scheme': 'local', 'path': file}, force=force) def _move(self, inp, outp): # moving in two stages to make last the move atomic in diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 3fbacb4ca..3ca2ac5e9 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -115,7 +115,7 @@ class RemoteS3(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, from_info, to_info, s3=None): + def copy(self, from_info, to_info, s3=None): s3 = s3 if s3 else self.s3 source = {'Bucket': from_info['bucket'], @@ -130,46 +130,10 @@ class RemoteS3(RemoteBase): path = self.checksum_to_path(etag) to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) return {self.PARAM_CHECKSUM: etag} - @staticmethod - def to_string(path_info): - return "s3://{}/{}".format(path_info['bucket'], path_info['path']) - - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 's3': - raise NotImplementedError - - etag = checksum_info.get(self.PARAM_CHECKSUM, None) - if not etag: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(etag): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(etag, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), etag)) - - path = self.checksum_to_path(etag) - from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - - self._copy(from_info, path_info) - def remove(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError @@ -262,7 +226,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError if to_info['scheme'] == 's3': - self._copy(from_info, to_info, s3=s3) + self.copy(from_info, to_info, s3=s3) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index b7da48978..8d576cad1 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -188,7 +188,7 @@ class RemoteSSH(RemoteBase): return md5 - def cp(self, from_info, to_info, ssh=None): + def copy(self, from_info, to_info, ssh=None): if from_info['scheme'] != 'ssh' or to_info['scheme'] != 'ssh': raise NotImplementedError @@ -212,13 +212,6 @@ class RemoteSSH(RemoteBase): return {self.PARAM_CHECKSUM: self.md5(path_info)} - @staticmethod - def to_string(path_info): - return "{}://{}@{}:{}".format(path_info['scheme'], - path_info['user'], - path_info['host'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -240,42 +233,10 @@ class RemoteSSH(RemoteBase): dest = path_info.copy() dest['path'] = self.checksum_to_path(md5) - self.cp(path_info, dest) + self.copy(path_info, dest) return {self.PARAM_CHECKSUM: md5} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'ssh': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_CHECKSUM, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - src = path_info.copy() - src['path'] = self.checksum_to_path(md5) - - self.cp(src, path_info) - def remove(self, path_info): if path_info['scheme'] != 'ssh': raise NotImplementedError @@ -309,7 +270,7 @@ class RemoteSSH(RemoteBase): assert from_info['host'] == to_info['host'] assert from_info['port'] == to_info['port'] assert from_info['user'] == to_info['user'] - self.cp(from_info, to_info, ssh=ssh) + self.copy(from_info, to_info, ssh=ssh) continue if to_info['scheme'] != 'local': From c8b39235bf8570b1f872b44e828fd81e58358386 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 22:16:14 +0200 Subject: [PATCH 10/11] remote: local: fix info msgs Signed-off-by: Ruslan Kuprieiev --- dvc/remote/local.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 5bce37e1a..0d8ced133 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -549,7 +549,7 @@ class RemoteLOCAL(RemoteBase): return list(by_md5.keys()), list(by_md5.values()) def status(self, checksum_infos, remote, jobs=None, show_checksums=False): - logger.info("Preparing to pull data from {}".format(remote.url)) + logger.info("Preparing to collect status from {}".format(remote.url)) title = "Collecting information" progress.set_n_total(1) @@ -579,6 +579,7 @@ class RemoteLOCAL(RemoteBase): remote, jobs=None, show_checksums=False): + logger.info("Preparing to pull data from {}".format(remote.url)) title = "Collecting information" progress.set_n_total(1) From 4029116b85b7e91002bfe87d8b081957b21d384b Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Fri, 21 Dec 2018 01:06:45 +0200 Subject: [PATCH 11/11] remote: RFC: unify push/pull Signed-off-by: Ruslan Kuprieiev --- dvc/project.py | 19 +-- dvc/remote/base.py | 11 +- dvc/remote/http.py | 11 +- dvc/remote/local.py | 257 +++++++++++++++++---------------------- tests/test_data_cloud.py | 26 ++-- 5 files changed, 144 insertions(+), 180 deletions(-) diff --git a/dvc/project.py b/dvc/project.py index 35f4afe91..93edfd8a9 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -896,12 +896,15 @@ class Project(object): remote=remote, jobs=jobs)['local'] - status = {} - for md5, ret in self.cloud.status(used, - jobs, - remote=remote, - show_checksums=show_checksums): - if ret == cloud.STATUS_OK: + ret = {} + status_info = self.cloud.status(used, + jobs, + remote=remote, + show_checksums=show_checksums) + for md5, info in status_info.items(): + name = info['name'] + status = info['status'] + if status == cloud.STATUS_OK: continue prefix_map = { @@ -909,9 +912,9 @@ class Project(object): cloud.STATUS_NEW: 'new', } - status[md5] = prefix_map[ret] + ret[name] = prefix_map[status] - return status + return ret def status(self, target=None, diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 8eb18829e..53eb27beb 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -224,15 +224,8 @@ class RemoteBase(object): # implementation of cache_exists(see http, local). # # Result of all() might be way too big, so we should walk through - # it in one pass. Also currently, cache_exists() should return - # a list of True/False that matches order in checksums list, - # so we need to use such an ugly logic. - ret = len(checksums) * [False] - for existing in self.all(): - for i, checksum in enumerate(checksums): - if checksum == existing: - ret[i] = True - return ret + # it in one pass. + return list(filter(lambda checksum: checksum in checksums, self.all())) def already_cached(self, path_info): current = self.save_info(path_info)[self.PARAM_CHECKSUM] diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 7147b9dd1..d2d80a0cf 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -89,12 +89,11 @@ class RemoteHTTP(RemoteBase): def cache_exists(self, md5s): assert isinstance(md5s, list) - ret = [] - for md5 in md5s: - url = self.checksum_to_path(md5) - exists = bool(self._request('HEAD', url)) - ret.append(exists) - return ret + + def func(md5): + return bool(self._request('HEAD', self.checksum_to_path(md5))) + + return list(filter(func, md5s)) def save_info(self, path_info): if path_info['scheme'] not in ['http', 'https']: diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 0d8ced133..18c53f34e 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -8,7 +8,7 @@ import posixpath from operator import itemgetter from dvc.system import System -from dvc.remote.base import RemoteBase, STATUS_MAP +from dvc.remote.base import RemoteBase, STATUS_MAP, STATUS_NEW, STATUS_DELETED from dvc.logger import logger from dvc.utils import remove, move, copyfile, dict_md5, to_chunks from dvc.utils import LARGE_DIR_SIZE @@ -458,11 +458,7 @@ class RemoteLOCAL(RemoteBase): def cache_exists(self, md5s): assert isinstance(md5s, list) - ret = [] - for md5 in md5s: - path = os.path.join(self.prefix, md5[0:2], md5[2:]) - ret.append(os.path.exists(path)) - return ret + return list(filter(lambda md5: not self.changed_cache_file(md5), md5s)) def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) @@ -531,7 +527,7 @@ class RemoteLOCAL(RemoteBase): md5 = info[self.PARAM_CHECKSUM] if show_checksums: - by_md5[md5] = md5 + by_md5[md5] = {'name': md5} continue name = info[self.PARAM_PATH] @@ -540,171 +536,136 @@ class RemoteLOCAL(RemoteBase): name += '({})'.format(branch) if md5 not in by_md5.keys(): - by_md5[md5] = '' + by_md5[md5] = {'name': name} else: - by_md5[md5] += ' ' + by_md5[md5]['name'] += ' ' + name - by_md5[md5] += name - - return list(by_md5.keys()), list(by_md5.values()) + return by_md5 def status(self, checksum_infos, remote, jobs=None, show_checksums=False): logger.info("Preparing to collect status from {}".format(remote.url)) title = "Collecting information" + ret = {} + progress.set_n_total(1) progress.update_target(title, 0, 100) progress.update_target(title, 10, 100) - md5s, names = self._group(checksum_infos, - show_checksums=show_checksums) + ret = self._group(checksum_infos, show_checksums=show_checksums) + md5s = list(ret.keys()) progress.update_target(title, 30, 100) - remote_exists = remote.cache_exists(md5s) + remote_exists = list(remote.cache_exists(md5s)) progress.update_target(title, 90, 100) - local_exists = [not self.changed_cache_file(md5) for md5 in md5s] + local_exists = self.cache_exists(md5s) progress.finish_target(title) - return [(name, STATUS_MAP[l, r]) for name, l, r in zip(names, - local_exists, - remote_exists)] + for md5, info in ret.items(): + info['status'] = STATUS_MAP[(md5 in local_exists, + md5 in remote_exists)] + + return ret + + def _get_chunks(self, download, remote, status_info, status, jobs): + title = "Analysing status." + + progress.set_n_total(1) + total = len(status_info) + current = 0 + + cache = [] + path_infos = [] + names = [] + for md5, info in status_info.items(): + if info['status'] == status: + cache.append(self.checksum_to_path_info(md5)) + path_infos.append(remote.checksum_to_path_info(md5)) + names.append(info['name']) + current += 1 + progress.update_target(title, current, total) + + progress.finish_target(title) + + progress.set_n_total(len(names)) + + if download: + to_infos = cache + from_infos = path_infos + else: + to_infos = path_infos + from_infos = cache + + return list(zip(to_chunks(from_infos, jobs), + to_chunks(to_infos, jobs), + to_chunks(names, jobs))) + + def _process(self, + checksum_infos, + remote, + jobs=None, + show_checksums=False, + download=False): + msg = "Preparing to {} data {} '{}'" + logger.info(msg.format('download' if download else 'upload', + 'from' if download else 'to', + remote.url)) + + if download: + func = remote.download + status = STATUS_DELETED + else: + func = remote.upload + status = STATUS_NEW + + if jobs is None: + jobs = remote.JOBS + + status_info = self.status(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums) + + chunks = self._get_chunks(download, remote, status_info, status, jobs) + + if len(chunks) == 0: + return + + futures = [] + with ThreadPoolExecutor(max_workers=jobs) as executor: + for from_infos, to_infos, names in chunks: + res = executor.submit(func, + from_infos, + to_infos, + names=names) + futures.append(res) + + for f in futures: + f.result() + + def push(self, + checksum_infos, + remote, + jobs=None, + show_checksums=False): + self._process(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums, + download=False) def pull(self, checksum_infos, remote, jobs=None, show_checksums=False): - logger.info("Preparing to pull data from {}".format(remote.url)) - title = "Collecting information" - - progress.set_n_total(1) - progress.update_target(title, 0, 100) - - grouped = zip(*self._group(checksum_infos, - show_checksums=show_checksums)) - - progress.update_target(title, 10, 100) - - md5s = [] - names = [] - # NOTE: filter files that are not corrupted - for md5, name in grouped: - if self.changed_cache_file(md5): - md5s.append(md5) - names.append(name) - - progress.update_target(title, 30, 100) - - cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s] - - progress.update_target(title, 50, 100) - - path_infos = remote.md5s_to_path_infos(md5s) - - progress.update_target(title, 60, 100) - - # NOTE: dummy call to try to establish a connection - # to see if we need to ask user for a password. - remote.cache_exists(['000']) - - progress.update_target(title, 70, 100) - - 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))) - - progress.finish_target(title) - - progress.set_n_total(len(names)) - - if len(chunks) == 0: - return - - futures = [] - with ThreadPoolExecutor(max_workers=len(chunks)) as executor: - for from_infos, to_infos, names in chunks: - res = executor.submit(remote.download, - from_infos, - to_infos, - names=names) - futures.append(res) - - for f in futures: - f.result() - - def push(self, checksum_infos, remote, jobs=None, show_checksums=False): - logger.info("Preparing to push data to {}".format(remote.url)) - title = "Collecting information" - - progress.set_n_total(1) - progress.update_target(title, 0, 100) - - # NOTE: verifying that our cache is not corrupted - def func(info): - return not self.changed_cache_file(info[self.PARAM_CHECKSUM]) - checksum_infos = list(filter(func, checksum_infos)) - - progress.update_target(title, 20, 100) - - # NOTE: filter files that are already uploaded - md5s = [i[self.PARAM_CHECKSUM] for i in checksum_infos] - exists = remote.cache_exists(md5s) - - progress.update_target(title, 30, 100) - - def func(entry): - return not entry[0] - - assert len(exists) == len(checksum_infos) - infos_exist = list(filter(func, zip(exists, checksum_infos))) - checksum_infos = [i for e, i in infos_exist] - - progress.update_target(title, 70, 100) - - md5s, names = self._group(checksum_infos, - show_checksums=show_checksums) - cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s] - - progress.update_target(title, 80, 100) - - path_infos = remote.md5s_to_path_infos(md5s) - - assert len(path_infos) == len(cache) == len(md5s) == len(names) - - 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))) - - progress.finish_target(title) - - progress.set_n_total(len(names)) - - if len(chunks) == 0: - return - - futures = [] - with ThreadPoolExecutor(max_workers=len(chunks)) as executor: - for to_infos, from_infos, names in chunks: - res = executor.submit(remote.upload, - from_infos, - to_infos, - names=names) - futures.append(res) - - for f in futures: - f.result() + self._process(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums, + download=True) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index a13872096..4b94f6407 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -233,10 +233,12 @@ class TestDataCloudBase(TestDvc): with self.cloud.project.state: # Check status status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_NEW)], status) + expected = {md5: {'name': md5, 'status': STATUS_NEW}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_NEW) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_NEW}} + self.assertEqual(status_dir, expected) # Push and check status self.cloud.push([info]) @@ -247,19 +249,23 @@ class TestDataCloudBase(TestDvc): self.assertTrue(os.path.isfile(cache_dir)) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_OK)], status) + expected = {md5: {'name': md5, 'status': STATUS_OK}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_OK) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}} + self.assertEqual(status_dir, expected) # Remove and check status shutil.rmtree(self.dvc.cache.local.cache_dir) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_DELETED)], status) + expected = {md5: {'name': md5, 'status': STATUS_DELETED}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertEqual([(md5_dir, STATUS_DELETED)], status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_DELETED}} + self.assertEqual(status_dir, expected) # Pull and check status self.cloud.pull([info]) @@ -272,10 +278,12 @@ class TestDataCloudBase(TestDvc): self.assertTrue(os.path.isfile(cache_dir)) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_OK)], status) + expected = {md5: {'name': md5, 'status': STATUS_OK}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_OK) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}} + self.assertTrue(status_dir, expected) def test(self): self._ensure_should_run() @@ -421,7 +429,7 @@ class TestDataCloudCLIBase(TestDvc): self.main(['fetch'] + args) - self.main(['pull'] + args) + self.main(['pull', '-f'] + args) self.assertTrue(os.path.exists(cache)) self.assertTrue(os.path.isfile(cache)) self.assertTrue(os.path.isfile(cache_dir))