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         self.invalidate()
 252
 253     def mtime(self):
 254         return self._mtime
 255
 256
 257 class CollectionDirectory(Directory):
 258     '''Represents the root of a directory tree holding a collection.'''
 259
 260     def __init__(self, parent_inode, inodes, api, num_retries, collection):
 261         super(CollectionDirectory, self).__init__(parent_inode)
 262         self.inodes = inodes
 263         self.api = api
 264         self.num_retries = num_retries
 265         self.collection_object_file = None
 266         self.collection_object = None
 267         if isinstance(collection, dict):
 268             self.collection_locator = collection['uuid']
 269         else:
 270             self.collection_locator = collection
 271
 272     def same(self, i):
 273         return i['uuid'] == self.collection_locator or i['portable_data_hash'] == self.collection_locator
 274
 275     def new_collection(self, new_collection_object, coll_reader):
 276         self.collection_object = new_collection_object
 277
 278         if self.collection_object_file is not None:
 279             self.collection_object_file.update(self.collection_object)
 280
 281         self.clear()
 282         for s in coll_reader.all_streams():
 283             cwd = self
 284             for part in s.name().split('/'):
 285                 if part != '' and part != '.':
 286                     partname = sanitize_filename(part)
 287                     if partname not in cwd._entries:
 288                         cwd._entries[partname] = self.inodes.add_entry(Directory(cwd.inode))
 289                     cwd = cwd._entries[partname]
 290             for k, v in s.files().items():
 291                 cwd._entries[sanitize_filename(k)] = self.inodes.add_entry(StreamReaderFile(cwd.inode, v, self.mtime()))
 292
 293     def update(self):
 294         try:
 295             if self.collection_object is not None and portable_data_hash_pattern.match(self.collection_locator):
 296                 return True
 297
 298             with llfuse.lock_released:
 299                 coll_reader = arvados.CollectionReader(
 300                     self.collection_locator, self.api, self.api.localkeep(),
 301                     num_retries=self.num_retries)
 302                 new_collection_object = coll_reader.api_response() or {}
 303                 # If the Collection only exists in Keep, there will be no API
 304                 # response.  Fill in the fields we need.
 305                 if 'uuid' not in new_collection_object:
 306                     new_collection_object['uuid'] = self.collection_locator
 307                 if "portable_data_hash" not in new_collection_object:
 308                     new_collection_object["portable_data_hash"] = new_collection_object["uuid"]
 309                 if 'manifest_text' not in new_collection_object:
 310                     new_collection_object['manifest_text'] = coll_reader.manifest_text()
 311                 coll_reader.normalize()
 312             # end with llfuse.lock_released, re-acquire lock
 313
 314             if self.collection_object is None or self.collection_object["portable_data_hash"] != new_collection_object["portable_data_hash"]:
 315                 self.new_collection(new_collection_object, coll_reader)
 316
 317             self.fresh()
 318             return True
 319         except apiclient.errors.NotFoundError:
 320             _logger.exception("arv-mount %s: error", self.collection_locator)
 321         except arvados.errors.ArgumentError as detail:
 322             _logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
 323             if self.collection_object is not None and "manifest_text" in self.collection_object:
 324                 _logger.warning("arv-mount manifest_text is: %s", self.collection_object["manifest_text"])
 325         except Exception:
 326             _logger.exception("arv-mount %s: error", self.collection_locator)
 327             if self.collection_object is not None and "manifest_text" in self.collection_object:
 328                 _logger.error("arv-mount manifest_text is: %s", self.collection_object["manifest_text"])
 329         return False
 330
 331     def __getitem__(self, item):
 332         self.checkupdate()
 333         if item == '.arvados#collection':
 334             if self.collection_object_file is None:
 335                 self.collection_object_file = ObjectFile(self.inode, self.collection_object)
 336                 self.inodes.add_entry(self.collection_object_file)
 337             return self.collection_object_file
 338         else:
 339             return super(CollectionDirectory, self).__getitem__(item)
 340
 341     def __contains__(self, k):
 342         if k == '.arvados#collection':
 343             return True
 344         else:
 345             return super(CollectionDirectory, self).__contains__(k)
 346
 347     def mtime(self):
 348         self.checkupdate()
 349         return convertTime(self.collection_object["modified_at"]) if self.collection_object is not None and 'modified_at' in self.collection_object else 0
 350
 351
 352 class MagicDirectory(Directory):
 353     '''A special directory that logically contains the set of all extant keep
 354     locators.  When a file is referenced by lookup(), it is tested to see if it
 355     is a valid keep locator to a manifest, and if so, loads the manifest
 356     contents as a subdirectory of this directory with the locator as the
 357     directory name.  Since querying a list of all extant keep locators is
 358     impractical, only collections that have already been accessed are visible
 359     to readdir().
 360     '''
 361
 362     README_TEXT = '''
 363 This directory provides access to Arvados collections as subdirectories listed
 364 by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
 365 the form '1234567890abcdefghijklmnopqrstuv+123').
 366
 367 Note that this directory will appear empty until you attempt to access a
 368 specific collection subdirectory (such as trying to 'cd' into it), at which
 369 point the collection will actually be looked up on the server and the directory
 370 will appear if it exists.
 371 '''.lstrip()
 372
 373     def __init__(self, parent_inode, inodes, api, num_retries):
 374         super(MagicDirectory, self).__init__(parent_inode)
 375         self.inodes = inodes
 376         self.api = api
 377         self.num_retries = num_retries
 378
 379     def __setattr__(self, name, value):
 380         super(MagicDirectory, self).__setattr__(name, value)
 381         # When we're assigned an inode, add a README.
 382         if ((name == 'inode') and (self.inode is not None) and
 383               (not self._entries)):
 384             self._entries['README'] = self.inodes.add_entry(
 385                 StringFile(self.inode, self.README_TEXT, time.time()))
 386             # If we're the root directory, add an identical by_id subdirectory.
 387             if self.inode == llfuse.ROOT_INODE:
 388                 self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
 389                         self.inode, self.inodes, self.api, self.num_retries))
 390
 391     def __contains__(self, k):
 392         if k in self._entries:
 393             return True
 394
 395         if not portable_data_hash_pattern.match(k) and not uuid_pattern.match(k):
 396             return False
 397
 398         try:
 399             e = self.inodes.add_entry(CollectionDirectory(
 400                     self.inode, self.inodes, self.api, self.num_retries, k))
 401             if e.update():
 402                 self._entries[k] = e
 403                 return True
 404             else:
 405                 return False
 406         except Exception as e:
 407             _logger.debug('arv-mount exception keep %s', e)
 408             return False
 409
 410     def __getitem__(self, item):
 411         if item in self:
 412             return self._entries[item]
 413         else:
 414             raise KeyError("No collection with id " + item)
 415
 416
 417 class RecursiveInvalidateDirectory(Directory):
 418     def invalidate(self):
 419         if self.inode == llfuse.ROOT_INODE:
 420             llfuse.lock.acquire()
 421         try:
 422             super(RecursiveInvalidateDirectory, self).invalidate()
 423             for a in self._entries:
 424                 self._entries[a].invalidate()
 425         except Exception:
 426             _logger.exception()
 427         finally:
 428             if self.inode == llfuse.ROOT_INODE:
 429                 llfuse.lock.release()
 430
 431
 432 class TagsDirectory(RecursiveInvalidateDirectory):
 433     '''A special directory that contains as subdirectories all tags visible to the user.'''
 434
 435     def __init__(self, parent_inode, inodes, api, num_retries, poll_time=60):
 436         super(TagsDirectory, self).__init__(parent_inode)
 437         self.inodes = inodes
 438         self.api = api
 439         self.num_retries = num_retries
 440         self._poll = True
 441         self._poll_time = poll_time
 442
 443     def update(self):
 444         with llfuse.lock_released:
 445             tags = self.api.links().list(
 446                 filters=[['link_class', '=', 'tag']],
 447                 select=['name'], distinct=True
 448                 ).execute(num_retries=self.num_retries)
 449         if "items" in tags:
 450             self.merge(tags['items'],
 451                        lambda i: i['name'] if 'name' in i else i['uuid'],
 452                        lambda a, i: a.tag == i,
 453                        lambda i: TagDirectory(self.inode, self.inodes, self.api, self.num_retries, i['name'], poll=self._poll, poll_time=self._poll_time))
 454
 455
 456 class TagDirectory(Directory):
 457     '''A special directory that contains as subdirectories all collections visible
 458     to the user that are tagged with a particular tag.
 459     '''
 460
 461     def __init__(self, parent_inode, inodes, api, num_retries, tag,
 462                  poll=False, poll_time=60):
 463         super(TagDirectory, self).__init__(parent_inode)
 464         self.inodes = inodes
 465         self.api = api
 466         self.num_retries = num_retries
 467         self.tag = tag
 468         self._poll = poll
 469         self._poll_time = poll_time
 470
 471     def update(self):
 472         with llfuse.lock_released:
 473             taggedcollections = self.api.links().list(
 474                 filters=[['link_class', '=', 'tag'],
 475                          ['name', '=', self.tag],
 476                          ['head_uuid', 'is_a', 'arvados#collection']],
 477                 select=['head_uuid']
 478                 ).execute(num_retries=self.num_retries)
 479         self.merge(taggedcollections['items'],
 480                    lambda i: i['head_uuid'],
 481                    lambda a, i: a.collection_locator == i['head_uuid'],
 482                    lambda i: CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid']))
 483
 484
 485 class ProjectDirectory(Directory):
 486     '''A special directory that contains the contents of a project.'''
 487
 488     def __init__(self, parent_inode, inodes, api, num_retries, project_object,
 489                  poll=False, poll_time=60):
 490         super(ProjectDirectory, self).__init__(parent_inode)
 491         self.inodes = inodes
 492         self.api = api
 493         self.num_retries = num_retries
 494         self.project_object = project_object
 495         self.project_object_file = None
 496         self.uuid = project_object['uuid']
 497         self._poll = poll
 498         self._poll_time = poll_time
 499
 500     def createDirectory(self, i):
 501         if collection_uuid_pattern.match(i['uuid']):
 502             return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i)
 503         elif group_uuid_pattern.match(i['uuid']):
 504             return ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i, self._poll, self._poll_time)
 505         elif link_uuid_pattern.match(i['uuid']):
 506             if i['head_kind'] == 'arvados#collection' or portable_data_hash_pattern.match(i['head_uuid']):
 507                 return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid'])
 508             else:
 509                 return None
 510         elif uuid_pattern.match(i['uuid']):
 511             return ObjectFile(self.parent_inode, i)
 512         else:
 513             return None
 514
 515     def update(self):
 516         if self.project_object_file == None:
 517             self.project_object_file = ObjectFile(self.inode, self.project_object)
 518             self.inodes.add_entry(self.project_object_file)
 519
 520         def namefn(i):
 521             if 'name' in i:
 522                 if i['name'] is None or len(i['name']) == 0:
 523                     return None
 524                 elif collection_uuid_pattern.match(i['uuid']) or group_uuid_pattern.match(i['uuid']):
 525                     # collection or subproject
 526                     return i['name']
 527                 elif link_uuid_pattern.match(i['uuid']) and i['head_kind'] == 'arvados#collection':
 528                     # name link
 529                     return i['name']
 530                 elif 'kind' in i and i['kind'].startswith('arvados#'):
 531                     # something else
 532                     return "{}.{}".format(i['name'], i['kind'][8:])
 533             else:
 534                 return None
 535
 536         def samefn(a, i):
 537             if isinstance(a, CollectionDirectory):
 538                 return a.collection_locator == i['uuid']
 539             elif isinstance(a, ProjectDirectory):
 540                 return a.uuid == i['uuid']
 541             elif isinstance(a, ObjectFile):
 542                 return a.uuid == i['uuid'] and not a.stale()
 543             return False
 544
 545         with llfuse.lock_released:
 546             if group_uuid_pattern.match(self.uuid):
 547                 self.project_object = self.api.groups().get(
 548                     uuid=self.uuid).execute(num_retries=self.num_retries)
 549             elif user_uuid_pattern.match(self.uuid):
 550                 self.project_object = self.api.users().get(
 551                     uuid=self.uuid).execute(num_retries=self.num_retries)
 552
 553             contents = arvados.util.list_all(self.api.groups().contents,
 554                                              self.num_retries, uuid=self.uuid)
 555
 556         # end with llfuse.lock_released, re-acquire lock
 557
 558         self.merge(contents,
 559                    namefn,
 560                    samefn,
 561                    self.createDirectory)
 562
 563     def __getitem__(self, item):
 564         self.checkupdate()
 565         if item == '.arvados#project':
 566             return self.project_object_file
 567         else:
 568             return super(ProjectDirectory, self).__getitem__(item)
 569
 570     def __contains__(self, k):
 571         if k == '.arvados#project':
 572             return True
 573         else:
 574             return super(ProjectDirectory, self).__contains__(k)
 575
 576
 577 class SharedDirectory(Directory):
 578     '''A special directory that represents users or groups who have shared projects with me.'''
 579
 580     def __init__(self, parent_inode, inodes, api, num_retries, exclude,
 581                  poll=False, poll_time=60):
 582         super(SharedDirectory, self).__init__(parent_inode)
 583         self.inodes = inodes
 584         self.api = api
 585         self.num_retries = num_retries
 586         self.current_user = api.users().current().execute(num_retries=num_retries)
 587         self._poll = True
 588         self._poll_time = poll_time
 589
 590     def update(self):
 591         with llfuse.lock_released:
 592             all_projects = arvados.util.list_all(
 593                 self.api.groups().list, self.num_retries,
 594                 filters=[['group_class','=','project']])
 595             objects = {}
 596             for ob in all_projects:
 597                 objects[ob['uuid']] = ob
 598
 599             roots = []
 600             root_owners = {}
 601             for ob in all_projects:
 602                 if ob['owner_uuid'] != self.current_user['uuid'] and ob['owner_uuid'] not in objects:
 603                     roots.append(ob)
 604                     root_owners[ob['owner_uuid']] = True
 605
 606             lusers = arvados.util.list_all(
 607                 self.api.users().list, self.num_retries,
 608                 filters=[['uuid','in', list(root_owners)]])
 609             lgroups = arvados.util.list_all(
 610                 self.api.groups().list, self.num_retries,
 611                 filters=[['uuid','in', list(root_owners)]])
 612
 613             users = {}
 614             groups = {}
 615
 616             for l in lusers:
 617                 objects[l["uuid"]] = l
 618             for l in lgroups:
 619                 objects[l["uuid"]] = l
 620
 621             contents = {}
 622             for r in root_owners:
 623                 if r in objects:
 624                     obr = objects[r]
 625                     if "name" in obr:
 626                         contents[obr["name"]] = obr
 627                     if "first_name" in obr:
 628                         contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
 629
 630             for r in roots:
 631                 if r['owner_uuid'] not in objects:
 632                     contents[r['name']] = r
 633
 634         # end with llfuse.lock_released, re-acquire lock
 635
 636         try:
 637             self.merge(contents.items(),
 638                        lambda i: i[0],
 639                        lambda a, i: a.uuid == i[1]['uuid'],
 640                        lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
 641         except Exception:
 642             _logger.exception()
 643
 644
 645 class FileHandle(object):
 646     '''Connects a numeric file handle to a File or Directory object that has
 647     been opened by the client.'''
 648
 649     def __init__(self, fh, entry):
 650         self.fh = fh
 651         self.entry = entry
 652
 653
 654 class Inodes(object):
 655     '''Manage the set of inodes.  This is the mapping from a numeric id
 656     to a concrete File or Directory object'''
 657
 658     def __init__(self):
 659         self._entries = {}
 660         self._counter = itertools.count(llfuse.ROOT_INODE)
 661
 662     def __getitem__(self, item):
 663         return self._entries[item]
 664
 665     def __setitem__(self, key, item):
 666         self._entries[key] = item
 667
 668     def __iter__(self):
 669         return self._entries.iterkeys()
 670
 671     def items(self):
 672         return self._entries.items()
 673
 674     def __contains__(self, k):
 675         return k in self._entries
 676
 677     def add_entry(self, entry):
 678         entry.inode = next(self._counter)
 679         self._entries[entry.inode] = entry
 680         return entry
 681
 682     def del_entry(self, entry):
 683         llfuse.invalidate_inode(entry.inode)
 684         del self._entries[entry.inode]
 685
 686 class Operations(llfuse.Operations):
 687     '''This is the main interface with llfuse.  The methods on this object are
 688     called by llfuse threads to service FUSE events to query and read from
 689     the file system.
 690
 691     llfuse has its own global lock which is acquired before calling a request handler,
 692     so request handlers do not run concurrently unless the lock is explicitly released
 693     using "with llfuse.lock_released:"'''
 694
 695     def __init__(self, uid, gid, encoding="utf-8"):
 696         super(Operations, self).__init__()
 697
 698         self.inodes = Inodes()
 699         self.uid = uid
 700         self.gid = gid
 701         self.encoding = encoding
 702
 703         # dict of inode to filehandle
 704         self._filehandles = {}
 705         self._filehandles_counter = 1
 706
 707         # Other threads that need to wait until the fuse driver
 708         # is fully initialized should wait() on this event object.
 709         self.initlock = threading.Event()
 710
 711     def init(self):
 712         # Allow threads that are waiting for the driver to be finished
 713         # initializing to continue
 714         self.initlock.set()
 715
 716     def access(self, inode, mode, ctx):
 717         return True
 718
 719     def getattr(self, inode):
 720         if inode not in self.inodes:
 721             raise llfuse.FUSEError(errno.ENOENT)
 722
 723         e = self.inodes[inode]
 724
 725         entry = llfuse.EntryAttributes()
 726         entry.st_ino = inode
 727         entry.generation = 0
 728         entry.entry_timeout = 300
 729         entry.attr_timeout = 300
 730
 731         entry.st_mode = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
 732         if isinstance(e, Directory):
 733             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFDIR
 734         elif isinstance(e, StreamReaderFile):
 735             entry.st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH | stat.S_IFREG
 736         else:
 737             entry.st_mode |= stat.S_IFREG
 738
 739         entry.st_nlink = 1
 740         entry.st_uid = self.uid
 741         entry.st_gid = self.gid
 742         entry.st_rdev = 0
 743
 744         entry.st_size = e.size()
 745
 746         entry.st_blksize = 512
 747         entry.st_blocks = (e.size()/512)+1
 748         entry.st_atime = int(e.atime())
 749         entry.st_mtime = int(e.mtime())
 750         entry.st_ctime = int(e.mtime())
 751
 752         return entry
 753
 754     def lookup(self, parent_inode, name):
 755         name = unicode(name, self.encoding)
 756         _logger.debug("arv-mount lookup: parent_inode %i name %s",
 757                       parent_inode, name)
 758         inode = None
 759
 760         if name == '.':
 761             inode = parent_inode
 762         else:
 763             if parent_inode in self.inodes:
 764                 p = self.inodes[parent_inode]
 765                 if name == '..':
 766                     inode = p.parent_inode
 767                 elif isinstance(p, Directory) and name in p:
 768                     inode = p[name].inode
 769
 770         if inode != None:
 771             return self.getattr(inode)
 772         else:
 773             raise llfuse.FUSEError(errno.ENOENT)
 774
 775     def open(self, inode, flags):
 776         if inode in self.inodes:
 777             p = self.inodes[inode]
 778         else:
 779             raise llfuse.FUSEError(errno.ENOENT)
 780
 781         if (flags & os.O_WRONLY) or (flags & os.O_RDWR):
 782             raise llfuse.FUSEError(errno.EROFS)
 783
 784         if isinstance(p, Directory):
 785             raise llfuse.FUSEError(errno.EISDIR)
 786
 787         fh = self._filehandles_counter
 788         self._filehandles_counter += 1
 789         self._filehandles[fh] = FileHandle(fh, p)
 790         return fh
 791
 792     def read(self, fh, off, size):
 793         _logger.debug("arv-mount read %i %i %i", fh, off, size)
 794         if fh in self._filehandles:
 795             handle = self._filehandles[fh]
 796         else:
 797             raise llfuse.FUSEError(errno.EBADF)
 798
 799         # update atime
 800         handle.entry._atime = time.time()
 801
 802         try:
 803             with llfuse.lock_released:
 804                 return handle.entry.readfrom(off, size)
 805         except arvados.errors.NotFoundError as e:
 806             _logger.warning("Block not found: " + str(e))
 807             raise llfuse.FUSEError(errno.EIO)
 808         except Exception:
 809             _logger.exception()
 810             raise llfuse.FUSEError(errno.EIO)
 811
 812     def release(self, fh):
 813         if fh in self._filehandles:
 814             del self._filehandles[fh]
 815
 816     def opendir(self, inode):
 817         _logger.debug("arv-mount opendir: inode %i", inode)
 818
 819         if inode in self.inodes:
 820             p = self.inodes[inode]
 821         else:
 822             raise llfuse.FUSEError(errno.ENOENT)
 823
 824         if not isinstance(p, Directory):
 825             raise llfuse.FUSEError(errno.ENOTDIR)
 826
 827         fh = self._filehandles_counter
 828         self._filehandles_counter += 1
 829         if p.parent_inode in self.inodes:
 830             parent = self.inodes[p.parent_inode]
 831         else:
 832             raise llfuse.FUSEError(errno.EIO)
 833
 834         # update atime
 835         p._atime = time.time()
 836
 837         self._filehandles[fh] = FileHandle(fh, [('.', p), ('..', parent)] + list(p.items()))
 838         return fh
 839
 840     def readdir(self, fh, off):
 841         _logger.debug("arv-mount readdir: fh %i off %i", fh, off)
 842
 843         if fh in self._filehandles:
 844             handle = self._filehandles[fh]
 845         else:
 846             raise llfuse.FUSEError(errno.EBADF)
 847
 848         _logger.debug("arv-mount handle.entry %s", handle.entry)
 849
 850         e = off
 851         while e < len(handle.entry):
 852             if handle.entry[e][1].inode in self.inodes:
 853                 try:
 854                     yield (handle.entry[e][0].encode(self.encoding), self.getattr(handle.entry[e][1].inode), e+1)
 855                 except UnicodeEncodeError:
 856                     pass
 857             e += 1
 858
 859     def releasedir(self, fh):
 860         del self._filehandles[fh]
 861
 862     def statfs(self):
 863         st = llfuse.StatvfsData()
 864         st.f_bsize = 64 * 1024
 865         st.f_blocks = 0
 866         st.f_files = 0
 867
 868         st.f_bfree = 0
 869         st.f_bavail = 0
 870
 871         st.f_ffree = 0
 872         st.f_favail = 0
 873
 874         st.f_frsize = 0
 875         return st
 876
 877     # The llfuse documentation recommends only overloading functions that
 878     # are actually implemented, as the default implementation will raise ENOSYS.
 879     # However, there is a bug in the llfuse default implementation of create()
 880     # "create() takes exactly 5 positional arguments (6 given)" which will crash
 881     # arv-mount.
 882     # The workaround is to implement it with the proper number of parameters,
 883     # and then everything works out.
 884     def create(self, inode_parent, name, mode, flags, ctx):
 885         raise llfuse.FUSEError(errno.EROFS)