X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/05ff1098f0e9eda5d642a1249f8b3a236656320c..8b3478bda6764b3f30aef69ec0a93729495296c0:/services/fuse/arvados_fuse/__init__.py

diff --git a/services/fuse/arvados_fuse/__init__.py b/services/fuse/arvados_fuse/__init__.py
index 4f91de90da..fd25aa9b5e 100644
--- a/services/fuse/arvados_fuse/__init__.py
+++ b/services/fuse/arvados_fuse/__init__.py
@@ -66,6 +66,15 @@ import itertools
 import ciso8601
 import collections
 import functools
+import arvados.keep
+
+import Queue
+
+# Default _notify_queue has a limit of 1000 items, but it really needs to be
+# unlimited to avoid deadlocks, see https://arvados.org/issues/3198#note-43 for
+# details.
+
+llfuse.capi._notify_queue = Queue.Queue()
 
 from fusedir import sanitize_filename, Directory, CollectionDirectory, MagicDirectory, TagsDirectory, ProjectDirectory, SharedDirectory, CollectionDirectoryBase
 from fusefile import StringFile, FuseArvadosFile
@@ -91,7 +100,8 @@ class Handle(object):
         self.obj.dec_use()
 
     def flush(self):
-        return self.obj.flush()
+        if self.obj.writable():
+            return self.obj.flush()
 
 
 class FileHandle(Handle):
@@ -139,7 +149,9 @@ class InodeCache(object):
         self._total -= obj.cache_size
         del self._entries[obj.cache_priority]
         if obj.cache_uuid:
-            del self._by_uuid[obj.cache_uuid]
+            self._by_uuid[obj.cache_uuid].remove(obj)
+            if not self._by_uuid[obj.cache_uuid]:
+                del self._by_uuid[obj.cache_uuid]
             obj.cache_uuid = None
         if clear:
             _logger.debug("InodeCache cleared %i total now %i", obj.inode, self._total)
@@ -159,9 +171,13 @@ class InodeCache(object):
             self._entries[obj.cache_priority] = obj
             obj.cache_uuid = obj.uuid()
             if obj.cache_uuid:
-                self._by_uuid[obj.cache_uuid] = obj
+                if obj.cache_uuid not in self._by_uuid:
+                    self._by_uuid[obj.cache_uuid] = [obj]
+                else:
+                    if obj not in self._by_uuid[obj.cache_uuid]:
+                        self._by_uuid[obj.cache_uuid].append(obj)
             self._total += obj.objsize()
-            _logger.debug("InodeCache touched %i (size %i) total now %i", obj.inode, obj.objsize(), self._total)
+            _logger.debug("InodeCache touched %i (size %i) (uuid %s) total now %i", obj.inode, obj.objsize(), obj.cache_uuid, self._total)
             self.cap_cache()
         else:
             obj.cache_priority = None
@@ -179,6 +195,11 @@ class InodeCache(object):
     def find(self, uuid):
         return self._by_uuid.get(uuid)
 
+    def clear(self):
+        self._entries.clear()
+        self._by_uuid.clear()
+        self._total = 0
+
 class Inodes(object):
     """Manage the set of inodes.  This is the mapping from a numeric id
     to a concrete File or Directory object"""
@@ -188,6 +209,7 @@ class Inodes(object):
         self._counter = itertools.count(llfuse.ROOT_INODE)
         self.inode_cache = inode_cache
         self.encoding = encoding
+        self.deferred_invalidations = []
 
     def __getitem__(self, item):
         return self._entries[item]
@@ -218,16 +240,33 @@ class Inodes(object):
 
     def del_entry(self, entry):
         if entry.ref_count == 0:
-            _logger.debug("Deleting inode %i", entry.inode)
             self.inode_cache.unmanage(entry)
-            llfuse.invalidate_inode(entry.inode)
-            entry.finalize()
             del self._entries[entry.inode]
+            with llfuse.lock_released:
+                entry.finalize()
+            self.invalidate_inode(entry.inode)
             entry.inode = None
         else:
             entry.dead = True
             _logger.debug("del_entry on inode %i with refcount %i", entry.inode, entry.ref_count)
 
+    def invalidate_inode(self, inode):
+        llfuse.invalidate_inode(inode)
+
+    def invalidate_entry(self, inode, name):
+        llfuse.invalidate_entry(inode, name)
+
+    def clear(self):
+        self.inode_cache.clear()
+
+        for k,v in self._entries.items():
+            try:
+                v.finalize()
+            except Exception as e:
+                _logger.exception("Error during finalize of inode %i", k)
+
+        self._entries.clear()
+
 
 def catch_exceptions(orig_func):
     """Catch uncaught exceptions and log them consistently."""
@@ -240,6 +279,12 @@ def catch_exceptions(orig_func):
             raise
         except EnvironmentError as e:
             raise llfuse.FUSEError(e.errno)
+        except arvados.errors.KeepWriteError as e:
+            _logger.error("Keep write error: " + str(e))
+            raise llfuse.FUSEError(errno.EIO)
+        except arvados.errors.NotFoundError as e:
+            _logger.error("Block not found error: " + str(e))
+            raise llfuse.FUSEError(errno.EIO)
         except:
             _logger.exception("Unhandled exception during FUSE operation")
             raise llfuse.FUSEError(errno.EIO)
@@ -259,7 +304,7 @@ class Operations(llfuse.Operations):
 
     """
 
-    def __init__(self, uid, gid, encoding="utf-8", inode_cache=None, num_retries=4):
+    def __init__(self, uid, gid, encoding="utf-8", inode_cache=None, num_retries=4, enable_write=False):
         super(Operations, self).__init__()
 
         if not inode_cache:
@@ -267,6 +312,7 @@ class Operations(llfuse.Operations):
         self.inodes = Inodes(inode_cache, encoding=encoding)
         self.uid = uid
         self.gid = gid
+        self.enable_write = enable_write
 
         # dict of inode to filehandle
         self._filehandles = {}
@@ -278,6 +324,11 @@ class Operations(llfuse.Operations):
 
         self.num_retries = num_retries
 
+        self.read_counter = arvados.keep.Counter()
+        self.write_counter = arvados.keep.Counter()
+        self.read_ops_counter = arvados.keep.Counter()
+        self.write_ops_counter = arvados.keep.Counter()
+
         self.events = None
 
     def init(self):
@@ -291,9 +342,7 @@ class Operations(llfuse.Operations):
             self.events.close()
             self.events = None
 
-        for k,v in self.inodes.items():
-            v.finalize()
-        self.inodes = None
+        self.inodes.clear()
 
     def access(self, inode, mode, ctx):
         return True
@@ -307,15 +356,21 @@ class Operations(llfuse.Operations):
     def on_event(self, ev):
         if 'event_type' in ev:
             with llfuse.lock:
-                item = self.inodes.inode_cache.find(ev["object_uuid"])
-                if item is not None:
-                    item.invalidate()
-                    if ev["object_kind"] == "arvados#collection":
-                        new_attr = ev.get("properties") and ev["properties"].get("new_attributes") and ev["properties"]["new_attributes"]
-                        record_version = (new_attr["modified_at"], new_attr["portable_data_hash"]) if new_attr else None
-                        item.update(to_record_version=record_version)
-                    else:
-                        item.update()
+                items = self.inodes.inode_cache.find(ev["object_uuid"])
+                if items is not None:
+                    for item in items:
+                        item.invalidate()
+                        if ev["object_kind"] == "arvados#collection":
+                            new_attr = ev.get("properties") and ev["properties"].get("new_attributes") and ev["properties"]["new_attributes"]
+
+                            # new_attributes.modified_at currently lacks subsecond precision (see #6347) so use event_at which
+                            # should always be the same.
+                            #record_version = (new_attr["modified_at"], new_attr["portable_data_hash"]) if new_attr else None
+                            record_version = (ev["event_at"], new_attr["portable_data_hash"]) if new_attr else None
+
+                            item.update(to_record_version=record_version)
+                        else:
+                            item.update()
 
                 oldowner = ev.get("properties") and ev["properties"].get("old_attributes") and ev["properties"]["old_attributes"].get("owner_uuid")
                 olditemparent = self.inodes.inode_cache.find(oldowner)
@@ -328,6 +383,7 @@ class Operations(llfuse.Operations):
                     itemparent.invalidate()
                     itemparent.update()
 
+
     @catch_exceptions
     def getattr(self, inode):
         if inode not in self.inodes:
@@ -349,7 +405,7 @@ class Operations(llfuse.Operations):
             if isinstance(e, FuseArvadosFile):
                 entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
 
-        if e.writable():
+        if self.enable_write and e.writable():
             entry.st_mode |= stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH
 
         entry.st_nlink = 1
@@ -434,6 +490,7 @@ class Operations(llfuse.Operations):
     @catch_exceptions
     def read(self, fh, off, size):
         _logger.debug("arv-mount read %i %i %i", fh, off, size)
+        self.read_ops_counter.add(1)
 
         if fh in self._filehandles:
             handle = self._filehandles[fh]
@@ -442,15 +499,16 @@ class Operations(llfuse.Operations):
 
         self.inodes.touch(handle.obj)
 
-        try:
-            return handle.obj.readfrom(off, size, self.num_retries)
-        except arvados.errors.NotFoundError as e:
-            _logger.error("Block not found: " + str(e))
-            raise llfuse.FUSEError(errno.EIO)
+        r = handle.obj.readfrom(off, size, self.num_retries)
+        if r:
+            self.read_counter.add(len(r))
+        return r
 
     @catch_exceptions
     def write(self, fh, off, buf):
         _logger.debug("arv-mount write %i %i %i", fh, off, len(buf))
+        self.write_ops_counter.add(1)
+
         if fh in self._filehandles:
             handle = self._filehandles[fh]
         else:
@@ -461,19 +519,21 @@ class Operations(llfuse.Operations):
 
         self.inodes.touch(handle.obj)
 
-        return handle.obj.writeto(off, buf, self.num_retries)
+        w = handle.obj.writeto(off, buf, self.num_retries)
+        if w:
+            self.write_counter.add(w)
+        return w
 
     @catch_exceptions
     def release(self, fh):
         if fh in self._filehandles:
             try:
                 self._filehandles[fh].flush()
-            except EnvironmentError as e:
-                raise llfuse.FUSEError(e.errno)
             except Exception:
-                _logger.exception("Flush error")
-            self._filehandles[fh].release()
-            del self._filehandles[fh]
+                raise
+            finally:
+                self._filehandles[fh].release()
+                del self._filehandles[fh]
         self.inodes.inode_cache.cap_cache()
 
     def releasedir(self, fh):
@@ -537,6 +597,9 @@ class Operations(llfuse.Operations):
         return st
 
     def _check_writable(self, inode_parent):
+        if not self.enable_write:
+            raise llfuse.FUSEError(errno.EROFS)
+
         if inode_parent in self.inodes:
             p = self.inodes[inode_parent]
         else:
@@ -552,6 +615,8 @@ class Operations(llfuse.Operations):
 
     @catch_exceptions
     def create(self, inode_parent, name, mode, flags, ctx):
+        _logger.debug("arv-mount create: %i '%s' %o", inode_parent, name, mode)
+
         p = self._check_writable(inode_parent)
         p.create(name)