Merge branch '21678-installer-diagnostics-internal'. Closes #21678
[arvados.git] / services / fuse / arvados_fuse / fusedir.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: AGPL-3.0
4
5 import apiclient
6 import arvados
7 import errno
8 import functools
9 import llfuse
10 import logging
11 import re
12 import sys
13 import threading
14 import time
15 from apiclient import errors as apiclient_errors
16
17 from .fusefile import StringFile, ObjectFile, FuncToJSONFile, FuseArvadosFile
18 from .fresh import FreshBase, convertTime, use_counter, check_update
19
20 import arvados.collection
21 from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
22
23 _logger = logging.getLogger('arvados.arvados_fuse')
24
25
26 # Match any character which FUSE or Linux cannot accommodate as part
27 # of a filename. (If present in a collection filename, they will
28 # appear as underscores in the fuse mount.)
29 _disallowed_filename_characters = re.compile(r'[\x00/]')
30
31
32 class Directory(FreshBase):
33     """Generic directory object, backed by a dict.
34
35     Consists of a set of entries with the key representing the filename
36     and the value referencing a File or Directory object.
37     """
38
39     __slots__ = ("inode", "parent_inode", "inodes", "_entries", "_mtime", "_enable_write", "_filters")
40
41     def __init__(self, parent_inode, inodes, enable_write, filters):
42         """parent_inode is the integer inode number"""
43
44         super(Directory, self).__init__()
45
46         self.inode = None
47         if not isinstance(parent_inode, int):
48             raise Exception("parent_inode should be an int")
49         self.parent_inode = parent_inode
50         self.inodes = inodes
51         self._entries = {}
52         self._mtime = time.time()
53         self._enable_write = enable_write
54         self._filters = filters or []
55
56     def _filters_for(self, subtype, *, qualified):
57         for f in self._filters:
58             f_type, _, f_name = f[0].partition('.')
59             if not f_name:
60                 yield f
61             elif f_type != subtype:
62                 pass
63             elif qualified:
64                 yield f
65             else:
66                 yield [f_name, *f[1:]]
67
68     def unsanitize_filename(self, incoming):
69         """Replace ForwardSlashNameSubstitution value with /"""
70         fsns = self.inodes.forward_slash_subst()
71         if isinstance(fsns, str):
72             return incoming.replace(fsns, '/')
73         else:
74             return incoming
75
76     def sanitize_filename(self, dirty):
77         """Replace disallowed filename characters according to
78         ForwardSlashNameSubstitution in self.api_config."""
79         # '.' and '..' are not reachable if API server is newer than #6277
80         if dirty is None:
81             return None
82         elif dirty == '':
83             return '_'
84         elif dirty == '.':
85             return '_'
86         elif dirty == '..':
87             return '__'
88         else:
89             fsns = self.inodes.forward_slash_subst()
90             if isinstance(fsns, str):
91                 dirty = dirty.replace('/', fsns)
92             return _disallowed_filename_characters.sub('_', dirty)
93
94
95     #  Overridden by subclasses to implement logic to update the
96     #  entries dict when the directory is stale
97     @use_counter
98     def update(self):
99         pass
100
101     # Only used when computing the size of the disk footprint of the directory
102     # (stub)
103     def size(self):
104         return 0
105
106     def persisted(self):
107         return False
108
109     def checkupdate(self):
110         if self.stale():
111             try:
112                 self.update()
113             except apiclient.errors.HttpError as e:
114                 _logger.warn(e)
115
116     @use_counter
117     @check_update
118     def __getitem__(self, item):
119         return self._entries[item]
120
121     @use_counter
122     @check_update
123     def items(self):
124         return list(self._entries.items())
125
126     @use_counter
127     @check_update
128     def __contains__(self, k):
129         return k in self._entries
130
131     @use_counter
132     @check_update
133     def __len__(self):
134         return len(self._entries)
135
136     def fresh(self):
137         self.inodes.touch(self)
138         super(Directory, self).fresh()
139
140     def objsize(self):
141         # Rough estimate of memory footprint based on using pympler
142         return len(self._entries) * 1024
143
144     def merge(self, items, fn, same, new_entry):
145         """Helper method for updating the contents of the directory.
146
147         Takes a list describing the new contents of the directory, reuse
148         entries that are the same in both the old and new lists, create new
149         entries, and delete old entries missing from the new list.
150
151         Arguments:
152         * items: Iterable --- New directory contents
153
154         * fn: Callable --- Takes an entry in 'items' and return the desired file or
155         directory name, or None if this entry should be skipped
156
157         * same: Callable --- Compare an existing entry (a File or Directory
158         object) with an entry in the items list to determine whether to keep
159         the existing entry.
160
161         * new_entry: Callable --- Create a new directory entry (File or Directory
162         object) from an entry in the items list.
163
164         """
165
166         oldentries = self._entries
167         self._entries = {}
168         changed = False
169         for i in items:
170             name = self.sanitize_filename(fn(i))
171             if not name:
172                 continue
173             if name in oldentries:
174                 ent = oldentries[name]
175                 if same(ent, i) and ent.parent_inode == self.inode:
176                     # move existing directory entry over
177                     self._entries[name] = ent
178                     del oldentries[name]
179                     self.inodes.inode_cache.touch(ent)
180
181         for i in items:
182             name = self.sanitize_filename(fn(i))
183             if not name:
184                 continue
185             if name not in self._entries:
186                 # create new directory entry
187                 ent = new_entry(i)
188                 if ent is not None:
189                     self._entries[name] = self.inodes.add_entry(ent)
190                     # need to invalidate this just in case there was a
191                     # previous entry that couldn't be moved over or a
192                     # lookup that returned file not found and cached
193                     # a negative result
194                     self.inodes.invalidate_entry(self, name)
195                     changed = True
196                 _logger.debug("Added entry '%s' as inode %i to parent inode %i", name, ent.inode, self.inode)
197
198         # delete any other directory entries that were not in found in 'items'
199         for name, ent in oldentries.items():
200             _logger.debug("Detaching entry '%s' from parent_inode %i", name, self.inode)
201             self.inodes.invalidate_entry(self, name)
202             self.inodes.del_entry(ent)
203             changed = True
204
205         if changed:
206             self._mtime = time.time()
207             self.inodes.inode_cache.update_cache_size(self)
208
209         self.fresh()
210
211     def in_use(self):
212         if super(Directory, self).in_use():
213             return True
214         for v in self._entries.values():
215             if v.in_use():
216                 return True
217         return False
218
219     def clear(self):
220         """Delete all entries"""
221         if not self._entries:
222             return
223         oldentries = self._entries
224         self._entries = {}
225         self.invalidate()
226         for name, ent in oldentries.items():
227             ent.clear()
228             self.inodes.invalidate_entry(self, name)
229             self.inodes.del_entry(ent)
230         self.inodes.inode_cache.update_cache_size(self)
231
232     def kernel_invalidate(self):
233         # Invalidating the dentry on the parent implies invalidating all paths
234         # below it as well.
235         if self.parent_inode in self.inodes:
236             parent = self.inodes[self.parent_inode]
237         else:
238             # parent was removed already.
239             return
240
241         # Find self on the parent in order to invalidate this path.
242         # Calling the public items() method might trigger a refresh,
243         # which we definitely don't want, so read the internal dict directly.
244         for k,v in parent._entries.items():
245             if v is self:
246                 self.inodes.invalidate_entry(parent, k)
247                 break
248
249     def mtime(self):
250         return self._mtime
251
252     def writable(self):
253         return False
254
255     def flush(self):
256         pass
257
258     def want_event_subscribe(self):
259         raise NotImplementedError()
260
261     def create(self, name):
262         raise NotImplementedError()
263
264     def mkdir(self, name):
265         raise NotImplementedError()
266
267     def unlink(self, name):
268         raise NotImplementedError()
269
270     def rmdir(self, name):
271         raise NotImplementedError()
272
273     def rename(self, name_old, name_new, src):
274         raise NotImplementedError()
275
276
277 class CollectionDirectoryBase(Directory):
278     """Represent an Arvados Collection as a directory.
279
280     This class is used for Subcollections, and is also the base class for
281     CollectionDirectory, which implements collection loading/saving on
282     Collection records.
283
284     Most operations act only the underlying Arvados `Collection` object.  The
285     `Collection` object signals via a notify callback to
286     `CollectionDirectoryBase.on_event` that an item was added, removed or
287     modified.  FUSE inodes and directory entries are created, deleted or
288     invalidated in response to these events.
289
290     """
291
292     __slots__ = ("collection", "collection_root", "collection_record_file")
293
294     def __init__(self, parent_inode, inodes, enable_write, filters, collection, collection_root):
295         super(CollectionDirectoryBase, self).__init__(parent_inode, inodes, enable_write, filters)
296         self.collection = collection
297         self.collection_root = collection_root
298         self.collection_record_file = None
299
300     def new_entry(self, name, item, mtime):
301         name = self.sanitize_filename(name)
302         if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
303             if item.fuse_entry.parent_inode is not None:
304                 raise Exception("Can only reparent unparented inode entry")
305             if item.fuse_entry.inode is None:
306                 raise Exception("Reparented entry must still have valid inode")
307             item.fuse_entry.parent_inode = self.inode
308             self._entries[name] = item.fuse_entry
309         elif isinstance(item, arvados.collection.RichCollectionBase):
310             self._entries[name] = self.inodes.add_entry(CollectionDirectoryBase(
311                 self.inode,
312                 self.inodes,
313                 self._enable_write,
314                 self._filters,
315                 item,
316                 self.collection_root,
317             ))
318             self._entries[name].populate(mtime)
319         else:
320             self._entries[name] = self.inodes.add_entry(FuseArvadosFile(self.inode, item, mtime, self._enable_write))
321         item.fuse_entry = self._entries[name]
322
323     def on_event(self, event, collection, name, item):
324         # These are events from the Collection object (ADD/DEL/MOD)
325         # emitted by operations on the Collection object (like
326         # "mkdirs" or "remove"), and by "update", which we need to
327         # synchronize with our FUSE objects that are assigned inodes.
328         if collection == self.collection:
329             name = self.sanitize_filename(name)
330
331             #
332             # It's possible for another thread to have llfuse.lock and
333             # be waiting on collection.lock.  Meanwhile, we released
334             # llfuse.lock earlier in the stack, but are still holding
335             # on to the collection lock, and now we need to re-acquire
336             # llfuse.lock.  If we don't release the collection lock,
337             # we'll deadlock where we're holding the collection lock
338             # waiting for llfuse.lock and the other thread is holding
339             # llfuse.lock and waiting for the collection lock.
340             #
341             # The correct locking order here is to take llfuse.lock
342             # first, then the collection lock.
343             #
344             # Since collection.lock is an RLock, it might be locked
345             # multiple times, so we need to release it multiple times,
346             # keep a count, then re-lock it the correct number of
347             # times.
348             #
349             lockcount = 0
350             try:
351                 while True:
352                     self.collection.lock.release()
353                     lockcount += 1
354             except RuntimeError:
355                 pass
356
357             try:
358                 with llfuse.lock:
359                     with self.collection.lock:
360                         if event == arvados.collection.ADD:
361                             self.new_entry(name, item, self.mtime())
362                         elif event == arvados.collection.DEL:
363                             ent = self._entries[name]
364                             del self._entries[name]
365                             self.inodes.invalidate_entry(self, name)
366                             self.inodes.del_entry(ent)
367                         elif event == arvados.collection.MOD:
368                             if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
369                                 self.inodes.invalidate_inode(item.fuse_entry)
370                             elif name in self._entries:
371                                 self.inodes.invalidate_inode(self._entries[name])
372
373                         if self.collection_record_file is not None:
374                             self.collection_record_file.invalidate()
375                             self.inodes.invalidate_inode(self.collection_record_file)
376             finally:
377                 while lockcount > 0:
378                     self.collection.lock.acquire()
379                     lockcount -= 1
380
381     def populate(self, mtime):
382         self._mtime = mtime
383         with self.collection.lock:
384             self.collection.subscribe(self.on_event)
385             for entry, item in self.collection.items():
386                 self.new_entry(entry, item, self.mtime())
387
388     def writable(self):
389         return self._enable_write and self.collection.writable()
390
391     @use_counter
392     def flush(self):
393         self.collection_root.flush()
394
395     @use_counter
396     @check_update
397     def create(self, name):
398         if not self.writable():
399             raise llfuse.FUSEError(errno.EROFS)
400         with llfuse.lock_released:
401             self.collection.open(name, "w").close()
402
403     @use_counter
404     @check_update
405     def mkdir(self, name):
406         if not self.writable():
407             raise llfuse.FUSEError(errno.EROFS)
408         with llfuse.lock_released:
409             self.collection.mkdirs(name)
410
411     @use_counter
412     @check_update
413     def unlink(self, name):
414         if not self.writable():
415             raise llfuse.FUSEError(errno.EROFS)
416         with llfuse.lock_released:
417             self.collection.remove(name)
418         self.flush()
419
420     @use_counter
421     @check_update
422     def rmdir(self, name):
423         if not self.writable():
424             raise llfuse.FUSEError(errno.EROFS)
425         with llfuse.lock_released:
426             self.collection.remove(name)
427         self.flush()
428
429     @use_counter
430     @check_update
431     def rename(self, name_old, name_new, src):
432         if not self.writable():
433             raise llfuse.FUSEError(errno.EROFS)
434
435         if not isinstance(src, CollectionDirectoryBase):
436             raise llfuse.FUSEError(errno.EPERM)
437
438         if name_new in self:
439             ent = src[name_old]
440             tgt = self[name_new]
441             if isinstance(ent, FuseArvadosFile) and isinstance(tgt, FuseArvadosFile):
442                 pass
443             elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, CollectionDirectoryBase):
444                 if len(tgt) > 0:
445                     raise llfuse.FUSEError(errno.ENOTEMPTY)
446             elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, FuseArvadosFile):
447                 raise llfuse.FUSEError(errno.ENOTDIR)
448             elif isinstance(ent, FuseArvadosFile) and isinstance(tgt, CollectionDirectoryBase):
449                 raise llfuse.FUSEError(errno.EISDIR)
450
451         with llfuse.lock_released:
452             self.collection.rename(name_old, name_new, source_collection=src.collection, overwrite=True)
453         self.flush()
454         src.flush()
455
456     def clear(self):
457         super(CollectionDirectoryBase, self).clear()
458         if self.collection is not None:
459             self.collection.unsubscribe()
460         self.collection = None
461
462     def objsize(self):
463         # objsize for the whole collection is represented at the root,
464         # don't double-count it
465         return 0
466
467 class CollectionDirectory(CollectionDirectoryBase):
468     """Represents the root of a directory tree representing a collection."""
469
470     __slots__ = ("api", "num_retries", "collection_locator",
471                  "_manifest_size", "_writable", "_updating_lock")
472
473     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters=None, collection_record=None, explicit_collection=None):
474         super(CollectionDirectory, self).__init__(parent_inode, inodes, enable_write, filters, None, self)
475         self.api = api
476         self.num_retries = num_retries
477         self._poll = True
478         try:
479             self._poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2) // 2)
480         except:
481             _logger.debug("Error getting blobSignatureTtl from discovery document: %s", sys.exc_info()[0])
482             self._poll_time = 60*60
483
484         if isinstance(collection_record, dict):
485             self.collection_locator = collection_record['uuid']
486             self._mtime = convertTime(collection_record.get('modified_at'))
487         else:
488             self.collection_locator = collection_record
489             self._mtime = 0
490         self._manifest_size = 0
491         if self.collection_locator:
492             self._writable = (uuid_pattern.match(self.collection_locator) is not None) and enable_write
493         self._updating_lock = threading.Lock()
494
495     def same(self, i):
496         return i['uuid'] == self.collection_locator or i['portable_data_hash'] == self.collection_locator
497
498     def writable(self):
499         return self._enable_write and (self.collection.writable() if self.collection is not None else self._writable)
500
501     @use_counter
502     def flush(self):
503         if not self.writable():
504             return
505         with llfuse.lock_released:
506             with self._updating_lock:
507                 if self.collection.committed():
508                     self.collection.update()
509                 else:
510                     self.collection.save()
511                 self.new_collection_record(self.collection.api_response())
512
513     def want_event_subscribe(self):
514         return (uuid_pattern.match(self.collection_locator) is not None)
515
516     def new_collection(self, new_collection_record, coll_reader):
517         if self.inode:
518             self.clear()
519         self.collection = coll_reader
520         self.new_collection_record(new_collection_record)
521         self.populate(self.mtime())
522
523     def new_collection_record(self, new_collection_record):
524         if not new_collection_record:
525             raise Exception("invalid new_collection_record")
526         self._mtime = convertTime(new_collection_record.get('modified_at'))
527         self._manifest_size = len(new_collection_record["manifest_text"])
528         self.collection_locator = new_collection_record["uuid"]
529         if self.collection_record_file is not None:
530             self.collection_record_file.invalidate()
531             self.inodes.invalidate_inode(self.collection_record_file)
532             _logger.debug("parent_inode %s invalidated collection record file inode %s", self.inode,
533                           self.collection_record_file.inode)
534         self.inodes.update_uuid(self)
535         self.inodes.inode_cache.update_cache_size(self)
536         self.fresh()
537
538     def uuid(self):
539         return self.collection_locator
540
541     @use_counter
542     def update(self):
543         try:
544             if self.collection is not None and portable_data_hash_pattern.match(self.collection_locator):
545                 # It's immutable, nothing to update
546                 return True
547
548             if self.collection_locator is None:
549                 # No collection locator to retrieve from
550                 self.fresh()
551                 return True
552
553             new_collection_record = None
554             try:
555                 with llfuse.lock_released:
556                     self._updating_lock.acquire()
557                     if not self.stale():
558                         return True
559
560                     _logger.debug("Updating collection %s inode %s", self.collection_locator, self.inode)
561                     coll_reader = None
562                     if self.collection is not None:
563                         # Already have a collection object
564                         self.collection.update()
565                         new_collection_record = self.collection.api_response()
566                     else:
567                         # Create a new collection object
568                         if uuid_pattern.match(self.collection_locator):
569                             coll_reader = arvados.collection.Collection(
570                                 self.collection_locator, self.api, self.api.keep,
571                                 num_retries=self.num_retries)
572                         else:
573                             coll_reader = arvados.collection.CollectionReader(
574                                 self.collection_locator, self.api, self.api.keep,
575                                 num_retries=self.num_retries)
576                         new_collection_record = coll_reader.api_response() or {}
577                         # If the Collection only exists in Keep, there will be no API
578                         # response.  Fill in the fields we need.
579                         if 'uuid' not in new_collection_record:
580                             new_collection_record['uuid'] = self.collection_locator
581                         if "portable_data_hash" not in new_collection_record:
582                             new_collection_record["portable_data_hash"] = new_collection_record["uuid"]
583                         if 'manifest_text' not in new_collection_record:
584                             new_collection_record['manifest_text'] = coll_reader.manifest_text()
585                         if 'storage_classes_desired' not in new_collection_record:
586                             new_collection_record['storage_classes_desired'] = coll_reader.storage_classes_desired()
587
588                 # end with llfuse.lock_released, re-acquire lock
589
590                 if new_collection_record is not None:
591                     if coll_reader is not None:
592                         self.new_collection(new_collection_record, coll_reader)
593                     else:
594                         self.new_collection_record(new_collection_record)
595
596                 return True
597             finally:
598                 self._updating_lock.release()
599         except arvados.errors.NotFoundError as e:
600             _logger.error("Error fetching collection '%s': %s", self.collection_locator, e)
601         except arvados.errors.ArgumentError as detail:
602             _logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
603             if new_collection_record is not None and "manifest_text" in new_collection_record:
604                 _logger.warning("arv-mount manifest_text is: %s", new_collection_record["manifest_text"])
605         except Exception:
606             _logger.exception("arv-mount %s: error", self.collection_locator)
607             if new_collection_record is not None and "manifest_text" in new_collection_record:
608                 _logger.error("arv-mount manifest_text is: %s", new_collection_record["manifest_text"])
609         self.invalidate()
610         return False
611
612     @use_counter
613     @check_update
614     def collection_record(self):
615         self.flush()
616         return self.collection.api_response()
617
618     @use_counter
619     @check_update
620     def __getitem__(self, item):
621         if item == '.arvados#collection':
622             if self.collection_record_file is None:
623                 self.collection_record_file = FuncToJSONFile(
624                     self.inode, self.collection_record)
625                 self.inodes.add_entry(self.collection_record_file)
626             self.invalidate()  # use lookup as a signal to force update
627             return self.collection_record_file
628         else:
629             return super(CollectionDirectory, self).__getitem__(item)
630
631     def __contains__(self, k):
632         if k == '.arvados#collection':
633             return True
634         else:
635             return super(CollectionDirectory, self).__contains__(k)
636
637     def invalidate(self):
638         if self.collection_record_file is not None:
639             self.collection_record_file.invalidate()
640             self.inodes.invalidate_inode(self.collection_record_file)
641         super(CollectionDirectory, self).invalidate()
642
643     def persisted(self):
644         return (self.collection_locator is not None)
645
646     def objsize(self):
647         # This is a rough guess of the amount of overhead involved for
648         # a collection; the assumptions are that that each file
649         # averages 128 bytes in the manifest, but consume 1024 bytes
650         # of Python data structures, so 1024/128=8 means we estimate
651         # the RAM footprint at 8 times the size of bare manifest text.
652         return self._manifest_size * 8
653
654     def finalize(self):
655         if self.collection is None:
656             return
657
658         if self.writable():
659             try:
660                 self.collection.save()
661             except Exception as e:
662                 _logger.exception("Failed to save collection %s", self.collection_locator)
663         self.collection.stop_threads()
664
665     def clear(self):
666         if self.collection is not None:
667             self.collection.stop_threads()
668         self._manifest_size = 0
669         super(CollectionDirectory, self).clear()
670         if self.collection_record_file is not None:
671             self.inodes.del_entry(self.collection_record_file)
672         self.collection_record_file = None
673
674
675 class TmpCollectionDirectory(CollectionDirectoryBase):
676     """A directory backed by an Arvados collection that never gets saved.
677
678     This supports using Keep as scratch space. A userspace program can
679     read the .arvados#collection file to get a current manifest in
680     order to save a snapshot of the scratch data or use it as a crunch
681     job output.
682     """
683
684     class UnsaveableCollection(arvados.collection.Collection):
685         def save(self):
686             pass
687         def save_new(self):
688             pass
689
690     def __init__(self, parent_inode, inodes, api_client, num_retries, enable_write, filters=None, storage_classes=None):
691         collection = self.UnsaveableCollection(
692             api_client=api_client,
693             keep_client=api_client.keep,
694             num_retries=num_retries,
695             storage_classes_desired=storage_classes)
696         # This is always enable_write=True because it never tries to
697         # save to the backend
698         super(TmpCollectionDirectory, self).__init__(
699             parent_inode, inodes, True, filters, collection, self)
700         self.populate(self.mtime())
701
702     def on_event(self, *args, **kwargs):
703         super(TmpCollectionDirectory, self).on_event(*args, **kwargs)
704         if self.collection_record_file is None:
705             return
706
707         # See discussion in CollectionDirectoryBase.on_event
708         lockcount = 0
709         try:
710             while True:
711                 self.collection.lock.release()
712                 lockcount += 1
713         except RuntimeError:
714             pass
715
716         try:
717             with llfuse.lock:
718                 with self.collection.lock:
719                     self.collection_record_file.invalidate()
720                     self.inodes.invalidate_inode(self.collection_record_file)
721                     _logger.debug("%s invalidated collection record", self.inode)
722         finally:
723             while lockcount > 0:
724                 self.collection.lock.acquire()
725                 lockcount -= 1
726
727     def collection_record(self):
728         with llfuse.lock_released:
729             return {
730                 "uuid": None,
731                 "manifest_text": self.collection.manifest_text(),
732                 "portable_data_hash": self.collection.portable_data_hash(),
733                 "storage_classes_desired": self.collection.storage_classes_desired(),
734             }
735
736     def __contains__(self, k):
737         return (k == '.arvados#collection' or
738                 super(TmpCollectionDirectory, self).__contains__(k))
739
740     @use_counter
741     def __getitem__(self, item):
742         if item == '.arvados#collection':
743             if self.collection_record_file is None:
744                 self.collection_record_file = FuncToJSONFile(
745                     self.inode, self.collection_record)
746                 self.inodes.add_entry(self.collection_record_file)
747             return self.collection_record_file
748         return super(TmpCollectionDirectory, self).__getitem__(item)
749
750     def persisted(self):
751         return False
752
753     def writable(self):
754         return True
755
756     def flush(self):
757         pass
758
759     def want_event_subscribe(self):
760         return False
761
762     def finalize(self):
763         self.collection.stop_threads()
764
765     def invalidate(self):
766         if self.collection_record_file:
767             self.collection_record_file.invalidate()
768         super(TmpCollectionDirectory, self).invalidate()
769
770
771 class MagicDirectory(Directory):
772     """A special directory that logically contains the set of all extant keep locators.
773
774     When a file is referenced by lookup(), it is tested to see if it is a valid
775     keep locator to a manifest, and if so, loads the manifest contents as a
776     subdirectory of this directory with the locator as the directory name.
777     Since querying a list of all extant keep locators is impractical, only
778     collections that have already been accessed are visible to readdir().
779
780     """
781
782     README_TEXT = """
783 This directory provides access to Arvados collections as subdirectories listed
784 by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
785 the form '1234567890abcdef0123456789abcdef+123'), and Arvados projects by uuid
786 (in the form 'zzzzz-j7d0g-1234567890abcde').
787
788 Note that this directory will appear empty until you attempt to access a
789 specific collection or project subdirectory (such as trying to 'cd' into it),
790 at which point the collection or project will actually be looked up on the server
791 and the directory will appear if it exists.
792
793 """.lstrip()
794
795     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters, pdh_only=False, storage_classes=None):
796         super(MagicDirectory, self).__init__(parent_inode, inodes, enable_write, filters)
797         self.api = api
798         self.num_retries = num_retries
799         self.pdh_only = pdh_only
800         self.storage_classes = storage_classes
801
802     def __setattr__(self, name, value):
803         super(MagicDirectory, self).__setattr__(name, value)
804         # When we're assigned an inode, add a README.
805         if ((name == 'inode') and (self.inode is not None) and
806               (not self._entries)):
807             self._entries['README'] = self.inodes.add_entry(
808                 StringFile(self.inode, self.README_TEXT, time.time()))
809             # If we're the root directory, add an identical by_id subdirectory.
810             if self.inode == llfuse.ROOT_INODE:
811                 self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
812                     self.inode,
813                     self.inodes,
814                     self.api,
815                     self.num_retries,
816                     self._enable_write,
817                     self._filters,
818                     self.pdh_only,
819                 ))
820
821     def __contains__(self, k):
822         if k in self._entries:
823             return True
824
825         if not portable_data_hash_pattern.match(k) and (self.pdh_only or not uuid_pattern.match(k)):
826             return False
827
828         try:
829             e = None
830
831             if group_uuid_pattern.match(k):
832                 project = self.api.groups().list(
833                     filters=[
834                         ['group_class', 'in', ['project','filter']],
835                         ["uuid", "=", k],
836                         *self._filters_for('groups', qualified=False),
837                     ],
838                 ).execute(num_retries=self.num_retries)
839                 if project[u'items_available'] == 0:
840                     return False
841                 e = self.inodes.add_entry(ProjectDirectory(
842                     self.inode,
843                     self.inodes,
844                     self.api,
845                     self.num_retries,
846                     self._enable_write,
847                     self._filters,
848                     project[u'items'][0],
849                     storage_classes=self.storage_classes,
850                 ))
851             else:
852                 e = self.inodes.add_entry(CollectionDirectory(
853                     self.inode,
854                     self.inodes,
855                     self.api,
856                     self.num_retries,
857                     self._enable_write,
858                     self._filters,
859                     k,
860                 ))
861
862             if e.update():
863                 if k not in self._entries:
864                     self._entries[k] = e
865                 else:
866                     self.inodes.del_entry(e)
867                 return True
868             else:
869                 self.inodes.invalidate_entry(self, k)
870                 self.inodes.del_entry(e)
871                 return False
872         except Exception as ex:
873             _logger.exception("arv-mount lookup '%s':", k)
874             if e is not None:
875                 self.inodes.del_entry(e)
876             return False
877
878     def __getitem__(self, item):
879         if item in self:
880             return self._entries[item]
881         else:
882             raise KeyError("No collection with id " + item)
883
884     def clear(self):
885         pass
886
887     def want_event_subscribe(self):
888         return not self.pdh_only
889
890
891 class TagsDirectory(Directory):
892     """A special directory that contains as subdirectories all tags visible to the user."""
893
894     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters, poll_time=60):
895         super(TagsDirectory, self).__init__(parent_inode, inodes, enable_write, filters)
896         self.api = api
897         self.num_retries = num_retries
898         self._poll = True
899         self._poll_time = poll_time
900         self._extra = set()
901
902     def want_event_subscribe(self):
903         return True
904
905     @use_counter
906     def update(self):
907         with llfuse.lock_released:
908             tags = self.api.links().list(
909                 filters=[
910                     ['link_class', '=', 'tag'],
911                     ['name', '!=', ''],
912                     *self._filters_for('links', qualified=False),
913                 ],
914                 select=['name'],
915                 distinct=True,
916                 limit=1000,
917             ).execute(num_retries=self.num_retries)
918         if "items" in tags:
919             self.merge(
920                 tags['items']+[{"name": n} for n in self._extra],
921                 lambda i: i['name'],
922                 lambda a, i: a.tag == i['name'],
923                 lambda i: TagDirectory(
924                     self.inode,
925                     self.inodes,
926                     self.api,
927                     self.num_retries,
928                     self._enable_write,
929                     self._filters,
930                     i['name'],
931                     poll=self._poll,
932                     poll_time=self._poll_time,
933                 ),
934             )
935
936     @use_counter
937     @check_update
938     def __getitem__(self, item):
939         if super(TagsDirectory, self).__contains__(item):
940             return super(TagsDirectory, self).__getitem__(item)
941         with llfuse.lock_released:
942             tags = self.api.links().list(
943                 filters=[
944                     ['link_class', '=', 'tag'],
945                     ['name', '=', item],
946                     *self._filters_for('links', qualified=False),
947                 ],
948                 limit=1,
949             ).execute(num_retries=self.num_retries)
950         if tags["items"]:
951             self._extra.add(item)
952             self.update()
953         return super(TagsDirectory, self).__getitem__(item)
954
955     @use_counter
956     @check_update
957     def __contains__(self, k):
958         if super(TagsDirectory, self).__contains__(k):
959             return True
960         try:
961             self[k]
962             return True
963         except KeyError:
964             pass
965         return False
966
967
968 class TagDirectory(Directory):
969     """A special directory that contains as subdirectories all collections visible
970     to the user that are tagged with a particular tag.
971     """
972
973     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters, tag,
974                  poll=False, poll_time=60):
975         super(TagDirectory, self).__init__(parent_inode, inodes, enable_write, filters)
976         self.api = api
977         self.num_retries = num_retries
978         self.tag = tag
979         self._poll = poll
980         self._poll_time = poll_time
981
982     def want_event_subscribe(self):
983         return True
984
985     @use_counter
986     def update(self):
987         with llfuse.lock_released:
988             taggedcollections = self.api.links().list(
989                 filters=[
990                     ['link_class', '=', 'tag'],
991                     ['name', '=', self.tag],
992                     ['head_uuid', 'is_a', 'arvados#collection'],
993                     *self._filters_for('links', qualified=False),
994                 ],
995                 select=['head_uuid'],
996             ).execute(num_retries=self.num_retries)
997         self.merge(
998             taggedcollections['items'],
999             lambda i: i['head_uuid'],
1000             lambda a, i: a.collection_locator == i['head_uuid'],
1001             lambda i: CollectionDirectory(
1002                 self.inode,
1003                 self.inodes,
1004                 self.api,
1005                 self.num_retries,
1006                 self._enable_write,
1007                 self._filters,
1008                 i['head_uuid'],
1009             ),
1010         )
1011
1012
1013 class ProjectDirectory(Directory):
1014     """A special directory that contains the contents of a project."""
1015
1016     __slots__ = ("api", "num_retries", "project_object", "project_object_file",
1017                  "project_uuid", "_updating_lock",
1018                  "_current_user", "_full_listing", "storage_classes", "recursively_contained")
1019
1020     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters,
1021                  project_object, poll=True, poll_time=3, storage_classes=None):
1022         super(ProjectDirectory, self).__init__(parent_inode, inodes, enable_write, filters)
1023         self.api = api
1024         self.num_retries = num_retries
1025         self.project_object = project_object
1026         self.project_object_file = None
1027         self.project_uuid = project_object['uuid']
1028         self._poll = poll
1029         self._poll_time = poll_time
1030         self._updating_lock = threading.Lock()
1031         self._current_user = None
1032         self._full_listing = False
1033         self.storage_classes = storage_classes
1034         self.recursively_contained = False
1035
1036         # Filter groups can contain themselves, which causes tools
1037         # that walk the filesystem to get stuck in an infinite loop,
1038         # so suppress returning a listing in that case.
1039         if self.project_object.get("group_class") == "filter":
1040             iter_parent_inode = parent_inode
1041             while iter_parent_inode != llfuse.ROOT_INODE:
1042                 parent_dir = self.inodes[iter_parent_inode]
1043                 if isinstance(parent_dir, ProjectDirectory) and parent_dir.project_uuid == self.project_uuid:
1044                     self.recursively_contained = True
1045                     break
1046                 iter_parent_inode = parent_dir.parent_inode
1047
1048     def want_event_subscribe(self):
1049         return True
1050
1051     def createDirectory(self, i):
1052         common_args = (self.inode, self.inodes, self.api, self.num_retries, self._enable_write, self._filters)
1053         if collection_uuid_pattern.match(i['uuid']):
1054             return CollectionDirectory(*common_args, i)
1055         elif group_uuid_pattern.match(i['uuid']):
1056             return ProjectDirectory(*common_args, i, self._poll, self._poll_time, self.storage_classes)
1057         elif link_uuid_pattern.match(i['uuid']):
1058             if i['head_kind'] == 'arvados#collection' or portable_data_hash_pattern.match(i['head_uuid']):
1059                 return CollectionDirectory(*common_args, i['head_uuid'])
1060             else:
1061                 return None
1062         elif uuid_pattern.match(i['uuid']):
1063             return ObjectFile(self.parent_inode, i)
1064         else:
1065             return None
1066
1067     def uuid(self):
1068         return self.project_uuid
1069
1070     def items(self):
1071         self._full_listing = True
1072         return super(ProjectDirectory, self).items()
1073
1074     def namefn(self, i):
1075         if 'name' in i:
1076             if i['name'] is None or len(i['name']) == 0:
1077                 return None
1078             elif "uuid" in i and (collection_uuid_pattern.match(i['uuid']) or group_uuid_pattern.match(i['uuid'])):
1079                 # collection or subproject
1080                 return i['name']
1081             elif link_uuid_pattern.match(i['uuid']) and i['head_kind'] == 'arvados#collection':
1082                 # name link
1083                 return i['name']
1084             elif 'kind' in i and i['kind'].startswith('arvados#'):
1085                 # something else
1086                 return "{}.{}".format(i['name'], i['kind'][8:])
1087         else:
1088             return None
1089
1090
1091     @use_counter
1092     def update(self):
1093         if self.project_object_file == None:
1094             self.project_object_file = ObjectFile(self.inode, self.project_object)
1095             self.inodes.add_entry(self.project_object_file)
1096
1097         if self.recursively_contained or not self._full_listing:
1098             return True
1099
1100         def samefn(a, i):
1101             if isinstance(a, CollectionDirectory) or isinstance(a, ProjectDirectory):
1102                 return a.uuid() == i['uuid']
1103             elif isinstance(a, ObjectFile):
1104                 return a.uuid() == i['uuid'] and not a.stale()
1105             return False
1106
1107         try:
1108             with llfuse.lock_released:
1109                 self._updating_lock.acquire()
1110                 if not self.stale():
1111                     return
1112
1113                 if group_uuid_pattern.match(self.project_uuid):
1114                     self.project_object = self.api.groups().get(
1115                         uuid=self.project_uuid).execute(num_retries=self.num_retries)
1116                 elif user_uuid_pattern.match(self.project_uuid):
1117                     self.project_object = self.api.users().get(
1118                         uuid=self.project_uuid).execute(num_retries=self.num_retries)
1119                 # do this in 2 steps until #17424 is fixed
1120                 contents = list(arvados.util.keyset_list_all(
1121                     self.api.groups().contents,
1122                     order_key='uuid',
1123                     num_retries=self.num_retries,
1124                     uuid=self.project_uuid,
1125                     filters=[
1126                         ['uuid', 'is_a', 'arvados#group'],
1127                         ['groups.group_class', 'in', ['project', 'filter']],
1128                         *self._filters_for('groups', qualified=True),
1129                     ],
1130                 ))
1131                 contents.extend(obj for obj in arvados.util.keyset_list_all(
1132                     self.api.groups().contents,
1133                     order_key='uuid',
1134                     num_retries=self.num_retries,
1135                     uuid=self.project_uuid,
1136                     filters=[
1137                         ['uuid', 'is_a', 'arvados#collection'],
1138                         *self._filters_for('collections', qualified=True),
1139                     ],
1140                 ) if obj['current_version_uuid'] == obj['uuid'])
1141             # end with llfuse.lock_released, re-acquire lock
1142
1143             self.merge(contents,
1144                        self.namefn,
1145                        samefn,
1146                        self.createDirectory)
1147             return True
1148         finally:
1149             self._updating_lock.release()
1150
1151     def _add_entry(self, i, name):
1152         ent = self.createDirectory(i)
1153         self._entries[name] = self.inodes.add_entry(ent)
1154         return self._entries[name]
1155
1156     @use_counter
1157     @check_update
1158     def __getitem__(self, k):
1159         if k == '.arvados#project':
1160             return self.project_object_file
1161         elif self._full_listing or super(ProjectDirectory, self).__contains__(k):
1162             return super(ProjectDirectory, self).__getitem__(k)
1163         with llfuse.lock_released:
1164             k2 = self.unsanitize_filename(k)
1165             if k2 == k:
1166                 namefilter = ["name", "=", k]
1167             else:
1168                 namefilter = ["name", "in", [k, k2]]
1169             contents = self.api.groups().list(
1170                 filters=[
1171                     ["owner_uuid", "=", self.project_uuid],
1172                     ["group_class", "in", ["project","filter"]],
1173                     namefilter,
1174                     *self._filters_for('groups', qualified=False),
1175                 ],
1176                 limit=2,
1177             ).execute(num_retries=self.num_retries)["items"]
1178             if not contents:
1179                 contents = self.api.collections().list(
1180                     filters=[
1181                         ["owner_uuid", "=", self.project_uuid],
1182                         namefilter,
1183                         *self._filters_for('collections', qualified=False),
1184                     ],
1185                     limit=2,
1186                 ).execute(num_retries=self.num_retries)["items"]
1187         if contents:
1188             if len(contents) > 1 and contents[1]['name'] == k:
1189                 # If "foo/bar" and "foo[SUBST]bar" both exist, use
1190                 # "foo[SUBST]bar".
1191                 contents = [contents[1]]
1192             name = self.sanitize_filename(self.namefn(contents[0]))
1193             if name != k:
1194                 raise KeyError(k)
1195             return self._add_entry(contents[0], name)
1196
1197         # Didn't find item
1198         raise KeyError(k)
1199
1200     def __contains__(self, k):
1201         if k == '.arvados#project':
1202             return True
1203         try:
1204             self[k]
1205             return True
1206         except KeyError:
1207             pass
1208         return False
1209
1210     @use_counter
1211     @check_update
1212     def writable(self):
1213         if not self._enable_write:
1214             return False
1215         with llfuse.lock_released:
1216             if not self._current_user:
1217                 self._current_user = self.api.users().current().execute(num_retries=self.num_retries)
1218             return self._current_user["uuid"] in self.project_object.get("writable_by", [])
1219
1220     def persisted(self):
1221         return True
1222
1223     def clear(self):
1224         super(ProjectDirectory, self).clear()
1225         if self.project_object_file is not None:
1226             self.inodes.del_entry(self.project_object_file)
1227         self.project_object_file = None
1228
1229     @use_counter
1230     @check_update
1231     def mkdir(self, name):
1232         if not self.writable():
1233             raise llfuse.FUSEError(errno.EROFS)
1234
1235         try:
1236             with llfuse.lock_released:
1237                 c = {
1238                     "owner_uuid": self.project_uuid,
1239                     "name": name,
1240                     "manifest_text": "" }
1241                 if self.storage_classes is not None:
1242                     c["storage_classes_desired"] = self.storage_classes
1243                 try:
1244                     self.api.collections().create(body=c).execute(num_retries=self.num_retries)
1245                 except Exception as e:
1246                     raise
1247             self.invalidate()
1248         except apiclient_errors.Error as error:
1249             _logger.error(error)
1250             raise llfuse.FUSEError(errno.EEXIST)
1251
1252     @use_counter
1253     @check_update
1254     def rmdir(self, name):
1255         if not self.writable():
1256             raise llfuse.FUSEError(errno.EROFS)
1257
1258         if name not in self:
1259             raise llfuse.FUSEError(errno.ENOENT)
1260         if not isinstance(self[name], CollectionDirectory):
1261             raise llfuse.FUSEError(errno.EPERM)
1262         if len(self[name]) > 0:
1263             raise llfuse.FUSEError(errno.ENOTEMPTY)
1264         with llfuse.lock_released:
1265             self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
1266         self.invalidate()
1267
1268     @use_counter
1269     @check_update
1270     def rename(self, name_old, name_new, src):
1271         if not self.writable():
1272             raise llfuse.FUSEError(errno.EROFS)
1273
1274         if not isinstance(src, ProjectDirectory):
1275             raise llfuse.FUSEError(errno.EPERM)
1276
1277         ent = src[name_old]
1278
1279         if not isinstance(ent, CollectionDirectory):
1280             raise llfuse.FUSEError(errno.EPERM)
1281
1282         if name_new in self:
1283             # POSIX semantics for replacing one directory with another is
1284             # tricky (the target directory must be empty, the operation must be
1285             # atomic which isn't possible with the Arvados API as of this
1286             # writing) so don't support that.
1287             raise llfuse.FUSEError(errno.EPERM)
1288
1289         self.api.collections().update(uuid=ent.uuid(),
1290                                       body={"owner_uuid": self.uuid(),
1291                                             "name": name_new}).execute(num_retries=self.num_retries)
1292
1293         # Acually move the entry from source directory to this directory.
1294         del src._entries[name_old]
1295         self._entries[name_new] = ent
1296         self.inodes.invalidate_entry(src, name_old)
1297
1298     @use_counter
1299     def child_event(self, ev):
1300         properties = ev.get("properties") or {}
1301         old_attrs = properties.get("old_attributes") or {}
1302         new_attrs = properties.get("new_attributes") or {}
1303         old_attrs["uuid"] = ev["object_uuid"]
1304         new_attrs["uuid"] = ev["object_uuid"]
1305         old_name = self.sanitize_filename(self.namefn(old_attrs))
1306         new_name = self.sanitize_filename(self.namefn(new_attrs))
1307
1308         # create events will have a new name, but not an old name
1309         # delete events will have an old name, but not a new name
1310         # update events will have an old and new name, and they may be same or different
1311         # if they are the same, an unrelated field changed and there is nothing to do.
1312
1313         if old_attrs.get("owner_uuid") != self.project_uuid:
1314             # Was moved from somewhere else, so don't try to remove entry.
1315             old_name = None
1316         if ev.get("object_owner_uuid") != self.project_uuid:
1317             # Was moved to somewhere else, so don't try to add entry
1318             new_name = None
1319
1320         if old_attrs.get("is_trashed"):
1321             # Was previously deleted
1322             old_name = None
1323         if new_attrs.get("is_trashed"):
1324             # Has been deleted
1325             new_name = None
1326
1327         if new_name != old_name:
1328             ent = None
1329             if old_name in self._entries:
1330                 ent = self._entries[old_name]
1331                 del self._entries[old_name]
1332                 self.inodes.invalidate_entry(self, old_name)
1333
1334             if new_name:
1335                 if ent is not None:
1336                     self._entries[new_name] = ent
1337                 else:
1338                     self._add_entry(new_attrs, new_name)
1339             elif ent is not None:
1340                 self.inodes.del_entry(ent)
1341
1342
1343 class SharedDirectory(Directory):
1344     """A special directory that represents users or groups who have shared projects with me."""
1345
1346     def __init__(self, parent_inode, inodes, api, num_retries, enable_write, filters,
1347                  exclude, poll=False, poll_time=60, storage_classes=None):
1348         super(SharedDirectory, self).__init__(parent_inode, inodes, enable_write, filters)
1349         self.api = api
1350         self.num_retries = num_retries
1351         self.current_user = api.users().current().execute(num_retries=num_retries)
1352         self._poll = True
1353         self._poll_time = poll_time
1354         self._updating_lock = threading.Lock()
1355         self.storage_classes = storage_classes
1356
1357     @use_counter
1358     def update(self):
1359         try:
1360             with llfuse.lock_released:
1361                 self._updating_lock.acquire()
1362                 if not self.stale():
1363                     return
1364
1365                 contents = {}
1366                 roots = []
1367                 root_owners = set()
1368                 objects = {}
1369
1370                 methods = self.api._rootDesc.get('resources')["groups"]['methods']
1371                 if 'httpMethod' in methods.get('shared', {}):
1372                     page = []
1373                     while True:
1374                         resp = self.api.groups().shared(
1375                             filters=[
1376                                 ['group_class', 'in', ['project','filter']],
1377                                 *page,
1378                                 *self._filters_for('groups', qualified=False),
1379                             ],
1380                             order="uuid",
1381                             limit=10000,
1382                             count="none",
1383                             include="owner_uuid",
1384                         ).execute()
1385                         if not resp["items"]:
1386                             break
1387                         page = [["uuid", ">", resp["items"][len(resp["items"])-1]["uuid"]]]
1388                         for r in resp["items"]:
1389                             objects[r["uuid"]] = r
1390                             roots.append(r["uuid"])
1391                         for r in resp["included"]:
1392                             objects[r["uuid"]] = r
1393                             root_owners.add(r["uuid"])
1394                 else:
1395                     all_projects = list(arvados.util.keyset_list_all(
1396                         self.api.groups().list,
1397                         order_key="uuid",
1398                         num_retries=self.num_retries,
1399                         filters=[
1400                             ['group_class', 'in', ['project','filter']],
1401                             *self._filters_for('groups', qualified=False),
1402                         ],
1403                         select=["uuid", "owner_uuid"],
1404                     ))
1405                     for ob in all_projects:
1406                         objects[ob['uuid']] = ob
1407
1408                     current_uuid = self.current_user['uuid']
1409                     for ob in all_projects:
1410                         if ob['owner_uuid'] != current_uuid and ob['owner_uuid'] not in objects:
1411                             roots.append(ob['uuid'])
1412                             root_owners.add(ob['owner_uuid'])
1413
1414                     lusers = arvados.util.keyset_list_all(
1415                         self.api.users().list,
1416                         order_key="uuid",
1417                         num_retries=self.num_retries,
1418                         filters=[
1419                             ['uuid', 'in', list(root_owners)],
1420                             *self._filters_for('users', qualified=False),
1421                         ],
1422                     )
1423                     lgroups = arvados.util.keyset_list_all(
1424                         self.api.groups().list,
1425                         order_key="uuid",
1426                         num_retries=self.num_retries,
1427                         filters=[
1428                             ['uuid', 'in', list(root_owners)+roots],
1429                             *self._filters_for('groups', qualified=False),
1430                         ],
1431                     )
1432                     for l in lusers:
1433                         objects[l["uuid"]] = l
1434                     for l in lgroups:
1435                         objects[l["uuid"]] = l
1436
1437                 for r in root_owners:
1438                     if r in objects:
1439                         obr = objects[r]
1440                         if obr.get("name"):
1441                             contents[obr["name"]] = obr
1442                         elif "first_name" in obr:
1443                             contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
1444
1445                 for r in roots:
1446                     if r in objects:
1447                         obr = objects[r]
1448                         if obr['owner_uuid'] not in objects:
1449                             contents[obr["name"]] = obr
1450
1451             # end with llfuse.lock_released, re-acquire lock
1452
1453             self.merge(
1454                 contents.items(),
1455                 lambda i: i[0],
1456                 lambda a, i: a.uuid() == i[1]['uuid'],
1457                 lambda i: ProjectDirectory(
1458                     self.inode,
1459                     self.inodes,
1460                     self.api,
1461                     self.num_retries,
1462                     self._enable_write,
1463                     self._filters,
1464                     i[1],
1465                     poll=self._poll,
1466                     poll_time=self._poll_time,
1467                     storage_classes=self.storage_classes,
1468                 ),
1469             )
1470         except Exception:
1471             _logger.exception("arv-mount shared dir error")
1472         finally:
1473             self._updating_lock.release()
1474
1475     def want_event_subscribe(self):
1476         return True