X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/349e1ee218d7e888c6c1bcb07f6537f0bdc85012..0f53d219c91812baeac4cff3387196c6501ec0c0:/sdk/python/arvados/collection.py?ds=sidebyside diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 5dde8f70d2..56d8b23933 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -474,6 +474,7 @@ class ResumableCollectionWriter(CollectionWriter): ADD = "add" DEL = "del" MOD = "mod" +TOK = "tok" FILE = "file" COLLECTION = "collection" @@ -644,14 +645,13 @@ class RichCollectionBase(CollectionBase): return ArvadosFileWriter(arvfile, mode, num_retries=self.num_retries) def modified(self): + """Determine if the collection has been modified since last commited.""" return not self.committed() - def set_unmodified(self): - self.set_committed() - @synchronized def committed(self): - """Test if the collection and all subcollection and files are committed.""" + """Determine if the collection has been committed to the API server.""" + if self._committed is False: return False for v in self._items.values(): @@ -661,7 +661,7 @@ class RichCollectionBase(CollectionBase): @synchronized def set_committed(self): - """Recursively set committed flag.""" + """Recursively set committed flag to True.""" self._committed = True for k,v in self._items.items(): v.set_committed() @@ -922,7 +922,7 @@ class RichCollectionBase(CollectionBase): return self._get_manifest_text(stream_name, strip, normalize) @synchronized - def _get_manifest_text(self, stream_name, strip, normalize): + def _get_manifest_text(self, stream_name, strip, normalize, only_committed=False): """Get the manifest text for this collection, sub collections and files. :stream_name: @@ -938,6 +938,9 @@ class RichCollectionBase(CollectionBase): is not modified, return the original manifest text even if it is not in normalized form. + :only_committed: + If True, only include blocks that were already committed to Keep. + """ if not self.committed() or self._manifest_text is None or normalize: @@ -951,6 +954,8 @@ class RichCollectionBase(CollectionBase): for segment in arvfile.segments(): loc = segment.locator if arvfile.parent._my_block_manager().is_bufferblock(loc): + if only_committed: + continue loc = arvfile.parent._my_block_manager().get_bufferblock(loc).locator() if strip: loc = KeepLocator(loc).stripped() @@ -987,6 +992,8 @@ class RichCollectionBase(CollectionBase): changes.extend(self[k].diff(end_collection[k], os.path.join(prefix, k), holding_collection)) elif end_collection[k] != self[k]: changes.append((MOD, os.path.join(prefix, k), self[k].clone(holding_collection, ""), end_collection[k].clone(holding_collection, ""))) + else: + changes.append((TOK, os.path.join(prefix, k), self[k].clone(holding_collection, ""), end_collection[k].clone(holding_collection, ""))) else: changes.append((ADD, os.path.join(prefix, k), end_collection[k].clone(holding_collection, ""))) return changes @@ -1017,7 +1024,7 @@ class RichCollectionBase(CollectionBase): # There is already local file and it is different: # save change to conflict file. self.copy(initial, conflictpath) - elif event_type == MOD: + elif event_type == MOD or event_type == TOK: final = change[3] if local == initial: # Local matches the "initial" item so it has not @@ -1133,7 +1140,8 @@ class Collection(RichCollectionBase): num_retries=None, parent=None, apiconfig=None, - block_manager=None): + block_manager=None, + replication_desired=None): """Collection constructor. :manifest_locator_or_text: @@ -1141,24 +1149,35 @@ class Collection(RichCollectionBase): a manifest, raw manifest text, or None (to create an empty collection). :parent: the parent Collection, may be None. + :apiconfig: A dict containing keys for ARVADOS_API_HOST and ARVADOS_API_TOKEN. Prefer this over supplying your own api_client and keep_client (except in testing). Will use default config settings if not specified. + :api_client: The API client object to use for requests. If not specified, create one using `apiconfig`. + :keep_client: the Keep client to use for requests. If not specified, create one using `apiconfig`. + :num_retries: the number of retries for API and Keep requests. + :block_manager: the block manager to use. If not specified, create one. + :replication_desired: + How many copies should Arvados maintain. If None, API server default + configuration applies. If not None, this value will also be used + for determining the number of block copies being written. + """ super(Collection, self).__init__(parent) self._api_client = api_client self._keep_client = keep_client self._block_manager = block_manager + self.replication_desired = replication_desired if apiconfig: self._config = apiconfig @@ -1169,6 +1188,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 @@ -1198,6 +1218,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): @@ -1207,6 +1231,15 @@ 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"))) and + response.get("portable_data_hash") != self.portable_data_hash()): + # The record on the server is different from our current one, but we've seen it before, + # so ignore it because it's already been merged. + # However, if it's the same as our current record, proceed with the update, because we want to update + # our tokens. + 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)) @@ -1216,7 +1249,8 @@ class Collection(RichCollectionBase): def _my_api(self): if self._api_client is None: self._api_client = ThreadSafeApiCache(self._config) - self._keep_client = self._api_client.keep + if self._keep_client is None: + self._keep_client = self._api_client.keep return self._api_client @synchronized @@ -1231,9 +1265,16 @@ class Collection(RichCollectionBase): @synchronized def _my_block_manager(self): if self._block_manager is None: - self._block_manager = _BlockManager(self._my_keep()) + copies = (self.replication_desired or + self._my_api()._rootDesc.get('defaultCollectionReplication', + 2)) + self._block_manager = _BlockManager(self._my_keep(), copies=copies) 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 @@ -1243,10 +1284,14 @@ 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'] + # If not overriden via kwargs, we should try to load the + # replication_desired from the API server + if self.replication_desired is None: + self.replication_desired = self._api_response.get('replication_desired', None) return None except Exception as e: return e @@ -1402,11 +1447,11 @@ 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_committed() @@ -1453,15 +1498,16 @@ 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, - "name": name} + "name": name, + "replication_desired": self.replication_desired} 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"] @@ -1544,7 +1590,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. """