X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/3809452aef74876da9d2644fe6c824a22527d6ac..7563fb986662a066f0aa3a9c4c1dd35159fb69cc:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index eea0717964..38e794c24a 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -487,7 +487,7 @@ class RichCollectionBase(CollectionBase): def __init__(self, parent=None): self.parent = parent - self._modified = True + self._committed = False self._callback = None self._items = {} @@ -542,7 +542,7 @@ class RichCollectionBase(CollectionBase): else: item = ArvadosFile(self, pathcomponents[0]) self._items[pathcomponents[0]] = item - self._modified = True + self._committed = False self.notify(ADD, self, pathcomponents[0], item) return item else: @@ -550,12 +550,12 @@ class RichCollectionBase(CollectionBase): # create new collection item = Subcollection(self, pathcomponents[0]) self._items[pathcomponents[0]] = item - self._modified = True + self._committed = False self.notify(ADD, self, pathcomponents[0], item) if isinstance(item, RichCollectionBase): return item.find_or_create(pathcomponents[1], create_type) else: - raise IOError(errno.ENOTDIR, "Interior path components must be subcollection") + raise IOError(errno.ENOTDIR, "Not a directory: '%s'" % pathcomponents[0]) else: return self @@ -568,7 +568,7 @@ class RichCollectionBase(CollectionBase): """ if not path: - raise errors.ArgumentError("Parameter 'path' must not be empty.") + raise errors.ArgumentError("Parameter 'path' is empty.") pathcomponents = path.split("/", 1) item = self._items.get(pathcomponents[0]) @@ -581,15 +581,20 @@ class RichCollectionBase(CollectionBase): else: return item else: - raise IOError(errno.ENOTDIR, "Interior path components must be subcollection") + raise IOError(errno.ENOTDIR, "Is not a directory: %s" % pathcomponents[0]) + @synchronized def mkdirs(self, path): """Recursive subcollection create. - Like `os.mkdirs()`. Will create intermediate subcollections needed to - contain the leaf subcollection path. + Like `os.makedirs()`. Will create intermediate subcollections needed + to contain the leaf subcollection path. """ + + if self.find(path) != None: + raise IOError(errno.EEXIST, "Directory or file exists: '%s'" % path) + return self.find_or_create(path, COLLECTION) def open(self, path, mode="r"): @@ -626,7 +631,7 @@ class RichCollectionBase(CollectionBase): if arvfile is None: raise IOError(errno.ENOENT, "File not found") if not isinstance(arvfile, ArvadosFile): - raise IOError(errno.EISDIR, "Path must refer to a file.") + raise IOError(errno.EISDIR, "Is a directory: %s" % path) if mode[0] == "w": arvfile.truncate(0) @@ -634,26 +639,31 @@ class RichCollectionBase(CollectionBase): name = os.path.basename(path) if mode == "r": - return ArvadosFileReader(arvfile, mode, num_retries=self.num_retries) + return ArvadosFileReader(arvfile, num_retries=self.num_retries) else: return ArvadosFileWriter(arvfile, mode, num_retries=self.num_retries) - @synchronized def modified(self): - """Test if the collection (or any subcollection or file) has been modified.""" - if self._modified: - return True + """Determine if the collection has been modified since last commited.""" + return not self.committed() + + @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.modified(): - return True - return False + if v.committed() is False: + return False + return True @synchronized - def set_unmodified(self): - """Recursively clear modified flag.""" - self._modified = False + def set_committed(self): + """Recursively set committed flag to True.""" + self._committed = True for k,v in self._items.items(): - v.set_unmodified() + v.set_committed() @synchronized def __iter__(self): @@ -684,7 +694,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._modified = True + self._committed = False self.notify(DEL, self, p, None) @synchronized @@ -716,7 +726,7 @@ class RichCollectionBase(CollectionBase): """ if not path: - raise errors.ArgumentError("Parameter 'path' must not be empty.") + raise errors.ArgumentError("Parameter 'path' is empty.") pathcomponents = path.split("/", 1) item = self._items.get(pathcomponents[0]) @@ -727,7 +737,7 @@ class RichCollectionBase(CollectionBase): raise IOError(errno.ENOTEMPTY, "Subcollection not empty") deleteditem = self._items[pathcomponents[0]] del self._items[pathcomponents[0]] - self._modified = True + self._committed = False self.notify(DEL, self, pathcomponents[0], deleteditem) else: item.remove(pathcomponents[1]) @@ -776,7 +786,7 @@ class RichCollectionBase(CollectionBase): item = source_obj.clone(self, target_name) self._items[target_name] = item - self._modified = True + self._committed = False if modified_from: self.notify(MOD, self, target_name, (modified_from, item)) @@ -870,7 +880,7 @@ class RichCollectionBase(CollectionBase): source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, False) if not source_obj.writable(): - raise IOError(errno.EROFS, "Source collection must be writable.") + raise IOError(errno.EROFS, "Source collection is read only.") target_dir.add(source_obj, target_name, overwrite, True) def portable_manifest_text(self, stream_name="."): @@ -885,6 +895,7 @@ class RichCollectionBase(CollectionBase): """ return self._get_manifest_text(stream_name, True, True) + @synchronized def manifest_text(self, stream_name=".", strip=False, normalize=False): """Get the manifest text for this collection, sub collections and files. @@ -928,7 +939,7 @@ class RichCollectionBase(CollectionBase): """ - if self.modified() or self._manifest_text is None or normalize: + if not self.committed() or self._manifest_text is None or normalize: stream = {} buf = [] sorted_keys = sorted(self.keys()) @@ -989,13 +1000,13 @@ class RichCollectionBase(CollectionBase): """ if changes: - self._modified = True + self._committed = False for change in changes: event_type = change[0] path = change[1] initial = change[2] local = self.find(path) - conflictpath = "%s~conflict-%s~" % (path, time.strftime("%Y-%m-%d-%H:%M:%S", + conflictpath = "%s~%s~conflict~" % (path, time.strftime("%Y%m%d-%H%M%S", time.gmtime())) if event_type == ADD: if local is None: @@ -1157,6 +1168,7 @@ class Collection(RichCollectionBase): self._manifest_locator = None self._manifest_text = None self._api_response = None + self._past_versions = set() self.lock = threading.RLock() self.events = None @@ -1170,7 +1182,7 @@ class Collection(RichCollectionBase): self._manifest_text = manifest_locator_or_text else: raise errors.ArgumentError( - "Argument to CollectionReader must be a manifest or a collection UUID") + "Argument to CollectionReader is not a manifest or a collection UUID") try: self._populate() @@ -1186,6 +1198,10 @@ class Collection(RichCollectionBase): def writable(self): return True + @synchronized + def known_past_version(self, modified_at_and_portable_data_hash): + return modified_at_and_portable_data_hash in self._past_versions + @synchronized @retry_method def update(self, other=None, num_retries=None): @@ -1195,6 +1211,11 @@ class Collection(RichCollectionBase): if self._manifest_locator is None: raise errors.ArgumentError("`other` is None but collection does not have a manifest_locator uuid") response = self._my_api().collections().get(uuid=self._manifest_locator).execute(num_retries=num_retries) + if self.known_past_version((response.get("modified_at"), response.get("portable_data_hash"))): + # We've merged this record this before. Don't do anything. + return + else: + self._past_versions.add((response.get("modified_at"), response.get("portable_data_hash"))) other = CollectionReader(response["manifest_text"]) baseline = CollectionReader(self._manifest_text) self.apply(baseline.diff(other)) @@ -1222,6 +1243,10 @@ class Collection(RichCollectionBase): self._block_manager = _BlockManager(self._my_keep()) return self._block_manager + def _remember_api_response(self, response): + self._api_response = response + self._past_versions.add((response.get("modified_at"), response.get("portable_data_hash"))) + def _populate_from_api_server(self): # As in KeepClient itself, we must wait until the last # possible moment to instantiate an API client, in order to @@ -1231,9 +1256,9 @@ class Collection(RichCollectionBase): # clause, just like any other Collection lookup # failure. Return an exception, or None if successful. try: - self._api_response = self._my_api().collections().get( + self._remember_api_response(self._my_api().collections().get( uuid=self._manifest_locator).execute( - num_retries=self.num_retries) + num_retries=self.num_retries)) self._manifest_text = self._api_response['manifest_text'] return None except Exception as e: @@ -1295,6 +1320,9 @@ class Collection(RichCollectionBase): if exc_type is None: if self.writable() and self._has_collection_uuid(): self.save() + self.stop_threads() + + def stop_threads(self): if self._block_manager is not None: self._block_manager.stop_threads() @@ -1377,9 +1405,9 @@ class Collection(RichCollectionBase): Retry count on API calls (if None, use the collection default) """ - if self.modified(): + if not self.committed(): if not self._has_collection_uuid(): - raise AssertionError("Collection manifest_locator must be a collection uuid. Use save_new() for new collections.") + raise AssertionError("Collection manifest_locator is not a collection uuid. Use save_new() for new collections.") self._my_block_manager().commit_all() @@ -1387,13 +1415,13 @@ class Collection(RichCollectionBase): self.update() text = self.manifest_text(strip=False) - self._api_response = self._my_api().collections().update( + self._remember_api_response(self._my_api().collections().update( uuid=self._manifest_locator, body={'manifest_text': text} ).execute( - num_retries=num_retries) + num_retries=num_retries)) self._manifest_text = self._api_response["manifest_text"] - self.set_unmodified() + self.set_committed() return self._manifest_text @@ -1438,7 +1466,7 @@ class Collection(RichCollectionBase): if create_collection_record: if name is None: - name = "Collection created %s" % (time.strftime("%Y-%m-%d %H:%M:%S %Z", time.localtime())) + name = "New collection" ensure_unique_name = True body = {"manifest_text": text, @@ -1446,13 +1474,13 @@ class Collection(RichCollectionBase): if owner_uuid: body["owner_uuid"] = owner_uuid - self._api_response = self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries) + self._remember_api_response(self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries)) text = self._api_response["manifest_text"] self._manifest_locator = self._api_response["uuid"] self._manifest_text = text - self.set_unmodified() + self.set_committed() return text @@ -1485,7 +1513,7 @@ class Collection(RichCollectionBase): segments = [] streamoffset = 0L state = BLOCKS - self.mkdirs(stream_name) + self.find_or_create(stream_name, COLLECTION) continue if state == BLOCKS: @@ -1511,13 +1539,13 @@ class Collection(RichCollectionBase): raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath) else: # error! - raise errors.SyntaxError("Invalid manifest format") + raise errors.SyntaxError("Invalid manifest format, expected file segment but did not match format: '%s'" % tok) if sep == "\n": stream_name = None state = STREAM_NAME - self.set_unmodified() + self.set_committed() @synchronized def notify(self, event, collection, name, item): @@ -1529,7 +1557,7 @@ class Subcollection(RichCollectionBase): """This is a subdirectory within a collection that doesn't have its own API server record. - It falls under the umbrella of the root collection. + Subcollection locking falls under the umbrella lock of its root collection. """ @@ -1567,7 +1595,7 @@ class Subcollection(RichCollectionBase): @must_be_writable @synchronized def _reparent(self, newparent, newname): - self._modified = True + self._committed = False self.flush() self.parent.remove(self.name, recursive=True) self.parent = newparent