Merge branch '3198-inode-cache' into 3198-writable-fuse

[arvados.git] / services / fuse / arvados_fuse / __init__.py

diff --git a/services/fuse/arvados_fuse/__init__.py b/services/fuse/arvados_fuse/__init__.py
index 83b4710da818571280eb6544f3fb462f4d8ce787..58fc5718cccd3405be2c9d48dc6d9447d0e855de 100644 (file)
--- a/services/fuse/arvados_fuse/__init__.py
+++ b/services/fuse/arvados_fuse/__init__.py
@@ -24,7 +24,7 @@ import ciso8601
 import collections
 
 from fusedir import sanitize_filename, Directory, CollectionDirectory, MagicDirectory, TagsDirectory, ProjectDirectory, SharedDirectory
-from fusefile import StreamReaderFile, StringFile
+from fusefile import StringFile, FuseArvadosFile
 
 _logger = logging.getLogger('arvados.arvados_fuse')
 
@@ -65,15 +65,15 @@ class InodeCache(object):
 
     def _remove(self, obj, clear):
         if clear and not obj.clear():
-            _logger.warn("Could not clear %s in_use %s", obj, obj.in_use())
+            _logger.debug("Could not clear %s in_use %s", obj, obj.in_use())
             return False
         self._total -= obj._cache_size
         del self._entries[obj._cache_priority]
-        _logger.warn("Cleared %s total now %i", obj, self._total)
+        _logger.debug("Cleared %s total now %i", obj, self._total)
         return True
 
     def cap_cache(self):
-        _logger.warn("total is %i cap is %i", self._total, self.cap)
+        _logger.debug("total is %i cap is %i", self._total, self.cap)
         if self._total > self.cap:
             need_gc = False
             for key in list(self._entries.keys()):
@@ -88,7 +88,7 @@ class InodeCache(object):
             obj._cache_size = obj.objsize()
             self._entries[obj._cache_priority] = obj
             self._total += obj.objsize()
-            _logger.warn("Managing %s total now %i", obj, self._total)
+            _logger.debug("Managing %s total now %i", obj, self._total)
             self.cap_cache()
 
     def touch(self, obj):
@@ -96,7 +96,7 @@ class InodeCache(object):
             if obj._cache_priority in self._entries:
                 self._remove(obj, False)
             self.manage(obj)
-            _logger.warn("Touched %s (%i) total now %i", obj, obj.objsize(), self._total)
+            _logger.debug("Touched %s (%i) total now %i", obj, obj.objsize(), self._total)
 
     def unmanage(self, obj):
         if obj.persisted() and obj._cache_priority in self._entries:
@@ -157,7 +157,7 @@ class Operations(llfuse.Operations):
 
     """
 
-    def __init__(self, uid, gid, encoding="utf-8", inode_cache=1000):
+    def __init__(self, uid, gid, encoding="utf-8", inode_cache=1000, num_retries=7):
         super(Operations, self).__init__()
 
         self.inodes = Inodes(inode_cache)
@@ -173,6 +173,8 @@ class Operations(llfuse.Operations):
         # is fully initialized should wait() on this event object.
         self.initlock = threading.Event()
 
+        self.num_retries = num_retries
+
     def init(self):
         # Allow threads that are waiting for the driver to be finished
         # initializing to continue
@@ -196,7 +198,7 @@ class Operations(llfuse.Operations):
         entry.st_mode = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
         if isinstance(e, Directory):
             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFDIR
-        elif isinstance(e, StreamReaderFile):
+        elif isinstance(e, FuseArvadosFile):
             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFREG
         else:
             entry.st_mode |= stat.S_IFREG
@@ -266,12 +268,12 @@ class Operations(llfuse.Operations):
 
         try:
             with llfuse.lock_released:
-                return handle.fileobj.readfrom(off, size)
+                return handle.fileobj.readfrom(off, size, self.num_retries)
         except arvados.errors.NotFoundError as e:
             _logger.warning("Block not found: " + str(e))
             raise llfuse.FUSEError(errno.EIO)
         except Exception:
-            _logger.exception()
+            _logger.exception("Read error")
             raise llfuse.FUSEError(errno.EIO)
 
     def release(self, fh):