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