import threading
from apiclient import errors as apiclient_errors
import errno
+import time
-from fusefile import StringFile, ObjectFile, FuseArvadosFile
+from fusefile import StringFile, ObjectFile, FuncToJSONFile, FuseArvadosFile
from fresh import FreshBase, convertTime, use_counter, check_update
import arvados.collection
def flush(self):
pass
+ def want_event_subscribe(self):
+ raise NotImplementedError()
+
def create(self, name):
raise NotImplementedError()
self.flush()
src.flush()
+ def clear(self, force=False):
+ r = super(CollectionDirectoryBase, self).clear(force)
+ self.collection = None
+ return r
+
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
+ try:
+ 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
+
if isinstance(collection_record, dict):
self.collection_locator = collection_record['uuid']
self._mtime = convertTime(collection_record.get('modified_at'))
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.
_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
job output.
"""
+ class UnsaveableCollection(arvados.collection.Collection):
+ def save(self):
+ pass
+ def save_new(self):
+ pass
+
def __init__(self, parent_inode, inodes, api_client, num_retries):
- collection = arvados.collection.Collection(
+ collection = self.UnsaveableCollection(
api_client=api_client,
- keep_client=api_client.keep)
- collection.save = self._commit_collection
- collection.save_new = self._commit_collection
+ keep_client=api_client.keep,
+ num_retries=num_retries)
super(TmpCollectionDirectory, self).__init__(
parent_inode, inodes, collection)
self.collection_record_file = None
- self._subscribed = False
- self._update_collection_record()
-
- def update(self, *args, **kwargs):
- if not self._subscribed:
- with llfuse.lock_released:
- self.populate(self.mtime())
- self._subscribed = True
-
- @use_counter
- def _commit_collection(self):
- """Commit the data blocks, but don't save the collection to API.
+ self.populate(self.mtime())
- Update the content of the special .arvados#collection file, if
- it has been instantiated.
- """
- self.collection.flush()
- self._update_collection_record()
- if self.collection_record_file is not None:
- self.collection_record_file.update(self.collection_record)
+ def on_event(self, *args, **kwargs):
+ super(TmpCollectionDirectory, self).on_event(*args, **kwargs)
+ if self.collection_record_file:
+ with llfuse.lock:
+ self.collection_record_file.invalidate()
self.inodes.invalidate_inode(self.collection_record_file.inode)
+ _logger.debug("%s invalidated collection record", self)
- def _update_collection_record(self):
- self.collection_record = {
- "uuid": None,
- "manifest_text": self.collection.manifest_text(),
- "portable_data_hash": self.collection.portable_data_hash(),
- }
+ def collection_record(self):
+ with llfuse.lock_released:
+ return {
+ "uuid": None,
+ "manifest_text": self.collection.manifest_text(),
+ "portable_data_hash": self.collection.portable_data_hash(),
+ }
def __contains__(self, k):
return (k == '.arvados#collection' or
def __getitem__(self, item):
if item == '.arvados#collection':
if self.collection_record_file is None:
- self.collection_record_file = ObjectFile(
+ self.collection_record_file = FuncToJSONFile(
self.inode, self.collection_record)
self.inodes.add_entry(self.collection_record_file)
return self.collection_record_file
return super(TmpCollectionDirectory, self).__getitem__(item)
+ def persisted(self):
+ return False
+
def writable(self):
return True
+ def want_event_subscribe(self):
+ return False
+
def finalize(self):
self.collection.stop_threads()
+ def invalidate(self):
+ if self.collection_record_file:
+ self.collection_record_file.invalidate()
+ super(TmpCollectionDirectory, self).invalidate()
+
class MagicDirectory(Directory):
"""A special directory that logically contains the set of all extant keep locators.
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
def clear(self, force=False):
pass
+ def want_event_subscribe(self):
+ return not self.pdh_only
+
class RecursiveInvalidateDirectory(Directory):
def invalidate(self):
self._poll = True
self._poll_time = poll_time
+ def want_event_subscribe(self):
+ return True
+
@use_counter
def update(self):
with llfuse.lock_released:
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