services/fuse/arvados_fuse/__init__.py

   1 #
   2 # FUSE driver for Arvados Keep
   3 #
   4
   5 import os
   6 import sys
   7 import llfuse
   8 import errno
   9 import stat
  10 import threading
  11 import arvados
  12 import pprint
  13 import arvados.events
  14 import re
  15 import apiclient
  16 import json
  17 import logging
  18 import time
  19 import _strptime
  20 import calendar
  21 import threading
  22 import itertools
  23
  24 from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
  25
  26 _logger = logging.getLogger('arvados.arvados_fuse')
  27
  28 # Match any character which FUSE or Linux cannot accommodate as part
  29 # of a filename. (If present in a collection filename, they will
  30 # appear as underscores in the fuse mount.)
  31 _disallowed_filename_characters = re.compile('[\x00/]')
  32
  33 def convertTime(t):
  34     """Parse Arvados timestamp to unix time."""
  35     if not t:
  36         return 0
  37     try:
  38         return calendar.timegm(time.strptime(t, "%Y-%m-%dT%H:%M:%SZ"))
  39     except (TypeError, ValueError):
  40         return 0
  41
  42 def sanitize_filename(dirty):
  43     '''Replace disallowed filename characters with harmless "_".'''
  44     if dirty is None:
  45         return None
  46     elif dirty == '':
  47         return '_'
  48     elif dirty == '.':
  49         return '_'
  50     elif dirty == '..':
  51         return '__'
  52     else:
  53         return _disallowed_filename_characters.sub('_', dirty)
  54
  55
  56 class FreshBase(object):
  57     '''Base class for maintaining fresh/stale state to determine when to update.'''
  58     def __init__(self):
  59         self._stale = True
  60         self._poll = False
  61         self._last_update = time.time()
  62         self._atime = time.time()
  63         self._poll_time = 60
  64
  65     # Mark the value as stale
  66     def invalidate(self):
  67         self._stale = True
  68
  69     # Test if the entries dict is stale.
  70     def stale(self):
  71         if self._stale:
  72             return True
  73         if self._poll:
  74             return (self._last_update + self._poll_time) < self._atime
  75         return False
  76
  77     def fresh(self):
  78         self._stale = False
  79         self._last_update = time.time()
  80
  81     def atime(self):
  82         return self._atime
  83
  84 class File(FreshBase):
  85     '''Base for file objects.'''
  86
  87     def __init__(self, parent_inode, _mtime=0):
  88         super(File, self).__init__()
  89         self.inode = None
  90         self.parent_inode = parent_inode
  91         self._mtime = _mtime
  92
  93     def size(self):
  94         return 0
  95
  96     def readfrom(self, off, size):
  97         return ''
  98
  99     def mtime(self):
 100         return self._mtime
 101
 102
 103 class StreamReaderFile(File):
 104     '''Wraps a StreamFileReader as a file.'''
 105
 106     def __init__(self, parent_inode, reader, _mtime):
 107         super(StreamReaderFile, self).__init__(parent_inode, _mtime)
 108         self.reader = reader
 109
 110     def size(self):
 111         return self.reader.size()
 112
 113     def readfrom(self, off, size):
 114         return self.reader.readfrom(off, size)
 115
 116     def stale(self):
 117         return False
 118
 119
 120 class StringFile(File):
 121     '''Wrap a simple string as a file'''
 122     def __init__(self, parent_inode, contents, _mtime):
 123         super(StringFile, self).__init__(parent_inode, _mtime)
 124         self.contents = contents
 125
 126     def size(self):
 127         return len(self.contents)
 128
 129     def readfrom(self, off, size):
 130         return self.contents[off:(off+size)]
 131
 132
 133 class ObjectFile(StringFile):
 134     '''Wrap a dict as a serialized json object.'''
 135
 136     def __init__(self, parent_inode, obj):
 137         super(ObjectFile, self).__init__(parent_inode, "", 0)
 138         self.uuid = obj['uuid']
 139         self.update(obj)
 140
 141     def update(self, obj):
 142         self._mtime = convertTime(obj['modified_at']) if 'modified_at' in obj else 0
 143         self.contents = json.dumps(obj, indent=4, sort_keys=True) + "\n"
 144
 145
 146 class Directory(FreshBase):
 147     '''Generic directory object, backed by a dict.
 148     Consists of a set of entries with the key representing the filename
 149     and the value referencing a File or Directory object.
 150     '''
 151
 152     def __init__(self, parent_inode):
 153         super(Directory, self).__init__()
 154
 155         '''parent_inode is the integer inode number'''
 156         self.inode = None
 157         if not isinstance(parent_inode, int):
 158             raise Exception("parent_inode should be an int")
 159         self.parent_inode = parent_inode
 160         self._entries = {}
 161         self._mtime = time.time()
 162
 163     #  Overriden by subclasses to implement logic to update the entries dict
 164     #  when the directory is stale
 165     def update(self):
 166         pass
 167
 168     # Only used when computing the size of the disk footprint of the directory
 169     # (stub)
 170     def size(self):
 171         return 0
 172
 173     def checkupdate(self):
 174         if self.stale():
 175             try:
 176                 self.update()
 177             except apiclient.errors.HttpError as e:
 178                 _logger.debug(e)
 179
 180     def __getitem__(self, item):
 181         self.checkupdate()
 182         return self._entries[item]
 183
 184     def items(self):
 185         self.checkupdate()
 186         return self._entries.items()
 187
 188     def __iter__(self):
 189         self.checkupdate()
 190         return self._entries.iterkeys()
 191
 192     def __contains__(self, k):
 193         self.checkupdate()
 194         return k in self._entries
 195
 196     def merge(self, items, fn, same, new_entry):
 197         '''Helper method for updating the contents of the directory.  Takes a list
 198         describing the new contents of the directory, reuse entries that are
 199         the same in both the old and new lists, create new entries, and delete
 200         old entries missing from the new list.
 201
 202         items: iterable with new directory contents
 203
 204         fn: function to take an entry in 'items' and return the desired file or
 205         directory name, or None if this entry should be skipped
 206
 207         same: function to compare an existing entry (a File or Directory
 208         object) with an entry in the items list to determine whether to keep
 209         the existing entry.
 210
 211         new_entry: function to create a new directory entry (File or Directory
 212         object) from an entry in the items list.
 213
 214         '''
 215
 216         oldentries = self._entries
 217         self._entries = {}
 218         changed = False
 219         for i in items:
 220             name = sanitize_filename(fn(i))
 221             if name:
 222                 if name in oldentries and same(oldentries[name], i):
 223                     # move existing directory entry over
 224                     self._entries[name] = oldentries[name]
 225                     del oldentries[name]
 226                 else:
 227                     # create new directory entry
 228                     ent = new_entry(i)
 229                     if ent is not None:
 230                         self._entries[name] = self.inodes.add_entry(ent)
 231                         changed = True
 232
 233         # delete any other directory entries that were not in found in 'items'
 234         for i in oldentries:
 235             llfuse.invalidate_entry(self.inode, str(i))
 236             self.inodes.del_entry(oldentries[i])
 237             changed = True
 238
 239         if changed:
 240             self._mtime = time.time()
 241
 242         self.fresh()
 243
 244     def clear(self):
 245         '''Delete all entries'''
 246         oldentries = self._entries
 247         self._entries = {}
 248         for n in oldentries:
 249             if isinstance(n, Directory):
 250                 n.clear()
 251             llfuse.invalidate_entry(self.inode, str(n))
 252             self.inodes.del_entry(oldentries[n])
 253         llfuse.invalidate_inode(self.inode)
 254         self.invalidate()
 255
 256     def mtime(self):
 257         return self._mtime
 258
 259
 260 class CollectionDirectory(Directory):
 261     '''Represents the root of a directory tree holding a collection.'''
 262
 263     def __init__(self, parent_inode, inodes, api, num_retries, collection):
 264         super(CollectionDirectory, self).__init__(parent_inode)
 265         self.inodes = inodes
 266         self.api = api
 267         self.num_retries = num_retries
 268         self.collection_object_file = None
 269         self.collection_object = None
 270         if isinstance(collection, dict):
 271             self.collection_locator = collection['uuid']
 272             self._mtime = convertTime(collection.get('modified_at'))
 273         else:
 274             self.collection_locator = collection
 275             self._mtime = 0
 276
 277     def same(self, i):
 278         return i['uuid'] == self.collection_locator or i['portable_data_hash'] == self.collection_locator
 279
 280     # Used by arv-web.py to switch the contents of the CollectionDirectory
 281     def change_collection(self, new_locator):
 282         """Switch the contents of the CollectionDirectory.  Must be called with llfuse.lock held."""
 283         self.collection_locator = new_locator
 284         self.collection_object = None
 285         self.update()
 286
 287     def new_collection(self, new_collection_object, coll_reader):
 288         self.collection_object = new_collection_object
 289
 290         self._mtime = convertTime(self.collection_object.get('modified_at'))
 291
 292         if self.collection_object_file is not None:
 293             self.collection_object_file.update(self.collection_object)
 294
 295         self.clear()
 296         for s in coll_reader.all_streams():
 297             cwd = self
 298             for part in s.name().split('/'):
 299                 if part != '' and part != '.':
 300                     partname = sanitize_filename(part)
 301                     if partname not in cwd._entries:
 302                         cwd._entries[partname] = self.inodes.add_entry(Directory(cwd.inode))
 303                     cwd = cwd._entries[partname]
 304             for k, v in s.files().items():
 305                 cwd._entries[sanitize_filename(k)] = self.inodes.add_entry(StreamReaderFile(cwd.inode, v, self.mtime()))
 306
 307     def update(self):
 308         try:
 309             if self.collection_object is not None and portable_data_hash_pattern.match(self.collection_locator):
 310                 return True
 311
 312             if self.collection_locator is None:
 313                 self.fresh()
 314                 return True
 315
 316             with llfuse.lock_released:
 317                 coll_reader = arvados.CollectionReader(
 318                     self.collection_locator, self.api, self.api.localkeep(),
 319                     num_retries=self.num_retries)
 320                 new_collection_object = coll_reader.api_response() or {}
 321                 # If the Collection only exists in Keep, there will be no API
 322                 # response.  Fill in the fields we need.
 323                 if 'uuid' not in new_collection_object:
 324                     new_collection_object['uuid'] = self.collection_locator
 325                 if "portable_data_hash" not in new_collection_object:
 326                     new_collection_object["portable_data_hash"] = new_collection_object["uuid"]
 327                 if 'manifest_text' not in new_collection_object:
 328                     new_collection_object['manifest_text'] = coll_reader.manifest_text()
 329                 coll_reader.normalize()
 330             # end with llfuse.lock_released, re-acquire lock
 331
 332             if self.collection_object is None or self.collection_object["portable_data_hash"] != new_collection_object["portable_data_hash"]:
 333                 self.new_collection(new_collection_object, coll_reader)
 334
 335             self.fresh()
 336             return True
 337         except arvados.errors.NotFoundError:
 338             _logger.exception("arv-mount %s: error", self.collection_locator)
 339         except arvados.errors.ArgumentError as detail:
 340             _logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
 341             if self.collection_object is not None and "manifest_text" in self.collection_object:
 342                 _logger.warning("arv-mount manifest_text is: %s", self.collection_object["manifest_text"])
 343         except Exception:
 344             _logger.exception("arv-mount %s: error", self.collection_locator)
 345             if self.collection_object is not None and "manifest_text" in self.collection_object:
 346                 _logger.error("arv-mount manifest_text is: %s", self.collection_object["manifest_text"])
 347         return False
 348
 349     def __getitem__(self, item):
 350         self.checkupdate()
 351         if item == '.arvados#collection':
 352             if self.collection_object_file is None:
 353                 self.collection_object_file = ObjectFile(self.inode, self.collection_object)
 354                 self.inodes.add_entry(self.collection_object_file)
 355             return self.collection_object_file
 356         else:
 357             return super(CollectionDirectory, self).__getitem__(item)
 358
 359     def __contains__(self, k):
 360         if k == '.arvados#collection':
 361             return True
 362         else:
 363             return super(CollectionDirectory, self).__contains__(k)
 364
 365
 366 class MagicDirectory(Directory):
 367     '''A special directory that logically contains the set of all extant keep
 368     locators.  When a file is referenced by lookup(), it is tested to see if it
 369     is a valid keep locator to a manifest, and if so, loads the manifest
 370     contents as a subdirectory of this directory with the locator as the
 371     directory name.  Since querying a list of all extant keep locators is
 372     impractical, only collections that have already been accessed are visible
 373     to readdir().
 374     '''
 375
 376     README_TEXT = '''
 377 This directory provides access to Arvados collections as subdirectories listed
 378 by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
 379 the form '1234567890abcdefghijklmnopqrstuv+123').
 380
 381 Note that this directory will appear empty until you attempt to access a
 382 specific collection subdirectory (such as trying to 'cd' into it), at which
 383 point the collection will actually be looked up on the server and the directory
 384 will appear if it exists.
 385 '''.lstrip()
 386
 387     def __init__(self, parent_inode, inodes, api, num_retries):
 388         super(MagicDirectory, self).__init__(parent_inode)
 389         self.inodes = inodes
 390         self.api = api
 391         self.num_retries = num_retries
 392
 393     def __setattr__(self, name, value):
 394         super(MagicDirectory, self).__setattr__(name, value)
 395         # When we're assigned an inode, add a README.
 396         if ((name == 'inode') and (self.inode is not None) and
 397               (not self._entries)):
 398             self._entries['README'] = self.inodes.add_entry(
 399                 StringFile(self.inode, self.README_TEXT, time.time()))
 400             # If we're the root directory, add an identical by_id subdirectory.
 401             if self.inode == llfuse.ROOT_INODE:
 402                 self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
 403                         self.inode, self.inodes, self.api, self.num_retries))
 404
 405     def __contains__(self, k):
 406         if k in self._entries:
 407             return True
 408
 409         if not portable_data_hash_pattern.match(k) and not uuid_pattern.match(k):
 410             return False
 411
 412         try:
 413             e = self.inodes.add_entry(CollectionDirectory(
 414                     self.inode, self.inodes, self.api, self.num_retries, k))
 415             if e.update():
 416                 self._entries[k] = e
 417                 return True
 418             else:
 419                 return False
 420         except Exception as e:
 421             _logger.debug('arv-mount exception keep %s', e)
 422             return False
 423
 424     def __getitem__(self, item):
 425         if item in self:
 426             return self._entries[item]
 427         else:
 428             raise KeyError("No collection with id " + item)
 429
 430
 431 class RecursiveInvalidateDirectory(Directory):
 432     def invalidate(self):
 433         if self.inode == llfuse.ROOT_INODE:
 434             llfuse.lock.acquire()
 435         try:
 436             super(RecursiveInvalidateDirectory, self).invalidate()
 437             for a in self._entries:
 438                 self._entries[a].invalidate()
 439         except Exception:
 440             _logger.exception()
 441         finally:
 442             if self.inode == llfuse.ROOT_INODE:
 443                 llfuse.lock.release()
 444
 445
 446 class TagsDirectory(RecursiveInvalidateDirectory):
 447     '''A special directory that contains as subdirectories all tags visible to the user.'''
 448
 449     def __init__(self, parent_inode, inodes, api, num_retries, poll_time=60):
 450         super(TagsDirectory, self).__init__(parent_inode)
 451         self.inodes = inodes
 452         self.api = api
 453         self.num_retries = num_retries
 454         self._poll = True
 455         self._poll_time = poll_time
 456
 457     def update(self):
 458         with llfuse.lock_released:
 459             tags = self.api.links().list(
 460                 filters=[['link_class', '=', 'tag']],
 461                 select=['name'], distinct=True
 462                 ).execute(num_retries=self.num_retries)
 463         if "items" in tags:
 464             self.merge(tags['items'],
 465                        lambda i: i['name'],
 466                        lambda a, i: a.tag == i['name'],
 467                        lambda i: TagDirectory(self.inode, self.inodes, self.api, self.num_retries, i['name'], poll=self._poll, poll_time=self._poll_time))
 468
 469
 470 class TagDirectory(Directory):
 471     '''A special directory that contains as subdirectories all collections visible
 472     to the user that are tagged with a particular tag.
 473     '''
 474
 475     def __init__(self, parent_inode, inodes, api, num_retries, tag,
 476                  poll=False, poll_time=60):
 477         super(TagDirectory, self).__init__(parent_inode)
 478         self.inodes = inodes
 479         self.api = api
 480         self.num_retries = num_retries
 481         self.tag = tag
 482         self._poll = poll
 483         self._poll_time = poll_time
 484
 485     def update(self):
 486         with llfuse.lock_released:
 487             taggedcollections = self.api.links().list(
 488                 filters=[['link_class', '=', 'tag'],
 489                          ['name', '=', self.tag],
 490                          ['head_uuid', 'is_a', 'arvados#collection']],
 491                 select=['head_uuid']
 492                 ).execute(num_retries=self.num_retries)
 493         self.merge(taggedcollections['items'],
 494                    lambda i: i['head_uuid'],
 495                    lambda a, i: a.collection_locator == i['head_uuid'],
 496                    lambda i: CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid']))
 497
 498
 499 class ProjectDirectory(Directory):
 500     '''A special directory that contains the contents of a project.'''
 501
 502     def __init__(self, parent_inode, inodes, api, num_retries, project_object,
 503                  poll=False, poll_time=60):
 504         super(ProjectDirectory, self).__init__(parent_inode)
 505         self.inodes = inodes
 506         self.api = api
 507         self.num_retries = num_retries
 508         self.project_object = project_object
 509         self.project_object_file = None
 510         self.uuid = project_object['uuid']
 511         self._poll = poll
 512         self._poll_time = poll_time
 513
 514     def createDirectory(self, i):
 515         if collection_uuid_pattern.match(i['uuid']):
 516             return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i)
 517         elif group_uuid_pattern.match(i['uuid']):
 518             return ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i, self._poll, self._poll_time)
 519         elif link_uuid_pattern.match(i['uuid']):
 520             if i['head_kind'] == 'arvados#collection' or portable_data_hash_pattern.match(i['head_uuid']):
 521                 return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid'])
 522             else:
 523                 return None
 524         elif uuid_pattern.match(i['uuid']):
 525             return ObjectFile(self.parent_inode, i)
 526         else:
 527             return None
 528
 529     def update(self):
 530         if self.project_object_file == None:
 531             self.project_object_file = ObjectFile(self.inode, self.project_object)
 532             self.inodes.add_entry(self.project_object_file)
 533
 534         def namefn(i):
 535             if 'name' in i:
 536                 if i['name'] is None or len(i['name']) == 0:
 537                     return None
 538                 elif collection_uuid_pattern.match(i['uuid']) or group_uuid_pattern.match(i['uuid']):
 539                     # collection or subproject
 540                     return i['name']
 541                 elif link_uuid_pattern.match(i['uuid']) and i['head_kind'] == 'arvados#collection':
 542                     # name link
 543                     return i['name']
 544                 elif 'kind' in i and i['kind'].startswith('arvados#'):
 545                     # something else
 546                     return "{}.{}".format(i['name'], i['kind'][8:])
 547             else:
 548                 return None
 549
 550         def samefn(a, i):
 551             if isinstance(a, CollectionDirectory):
 552                 return a.collection_locator == i['uuid']
 553             elif isinstance(a, ProjectDirectory):
 554                 return a.uuid == i['uuid']
 555             elif isinstance(a, ObjectFile):
 556                 return a.uuid == i['uuid'] and not a.stale()
 557             return False
 558
 559         with llfuse.lock_released:
 560             if group_uuid_pattern.match(self.uuid):
 561                 self.project_object = self.api.groups().get(
 562                     uuid=self.uuid).execute(num_retries=self.num_retries)
 563             elif user_uuid_pattern.match(self.uuid):
 564                 self.project_object = self.api.users().get(
 565                     uuid=self.uuid).execute(num_retries=self.num_retries)
 566
 567             contents = arvados.util.list_all(self.api.groups().contents,
 568                                              self.num_retries, uuid=self.uuid)
 569
 570         # end with llfuse.lock_released, re-acquire lock
 571
 572         self.merge(contents,
 573                    namefn,
 574                    samefn,
 575                    self.createDirectory)
 576
 577     def __getitem__(self, item):
 578         self.checkupdate()
 579         if item == '.arvados#project':
 580             return self.project_object_file
 581         else:
 582             return super(ProjectDirectory, self).__getitem__(item)
 583
 584     def __contains__(self, k):
 585         if k == '.arvados#project':
 586             return True
 587         else:
 588             return super(ProjectDirectory, self).__contains__(k)
 589
 590
 591 class SharedDirectory(Directory):
 592     '''A special directory that represents users or groups who have shared projects with me.'''
 593
 594     def __init__(self, parent_inode, inodes, api, num_retries, exclude,
 595                  poll=False, poll_time=60):
 596         super(SharedDirectory, self).__init__(parent_inode)
 597         self.inodes = inodes
 598         self.api = api
 599         self.num_retries = num_retries
 600         self.current_user = api.users().current().execute(num_retries=num_retries)
 601         self._poll = True
 602         self._poll_time = poll_time
 603
 604     def update(self):
 605         with llfuse.lock_released:
 606             all_projects = arvados.util.list_all(
 607                 self.api.groups().list, self.num_retries,
 608                 filters=[['group_class','=','project']])
 609             objects = {}
 610             for ob in all_projects:
 611                 objects[ob['uuid']] = ob
 612
 613             roots = []
 614             root_owners = {}
 615             for ob in all_projects:
 616                 if ob['owner_uuid'] != self.current_user['uuid'] and ob['owner_uuid'] not in objects:
 617                     roots.append(ob)
 618                     root_owners[ob['owner_uuid']] = True
 619
 620             lusers = arvados.util.list_all(
 621                 self.api.users().list, self.num_retries,
 622                 filters=[['uuid','in', list(root_owners)]])
 623             lgroups = arvados.util.list_all(
 624                 self.api.groups().list, self.num_retries,
 625                 filters=[['uuid','in', list(root_owners)]])
 626
 627             users = {}
 628             groups = {}
 629
 630             for l in lusers:
 631                 objects[l["uuid"]] = l
 632             for l in lgroups:
 633                 objects[l["uuid"]] = l
 634
 635             contents = {}
 636             for r in root_owners:
 637                 if r in objects:
 638                     obr = objects[r]
 639                     if "name" in obr:
 640                         contents[obr["name"]] = obr
 641                     if "first_name" in obr:
 642                         contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
 643
 644             for r in roots:
 645                 if r['owner_uuid'] not in objects:
 646                     contents[r['name']] = r
 647
 648         # end with llfuse.lock_released, re-acquire lock
 649
 650         try:
 651             self.merge(contents.items(),
 652                        lambda i: i[0],
 653                        lambda a, i: a.uuid == i[1]['uuid'],
 654                        lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
 655         except Exception:
 656             _logger.exception()
 657
 658
 659 class FileHandle(object):
 660     '''Connects a numeric file handle to a File or Directory object that has
 661     been opened by the client.'''
 662
 663     def __init__(self, fh, entry):
 664         self.fh = fh
 665         self.entry = entry
 666
 667
 668 class Inodes(object):
 669     '''Manage the set of inodes.  This is the mapping from a numeric id
 670     to a concrete File or Directory object'''
 671
 672     def __init__(self):
 673         self._entries = {}
 674         self._counter = itertools.count(llfuse.ROOT_INODE)
 675
 676     def __getitem__(self, item):
 677         return self._entries[item]
 678
 679     def __setitem__(self, key, item):
 680         self._entries[key] = item
 681
 682     def __iter__(self):
 683         return self._entries.iterkeys()
 684
 685     def items(self):
 686         return self._entries.items()
 687
 688     def __contains__(self, k):
 689         return k in self._entries
 690
 691     def add_entry(self, entry):
 692         entry.inode = next(self._counter)
 693         self._entries[entry.inode] = entry
 694         return entry
 695
 696     def del_entry(self, entry):
 697         llfuse.invalidate_inode(entry.inode)
 698         del self._entries[entry.inode]
 699
 700 class Operations(llfuse.Operations):
 701     '''This is the main interface with llfuse.  The methods on this object are
 702     called by llfuse threads to service FUSE events to query and read from
 703     the file system.
 704
 705     llfuse has its own global lock which is acquired before calling a request handler,
 706     so request handlers do not run concurrently unless the lock is explicitly released
 707     using "with llfuse.lock_released:"'''
 708
 709     def __init__(self, uid, gid, encoding="utf-8"):
 710         super(Operations, self).__init__()
 711
 712         self.inodes = Inodes()
 713         self.uid = uid
 714         self.gid = gid
 715         self.encoding = encoding
 716
 717         # dict of inode to filehandle
 718         self._filehandles = {}
 719         self._filehandles_counter = 1
 720
 721         # Other threads that need to wait until the fuse driver
 722         # is fully initialized should wait() on this event object.
 723         self.initlock = threading.Event()
 724
 725     def init(self):
 726         # Allow threads that are waiting for the driver to be finished
 727         # initializing to continue
 728         self.initlock.set()
 729
 730     def access(self, inode, mode, ctx):
 731         return True
 732
 733     def getattr(self, inode):
 734         if inode not in self.inodes:
 735             raise llfuse.FUSEError(errno.ENOENT)
 736
 737         e = self.inodes[inode]
 738
 739         entry = llfuse.EntryAttributes()
 740         entry.st_ino = inode
 741         entry.generation = 0
 742         entry.entry_timeout = 300
 743         entry.attr_timeout = 300
 744
 745         entry.st_mode = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
 746         if isinstance(e, Directory):
 747             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFDIR
 748         elif isinstance(e, StreamReaderFile):
 749             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFREG
 750         else:
 751             entry.st_mode |= stat.S_IFREG
 752
 753         entry.st_nlink = 1
 754         entry.st_uid = self.uid
 755         entry.st_gid = self.gid
 756         entry.st_rdev = 0
 757
 758         entry.st_size = e.size()
 759
 760         entry.st_blksize = 512
 761         entry.st_blocks = (e.size()/512)+1
 762         entry.st_atime = int(e.atime())
 763         entry.st_mtime = int(e.mtime())
 764         entry.st_ctime = int(e.mtime())
 765
 766         return entry
 767
 768     def lookup(self, parent_inode, name):
 769         name = unicode(name, self.encoding)
 770         _logger.debug("arv-mount lookup: parent_inode %i name %s",
 771                       parent_inode, name)
 772         inode = None
 773
 774         if name == '.':
 775             inode = parent_inode
 776         else:
 777             if parent_inode in self.inodes:
 778                 p = self.inodes[parent_inode]
 779                 if name == '..':
 780                     inode = p.parent_inode
 781                 elif isinstance(p, Directory) and name in p:
 782                     inode = p[name].inode
 783
 784         if inode != None:
 785             return self.getattr(inode)
 786         else:
 787             raise llfuse.FUSEError(errno.ENOENT)
 788
 789     def open(self, inode, flags):
 790         if inode in self.inodes:
 791             p = self.inodes[inode]
 792         else:
 793             raise llfuse.FUSEError(errno.ENOENT)
 794
 795         if (flags & os.O_WRONLY) or (flags & os.O_RDWR):
 796             raise llfuse.FUSEError(errno.EROFS)
 797
 798         if isinstance(p, Directory):
 799             raise llfuse.FUSEError(errno.EISDIR)
 800
 801         fh = self._filehandles_counter
 802         self._filehandles_counter += 1
 803         self._filehandles[fh] = FileHandle(fh, p)
 804         return fh
 805
 806     def read(self, fh, off, size):
 807         _logger.debug("arv-mount read %i %i %i", fh, off, size)
 808         if fh in self._filehandles:
 809             handle = self._filehandles[fh]
 810         else:
 811             raise llfuse.FUSEError(errno.EBADF)
 812
 813         # update atime
 814         handle.entry._atime = time.time()
 815
 816         try:
 817             with llfuse.lock_released:
 818                 return handle.entry.readfrom(off, size)
 819         except arvados.errors.NotFoundError as e:
 820             _logger.warning("Block not found: " + str(e))
 821             raise llfuse.FUSEError(errno.EIO)
 822         except Exception:
 823             _logger.exception()
 824             raise llfuse.FUSEError(errno.EIO)
 825
 826     def release(self, fh):
 827         if fh in self._filehandles:
 828             del self._filehandles[fh]
 829
 830     def opendir(self, inode):
 831         _logger.debug("arv-mount opendir: inode %i", inode)
 832
 833         if inode in self.inodes:
 834             p = self.inodes[inode]
 835         else:
 836             raise llfuse.FUSEError(errno.ENOENT)
 837
 838         if not isinstance(p, Directory):
 839             raise llfuse.FUSEError(errno.ENOTDIR)
 840
 841         fh = self._filehandles_counter
 842         self._filehandles_counter += 1
 843         if p.parent_inode in self.inodes:
 844             parent = self.inodes[p.parent_inode]
 845         else:
 846             raise llfuse.FUSEError(errno.EIO)
 847
 848         # update atime
 849         p._atime = time.time()
 850
 851         self._filehandles[fh] = FileHandle(fh, [('.', p), ('..', parent)] + list(p.items()))
 852         return fh
 853
 854     def readdir(self, fh, off):
 855         _logger.debug("arv-mount readdir: fh %i off %i", fh, off)
 856
 857         if fh in self._filehandles:
 858             handle = self._filehandles[fh]
 859         else:
 860             raise llfuse.FUSEError(errno.EBADF)
 861
 862         _logger.debug("arv-mount handle.entry %s", handle.entry)
 863
 864         e = off
 865         while e < len(handle.entry):
 866             if handle.entry[e][1].inode in self.inodes:
 867                 try:
 868                     yield (handle.entry[e][0].encode(self.encoding), self.getattr(handle.entry[e][1].inode), e+1)
 869                 except UnicodeEncodeError:
 870                     pass
 871             e += 1
 872
 873     def releasedir(self, fh):
 874         del self._filehandles[fh]
 875
 876     def statfs(self):
 877         st = llfuse.StatvfsData()
 878         st.f_bsize = 64 * 1024
 879         st.f_blocks = 0
 880         st.f_files = 0
 881
 882         st.f_bfree = 0
 883         st.f_bavail = 0
 884
 885         st.f_ffree = 0
 886         st.f_favail = 0
 887
 888         st.f_frsize = 0
 889         return st
 890
 891     # The llfuse documentation recommends only overloading functions that
 892     # are actually implemented, as the default implementation will raise ENOSYS.
 893     # However, there is a bug in the llfuse default implementation of create()
 894     # "create() takes exactly 5 positional arguments (6 given)" which will crash
 895     # arv-mount.
 896     # The workaround is to implement it with the proper number of parameters,
 897     # and then everything works out.
 898     def create(self, inode_parent, name, mode, flags, ctx):
 899         raise llfuse.FUSEError(errno.EROFS)