-#
-# FUSE driver for Arvados Keep
-#
+"""FUSE driver for Arvados Keep
+
+Architecture:
+
+There is one `Operations` object per mount point. It is the entry point for all
+read and write requests from the llfuse module.
+
+The operations object owns an `Inodes` object. The inodes object stores the
+mapping from numeric inode (used throughout the file system API to uniquely
+identify files) to the Python objects that implement files and directories.
+
+The `Inodes` object owns an `InodeCache` object. The inode cache records the
+memory footprint of file system objects and when they are last used. When the
+cache limit is exceeded, the least recently used objects are cleared.
+
+File system objects inherit from `fresh.FreshBase` which manages the object lifecycle.
+
+File objects inherit from `fusefile.File`. Key methods are `readfrom` and `writeto`
+which implement actual reads and writes.
+
+Directory objects inherit from `fusedir.Directory`. The directory object wraps
+a Python dict which stores the mapping from filenames to directory entries.
+Directory contents can be accessed through the Python operators such as `[]`
+and `in`. These methods automatically check if the directory is fresh (up to
+date) or stale (needs update) and will call `update` if necessary before
+returing a result.
+
+The general FUSE operation flow is as follows:
+
+- The request handler is called with either an inode or file handle that is the
+ subject of the operation.
+
+- Look up the inode using the Inodes table or the file handle in the
+ filehandles table to get the file system object.
+
+- For methods that alter files or directories, check that the operation is
+ valid and permitted using _check_writable().
+
+- Call the relevant method on the file system object.
+
+- Return the result.
+
+The FUSE driver supports the Arvados event bus. When an event is received for
+an object that is live in the inode cache, that object is immediately updated.
+
+"""
import os
import sys
class InodeCache(object):
+ """Records the memory footprint of objects and when they are last used.
+
+ When the cache limit is exceeded, the least recently used objects are
+ cleared. Clearing the object means discarding its contents to release
+ memory. The next time the object is accessed, it must be re-fetched from
+ the server. Note that the inode cache limit is a soft limit; the cache
+ limit may be exceeded if necessary to load very large objects, it may also
+ be exceeded if open file handles prevent objects from being cleared.
+
+ """
+
def __init__(self, cap, min_entries=4):
self._entries = collections.OrderedDict()
self._by_uuid = {}
entry.dead = True
_logger.debug("del_entry on inode %i with refcount %i", entry.inode, entry.ref_count)
+
def catch_exceptions(orig_func):
+ """Catch uncaught exceptions and log them consistently."""
+
@functools.wraps(orig_func)
def catch_exceptions_wrapper(self, *args, **kwargs):
try:
@catch_exceptions
def statfs(self):
st = llfuse.StatvfsData()
- st.f_bsize = 64 * 1024
+ st.f_bsize = 128 * 1024
st.f_blocks = 0
st.f_files = 0
self.dec_use()
return use_counter_wrapper
+def check_update(orig_func):
+ @functools.wraps(orig_func)
+ def check_update_wrapper(self, *args, **kwargs):
+ self.checkupdate()
+ return orig_func(self, *args, **kwargs)
+ return check_update_wrapper
+
class FreshBase(object):
- """Base class for maintaining fresh/stale state to determine when to update."""
+ """Base class for maintaining object lifecycle.
+
+ Functions include:
+
+ * Indicate if an object is up to date (stale() == false) or needs to be
+ updated sets stale() == True). Use invalidate() to mark the object as
+ stale. An object is also automatically stale if it has not been updated
+ in `_poll_time` seconds.
+
+ * Record access time (atime) timestamp
+
+ * Manage internal use count used by the inode cache ("inc_use" and
+ "dec_use"). An object which is in use cannot be cleared by the inode
+ cache.
+
+ * Manage the kernel reference count ("inc_ref" and "dec_ref"). An object
+ which is referenced by the kernel cannot have its inode entry deleted.
+
+ * Record cache footprint, cache priority
+
+ * Record Arvados uuid at the time the object is placed in the cache
+
+ * Clear the object contents (invalidates the object)
+
+ """
def __init__(self):
self._stale = True
self._poll = False
import errno
from fusefile import StringFile, ObjectFile, FuseArvadosFile
-from fresh import FreshBase, convertTime, use_counter
+from fresh import FreshBase, convertTime, use_counter, check_update
import arvados.collection
from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
"""
def __init__(self, parent_inode, inodes):
+ """parent_inode is the integer inode number"""
+
super(Directory, self).__init__()
- """parent_inode is the integer inode number"""
self.inode = None
if not isinstance(parent_inode, int):
raise Exception("parent_inode should be an int")
_logger.warn(e)
@use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
return self._entries[item]
@use_counter
+ @check_update
def items(self):
- self.checkupdate()
return list(self._entries.items())
@use_counter
+ @check_update
def __contains__(self, k):
- self.checkupdate()
return k in self._entries
@use_counter
+ @check_update
def __len__(self):
- self.checkupdate()
return len(self._entries)
def fresh(self):
def rename(self, name_old, name_new, src):
raise NotImplementedError()
+
class CollectionDirectoryBase(Directory):
+ """Represent an Arvados Collection as a directory.
+
+ This class is used for Subcollections, and is also the base class for
+ CollectionDirectory, which implements collection loading/saving on
+ Collection records.
+
+ Most operations act only the underlying Arvados `Collection` object. The
+ `Collection` object signals via a notify callback to
+ `CollectionDirectoryBase.on_event` that an item was added, removed or
+ modified. FUSE inodes and directory entries are created, deleted or
+ invalidated in response to these events.
+
+ """
+
def __init__(self, parent_inode, inodes, collection):
super(CollectionDirectoryBase, self).__init__(parent_inode, inodes)
self.collection = collection
def writable(self):
return self.collection.writable()
+ @use_counter
def flush(self):
with llfuse.lock_released:
self.collection.root_collection().save()
+ @use_counter
+ @check_update
def create(self, name):
with llfuse.lock_released:
self.collection.open(name, "w").close()
+ @use_counter
+ @check_update
def mkdir(self, name):
with llfuse.lock_released:
self.collection.mkdirs(name)
+ @use_counter
+ @check_update
def unlink(self, name):
with llfuse.lock_released:
self.collection.remove(name)
self.flush()
+ @use_counter
+ @check_update
def rmdir(self, name):
with llfuse.lock_released:
self.collection.remove(name)
self.flush()
+ @use_counter
+ @check_update
def rename(self, name_old, name_new, src):
if not isinstance(src, CollectionDirectoryBase):
raise llfuse.FUSEError(errno.EPERM)
class CollectionDirectory(CollectionDirectoryBase):
- """Represents the root of a directory tree holding a collection."""
+ """Represents the root of a directory tree representing a collection."""
def __init__(self, parent_inode, inodes, api, num_retries, collection_record=None, explicit_collection=None):
super(CollectionDirectory, self).__init__(parent_inode, inodes, None)
def uuid(self):
return self.collection_locator
+ @use_counter
def update(self):
try:
if self.collection_record is not None and portable_data_hash_pattern.match(self.collection_locator):
_logger.error("arv-mount manifest_text is: %s", self.collection_record["manifest_text"])
return False
+ @use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
if item == '.arvados#collection':
if self.collection_record_file is None:
self.collection_record_file = ObjectFile(self.inode, self.collection_record)
# footprint directly would be more accurate, but also more complicated.
return self._manifest_size * 128
+
class MagicDirectory(Directory):
"""A special directory that logically contains the set of all extant keep locators.
self._poll = True
self._poll_time = poll_time
+ @use_counter
def update(self):
with llfuse.lock_released:
tags = self.api.links().list(
self._poll = poll
self._poll_time = poll_time
+ @use_counter
def update(self):
with llfuse.lock_released:
taggedcollections = self.api.links().list(
def uuid(self):
return self.project_uuid
+ @use_counter
def update(self):
if self.project_object_file == None:
self.project_object_file = ObjectFile(self.inode, self.project_object)
finally:
self._updating_lock.release()
+ @use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
if item == '.arvados#project':
return self.project_object_file
else:
else:
return super(ProjectDirectory, self).__contains__(k)
+ @use_counter
+ @check_update
def writable(self):
with llfuse.lock_released:
if not self._current_user:
def persisted(self):
return True
+ @use_counter
+ @check_update
def mkdir(self, name):
try:
with llfuse.lock_released:
_logger.error(error)
raise llfuse.FUSEError(errno.EEXIST)
+ @use_counter
+ @check_update
def rmdir(self, name):
if name not in self:
raise llfuse.FUSEError(errno.ENOENT)
self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
self.invalidate()
+ @use_counter
+ @check_update
def rename(self, name_old, name_new, src):
if not isinstance(src, ProjectDirectory):
raise llfuse.FUSEError(errno.EPERM)
self._entries[name_new] = ent
llfuse.invalidate_entry(src.inode, name_old)
+
class SharedDirectory(Directory):
"""A special directory that represents users or groups who have shared projects with me."""
self._poll = True
self._poll_time = poll_time
+ @use_counter
def update(self):
with llfuse.lock_released:
all_projects = arvados.util.list_all(
projects / arvados.git / commitdiff