+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
import logging
import re
import time
self.fresh()
- def clear(self, force=False):
- """Delete all entries"""
+ def in_use(self):
+ if super(Directory, self).in_use():
+ return True
+ for v in self._entries.itervalues():
+ if v.in_use():
+ return True
+ return False
- if not self.in_use() or force:
- oldentries = self._entries
- self._entries = {}
- for n in oldentries:
- if not oldentries[n].clear(force):
- self._entries = oldentries
- return False
- for n in oldentries:
- self.inodes.invalidate_entry(self.inode, n.encode(self.inodes.encoding))
- self.inodes.del_entry(oldentries[n])
- self.inodes.invalidate_inode(self.inode)
- self.invalidate()
+ def has_ref(self, only_children):
+ if super(Directory, self).has_ref(only_children):
return True
- else:
- return False
+ for v in self._entries.itervalues():
+ if v.has_ref(False):
+ return True
+ return False
+
+ def clear(self):
+ """Delete all entries"""
+ oldentries = self._entries
+ self._entries = {}
+ for n in oldentries:
+ oldentries[n].clear()
+ self.inodes.invalidate_entry(self.inode, n.encode(self.inodes.encoding))
+ self.inodes.del_entry(oldentries[n])
+ self.inodes.invalidate_inode(self.inode)
+ self.invalidate()
+
+ def kernel_invalidate(self):
+ for n, e in self._entries.iteritems():
+ self.inodes.invalidate_entry(self.inode, n.encode(self.inodes.encoding))
+ e.kernel_invalidate()
+ self.inodes.invalidate_inode(self.inode)
def mtime(self):
return self._mtime
def flush(self):
pass
+ def want_event_subscribe(self):
+ raise NotImplementedError()
+
def create(self, name):
raise NotImplementedError()
self.flush()
src.flush()
+ def clear(self):
+ super(CollectionDirectoryBase, self).clear()
+ self.collection = None
+
class CollectionDirectory(CollectionDirectoryBase):
"""Represents the root of a directory tree representing a collection."""
self.num_retries = num_retries
self.collection_record_file = None
self.collection_record = None
- self.poll = True
+ self._poll = True
try:
- self.poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2)/2)
+ self._poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2)/2)
except:
_logger.debug("Error getting blobSignatureTtl from discovery document: %s", sys.exc_info()[0])
- self.poll_time = 60*60
+ self._poll_time = 60*60
if isinstance(collection_record, dict):
self.collection_locator = collection_record['uuid']
def writable(self):
return self.collection.writable() if self.collection is not None else self._writable
+ def want_event_subscribe(self):
+ return (uuid_pattern.match(self.collection_locator) is not None)
+
# Used by arv-web.py to switch the contents of the CollectionDirectory
def change_collection(self, new_locator):
"""Switch the contents of the CollectionDirectory.
def new_collection(self, new_collection_record, coll_reader):
if self.inode:
- self.clear(force=True)
+ self.clear()
self.collection_record = new_collection_record
if not self.stale():
return
- _logger.debug("Updating %s", to_record_version)
+ _logger.debug("Updating collection %s inode %s to record version %s", self.collection_locator, self.inode, to_record_version)
if self.collection is not None:
if self.collection.known_past_version(to_record_version):
_logger.debug("%s already processed %s", self.collection_locator, to_record_version)
_logger.exception("arv-mount %s: error", self.collection_locator)
if self.collection_record is not None and "manifest_text" in self.collection_record:
_logger.error("arv-mount manifest_text is: %s", self.collection_record["manifest_text"])
+ self.invalidate()
return False
@use_counter
self.collection.save()
self.collection.stop_threads()
+ def clear(self):
+ if self.collection is not None:
+ self.collection.stop_threads()
+ super(CollectionDirectory, self).clear()
+ self._manifest_size = 0
+
class TmpCollectionDirectory(CollectionDirectoryBase):
"""A directory backed by an Arvados collection that never gets saved.
def __init__(self, parent_inode, inodes, api_client, num_retries):
collection = self.UnsaveableCollection(
api_client=api_client,
- keep_client=api_client.keep)
+ keep_client=api_client.keep,
+ num_retries=num_retries)
super(TmpCollectionDirectory, self).__init__(
parent_inode, inodes, collection)
self.collection_record_file = None
def writable(self):
return True
+ def want_event_subscribe(self):
+ return False
+
def finalize(self):
self.collection.stop_threads()
self.inodes.del_entry(e)
return True
else:
+ self.inodes.invalidate_entry(self.inode, k)
self.inodes.del_entry(e)
return False
- except Exception as e:
- _logger.debug('arv-mount exception keep %s', e)
+ except Exception as ex:
+ _logger.debug('arv-mount exception keep %s', ex)
self.inodes.del_entry(e)
return False
else:
raise KeyError("No collection with id " + item)
- def clear(self, force=False):
+ def clear(self):
pass
+ def want_event_subscribe(self):
+ return not self.pdh_only
-class RecursiveInvalidateDirectory(Directory):
- def invalidate(self):
- try:
- super(RecursiveInvalidateDirectory, self).invalidate()
- for a in self._entries:
- self._entries[a].invalidate()
- except Exception:
- _logger.exception()
-
-class TagsDirectory(RecursiveInvalidateDirectory):
+class TagsDirectory(Directory):
"""A special directory that contains as subdirectories all tags visible to the user."""
def __init__(self, parent_inode, inodes, api, num_retries, poll_time=60):
self.num_retries = num_retries
self._poll = True
self._poll_time = poll_time
+ self._extra = set()
+
+ def want_event_subscribe(self):
+ return True
@use_counter
def update(self):
with llfuse.lock_released:
tags = self.api.links().list(
- filters=[['link_class', '=', 'tag']],
- select=['name'], distinct=True
+ filters=[['link_class', '=', 'tag'], ["name", "!=", ""]],
+ select=['name'], distinct=True, limit=1000
).execute(num_retries=self.num_retries)
if "items" in tags:
- self.merge(tags['items'],
+ self.merge(tags['items']+[{"name": n} for n in self._extra],
lambda i: i['name'],
lambda a, i: a.tag == i['name'],
lambda i: TagDirectory(self.inode, self.inodes, self.api, self.num_retries, i['name'], poll=self._poll, poll_time=self._poll_time))
+ @use_counter
+ @check_update
+ def __getitem__(self, item):
+ if super(TagsDirectory, self).__contains__(item):
+ return super(TagsDirectory, self).__getitem__(item)
+ with llfuse.lock_released:
+ tags = self.api.links().list(
+ filters=[['link_class', '=', 'tag'], ['name', '=', item]], limit=1
+ ).execute(num_retries=self.num_retries)
+ if tags["items"]:
+ self._extra.add(item)
+ self.update()
+ return super(TagsDirectory, self).__getitem__(item)
+
+ @use_counter
+ @check_update
+ def __contains__(self, k):
+ if super(TagsDirectory, self).__contains__(k):
+ return True
+ try:
+ self[k]
+ return True
+ except KeyError:
+ pass
+ return False
+
class TagDirectory(Directory):
"""A special directory that contains as subdirectories all collections visible
self._poll = poll
self._poll_time = poll_time
+ def want_event_subscribe(self):
+ return True
+
@use_counter
def update(self):
with llfuse.lock_released:
self._updating_lock = threading.Lock()
self._current_user = None
+ def want_event_subscribe(self):
+ return True
+
def createDirectory(self, i):
if collection_uuid_pattern.match(i['uuid']):
return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i)
lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
except Exception:
_logger.exception()
+
+ def want_event_subscribe(self):
+ return True
projects / arvados.git / blobdiff