Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def main():
logging.basicConfig()
args = parse_args()
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
argv = args.argv_prefix if args.argv_prefix else ("bash", "-c")
operations = BashFS(argv_prefix=argv, separator=args.separator.encode())
fuse_options = set(pyfuse3.default_options)
fuse_options.add("fsname=bashfs")
fuse_options.discard("default_permissions")
if args.debug_fuse:
fuse_options.add("debug")
pyfuse3.init(operations, args.mountpoint, fuse_options)
try:
trio.run(pyfuse3.main)
except:
pyfuse3.close(unmount=True)
raise
pyfuse3.close()
metadata_upload_task = MetadataUploadTask(backend_pool, param, db,
options.metadata_upload_interval)
block_cache = BlockCache(backend_pool, db, cachepath + '-cache',
options.cachesize * 1024, options.max_cache_entries)
commit_task = CommitTask(block_cache)
operations = fs.Operations(block_cache, db, max_obj_size=param['max_obj_size'],
inode_cache=InodeCache(db, param['inode_gen']),
upload_event=metadata_upload_task.event)
block_cache.fs = operations
metadata_upload_task.fs = operations
async with trio.open_nursery() as nursery:
with ExitStack() as cm:
log.info('Mounting %s at %s...', options.storage_url, options.mountpoint)
try:
pyfuse3.init(operations, options.mountpoint, get_fuse_opts(options))
except RuntimeError as exc:
raise QuietError(str(exc), exitcode=39)
unmount_clean = False
def unmount():
log.info("Unmounting file system...")
pyfuse3.close(unmount=unmount_clean)
cm.callback(unmount)
if options.fg or options.systemd:
faulthandler.enable()
faulthandler.register(signal.SIGUSR1)
else:
if stdout_log_handler:
logging.getLogger().removeHandler(stdout_log_handler)
crit_log_fd = os.open(os.path.join(options.cachedir, 'mount.s3ql_crit.log'),