services/fuse/arvados_fuse/__init__.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 """FUSE driver for Arvados Keep
   6
   7 Architecture:
   8
   9 There is one `Operations` object per mount point.  It is the entry point for all
  10 read and write requests from the llfuse module.
  11
  12 The operations object owns an `Inodes` object.  The inodes object stores the
  13 mapping from numeric inode (used throughout the file system API to uniquely
  14 identify files) to the Python objects that implement files and directories.
  15
  16 The `Inodes` object owns an `InodeCache` object.  The inode cache records the
  17 memory footprint of file system objects and when they are last used.  When the
  18 cache limit is exceeded, the least recently used objects are cleared.
  19
  20 File system objects inherit from `fresh.FreshBase` which manages the object lifecycle.
  21
  22 File objects inherit from `fusefile.File`.  Key methods are `readfrom` and `writeto`
  23 which implement actual reads and writes.
  24
  25 Directory objects inherit from `fusedir.Directory`.  The directory object wraps
  26 a Python dict which stores the mapping from filenames to directory entries.
  27 Directory contents can be accessed through the Python operators such as `[]`
  28 and `in`.  These methods automatically check if the directory is fresh (up to
  29 date) or stale (needs update) and will call `update` if necessary before
  30 returing a result.
  31
  32 The general FUSE operation flow is as follows:
  33
  34 - The request handler is called with either an inode or file handle that is the
  35   subject of the operation.
  36
  37 - Look up the inode using the Inodes table or the file handle in the
  38   filehandles table to get the file system object.
  39
  40 - For methods that alter files or directories, check that the operation is
  41   valid and permitted using _check_writable().
  42
  43 - Call the relevant method on the file system object.
  44
  45 - Return the result.
  46
  47 The FUSE driver supports the Arvados event bus.  When an event is received for
  48 an object that is live in the inode cache, that object is immediately updated.
  49
  50 """
  51
  52 from __future__ import absolute_import
  53 from __future__ import division
  54 from future.utils import viewitems
  55 from future.utils import native
  56 from future.utils import listvalues
  57 from future.utils import listitems
  58 from future import standard_library
  59 standard_library.install_aliases()
  60 from builtins import next
  61 from builtins import str
  62 from builtins import object
  63 import os
  64 import llfuse
  65 import errno
  66 import stat
  67 import threading
  68 import arvados
  69 import arvados.events
  70 import logging
  71 import time
  72 import threading
  73 import itertools
  74 import collections
  75 import functools
  76 import arvados.keep
  77 from prometheus_client import Summary
  78 import queue
  79
  80 # Default _notify_queue has a limit of 1000 items, but it really needs to be
  81 # unlimited to avoid deadlocks, see https://arvados.org/issues/3198#note-43 for
  82 # details.
  83
  84 if hasattr(llfuse, 'capi'):
  85     # llfuse < 0.42
  86     llfuse.capi._notify_queue = queue.Queue()
  87 else:
  88     # llfuse >= 0.42
  89     llfuse._notify_queue = queue.Queue()
  90
  91 LLFUSE_VERSION_0 = llfuse.__version__.startswith('0')
  92
  93 from .fusedir import Directory, CollectionDirectory, TmpCollectionDirectory, MagicDirectory, TagsDirectory, ProjectDirectory, SharedDirectory, CollectionDirectoryBase
  94 from .fusefile import StringFile, FuseArvadosFile
  95
  96 _logger = logging.getLogger('arvados.arvados_fuse')
  97
  98 # Uncomment this to enable llfuse debug logging.
  99 # log_handler = logging.StreamHandler()
 100 # llogger = logging.getLogger('llfuse')
 101 # llogger.addHandler(log_handler)
 102 # llogger.setLevel(logging.DEBUG)
 103
 104 class Handle(object):
 105     """Connects a numeric file handle to a File or Directory object that has
 106     been opened by the client."""
 107
 108     def __init__(self, fh, obj):
 109         self.fh = fh
 110         self.obj = obj
 111         self.obj.inc_use()
 112
 113     def release(self):
 114         self.obj.dec_use()
 115
 116     def flush(self):
 117         pass
 118
 119
 120 class FileHandle(Handle):
 121     """Connects a numeric file handle to a File  object that has
 122     been opened by the client."""
 123
 124     def flush(self):
 125         if self.obj.writable():
 126             return self.obj.flush()
 127
 128
 129 class DirectoryHandle(Handle):
 130     """Connects a numeric file handle to a Directory object that has
 131     been opened by the client."""
 132
 133     def __init__(self, fh, dirobj, entries):
 134         super(DirectoryHandle, self).__init__(fh, dirobj)
 135         self.entries = entries
 136
 137
 138 class InodeCache(object):
 139     """Records the memory footprint of objects and when they are last used.
 140
 141     When the cache limit is exceeded, the least recently used objects are
 142     cleared.  Clearing the object means discarding its contents to release
 143     memory.  The next time the object is accessed, it must be re-fetched from
 144     the server.  Note that the inode cache limit is a soft limit; the cache
 145     limit may be exceeded if necessary to load very large objects, it may also
 146     be exceeded if open file handles prevent objects from being cleared.
 147
 148     """
 149
 150     def __init__(self, cap, min_entries=4):
 151         self._entries = collections.OrderedDict()
 152         self._by_uuid = {}
 153         self.cap = cap
 154         self._total = 0
 155         self.min_entries = min_entries
 156
 157     def total(self):
 158         return self._total
 159
 160     def _remove(self, obj, clear):
 161         if clear:
 162             # Kernel behavior seems to be that if a file is
 163             # referenced, its parents remain referenced too. This
 164             # means has_ref() exits early when a collection is not
 165             # candidate for eviction.
 166             #
 167             # By contrast, in_use() doesn't increment references on
 168             # parents, so it requires a full tree walk to determine if
 169             # a collection is a candidate for eviction.  This takes
 170             # .07s for 240000 files, which becomes a major drag when
 171             # cap_cache is being called several times a second and
 172             # there are multiple non-evictable collections in the
 173             # cache.
 174             #
 175             # So it is important for performance that we do the
 176             # has_ref() check first.
 177
 178             if obj.has_ref(True):
 179                 _logger.debug("InodeCache cannot clear inode %i, still referenced", obj.inode)
 180                 return
 181
 182             if obj.in_use():
 183                 _logger.debug("InodeCache cannot clear inode %i, in use", obj.inode)
 184                 return
 185
 186             obj.kernel_invalidate()
 187             _logger.debug("InodeCache sent kernel invalidate inode %i", obj.inode)
 188             obj.clear()
 189
 190         # The llfuse lock is released in del_entry(), which is called by
 191         # Directory.clear().  While the llfuse lock is released, it can happen
 192         # that a reentrant call removes this entry before this call gets to it.
 193         # Ensure that the entry is still valid before trying to remove it.
 194         if obj.inode not in self._entries:
 195             return
 196
 197         self._total -= obj.cache_size
 198         del self._entries[obj.inode]
 199         if obj.cache_uuid:
 200             self._by_uuid[obj.cache_uuid].remove(obj)
 201             if not self._by_uuid[obj.cache_uuid]:
 202                 del self._by_uuid[obj.cache_uuid]
 203             obj.cache_uuid = None
 204         if clear:
 205             _logger.debug("InodeCache cleared inode %i total now %i", obj.inode, self._total)
 206
 207     def cap_cache(self):
 208         if self._total > self.cap:
 209             for ent in listvalues(self._entries):
 210                 if self._total < self.cap or len(self._entries) < self.min_entries:
 211                     break
 212                 self._remove(ent, True)
 213
 214     def manage(self, obj):
 215         if obj.persisted():
 216             obj.cache_size = obj.objsize()
 217             self._entries[obj.inode] = obj
 218             obj.cache_uuid = obj.uuid()
 219             if obj.cache_uuid:
 220                 if obj.cache_uuid not in self._by_uuid:
 221                     self._by_uuid[obj.cache_uuid] = [obj]
 222                 else:
 223                     if obj not in self._by_uuid[obj.cache_uuid]:
 224                         self._by_uuid[obj.cache_uuid].append(obj)
 225             self._total += obj.objsize()
 226             _logger.debug("InodeCache touched inode %i (size %i) (uuid %s) total now %i (%i entries)",
 227                           obj.inode, obj.objsize(), obj.cache_uuid, self._total, len(self._entries))
 228             self.cap_cache()
 229
 230     def touch(self, obj):
 231         if obj.persisted():
 232             if obj.inode in self._entries:
 233                 self._remove(obj, False)
 234             self.manage(obj)
 235
 236     def unmanage(self, obj):
 237         if obj.persisted() and obj.inode in self._entries:
 238             self._remove(obj, True)
 239
 240     def find_by_uuid(self, uuid):
 241         return self._by_uuid.get(uuid, [])
 242
 243     def clear(self):
 244         self._entries.clear()
 245         self._by_uuid.clear()
 246         self._total = 0
 247
 248 class Inodes(object):
 249     """Manage the set of inodes.  This is the mapping from a numeric id
 250     to a concrete File or Directory object"""
 251
 252     def __init__(self, inode_cache, encoding="utf-8"):
 253         self._entries = {}
 254         self._counter = itertools.count(llfuse.ROOT_INODE)
 255         self.inode_cache = inode_cache
 256         self.encoding = encoding
 257         self.deferred_invalidations = []
 258
 259     def __getitem__(self, item):
 260         return self._entries[item]
 261
 262     def __setitem__(self, key, item):
 263         self._entries[key] = item
 264
 265     def __iter__(self):
 266         return iter(self._entries.keys())
 267
 268     def items(self):
 269         return viewitems(self._entries.items())
 270
 271     def __contains__(self, k):
 272         return k in self._entries
 273
 274     def touch(self, entry):
 275         entry._atime = time.time()
 276         self.inode_cache.touch(entry)
 277
 278     def add_entry(self, entry):
 279         entry.inode = next(self._counter)
 280         if entry.inode == llfuse.ROOT_INODE:
 281             entry.inc_ref()
 282         self._entries[entry.inode] = entry
 283         self.inode_cache.manage(entry)
 284         return entry
 285
 286     def del_entry(self, entry):
 287         if entry.ref_count == 0:
 288             self.inode_cache.unmanage(entry)
 289             del self._entries[entry.inode]
 290             with llfuse.lock_released:
 291                 entry.finalize()
 292             entry.inode = None
 293         else:
 294             entry.dead = True
 295             _logger.debug("del_entry on inode %i with refcount %i", entry.inode, entry.ref_count)
 296
 297     def invalidate_inode(self, entry):
 298         if entry.has_ref(False):
 299             # Only necessary if the kernel has previously done a lookup on this
 300             # inode and hasn't yet forgotten about it.
 301             llfuse.invalidate_inode(entry.inode)
 302
 303     def invalidate_entry(self, entry, name):
 304         if entry.has_ref(False):
 305             # Only necessary if the kernel has previously done a lookup on this
 306             # inode and hasn't yet forgotten about it.
 307             llfuse.invalidate_entry(entry.inode, native(name.encode(self.encoding)))
 308
 309     def clear(self):
 310         self.inode_cache.clear()
 311
 312         for k,v in viewitems(self._entries):
 313             try:
 314                 v.finalize()
 315             except Exception as e:
 316                 _logger.exception("Error during finalize of inode %i", k)
 317
 318         self._entries.clear()
 319
 320
 321 def catch_exceptions(orig_func):
 322     """Catch uncaught exceptions and log them consistently."""
 323
 324     @functools.wraps(orig_func)
 325     def catch_exceptions_wrapper(self, *args, **kwargs):
 326         try:
 327             return orig_func(self, *args, **kwargs)
 328         except llfuse.FUSEError:
 329             raise
 330         except EnvironmentError as e:
 331             raise llfuse.FUSEError(e.errno)
 332         except arvados.errors.KeepWriteError as e:
 333             _logger.error("Keep write error: " + str(e))
 334             raise llfuse.FUSEError(errno.EIO)
 335         except arvados.errors.NotFoundError as e:
 336             _logger.error("Block not found error: " + str(e))
 337             raise llfuse.FUSEError(errno.EIO)
 338         except:
 339             _logger.exception("Unhandled exception during FUSE operation")
 340             raise llfuse.FUSEError(errno.EIO)
 341
 342     return catch_exceptions_wrapper
 343
 344
 345 class Operations(llfuse.Operations):
 346     """This is the main interface with llfuse.
 347
 348     The methods on this object are called by llfuse threads to service FUSE
 349     events to query and read from the file system.
 350
 351     llfuse has its own global lock which is acquired before calling a request handler,
 352     so request handlers do not run concurrently unless the lock is explicitly released
 353     using 'with llfuse.lock_released:'
 354
 355     """
 356
 357     fuse_time = Summary('arvmount_fuse_operations_seconds', 'Time spent during FUSE operations', labelnames=['op'])
 358     read_time = fuse_time.labels(op='read')
 359     write_time = fuse_time.labels(op='write')
 360     destroy_time = fuse_time.labels(op='destroy')
 361     on_event_time = fuse_time.labels(op='on_event')
 362     getattr_time = fuse_time.labels(op='getattr')
 363     setattr_time = fuse_time.labels(op='setattr')
 364     lookup_time = fuse_time.labels(op='lookup')
 365     forget_time = fuse_time.labels(op='forget')
 366     open_time = fuse_time.labels(op='open')
 367     release_time = fuse_time.labels(op='release')
 368     opendir_time = fuse_time.labels(op='opendir')
 369     readdir_time = fuse_time.labels(op='readdir')
 370     statfs_time = fuse_time.labels(op='statfs')
 371     create_time = fuse_time.labels(op='create')
 372     mkdir_time = fuse_time.labels(op='mkdir')
 373     unlink_time = fuse_time.labels(op='unlink')
 374     rmdir_time = fuse_time.labels(op='rmdir')
 375     rename_time = fuse_time.labels(op='rename')
 376     flush_time = fuse_time.labels(op='flush')
 377
 378     def __init__(self, uid, gid, api_client, encoding="utf-8", inode_cache=None, num_retries=4, enable_write=False):
 379         super(Operations, self).__init__()
 380
 381         self._api_client = api_client
 382
 383         if not inode_cache:
 384             inode_cache = InodeCache(cap=256*1024*1024)
 385         self.inodes = Inodes(inode_cache, encoding=encoding)
 386         self.uid = uid
 387         self.gid = gid
 388         self.enable_write = enable_write
 389
 390         # dict of inode to filehandle
 391         self._filehandles = {}
 392         self._filehandles_counter = itertools.count(0)
 393
 394         # Other threads that need to wait until the fuse driver
 395         # is fully initialized should wait() on this event object.
 396         self.initlock = threading.Event()
 397
 398         # If we get overlapping shutdown events (e.g., fusermount -u
 399         # -z and operations.destroy()) llfuse calls forget() on inodes
 400         # that have already been deleted. To avoid this, we make
 401         # forget() a no-op if called after destroy().
 402         self._shutdown_started = threading.Event()
 403
 404         self.num_retries = num_retries
 405
 406         self.read_counter = arvados.keep.Counter()
 407         self.write_counter = arvados.keep.Counter()
 408         self.read_ops_counter = arvados.keep.Counter()
 409         self.write_ops_counter = arvados.keep.Counter()
 410
 411         self.events = None
 412
 413     def init(self):
 414         # Allow threads that are waiting for the driver to be finished
 415         # initializing to continue
 416         self.initlock.set()
 417
 418     def metric_samples(self):
 419         return self.fuse_time.collect()[0].samples
 420
 421     def metric_op_names(self):
 422         ops = []
 423         for cur_op in [sample.labels['op'] for sample in self.metric_samples()]:
 424             if cur_op not in ops:
 425                 ops.append(cur_op)
 426         return ops
 427
 428     def metric_value(self, opname, metric):
 429         op_value = [sample.value for sample in self.metric_samples()
 430                     if sample.name == metric and sample.labels['op'] == opname]
 431         return op_value[0] if len(op_value) == 1 else None
 432
 433     def metric_sum_func(self, opname):
 434         return lambda: self.metric_value(opname, "arvmount_fuse_operations_seconds_sum")
 435
 436     def metric_count_func(self, opname):
 437         return lambda: int(self.metric_value(opname, "arvmount_fuse_operations_seconds_count"))
 438
 439     @destroy_time.time()
 440     @catch_exceptions
 441     def destroy(self):
 442         self._shutdown_started.set()
 443         if self.events:
 444             self.events.close()
 445             self.events = None
 446
 447         # Different versions of llfuse require and forbid us to
 448         # acquire the lock here. See #8345#note-37, #10805#note-9.
 449         if LLFUSE_VERSION_0 and llfuse.lock.acquire():
 450             # llfuse < 0.42
 451             self.inodes.clear()
 452             llfuse.lock.release()
 453         else:
 454             # llfuse >= 0.42
 455             self.inodes.clear()
 456
 457     def access(self, inode, mode, ctx):
 458         return True
 459
 460     def listen_for_events(self):
 461         self.events = arvados.events.subscribe(
 462             self._api_client,
 463             [["event_type", "in", ["create", "update", "delete"]]],
 464             self.on_event)
 465
 466     @on_event_time.time()
 467     @catch_exceptions
 468     def on_event(self, ev):
 469         if 'event_type' not in ev or ev["event_type"] not in ("create", "update", "delete"):
 470             return
 471         with llfuse.lock:
 472             properties = ev.get("properties") or {}
 473             old_attrs = properties.get("old_attributes") or {}
 474             new_attrs = properties.get("new_attributes") or {}
 475
 476             for item in self.inodes.inode_cache.find_by_uuid(ev["object_uuid"]):
 477                 item.invalidate()
 478
 479             oldowner = old_attrs.get("owner_uuid")
 480             newowner = ev.get("object_owner_uuid")
 481             for parent in (
 482                     self.inodes.inode_cache.find_by_uuid(oldowner) +
 483                     self.inodes.inode_cache.find_by_uuid(newowner)):
 484                 parent.invalidate()
 485
 486     @getattr_time.time()
 487     @catch_exceptions
 488     def getattr(self, inode, ctx=None):
 489         if inode not in self.inodes:
 490             raise llfuse.FUSEError(errno.ENOENT)
 491
 492         e = self.inodes[inode]
 493
 494         entry = llfuse.EntryAttributes()
 495         entry.st_ino = inode
 496         entry.generation = 0
 497         entry.entry_timeout = 0
 498         entry.attr_timeout = e.time_to_next_poll() if e.allow_attr_cache else 0
 499
 500         entry.st_mode = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
 501         if isinstance(e, Directory):
 502             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFDIR
 503         else:
 504             entry.st_mode |= stat.S_IFREG
 505             if isinstance(e, FuseArvadosFile):
 506                 entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
 507
 508         if self.enable_write and e.writable():
 509             entry.st_mode |= stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH
 510
 511         entry.st_nlink = 1
 512         entry.st_uid = self.uid
 513         entry.st_gid = self.gid
 514         entry.st_rdev = 0
 515
 516         entry.st_size = e.size()
 517
 518         entry.st_blksize = 512
 519         entry.st_blocks = (entry.st_size // 512) + 1
 520         if hasattr(entry, 'st_atime_ns'):
 521             # llfuse >= 0.42
 522             entry.st_atime_ns = int(e.atime() * 1000000000)
 523             entry.st_mtime_ns = int(e.mtime() * 1000000000)
 524             entry.st_ctime_ns = int(e.mtime() * 1000000000)
 525         else:
 526             # llfuse < 0.42
 527             entry.st_atime = int(e.atime)
 528             entry.st_mtime = int(e.mtime)
 529             entry.st_ctime = int(e.mtime)
 530
 531         return entry
 532
 533     @setattr_time.time()
 534     @catch_exceptions
 535     def setattr(self, inode, attr, fields=None, fh=None, ctx=None):
 536         entry = self.getattr(inode)
 537
 538         if fh is not None and fh in self._filehandles:
 539             handle = self._filehandles[fh]
 540             e = handle.obj
 541         else:
 542             e = self.inodes[inode]
 543
 544         if fields is None:
 545             # llfuse < 0.42
 546             update_size = attr.st_size is not None
 547         else:
 548             # llfuse >= 0.42
 549             update_size = fields.update_size
 550         if update_size and isinstance(e, FuseArvadosFile):
 551             with llfuse.lock_released:
 552                 e.arvfile.truncate(attr.st_size)
 553                 entry.st_size = e.arvfile.size()
 554
 555         return entry
 556
 557     @lookup_time.time()
 558     @catch_exceptions
 559     def lookup(self, parent_inode, name, ctx=None):
 560         name = str(name, self.inodes.encoding)
 561         inode = None
 562
 563         if name == '.':
 564             inode = parent_inode
 565         else:
 566             if parent_inode in self.inodes:
 567                 p = self.inodes[parent_inode]
 568                 self.inodes.touch(p)
 569                 if name == '..':
 570                     inode = p.parent_inode
 571                 elif isinstance(p, Directory) and name in p:
 572                     inode = p[name].inode
 573
 574         if inode != None:
 575             _logger.debug("arv-mount lookup: parent_inode %i name '%s' inode %i",
 576                       parent_inode, name, inode)
 577             self.inodes[inode].inc_ref()
 578             return self.getattr(inode)
 579         else:
 580             _logger.debug("arv-mount lookup: parent_inode %i name '%s' not found",
 581                       parent_inode, name)
 582             raise llfuse.FUSEError(errno.ENOENT)
 583
 584     @forget_time.time()
 585     @catch_exceptions
 586     def forget(self, inodes):
 587         if self._shutdown_started.is_set():
 588             return
 589         for inode, nlookup in inodes:
 590             ent = self.inodes[inode]
 591             _logger.debug("arv-mount forget: inode %i nlookup %i ref_count %i", inode, nlookup, ent.ref_count)
 592             if ent.dec_ref(nlookup) == 0 and ent.dead:
 593                 self.inodes.del_entry(ent)
 594
 595     @open_time.time()
 596     @catch_exceptions
 597     def open(self, inode, flags, ctx=None):
 598         if inode in self.inodes:
 599             p = self.inodes[inode]
 600         else:
 601             raise llfuse.FUSEError(errno.ENOENT)
 602
 603         if isinstance(p, Directory):
 604             raise llfuse.FUSEError(errno.EISDIR)
 605
 606         if ((flags & os.O_WRONLY) or (flags & os.O_RDWR)) and not p.writable():
 607             raise llfuse.FUSEError(errno.EPERM)
 608
 609         fh = next(self._filehandles_counter)
 610         self._filehandles[fh] = FileHandle(fh, p)
 611         self.inodes.touch(p)
 612
 613         # Normally, we will have received an "update" event if the
 614         # parent collection is stale here. However, even if the parent
 615         # collection hasn't changed, the manifest might have been
 616         # fetched so long ago that the signatures on the data block
 617         # locators have expired. Calling checkupdate() on all
 618         # ancestors ensures the signatures will be refreshed if
 619         # necessary.
 620         while p.parent_inode in self.inodes:
 621             if p == self.inodes[p.parent_inode]:
 622                 break
 623             p = self.inodes[p.parent_inode]
 624             self.inodes.touch(p)
 625             p.checkupdate()
 626
 627         _logger.debug("arv-mount open inode %i flags %x fh %i", inode, flags, fh)
 628
 629         return fh
 630
 631     @read_time.time()
 632     @catch_exceptions
 633     def read(self, fh, off, size):
 634         _logger.debug("arv-mount read fh %i off %i size %i", fh, off, size)
 635         self.read_ops_counter.add(1)
 636
 637         if fh in self._filehandles:
 638             handle = self._filehandles[fh]
 639         else:
 640             raise llfuse.FUSEError(errno.EBADF)
 641
 642         self.inodes.touch(handle.obj)
 643
 644         r = handle.obj.readfrom(off, size, self.num_retries)
 645         if r:
 646             self.read_counter.add(len(r))
 647         return r
 648
 649     @write_time.time()
 650     @catch_exceptions
 651     def write(self, fh, off, buf):
 652         _logger.debug("arv-mount write %i %i %i", fh, off, len(buf))
 653         self.write_ops_counter.add(1)
 654
 655         if fh in self._filehandles:
 656             handle = self._filehandles[fh]
 657         else:
 658             raise llfuse.FUSEError(errno.EBADF)
 659
 660         if not handle.obj.writable():
 661             raise llfuse.FUSEError(errno.EPERM)
 662
 663         self.inodes.touch(handle.obj)
 664
 665         w = handle.obj.writeto(off, buf, self.num_retries)
 666         if w:
 667             self.write_counter.add(w)
 668         return w
 669
 670     @release_time.time()
 671     @catch_exceptions
 672     def release(self, fh):
 673         if fh in self._filehandles:
 674             _logger.debug("arv-mount release fh %i", fh)
 675             try:
 676                 self._filehandles[fh].flush()
 677             except Exception:
 678                 raise
 679             finally:
 680                 self._filehandles[fh].release()
 681                 del self._filehandles[fh]
 682         self.inodes.inode_cache.cap_cache()
 683
 684     def releasedir(self, fh):
 685         self.release(fh)
 686
 687     @opendir_time.time()
 688     @catch_exceptions
 689     def opendir(self, inode, ctx=None):
 690         _logger.debug("arv-mount opendir: inode %i", inode)
 691
 692         if inode in self.inodes:
 693             p = self.inodes[inode]
 694         else:
 695             raise llfuse.FUSEError(errno.ENOENT)
 696
 697         if not isinstance(p, Directory):
 698             raise llfuse.FUSEError(errno.ENOTDIR)
 699
 700         fh = next(self._filehandles_counter)
 701         if p.parent_inode in self.inodes:
 702             parent = self.inodes[p.parent_inode]
 703         else:
 704             raise llfuse.FUSEError(errno.EIO)
 705
 706         # update atime
 707         self.inodes.touch(p)
 708         self._filehandles[fh] = DirectoryHandle(fh, p, [('.', p), ('..', parent)] + listitems(p))
 709         return fh
 710
 711     @readdir_time.time()
 712     @catch_exceptions
 713     def readdir(self, fh, off):
 714         _logger.debug("arv-mount readdir: fh %i off %i", fh, off)
 715
 716         if fh in self._filehandles:
 717             handle = self._filehandles[fh]
 718         else:
 719             raise llfuse.FUSEError(errno.EBADF)
 720
 721         e = off
 722         while e < len(handle.entries):
 723             if handle.entries[e][1].inode in self.inodes:
 724                 yield (handle.entries[e][0].encode(self.inodes.encoding), self.getattr(handle.entries[e][1].inode), e+1)
 725             e += 1
 726
 727     @statfs_time.time()
 728     @catch_exceptions
 729     def statfs(self, ctx=None):
 730         st = llfuse.StatvfsData()
 731         st.f_bsize = 128 * 1024
 732         st.f_blocks = 0
 733         st.f_files = 0
 734
 735         st.f_bfree = 0
 736         st.f_bavail = 0
 737
 738         st.f_ffree = 0
 739         st.f_favail = 0
 740
 741         st.f_frsize = 0
 742         return st
 743
 744     def _check_writable(self, inode_parent):
 745         if not self.enable_write:
 746             raise llfuse.FUSEError(errno.EROFS)
 747
 748         if inode_parent in self.inodes:
 749             p = self.inodes[inode_parent]
 750         else:
 751             raise llfuse.FUSEError(errno.ENOENT)
 752
 753         if not isinstance(p, Directory):
 754             raise llfuse.FUSEError(errno.ENOTDIR)
 755
 756         if not p.writable():
 757             raise llfuse.FUSEError(errno.EPERM)
 758
 759         return p
 760
 761     @create_time.time()
 762     @catch_exceptions
 763     def create(self, inode_parent, name, mode, flags, ctx=None):
 764         name = name.decode(encoding=self.inodes.encoding)
 765         _logger.debug("arv-mount create: parent_inode %i '%s' %o", inode_parent, name, mode)
 766
 767         p = self._check_writable(inode_parent)
 768         p.create(name)
 769
 770         # The file entry should have been implicitly created by callback.
 771         f = p[name]
 772         fh = next(self._filehandles_counter)
 773         self._filehandles[fh] = FileHandle(fh, f)
 774         self.inodes.touch(p)
 775
 776         f.inc_ref()
 777         return (fh, self.getattr(f.inode))
 778
 779     @mkdir_time.time()
 780     @catch_exceptions
 781     def mkdir(self, inode_parent, name, mode, ctx=None):
 782         name = name.decode(encoding=self.inodes.encoding)
 783         _logger.debug("arv-mount mkdir: parent_inode %i '%s' %o", inode_parent, name, mode)
 784
 785         p = self._check_writable(inode_parent)
 786         p.mkdir(name)
 787
 788         # The dir entry should have been implicitly created by callback.
 789         d = p[name]
 790
 791         d.inc_ref()
 792         return self.getattr(d.inode)
 793
 794     @unlink_time.time()
 795     @catch_exceptions
 796     def unlink(self, inode_parent, name, ctx=None):
 797         name = name.decode(encoding=self.inodes.encoding)
 798         _logger.debug("arv-mount unlink: parent_inode %i '%s'", inode_parent, name)
 799         p = self._check_writable(inode_parent)
 800         p.unlink(name)
 801
 802     @rmdir_time.time()
 803     @catch_exceptions
 804     def rmdir(self, inode_parent, name, ctx=None):
 805         name = name.decode(encoding=self.inodes.encoding)
 806         _logger.debug("arv-mount rmdir: parent_inode %i '%s'", inode_parent, name)
 807         p = self._check_writable(inode_parent)
 808         p.rmdir(name)
 809
 810     @rename_time.time()
 811     @catch_exceptions
 812     def rename(self, inode_parent_old, name_old, inode_parent_new, name_new, ctx=None):
 813         name_old = name_old.decode(encoding=self.inodes.encoding)
 814         name_new = name_new.decode(encoding=self.inodes.encoding)
 815         _logger.debug("arv-mount rename: old_parent_inode %i '%s' new_parent_inode %i '%s'", inode_parent_old, name_old, inode_parent_new, name_new)
 816         src = self._check_writable(inode_parent_old)
 817         dest = self._check_writable(inode_parent_new)
 818         dest.rename(name_old, name_new, src)
 819
 820     @flush_time.time()
 821     @catch_exceptions
 822     def flush(self, fh):
 823         if fh in self._filehandles:
 824             self._filehandles[fh].flush()
 825
 826     def fsync(self, fh, datasync):
 827         self.flush(fh)
 828
 829     def fsyncdir(self, fh, datasync):
 830         self.flush(fh)