import llfuse
import logging
import os
+import resource
import signal
import subprocess
import sys
self.add_argument('--file-cache', type=int, help="File data cache size, in bytes (default 256MiB)", default=256*1024*1024)
self.add_argument('--directory-cache', type=int, help="Directory data cache size, in bytes (default 128MiB)", default=128*1024*1024)
+ self.add_argument('--disable-event-listening', action='store_true', help="Don't subscribe to events on the API server", dest="disable_event_listening", default=False)
+
self.add_argument('--read-only', action='store_false', help="Mount will be read only (default)", dest="enable_write", default=False)
self.add_argument('--read-write', action='store_true', help="Mount will be read-write", dest="enable_write", default=False)
self.add_argument('--crunchstat-interval', type=float, help="Write stats to stderr every N seconds (default disabled)", default=0)
+ self.add_argument('--unmount-timeout',
+ type=float, default=2.0,
+ help="Time to wait for graceful shutdown after --exec program exits and filesystem is unmounted")
+
self.add_argument('--exec', type=str, nargs=argparse.REMAINDER,
dest="exec_args", metavar=('command', 'args', '...', '--'),
help="""Mount, run a command, then unmount and exit""")
def __init__(self, args, logger=logging.getLogger('arvados.arv-mount')):
self.logger = logger
self.args = args
+ self.listen_for_events = False
self.args.mountpoint = os.path.realpath(self.args.mountpoint)
if self.args.logfile:
self.args.logfile = os.path.realpath(self.args.logfile)
- # Daemonize as early as possible, so we don't accidentally close
- # file descriptors we're using.
- self.daemon_ctx = None
- if not (self.args.exec_args or self.args.foreground):
- os.chdir(self.args.mountpoint)
- self.daemon_ctx = daemon.DaemonContext(working_directory='.')
- self.daemon_ctx.open()
-
try:
self._setup_logging()
self._setup_api()
def __enter__(self):
llfuse.init(self.operations, self.args.mountpoint, self._fuse_options())
- if self.args.mode != 'by_pdh':
+ if self.listen_for_events and not self.args.disable_event_listening:
self.operations.listen_for_events()
- t = threading.Thread(None, lambda: llfuse.main())
- t.start()
+ self.llfuse_thread = threading.Thread(None, lambda: self._llfuse_main())
+ self.llfuse_thread.daemon = True
+ self.llfuse_thread.start()
self.operations.initlock.wait()
+ return self
def __exit__(self, exc_type, exc_value, traceback):
subprocess.call(["fusermount", "-u", "-z", self.args.mountpoint])
- self.operations.destroy()
+ self.llfuse_thread.join(timeout=self.args.unmount_timeout)
+ if self.llfuse_thread.is_alive():
+ self.logger.warning("Mount.__exit__:"
+ " llfuse thread still alive %fs after umount"
+ " -- abandoning and exiting anyway",
+ self.args.unmount_timeout)
def run(self):
if self.args.exec_args:
- self._run_exec(self.args)
+ self._run_exec()
else:
- self._run_standalone(self.args)
+ self._run_standalone()
def _fuse_options(self):
"""FUSE mount options; see mount.fuse(8)"""
# Configure a log handler based on command-line switches.
if self.args.logfile:
log_handler = logging.FileHandler(self.args.logfile)
- elif self.daemon_ctx:
- log_handler = logging.NullHandler()
else:
log_handler = None
self.api = arvados.safeapi.ThreadSafeApiCache(
apiconfig=arvados.config.settings(),
keep_params={
- "block_cache": arvados.keep.KeepBlockCache(self.args.file_cache)
+ 'block_cache': arvados.keep.KeepBlockCache(self.args.file_cache),
+ 'num_retries': self.args.retries,
})
+ # Do a sanity check that we have a working arvados host + token.
+ self.api.users().current().execute()
def _setup_mount(self):
self.operations = Operations(
if self.args.collection is not None:
# Set up the request handler with the collection at the root
+ # First check that the collection is readable
+ self.api.collections().get(uuid=self.args.collection).execute()
self.args.mode = 'collection'
dir_class = CollectionDirectory
dir_args.append(self.args.collection)
mount_readme = True
if dir_class is not None:
- self.operations.inodes.add_entry(dir_class(*dir_args))
+ ent = dir_class(*dir_args)
+ self.operations.inodes.add_entry(ent)
+ self.listen_for_events = ent.want_event_subscribe()
return
e = self.operations.inodes.add_entry(Directory(
if name in ['', '.', '..'] or '/' in name:
sys.exit("Mount point '{}' is not supported.".format(name))
tld._entries[name] = self.operations.inodes.add_entry(ent)
+ self.listen_for_events = (self.listen_for_events or ent.want_event_subscribe())
def _readme_text(self, api_host, user_email):
return '''
'''.format(api_host, user_email)
def _run_exec(self):
- # Initialize the fuse connection
- llfuse.init(self.operations, self.args.mountpoint, self._fuse_options())
-
- # Subscribe to change events from API server
- if self.args.mode != 'by_pdh':
- self.operations.listen_for_events()
-
- t = threading.Thread(None, lambda: llfuse.main())
- t.start()
-
- # wait until the driver is finished initializing
- self.operations.initlock.wait()
-
rc = 255
- try:
- sp = subprocess.Popen(self.args.exec_args, shell=False)
-
- # forward signals to the process.
- signal.signal(signal.SIGINT, lambda signum, frame: sp.send_signal(signum))
- signal.signal(signal.SIGTERM, lambda signum, frame: sp.send_signal(signum))
- signal.signal(signal.SIGQUIT, lambda signum, frame: sp.send_signal(signum))
-
- # wait for process to complete.
- rc = sp.wait()
-
- # restore default signal handlers.
- signal.signal(signal.SIGINT, signal.SIG_DFL)
- signal.signal(signal.SIGTERM, signal.SIG_DFL)
- signal.signal(signal.SIGQUIT, signal.SIG_DFL)
- except Exception as e:
- self.logger.exception(
- 'arv-mount: exception during exec %s', self.args.exec_args)
+ with self:
try:
- rc = e.errno
- except AttributeError:
- pass
- finally:
- subprocess.call(["fusermount", "-u", "-z", self.args.mountpoint])
- self.operations.destroy()
+ sp = subprocess.Popen(self.args.exec_args, shell=False)
+
+ # forward signals to the process.
+ signal.signal(signal.SIGINT, lambda signum, frame: sp.send_signal(signum))
+ signal.signal(signal.SIGTERM, lambda signum, frame: sp.send_signal(signum))
+ signal.signal(signal.SIGQUIT, lambda signum, frame: sp.send_signal(signum))
+
+ # wait for process to complete.
+ rc = sp.wait()
+
+ # restore default signal handlers.
+ signal.signal(signal.SIGINT, signal.SIG_DFL)
+ signal.signal(signal.SIGTERM, signal.SIG_DFL)
+ signal.signal(signal.SIGQUIT, signal.SIG_DFL)
+ except Exception as e:
+ self.logger.exception(
+ 'arv-mount: exception during exec %s', self.args.exec_args)
+ try:
+ rc = e.errno
+ except AttributeError:
+ pass
exit(rc)
def _run_standalone(self):
try:
llfuse.init(self.operations, self.args.mountpoint, self._fuse_options())
+ if not self.args.foreground:
+ self.daemon_ctx = daemon.DaemonContext(
+ working_directory=os.path.dirname(self.args.mountpoint),
+ files_preserve=range(
+ 3, resource.getrlimit(resource.RLIMIT_NOFILE)[1]))
+ self.daemon_ctx.open()
+
# Subscribe to change events from API server
- self.operations.listen_for_events()
+ if self.listen_for_events and not self.args.disable_event_listening:
+ self.operations.listen_for_events()
- llfuse.main()
+ self._llfuse_main()
except Exception as e:
self.logger.exception('arv-mount: exception during mount: %s', e)
exit(getattr(e, 'errno', 1))
- finally:
- self.operations.destroy()
exit(0)
+
+ def _llfuse_main(self):
+ try:
+ llfuse.main()
+ except:
+ llfuse.close(unmount=False)
+ raise
+ llfuse.close()
projects / arvados.git / blobdiff