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 import os
  53 import sys
  54 import llfuse
  55 import errno
  56 import stat
  57 import threading
  58 import arvados
  59 import pprint
  60 import arvados.events
  61 import re
  62 import apiclient
  63 import json
  64 import logging
  65 import time
  66 import _strptime
  67 import calendar
  68 import threading
  69 import itertools
  70 import ciso8601
  71 import collections
  72 import functools
  73 import arvados.keep
  74
  75 import Queue
  76
  77 # Default _notify_queue has a limit of 1000 items, but it really needs to be
  78 # unlimited to avoid deadlocks, see https://arvados.org/issues/3198#note-43 for
  79 # details.
  80
  81 if hasattr(llfuse, 'capi'):
  82     # llfuse < 0.42
  83     llfuse.capi._notify_queue = Queue.Queue()
  84 else:
  85     # llfuse >= 0.42
  86     llfuse._notify_queue = Queue.Queue()
  87
  88 LLFUSE_VERSION_0 = llfuse.__version__.startswith('0')
  89
  90 from fusedir import sanitize_filename, Directory, CollectionDirectory, TmpCollectionDirectory, MagicDirectory, TagsDirectory, ProjectDirectory, SharedDirectory, CollectionDirectoryBase
  91 from fusefile import StringFile, FuseArvadosFile
  92
  93 _logger = logging.getLogger('arvados.arvados_fuse')
  94
  95 # Uncomment this to enable llfuse debug logging.
  96 # log_handler = logging.StreamHandler()
  97 # llogger = logging.getLogger('llfuse')
  98 # llogger.addHandler(log_handler)
  99 # llogger.setLevel(logging.DEBUG)
 100
 101 class Handle(object):
 102     """Connects a numeric file handle to a File or Directory object that has
 103     been opened by the client."""
 104
 105     def __init__(self, fh, obj):
 106         self.fh = fh
 107         self.obj = obj
 108         self.obj.inc_use()
 109
 110     def release(self):
 111         self.obj.dec_use()
 112
 113     def flush(self):
 114         pass
 115
 116
 117 class FileHandle(Handle):
 118     """Connects a numeric file handle to a File  object that has
 119     been opened by the client."""
 120
 121     def flush(self):
 122         if self.obj.writable():
 123             return self.obj.flush()
 124
 125
 126 class DirectoryHandle(Handle):
 127     """Connects a numeric file handle to a Directory object that has
 128     been opened by the client."""
 129
 130     def __init__(self, fh, dirobj, entries):
 131         super(DirectoryHandle, self).__init__(fh, dirobj)
 132         self.entries = entries
 133
 134
 135 class InodeCache(object):
 136     """Records the memory footprint of objects and when they are last used.
 137
 138     When the cache limit is exceeded, the least recently used objects are
 139     cleared.  Clearing the object means discarding its contents to release
 140     memory.  The next time the object is accessed, it must be re-fetched from
 141     the server.  Note that the inode cache limit is a soft limit; the cache
 142     limit may be exceeded if necessary to load very large objects, it may also
 143     be exceeded if open file handles prevent objects from being cleared.
 144
 145     """
 146
 147     def __init__(self, cap, min_entries=4):
 148         self._entries = collections.OrderedDict()
 149         self._by_uuid = {}
 150         self.cap = cap
 151         self._total = 0
 152         self.min_entries = min_entries
 153
 154     def total(self):
 155         return self._total
 156
 157     def _remove(self, obj, clear):
 158         if clear:
 159             if obj.in_use():
 160                 _logger.debug("InodeCache cannot clear inode %i, in use", obj.inode)
 161                 return
 162             obj.kernel_invalidate()
 163             if obj.has_ref(True):
 164                 _logger.debug("InodeCache sent kernel invalidate inode %i", obj.inode)
 165                 return
 166             obj.clear()
 167
 168         # The llfuse lock is released in del_entry(), which is called by
 169         # Directory.clear().  While the llfuse lock is released, it can happen
 170         # that a reentrant call removes this entry before this call gets to it.
 171         # Ensure that the entry is still valid before trying to remove it.
 172         if obj.inode not in self._entries:
 173             return
 174
 175         self._total -= obj.cache_size
 176         del self._entries[obj.inode]
 177         if obj.cache_uuid:
 178             self._by_uuid[obj.cache_uuid].remove(obj)
 179             if not self._by_uuid[obj.cache_uuid]:
 180                 del self._by_uuid[obj.cache_uuid]
 181             obj.cache_uuid = None
 182         if clear:
 183             _logger.debug("InodeCache cleared inode %i total now %i", obj.inode, self._total)
 184
 185     def cap_cache(self):
 186         if self._total > self.cap:
 187             for ent in self._entries.values():
 188                 if self._total < self.cap or len(self._entries) < self.min_entries:
 189                     break
 190                 self._remove(ent, True)
 191
 192     def manage(self, obj):
 193         if obj.persisted():
 194             obj.cache_size = obj.objsize()
 195             self._entries[obj.inode] = obj
 196             obj.cache_uuid = obj.uuid()
 197             if obj.cache_uuid:
 198                 if obj.cache_uuid not in self._by_uuid:
 199                     self._by_uuid[obj.cache_uuid] = [obj]
 200                 else:
 201                     if obj not in self._by_uuid[obj.cache_uuid]:
 202                         self._by_uuid[obj.cache_uuid].append(obj)
 203             self._total += obj.objsize()
 204             _logger.debug("InodeCache touched inode %i (size %i) (uuid %s) total now %i", obj.inode, obj.objsize(), obj.cache_uuid, self._total)
 205             self.cap_cache()
 206
 207     def touch(self, obj):
 208         if obj.persisted():
 209             if obj.inode in self._entries:
 210                 self._remove(obj, False)
 211             self.manage(obj)
 212
 213     def unmanage(self, obj):
 214         if obj.persisted() and obj.inode in self._entries:
 215             self._remove(obj, True)
 216
 217     def find_by_uuid(self, uuid):
 218         return self._by_uuid.get(uuid, [])
 219
 220     def clear(self):
 221         self._entries.clear()
 222         self._by_uuid.clear()
 223         self._total = 0
 224
 225 class Inodes(object):
 226     """Manage the set of inodes.  This is the mapping from a numeric id
 227     to a concrete File or Directory object"""
 228
 229     def __init__(self, inode_cache, encoding="utf-8"):
 230         self._entries = {}
 231         self._counter = itertools.count(llfuse.ROOT_INODE)
 232         self.inode_cache = inode_cache
 233         self.encoding = encoding
 234         self.deferred_invalidations = []
 235
 236     def __getitem__(self, item):
 237         return self._entries[item]
 238
 239     def __setitem__(self, key, item):
 240         self._entries[key] = item
 241
 242     def __iter__(self):
 243         return self._entries.iterkeys()
 244
 245     def items(self):
 246         return self._entries.items()
 247
 248     def __contains__(self, k):
 249         return k in self._entries
 250
 251     def touch(self, entry):
 252         entry._atime = time.time()
 253         self.inode_cache.touch(entry)
 254
 255     def add_entry(self, entry):
 256         entry.inode = next(self._counter)
 257         if entry.inode == llfuse.ROOT_INODE:
 258             entry.inc_ref()
 259         self._entries[entry.inode] = entry
 260         self.inode_cache.manage(entry)
 261         return entry
 262
 263     def del_entry(self, entry):
 264         if entry.ref_count == 0:
 265             self.inode_cache.unmanage(entry)
 266             del self._entries[entry.inode]
 267             with llfuse.lock_released:
 268                 entry.finalize()
 269             entry.inode = None
 270         else:
 271             entry.dead = True
 272             _logger.debug("del_entry on inode %i with refcount %i", entry.inode, entry.ref_count)
 273
 274     def invalidate_inode(self, entry):
 275         if entry.has_ref(False):
 276             # Only necessary if the kernel has previously done a lookup on this
 277             # inode and hasn't yet forgotten about it.
 278             llfuse.invalidate_inode(entry.inode)
 279
 280     def invalidate_entry(self, entry, name):
 281         if entry.has_ref(False):
 282             # Only necessary if the kernel has previously done a lookup on this
 283             # inode and hasn't yet forgotten about it.
 284             llfuse.invalidate_entry(entry.inode, name.encode(self.encoding))
 285
 286     def clear(self):
 287         self.inode_cache.clear()
 288
 289         for k,v in self._entries.items():
 290             try:
 291                 v.finalize()
 292             except Exception as e:
 293                 _logger.exception("Error during finalize of inode %i", k)
 294
 295         self._entries.clear()
 296
 297
 298 def catch_exceptions(orig_func):
 299     """Catch uncaught exceptions and log them consistently."""
 300
 301     @functools.wraps(orig_func)
 302     def catch_exceptions_wrapper(self, *args, **kwargs):
 303         try:
 304             return orig_func(self, *args, **kwargs)
 305         except llfuse.FUSEError:
 306             raise
 307         except EnvironmentError as e:
 308             raise llfuse.FUSEError(e.errno)
 309         except arvados.errors.KeepWriteError as e:
 310             _logger.error("Keep write error: " + str(e))
 311             raise llfuse.FUSEError(errno.EIO)
 312         except arvados.errors.NotFoundError as e:
 313             _logger.error("Block not found error: " + str(e))
 314             raise llfuse.FUSEError(errno.EIO)
 315         except:
 316             _logger.exception("Unhandled exception during FUSE operation")
 317             raise llfuse.FUSEError(errno.EIO)
 318
 319     return catch_exceptions_wrapper
 320
 321
 322 class Operations(llfuse.Operations):
 323     """This is the main interface with llfuse.
 324
 325     The methods on this object are called by llfuse threads to service FUSE
 326     events to query and read from the file system.
 327
 328     llfuse has its own global lock which is acquired before calling a request handler,
 329     so request handlers do not run concurrently unless the lock is explicitly released
 330     using 'with llfuse.lock_released:'
 331
 332     """
 333
 334     def __init__(self, uid, gid, api_client, encoding="utf-8", inode_cache=None, num_retries=4, enable_write=False):
 335         super(Operations, self).__init__()
 336
 337         self._api_client = api_client
 338
 339         if not inode_cache:
 340             inode_cache = InodeCache(cap=256*1024*1024)
 341         self.inodes = Inodes(inode_cache, encoding=encoding)
 342         self.uid = uid
 343         self.gid = gid
 344         self.enable_write = enable_write
 345
 346         # dict of inode to filehandle
 347         self._filehandles = {}
 348         self._filehandles_counter = itertools.count(0)
 349
 350         # Other threads that need to wait until the fuse driver
 351         # is fully initialized should wait() on this event object.
 352         self.initlock = threading.Event()
 353
 354         # If we get overlapping shutdown events (e.g., fusermount -u
 355         # -z and operations.destroy()) llfuse calls forget() on inodes
 356         # that have already been deleted. To avoid this, we make
 357         # forget() a no-op if called after destroy().
 358         self._shutdown_started = threading.Event()
 359
 360         self.num_retries = num_retries
 361
 362         self.read_counter = arvados.keep.Counter()
 363         self.write_counter = arvados.keep.Counter()
 364         self.read_ops_counter = arvados.keep.Counter()
 365         self.write_ops_counter = arvados.keep.Counter()
 366
 367         self.events = None
 368
 369     def init(self):
 370         # Allow threads that are waiting for the driver to be finished
 371         # initializing to continue
 372         self.initlock.set()
 373
 374     @catch_exceptions
 375     def destroy(self):
 376         self._shutdown_started.set()
 377         if self.events:
 378             self.events.close()
 379             self.events = None
 380
 381         # Different versions of llfuse require and forbid us to
 382         # acquire the lock here. See #8345#note-37, #10805#note-9.
 383         if LLFUSE_VERSION_0 and llfuse.lock.acquire():
 384             # llfuse < 0.42
 385             self.inodes.clear()
 386             llfuse.lock.release()
 387         else:
 388             # llfuse >= 0.42
 389             self.inodes.clear()
 390
 391     def access(self, inode, mode, ctx):
 392         return True
 393
 394     def listen_for_events(self):
 395         self.events = arvados.events.subscribe(
 396             self._api_client,
 397             [["event_type", "in", ["create", "update", "delete"]]],
 398             self.on_event)
 399
 400     @catch_exceptions
 401     def on_event(self, ev):
 402         if 'event_type' not in ev or ev["event_type"] not in ("create", "update", "delete"):
 403             return
 404         with llfuse.lock:
 405             properties = ev.get("properties") or {}
 406             old_attrs = properties.get("old_attributes") or {}
 407             new_attrs = properties.get("new_attributes") or {}
 408
 409             for item in self.inodes.inode_cache.find_by_uuid(ev["object_uuid"]):
 410                 item.invalidate()
 411                 if ev.get("object_kind") == "arvados#collection":
 412                     pdh = new_attrs.get("portable_data_hash")
 413                     # new_attributes.modified_at currently lacks
 414                     # subsecond precision (see #6347) so use event_at
 415                     # which should always be the same.
 416                     stamp = ev.get("event_at")
 417                     if (stamp and pdh and item.writable() and
 418                         item.collection is not None and
 419                         item.collection.modified() and
 420                         new_attrs.get("is_trashed") is not True):
 421                         item.update(to_record_version=(stamp, pdh))
 422
 423             oldowner = old_attrs.get("owner_uuid")
 424             newowner = ev.get("object_owner_uuid")
 425             for parent in (
 426                     self.inodes.inode_cache.find_by_uuid(oldowner) +
 427                     self.inodes.inode_cache.find_by_uuid(newowner)):
 428                 parent.child_event(ev)
 429
 430     @catch_exceptions
 431     def getattr(self, inode, ctx=None):
 432         if inode not in self.inodes:
 433             raise llfuse.FUSEError(errno.ENOENT)
 434
 435         e = self.inodes[inode]
 436
 437         entry = llfuse.EntryAttributes()
 438         entry.st_ino = inode
 439         entry.generation = 0
 440         entry.attr_timeout = e.time_to_next_poll() if e.allow_attr_cache else 0
 441
 442         entry.st_mode = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
 443         if isinstance(e, Directory):
 444             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFDIR
 445         else:
 446             entry.st_mode |= stat.S_IFREG
 447             if isinstance(e, FuseArvadosFile):
 448                 entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
 449
 450         if self.enable_write and e.writable():
 451             entry.st_mode |= stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH
 452
 453         entry.st_nlink = 1
 454         entry.st_uid = self.uid
 455         entry.st_gid = self.gid
 456         entry.st_rdev = 0
 457
 458         entry.st_size = e.size()
 459
 460         entry.st_blksize = 512
 461         entry.st_blocks = (entry.st_size/512)+1
 462         if hasattr(entry, 'st_atime_ns'):
 463             # llfuse >= 0.42
 464             entry.st_atime_ns = int(e.atime() * 1000000000)
 465             entry.st_mtime_ns = int(e.mtime() * 1000000000)
 466             entry.st_ctime_ns = int(e.mtime() * 1000000000)
 467         else:
 468             # llfuse < 0.42
 469             entry.st_atime = int(e.atime)
 470             entry.st_mtime = int(e.mtime)
 471             entry.st_ctime = int(e.mtime)
 472
 473         return entry
 474
 475     @catch_exceptions
 476     def setattr(self, inode, attr, fields=None, fh=None, ctx=None):
 477         entry = self.getattr(inode)
 478
 479         if fh is not None and fh in self._filehandles:
 480             handle = self._filehandles[fh]
 481             e = handle.obj
 482         else:
 483             e = self.inodes[inode]
 484
 485         if fields is None:
 486             # llfuse < 0.42
 487             update_size = attr.st_size is not None
 488         else:
 489             # llfuse >= 0.42
 490             update_size = fields.update_size
 491         if update_size and isinstance(e, FuseArvadosFile):
 492             with llfuse.lock_released:
 493                 e.arvfile.truncate(attr.st_size)
 494                 entry.st_size = e.arvfile.size()
 495
 496         return entry
 497
 498     @catch_exceptions
 499     def lookup(self, parent_inode, name, ctx=None):
 500         name = unicode(name, self.inodes.encoding)
 501         inode = None
 502
 503         if name == '.':
 504             inode = parent_inode
 505         else:
 506             if parent_inode in self.inodes:
 507                 p = self.inodes[parent_inode]
 508                 self.inodes.touch(p)
 509                 if name == '..':
 510                     inode = p.parent_inode
 511                 elif isinstance(p, Directory) and name in p:
 512                     inode = p[name].inode
 513
 514         if inode != None:
 515             _logger.debug("arv-mount lookup: parent_inode %i name '%s' inode %i",
 516                       parent_inode, name, inode)
 517             self.inodes[inode].inc_ref()
 518             return self.getattr(inode)
 519         else:
 520             _logger.debug("arv-mount lookup: parent_inode %i name '%s' not found",
 521                       parent_inode, name)
 522             raise llfuse.FUSEError(errno.ENOENT)
 523
 524     @catch_exceptions
 525     def forget(self, inodes):
 526         if self._shutdown_started.is_set():
 527             return
 528         for inode, nlookup in inodes:
 529             ent = self.inodes[inode]
 530             _logger.debug("arv-mount forget: inode %i nlookup %i ref_count %i", inode, nlookup, ent.ref_count)
 531             if ent.dec_ref(nlookup) == 0 and ent.dead:
 532                 self.inodes.del_entry(ent)
 533
 534     @catch_exceptions
 535     def open(self, inode, flags, ctx=None):
 536         if inode in self.inodes:
 537             p = self.inodes[inode]
 538         else:
 539             raise llfuse.FUSEError(errno.ENOENT)
 540
 541         if isinstance(p, Directory):
 542             raise llfuse.FUSEError(errno.EISDIR)
 543
 544         if ((flags & os.O_WRONLY) or (flags & os.O_RDWR)) and not p.writable():
 545             raise llfuse.FUSEError(errno.EPERM)
 546
 547         fh = next(self._filehandles_counter)
 548         self._filehandles[fh] = FileHandle(fh, p)
 549         self.inodes.touch(p)
 550
 551         # Normally, we will have received an "update" event if the
 552         # parent collection is stale here. However, even if the parent
 553         # collection hasn't changed, the manifest might have been
 554         # fetched so long ago that the signatures on the data block
 555         # locators have expired. Calling checkupdate() on all
 556         # ancestors ensures the signatures will be refreshed if
 557         # necessary.
 558         while p.parent_inode in self.inodes:
 559             if p == self.inodes[p.parent_inode]:
 560                 break
 561             p = self.inodes[p.parent_inode]
 562             self.inodes.touch(p)
 563             p.checkupdate()
 564
 565         _logger.debug("arv-mount open inode %i flags %x fh %i", inode, flags, fh)
 566
 567         return fh
 568
 569     @catch_exceptions
 570     def read(self, fh, off, size):
 571         _logger.debug("arv-mount read fh %i off %i size %i", fh, off, size)
 572         self.read_ops_counter.add(1)
 573
 574         if fh in self._filehandles:
 575             handle = self._filehandles[fh]
 576         else:
 577             raise llfuse.FUSEError(errno.EBADF)
 578
 579         self.inodes.touch(handle.obj)
 580
 581         r = handle.obj.readfrom(off, size, self.num_retries)
 582         if r:
 583             self.read_counter.add(len(r))
 584         return r
 585
 586     @catch_exceptions
 587     def write(self, fh, off, buf):
 588         _logger.debug("arv-mount write %i %i %i", fh, off, len(buf))
 589         self.write_ops_counter.add(1)
 590
 591         if fh in self._filehandles:
 592             handle = self._filehandles[fh]
 593         else:
 594             raise llfuse.FUSEError(errno.EBADF)
 595
 596         if not handle.obj.writable():
 597             raise llfuse.FUSEError(errno.EPERM)
 598
 599         self.inodes.touch(handle.obj)
 600
 601         w = handle.obj.writeto(off, buf, self.num_retries)
 602         if w:
 603             self.write_counter.add(w)
 604         return w
 605
 606     @catch_exceptions
 607     def release(self, fh):
 608         if fh in self._filehandles:
 609             _logger.debug("arv-mount release fh %i", fh)
 610             try:
 611                 self._filehandles[fh].flush()
 612             except Exception:
 613                 raise
 614             finally:
 615                 self._filehandles[fh].release()
 616                 del self._filehandles[fh]
 617         self.inodes.inode_cache.cap_cache()
 618
 619     def releasedir(self, fh):
 620         self.release(fh)
 621
 622     @catch_exceptions
 623     def opendir(self, inode, ctx=None):
 624         _logger.debug("arv-mount opendir: inode %i", inode)
 625
 626         if inode in self.inodes:
 627             p = self.inodes[inode]
 628         else:
 629             raise llfuse.FUSEError(errno.ENOENT)
 630
 631         if not isinstance(p, Directory):
 632             raise llfuse.FUSEError(errno.ENOTDIR)
 633
 634         fh = next(self._filehandles_counter)
 635         if p.parent_inode in self.inodes:
 636             parent = self.inodes[p.parent_inode]
 637         else:
 638             raise llfuse.FUSEError(errno.EIO)
 639
 640         # update atime
 641         self.inodes.touch(p)
 642
 643         self._filehandles[fh] = DirectoryHandle(fh, p, [('.', p), ('..', parent)] + list(p.items()))
 644         return fh
 645
 646     @catch_exceptions
 647     def readdir(self, fh, off):
 648         _logger.debug("arv-mount readdir: fh %i off %i", fh, off)
 649
 650         if fh in self._filehandles:
 651             handle = self._filehandles[fh]
 652         else:
 653             raise llfuse.FUSEError(errno.EBADF)
 654
 655         e = off
 656         while e < len(handle.entries):
 657             if handle.entries[e][1].inode in self.inodes:
 658                 yield (handle.entries[e][0].encode(self.inodes.encoding), self.getattr(handle.entries[e][1].inode), e+1)
 659             e += 1
 660
 661     @catch_exceptions
 662     def statfs(self, ctx=None):
 663         st = llfuse.StatvfsData()
 664         st.f_bsize = 128 * 1024
 665         st.f_blocks = 0
 666         st.f_files = 0
 667
 668         st.f_bfree = 0
 669         st.f_bavail = 0
 670
 671         st.f_ffree = 0
 672         st.f_favail = 0
 673
 674         st.f_frsize = 0
 675         return st
 676
 677     def _check_writable(self, inode_parent):
 678         if not self.enable_write:
 679             raise llfuse.FUSEError(errno.EROFS)
 680
 681         if inode_parent in self.inodes:
 682             p = self.inodes[inode_parent]
 683         else:
 684             raise llfuse.FUSEError(errno.ENOENT)
 685
 686         if not isinstance(p, Directory):
 687             raise llfuse.FUSEError(errno.ENOTDIR)
 688
 689         if not p.writable():
 690             raise llfuse.FUSEError(errno.EPERM)
 691
 692         return p
 693
 694     @catch_exceptions
 695     def create(self, inode_parent, name, mode, flags, ctx=None):
 696         _logger.debug("arv-mount create: parent_inode %i '%s' %o", inode_parent, name, mode)
 697
 698         p = self._check_writable(inode_parent)
 699         p.create(name)
 700
 701         # The file entry should have been implicitly created by callback.
 702         f = p[name]
 703         fh = next(self._filehandles_counter)
 704         self._filehandles[fh] = FileHandle(fh, f)
 705         self.inodes.touch(p)
 706
 707         f.inc_ref()
 708         return (fh, self.getattr(f.inode))
 709
 710     @catch_exceptions
 711     def mkdir(self, inode_parent, name, mode, ctx=None):
 712         _logger.debug("arv-mount mkdir: parent_inode %i '%s' %o", inode_parent, name, mode)
 713
 714         p = self._check_writable(inode_parent)
 715         p.mkdir(name)
 716
 717         # The dir entry should have been implicitly created by callback.
 718         d = p[name]
 719
 720         d.inc_ref()
 721         return self.getattr(d.inode)
 722
 723     @catch_exceptions
 724     def unlink(self, inode_parent, name, ctx=None):
 725         _logger.debug("arv-mount unlink: parent_inode %i '%s'", inode_parent, name)
 726         p = self._check_writable(inode_parent)
 727         p.unlink(name)
 728
 729     @catch_exceptions
 730     def rmdir(self, inode_parent, name, ctx=None):
 731         _logger.debug("arv-mount rmdir: parent_inode %i '%s'", inode_parent, name)
 732         p = self._check_writable(inode_parent)
 733         p.rmdir(name)
 734
 735     @catch_exceptions
 736     def rename(self, inode_parent_old, name_old, inode_parent_new, name_new, ctx=None):
 737         _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)
 738         src = self._check_writable(inode_parent_old)
 739         dest = self._check_writable(inode_parent_new)
 740         dest.rename(name_old, name_new, src)
 741
 742     @catch_exceptions
 743     def flush(self, fh):
 744         if fh in self._filehandles:
 745             self._filehandles[fh].flush()
 746
 747     def fsync(self, fh, datasync):
 748         self.flush(fh)
 749
 750     def fsyncdir(self, fh, datasync):
 751         self.flush(fh)