X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0662b235357dd40b5d27efd06b60044ddcec06f6..cd175e917038f68d80d29cc6fee24fc3b389069e:/services/fuse/arvados_fuse/command.py diff --git a/services/fuse/arvados_fuse/command.py b/services/fuse/arvados_fuse/command.py index 93bcd20a2d..3f89732bea 100644 --- a/services/fuse/arvados_fuse/command.py +++ b/services/fuse/arvados_fuse/command.py @@ -4,6 +4,7 @@ import daemon import llfuse import logging import os +import resource import signal import subprocess import sys @@ -76,11 +77,17 @@ class ArgumentParser(argparse.ArgumentParser): 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""") @@ -90,19 +97,12 @@ class Mount(object): 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() @@ -113,15 +113,22 @@ class Mount(object): 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: @@ -141,8 +148,6 @@ class Mount(object): # 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 @@ -160,8 +165,11 @@ class Mount(object): 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( @@ -189,6 +197,8 @@ class Mount(object): 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) @@ -234,7 +244,9 @@ class Mount(object): 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( @@ -264,6 +276,7 @@ class Mount(object): 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 ''' @@ -281,58 +294,57 @@ From here, the following directories are available: '''.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()