X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/29efe0926dde0ee85387846f31062f4c680cb49c..8c5f2973a5c5f042d1d12aef1c470b37519fd416:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 27aad033ae..f26d3a3d27 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -544,7 +544,7 @@ class RichCollectionBase(CollectionBase): else: item = ArvadosFile(self, pathcomponents[0]) self._items[pathcomponents[0]] = item - self._committed = False + self.set_committed(False) self.notify(ADD, self, pathcomponents[0], item) return item else: @@ -552,7 +552,7 @@ class RichCollectionBase(CollectionBase): # create new collection item = Subcollection(self, pathcomponents[0]) self._items[pathcomponents[0]] = item - self._committed = False + self.set_committed(False) self.notify(ADD, self, pathcomponents[0], item) if isinstance(item, RichCollectionBase): return item.find_or_create(pathcomponents[1], create_type) @@ -565,16 +565,23 @@ class RichCollectionBase(CollectionBase): def find(self, path): """Recursively search the specified file path. - May return either a Collection or ArvadosFile. Return None if not + May return either a Collection or ArvadosFile. Return None if not found. + If path is invalid (ex: starts with '/'), an IOError exception will be + raised. """ if not path: raise errors.ArgumentError("Parameter 'path' is empty.") pathcomponents = path.split("/", 1) + if pathcomponents[0] == '': + raise IOError(errno.ENOTDIR, "Not a directory", pathcomponents[0]) + item = self._items.get(pathcomponents[0]) - if len(pathcomponents) == 1: + if item is None: + return None + elif len(pathcomponents) == 1: return item else: if isinstance(item, RichCollectionBase): @@ -652,20 +659,26 @@ class RichCollectionBase(CollectionBase): @synchronized def committed(self): """Determine if the collection has been committed to the API server.""" - - if self._committed is False: - return False - for v in self._items.values(): - if v.committed() is False: - return False - return True + return self._committed @synchronized - def set_committed(self): - """Recursively set committed flag to True.""" - self._committed = True - for k,v in self._items.items(): - v.set_committed() + def set_committed(self, value=True): + """Recursively set committed flag. + + If value is True, set committed to be True for this and all children. + + If value is False, set committed to be False for this and all parents. + """ + if value == self._committed: + return + if value: + for k,v in self._items.items(): + v.set_committed(True) + self._committed = True + else: + self._committed = False + if self.parent is not None: + self.parent.set_committed(False) @synchronized def __iter__(self): @@ -696,7 +709,7 @@ class RichCollectionBase(CollectionBase): def __delitem__(self, p): """Delete an item by name which is directly contained by this collection.""" del self._items[p] - self._committed = False + self.set_committed(False) self.notify(DEL, self, p, None) @synchronized @@ -739,7 +752,7 @@ class RichCollectionBase(CollectionBase): raise IOError(errno.ENOTEMPTY, "Directory not empty", path) deleteditem = self._items[pathcomponents[0]] del self._items[pathcomponents[0]] - self._committed = False + self.set_committed(False) self.notify(DEL, self, pathcomponents[0], deleteditem) else: item.remove(pathcomponents[1]) @@ -788,7 +801,7 @@ class RichCollectionBase(CollectionBase): item = source_obj.clone(self, target_name) self._items[target_name] = item - self._committed = False + self.set_committed(False) if modified_from: self.notify(MOD, self, target_name, (modified_from, item)) @@ -829,7 +842,7 @@ class RichCollectionBase(CollectionBase): if target_dir is None: raise IOError(errno.ENOENT, "Target directory not found", target_name) - if target_name in target_dir and isinstance(self[target_name], RichCollectionBase) and sourcecomponents: + if target_name in target_dir and isinstance(target_dir[target_name], RichCollectionBase) and sourcecomponents: target_dir = target_dir[target_name] target_name = sourcecomponents[-1] @@ -898,7 +911,8 @@ class RichCollectionBase(CollectionBase): return self._get_manifest_text(stream_name, True, True) @synchronized - def manifest_text(self, stream_name=".", strip=False, normalize=False): + def manifest_text(self, stream_name=".", strip=False, normalize=False, + only_committed=False): """Get the manifest text for this collection, sub collections and files. This method will flush outstanding blocks to Keep. By default, it will @@ -917,10 +931,15 @@ class RichCollectionBase(CollectionBase): is not modified, return the original manifest text even if it is not in normalized form. + :only_committed: + If True, don't commit pending blocks. + """ - self._my_block_manager().commit_all() - return self._get_manifest_text(stream_name, strip, normalize) + if not only_committed: + self._my_block_manager().commit_all() + return self._get_manifest_text(stream_name, strip, normalize, + only_committed=only_committed) @synchronized def _get_manifest_text(self, stream_name, strip, normalize, only_committed=False): @@ -966,7 +985,7 @@ class RichCollectionBase(CollectionBase): if stream: buf.append(" ".join(normalize_stream(stream_name, stream)) + "\n") for dirname in [s for s in sorted_keys if isinstance(self[s], RichCollectionBase)]: - buf.append(self[dirname].manifest_text(stream_name=os.path.join(stream_name, dirname), strip=strip, normalize=True)) + buf.append(self[dirname].manifest_text(stream_name=os.path.join(stream_name, dirname), strip=strip, normalize=True, only_committed=only_committed)) return "".join(buf) else: if strip: @@ -1009,7 +1028,7 @@ class RichCollectionBase(CollectionBase): """ if changes: - self._committed = False + self.set_committed(False) for change in changes: event_type = change[0] path = change[1] @@ -1050,8 +1069,13 @@ class RichCollectionBase(CollectionBase): def portable_data_hash(self): """Get the portable data hash for this collection's manifest.""" - stripped = self.portable_manifest_text() - return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) + if self._manifest_locator and self.committed(): + # If the collection is already saved on the API server, and it's committed + # then return API server's PDH response. + return self._portable_data_hash + else: + stripped = self.portable_manifest_text() + return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) @synchronized def subscribe(self, callback): @@ -1142,7 +1166,8 @@ class Collection(RichCollectionBase): parent=None, apiconfig=None, block_manager=None, - replication_desired=None): + replication_desired=None, + put_threads=None): """Collection constructor. :manifest_locator_or_text: @@ -1179,6 +1204,7 @@ class Collection(RichCollectionBase): self._keep_client = keep_client self._block_manager = block_manager self.replication_desired = replication_desired + self.put_threads = put_threads if apiconfig: self._config = apiconfig @@ -1188,6 +1214,7 @@ class Collection(RichCollectionBase): self.num_retries = num_retries if num_retries is not None else 0 self._manifest_locator = None self._manifest_text = None + self._portable_data_hash = None self._api_response = None self._past_versions = set() @@ -1269,7 +1296,7 @@ class Collection(RichCollectionBase): copies = (self.replication_desired or self._my_api()._rootDesc.get('defaultCollectionReplication', 2)) - self._block_manager = _BlockManager(self._my_keep(), copies=copies) + self._block_manager = _BlockManager(self._my_keep(), copies=copies, put_threads=self.put_threads) return self._block_manager def _remember_api_response(self, response): @@ -1289,6 +1316,7 @@ class Collection(RichCollectionBase): uuid=self._manifest_locator).execute( num_retries=self.num_retries)) self._manifest_text = self._api_response['manifest_text'] + self._portable_data_hash = self._api_response['portable_data_hash'] # If not overriden via kwargs, we should try to load the # replication_desired from the API server if self.replication_desired is None: @@ -1454,7 +1482,8 @@ class Collection(RichCollectionBase): ).execute( num_retries=num_retries)) self._manifest_text = self._api_response["manifest_text"] - self.set_committed() + self._portable_data_hash = self._api_response["portable_data_hash"] + self.set_committed(True) return self._manifest_text @@ -1512,9 +1541,10 @@ class Collection(RichCollectionBase): text = self._api_response["manifest_text"] self._manifest_locator = self._api_response["uuid"] + self._portable_data_hash = self._api_response["portable_data_hash"] self._manifest_text = text - self.set_committed() + self.set_committed(True) return text @@ -1579,7 +1609,7 @@ class Collection(RichCollectionBase): stream_name = None state = STREAM_NAME - self.set_committed() + self.set_committed(True) @synchronized def notify(self, event, collection, name, item): @@ -1629,7 +1659,7 @@ class Subcollection(RichCollectionBase): @must_be_writable @synchronized def _reparent(self, newparent, newname): - self._committed = False + self.set_committed(False) self.flush() self.parent.remove(self.name, recursive=True) self.parent = newparent