From a21a3c8eb0bbb7116efd732f526e0ab594744b26 Mon Sep 17 00:00:00 2001 From: Michael MacDonald Date: Sat, 1 Aug 2026 21:37:44 -0400 Subject: [PATCH 1/2] DAOS-19386 test: Add stall watchdog to NLT FI tests Don't allow a stuck child process to tank the entire NLT run. When a stall is detected, attempt to capture stack traces and fail early. The /proc dump shows a real stall directly. From the hang that motivated this patch: the stalled child holds the parent-dir lock while waiting on its LOOKUP reply, and every dfuse worker is blocked on that same lock trying to write a dentry invalidation: TID 34163: comm=daos wchan=request_wait_answer state=S [<0>] request_wait_answer+0xfa/0x210 [fuse] [<0>] fuse_simple_request+0x1b8/0x330 [fuse] [<0>] fuse_lookup_name+0xa4/0x1c0 [fuse] [<0>] fuse_lookup+0x66/0x190 [fuse] [<0>] __lookup_hash+0x70/0xa0 [<0>] __filename_create+0x87/0x150 [<0>] do_mkdirat+0x4c/0x160 [<0>] __x64_sys_mkdir+0x47/0x70 TID 29212: comm=dfuse worker wchan=fuse_reverse_inval_entry state=D [<0>] fuse_reverse_inval_entry+0x40/0x210 [fuse] [<0>] fuse_notify+0x287/0x500 [fuse] [<0>] fuse_dev_do_write+0x305/0x4e0 [fuse] [<0>] fuse_dev_write+0x50/0x80 [fuse] TID 29213, 29214: identical (the whole worker pool) Signed-off-by: Michael MacDonald --- ci/unit/test_nlt_node.sh | 2 + utils/cq/words.dict | 1 + utils/node_local_test.py | 402 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 390 insertions(+), 15 deletions(-) diff --git a/ci/unit/test_nlt_node.sh b/ci/unit/test_nlt_node.sh index 77099b5bd2c..2b7725b67a8 100755 --- a/ci/unit/test_nlt_node.sh +++ b/ci/unit/test_nlt_node.sh @@ -51,7 +51,9 @@ mkdir -p nlt_logs sudo mount -t tmpfs tmpfs nlt_logs sudo chown jenkins:jenkins nlt_logs +# Unbuffered so the console shows exactly where a hang occurs. TMPDIR="$(pwd)/nlt_logs" \ + PYTHONUNBUFFERED=1 \ HTTPS_PROXY="${DAOS_HTTPS_PROXY:-}" \ NO_PROXY="${DAOS_NO_PROXY:-}" \ exec ./utils/node_local_test.py "$@" diff --git a/utils/cq/words.dict b/utils/cq/words.dict index 801a166ed92..0211097656f 100644 --- a/utils/cq/words.dict +++ b/utils/cq/words.dict @@ -184,6 +184,7 @@ epilog errored ethernet fallocate +faulthandler fchmod fcntl filename diff --git a/utils/node_local_test.py b/utils/node_local_test.py index 7d2e3191ab2..a8813982122 100755 --- a/utils/node_local_test.py +++ b/utils/node_local_test.py @@ -20,6 +20,7 @@ import argparse import copy import errno +import faulthandler import functools import importlib import json @@ -47,6 +48,19 @@ import xattr import yaml +# How long a process may show no sign of progress before it is declared stalled. +STALL_SECS = int(os.environ.get('NLT_STALL_SECS', '300')) +# Factor to be used for calculating timeouts when running with valgrind. +VALGRIND_SLOWDOWN = 6 + +# Bounds on the stall diagnostics themselves. +DIAG_TIMEOUT = 60 +STACK_READ_TIMEOUT = 15 +DUMP_DEADLINE_SECS = 120 + +# How long a killed process gets to shut down; longer than this is assumed to be hung. +KILL_GRACE = 30 + class NLTestFail(Exception): """Used to indicate test failure""" @@ -125,10 +139,30 @@ def __init__(self, json_file, args): os.makedirs(self.tmp_dir) self._compress_procs = [] + self._cleaned = False def __del__(self): + self.cleanup() + + def cleanup(self): + """Report any leaked dfuse mount, flush compression and remove the working directory""" + if self._cleaned: + return + self._cleaned = True + + leaked = {conn: mnt for conn, mnt in _dfuse_connection_ids().items() + if mnt.startswith(self.dfuse_parent_dir)} + if leaked: + print(f'Leaked dfuse mounts at exit: {sorted(leaked.values())}', flush=True) + for line in _abort_fuse_connections()[1]: + print(line, flush=True) + self.flush_bz2() - os.rmdir(self.dfuse_parent_dir) + + try: + os.rmdir(self.dfuse_parent_dir) + except OSError as err: + print(f'Could not remove {self.dfuse_parent_dir}: {err}', flush=True) def set_wf(self, wf): """Set the WarningsFactory object""" @@ -1519,7 +1553,8 @@ def stop(self, ignore_einval=False): print('Stopping fuse') if self.container: - self.run_query(use_json=True) + # This queries the mount that may itself be wedged. + self.run_query(use_json=True, timeout=120) ret = umount(self.dir) if ret: umount(self.dir, background=True) @@ -1606,10 +1641,10 @@ def il_cmd(self, cmd, check_read=True, check_write=True, check_fstat=True): assert ret.returncode == 0, ret return ret - def run_query(self, use_json=False, quiet=False): + def run_query(self, use_json=False, quiet=False, timeout=None): """Run filesystem query""" rc = run_daos_cmd(self.conf, ['filesystem', 'query', self.dir], - use_json=use_json, log_check=quiet, valgrind=quiet) + use_json=use_json, log_check=quiet, valgrind=quiet, timeout=timeout) print(rc) return rc @@ -1730,7 +1765,8 @@ def run_daos_cmd(conf, log_check=True, ignore_busy=False, use_json=False, - cwd=None): + cwd=None, + timeout=None): """Run a DAOS command Run a command, returning what subprocess.run() would. @@ -1777,8 +1813,24 @@ def run_daos_cmd(conf, cmd_env['DAOS_AGENT_DRPC_DIR'] = conf.agent_dir - rc = subprocess.run(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - env=cmd_env, check=False, cwd=cwd) + # Avoid getting stuck on child processes that are blocked in the kernel. + # pylint: disable-next=consider-using-with + proc = subprocess.Popen(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + env=cmd_env, cwd=cwd) + timed_out = False + try: + stdout, stderr = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + print(f'Timeout after {timeout}s running {" ".join(cmd)}', flush=True) + proc.kill() + try: + stdout, stderr = proc.communicate(timeout=KILL_GRACE) + except subprocess.TimeoutExpired: + print(f'Command did not exit after kill: {" ".join(cmd)}', flush=True) + stdout, stderr = b'', b'' + returncode = -signal.SIGKILL if proc.returncode is None else proc.returncode + rc = subprocess.CompletedProcess(exec_cmd, returncode, stdout, stderr) if rc.stderr != b'': print('Stderr from command') @@ -1807,7 +1859,14 @@ def run_daos_cmd(conf, conf.valgrind_errors = True rc.returncode = 0 if use_json: - rc.json = json.loads(rc.stdout.decode('utf-8')) + try: + rc.json = json.loads(rc.stdout.decode('utf-8')) + except json.JSONDecodeError: + if timed_out: + print(f'No JSON output from timed-out command: {" ".join(cmd)}') + rc.json = None + else: + raise dcr.rc = rc return dcr @@ -5616,6 +5675,216 @@ def test_pydaos_kv_obj_class(server, conf): # +def _run_diag(cmd, timeout=DIAG_TIMEOUT): + """Run a diagnostic command returning (exit code, output), never raising or blocking + + A command that outlives timeout is killed and reports a nonzero code. + """ + try: + # pylint: disable-next=consider-using-with + proc = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, start_new_session=True) + except OSError as err: + return 127, f'<{cmd[0]} failed: {err}>' + + def output(wait_secs): + return proc.communicate(timeout=wait_secs)[0].decode('utf-8', errors='replace').rstrip() + + try: + text = output(timeout) + except subprocess.TimeoutExpired: + # Overdue: kill the process group and salvage whatever it wrote. + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + pass + try: + return -1, f'{output(KILL_GRACE)}\n<{cmd[0]} exceeded {timeout}s>' + except subprocess.TimeoutExpired: + proc.stdout.close() + return -1, f'<{cmd[0]} exceeded {timeout}s and could not be reaped>' + return proc.returncode, text + + +def _read_file(path): + """Return the stripped content of a small /proc or /sys file, or None""" + try: + with open(path, encoding='utf-8', errors='replace') as pfile: + return pfile.read().strip() + except OSError: + return None + + +def _proc_state(pid): + """Return blocked system-call state for a pid from /proc""" + out = [] + for name in ('wchan', 'syscall'): + out.append(f'{name}={_read_file(f"/proc/{pid}/{name}") or ""}') + status = _read_file(f'/proc/{pid}/status') or '' + for line in status.splitlines(): + if line.startswith(('State:', 'Threads:')): + out.append(line.strip()) + return ' '.join(out) + + +def _proc_threads(pid, deadline=None): + """Report every thread of a process from /proc; debuggers cannot attach in D state""" + lines = [] + stack_err = None + try: + tids = sorted(os.listdir(f'/proc/{pid}/task'), key=int) + except OSError as err: + return f'' + for tid in tids: + if deadline is not None and time.monotonic() > deadline: + lines.append(f' ') + break + base = f'/proc/{pid}/task/{tid}' + fields = [] + for name in ('comm', 'wchan'): + fields.append(f'{name}={_read_file(f"{base}/{name}") or "?"}') + # The state field follows the parenthesized command name, which can itself + # contain ') ', so anchor on the last occurrence per proc(5). + state = (_read_file(f'{base}/stat') or '').rpartition(') ')[2].split() + if state: + fields.append(f'state={state[0]}') + lines.append(f' TID {tid}: ' + ' '.join(fields)) + rc, kstack = _run_diag(['sudo', 'cat', f'{base}/stack'], timeout=STACK_READ_TIMEOUT) + if rc == 0 and kstack: + for kline in kstack.splitlines()[:12]: + lines.append(f' {kline.strip()}') + elif rc != 0 and stack_err is None: + stack_err = (kstack or '').splitlines()[0].strip() + lines.append(f' ') + return '\n'.join(lines) + + +def dump_stalled(active, log_dir=None): + """Dump diagnostics for wedged child processes, to stdout and to a file""" + dump_file = None + if log_dir: + try: + # The dnt*.log name is what CI's log collection keeps; append across stalls. + # pylint: disable-next=consider-using-with + dump_file = open(join(log_dir, 'dnt_stall_dump.log'), 'a', encoding='utf-8') + except OSError: + dump_file = None + + def emit(text): + # File first; a stalled stdout consumer must not cost us the dump. + if dump_file: + try: + dump_file.write(f'{text}\n') + dump_file.flush() + except OSError: + pass + print(text, flush=True) + + try: + deadline = time.monotonic() + DUMP_DEADLINE_SECS + emit(f'\n===== NLT STALL DETECTED {time.strftime("%Y-%m-%d %H:%M:%S")} =====') + # /proc cannot block, so gather it before anything that can. + for child in active: + pid = child.pid() + emit(f'--- stalled child: loc={child.loc} pid={pid} elapsed={child.elapsed():.0f}s') + emit(_proc_threads(pid, deadline=deadline)) + daemons = _run_diag(['pgrep', '-a', 'daos_engine|daos_agent|dfuse'])[1] + emit(f'--- daemons:\n{daemons}') + daemon_pids = [int(line.split()[0]) for line in daemons.splitlines() + if line and line.split()[0].isdigit()] + for pid in daemon_pids: + if time.monotonic() > deadline: + emit('') + break + emit(f'--- daemon pid={pid}:') + emit(_proc_threads(pid, deadline=deadline)) + emit('--- process tree:') + emit(_run_diag(['ps', 'auxwwf'])[1]) + emit('===== NLT STALL DUMP COMPLETE =====') + except Exception as err: # pylint: disable=broad-except + emit(f'===== NLT STALL DUMP ABORTED: {err!r} =====') + traceback.print_exc(file=sys.stdout) + sys.stdout.flush() + finally: + if dump_file: + dump_file.close() + + +def _dfuse_connection_ids(): + """Return the FUSE connection ids of dfuse mounts, mapped to their mount points""" + ids = {} + # Safer than stat, which could block on a wedged mount + for line in (_read_file('/proc/self/mountinfo') or '').splitlines(): + fields = line.split() + if '-' in fields and 'fuse.daos' in fields[fields.index('-'):]: + ids[fields[2].split(':')[1]] = fields[4] + return ids + + +def _abort_fuse_connections(): + """Abort every backed-up dfuse connection, failing its requests so blocked processes can die""" + done = [] + aborted = 0 + dfuse_conns = _dfuse_connection_ids() + try: + conns = sorted(os.listdir('/sys/fs/fuse/connections')) + except OSError as err: + return 0, [f''] + for conn in conns: + if conn not in dfuse_conns: + continue + waiting = _read_file(f'/sys/fs/fuse/connections/{conn}/waiting') + if waiting is None: + done.append(f'') + continue + if waiting in ('', '0'): + continue + path = f'/sys/fs/fuse/connections/{conn}/abort' + rc, res = _run_diag(['sudo', 'sh', '-c', f'echo 1 > {path}']) + if rc == 0: + aborted += 1 + done.append(f'aborted dfuse connection {conn} (waiting={waiting}): {res or "ok"}') + else: + done.append(f'') + return aborted, (done or ['']) + + +def handle_stalled(active, log_dir=None): + """Dump diagnostics then clear the wedged children; returns whether a mount was aborted""" + dump_stalled(active, log_dir=log_dir) + aborted_mount = False + for child in active: + child.hang_kill() + + def _find_zombies(wait_sec): + deadline = time.monotonic() + wait_sec + alive = list(active) + while alive and time.monotonic() < deadline: + alive = [c for c in alive if not c.is_dead()] + if alive: + time.sleep(1) + return alive + + zombies = _find_zombies(KILL_GRACE) + if zombies: + # A child that outlives the kill grace is likely blocked in the kernel. + # We can try to free it up by failing its FUSE requests so that it + # can exit. + print(f'{len(zombies)} child(ren) survived the kill; ' + f'aborting backed-up FUSE connections to release them', flush=True) + aborted, lines = _abort_fuse_connections() + for line in lines: + print(line, flush=True) + if aborted: + aborted_mount = True + zombies = _find_zombies(KILL_GRACE) + + for child in zombies: + print(f'WARNING: pid {child.pid()} (loc {child.loc}) survived SIGKILL: ' + f'{_proc_state(child.pid())}', flush=True) + return aborted_mount + + class AllocFailTestRun(): """Class to run a fault injection command with a single fault""" @@ -5631,6 +5900,8 @@ def __init__(self, aft, cmd, env, loc, cwd): self.dir_handle = None self.stdout = None self.returncode = None + self.was_killed = False + self._start_time = None # Set this to disable memory leak checking if the command outputs a DER_BUSY message. This # is to allow tests to leak memory if there are errors during shutdown. @@ -5715,6 +5986,26 @@ def start(self): stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self._start_time = time.monotonic() + + def pid(self): + """Return the pid of the command""" + return self._sp.pid + + def elapsed(self): + """Return seconds since the command started""" + return time.monotonic() - self._start_time + + def hang_kill(self): + """Kill a wedged command; result checks are skipped on reap""" + if self._sp.poll() is not None: + return + self.was_killed = True + self._sp.kill() + + def is_dead(self): + """Return whether the command has exited, without blocking""" + return self._sp.poll() is not None def has_finished(self): """Check if the command has completed""" @@ -5724,15 +6015,45 @@ def has_finished(self): rc = self._sp.poll() if rc is None: return False - self._post(rc) + self._reap(rc) return True - def wait(self): + def wait(self, timeout=None): """Wait for the command to complete""" if self.returncode is not None: return - self._post(self._sp.wait()) + if timeout is not None: + try: + self._reap(self._sp.wait(timeout=timeout)) + except subprocess.TimeoutExpired: + handle_stalled([self], log_dir=self._aft.log_dir) + try: + self._reap(self._sp.wait(timeout=KILL_GRACE)) + except subprocess.TimeoutExpired: + # Blocked in the kernel; record the kill rather than joining the hang. + self._post_killed(-signal.SIGKILL) + return + + self._reap(self._sp.wait()) + + def _reap(self, rc): + """Process a completed command, bypassing result checks for watchdog kills""" + if self.was_killed: + self._post_killed(rc) + else: + self._post(rc) + + def _post_killed(self, rc): + """Reap after a watchdog kill; the positive returncode stops a valgrind re-run""" + print() + self.returncode = 128 - rc if rc < 0 else rc + try: + self.stdout, self._stderr = self._sp.communicate(timeout=KILL_GRACE) + except subprocess.TimeoutExpired: + self.stdout = b'' + self._stderr = b'' + self.fault_injected = True def _post(self, rc): """Helper function, called once after command is complete. @@ -5929,7 +6250,9 @@ def launch(self): def _prep(self): rc = self._run_cmd(None) - rc.wait() + rc.wait(timeout=STALL_SECS) + if rc.was_killed: + raise NLTestFail('prep run (no faults enabled) stalled and was killed') self.expected_stdout = rc.stdout assert not rc.fault_injected @@ -5965,6 +6288,10 @@ def _prep(self): max_load_avg = 100 + last_progress = time.monotonic() + stall_rounds = 0 + stalled_out = False + # Now run all iterations in parallel up to max_child. Iterations will be launched # in order but may not finish in order, rather they are processed in the order they # finish. After each repetition completes then check for re-launch new processes @@ -6005,6 +6332,7 @@ def _prep(self): if not ret.has_finished(): continue active.remove(ret) + last_progress = time.monotonic() print() print(ret) if ret.returncode < 0: @@ -6016,15 +6344,43 @@ def _prep(self): finished = True break + if active and time.monotonic() - last_progress > STALL_SECS: + stall_rounds += 1 + fatal_errors = True + if handle_stalled(active, log_dir=self.log_dir): + print('Mount was aborted to clear the stall; ending this sweep') + stalled_out = True + elif stall_rounds >= 2: + # A second stall means the mount did not really recover; new + # children will only stall again, forever. + print('Sweep stalled again after a cleared stall; giving up on it') + stalled_out = True + if stalled_out: + finished = True + # Reap the children that died; waiting for one that is stuck in + # the kernel would hang the run. + for child in active: + if not child.has_finished(): + print(f'Abandoning stuck pid {child.pid()} (loc {child.loc})', + flush=True) + active.clear() + last_progress = time.monotonic() + print(f'Completed, fid {fid}') print(f'Max in flight {max_count}/{max_child}') if to_rerun: print(f'Number of indexes to re-run {len(to_rerun)}') + if stalled_out: + print('Skipping valgrind re-runs; the mount did not survive the sweep') + to_rerun = [] for fid in to_rerun: rerun = self._run_cmd(fid, valgrind=True) print(rerun) - rerun.wait() + rerun.wait(timeout=STALL_SECS * VALGRIND_SLOWDOWN) + if rerun.was_killed and self.conf.args.failfast: + print(f'--failfast set; skipping remaining re-runs after stall at {fid}') + break return fatal_errors @@ -6805,6 +7161,9 @@ def run(wf, args): print(fs) if fs.returncode == 0: run_fi = True + elif fi_test or fi_test_dfuse: + raise NLTestFail('Unable to detect fault injection feature ' + '- cannot run requested FI tests') else: print("Unable to detect fault injection feature - skipping FI testing") @@ -6880,11 +7239,19 @@ def run(wf, args): wf_server.close() close_log_test(conf) + conf.cleanup() print(f'Total time in log analysis: {conf.log_timer.total:.2f} seconds') print(f'Total time in log compression: {conf.compress_timer.total:.2f} seconds') return fatal_errors +def _exit_now(code): + """Terminate without interpreter shutdown, whose cleanup can block on a wedged mount""" + sys.stdout.flush() + sys.stderr.flush() + os._exit(code) # pylint: disable=protected-access + + def _positive_int(value): """argparse type that rejects values below 1.""" ivalue = int(value) @@ -6902,6 +7269,10 @@ def main(): Test names can either be 'pure' or include suffices that encode a particular caching regimen (e.g., read_caching_off). """ + # SIGUSR1 dumps every thread's Python stack. sys.__stderr__ because NLT's stderr + # wrapper lacks the fileno() that faulthandler requires. + faulthandler.register(signal.SIGUSR1, file=sys.__stderr__, all_threads=True) + parser = argparse.ArgumentParser(description='Run DAOS client on local node') parser.add_argument('--server-debug', default=None) parser.add_argument('--dfuse-debug', default=None) @@ -6917,7 +7288,7 @@ def main(): parser.add_argument('--repeat', type=_positive_int, default=1, help='Repeat the test execution N times (soak/stability testing)') parser.add_argument('--failfast', action='store_true', - help='With --repeat, stop after the first failing iteration') + help='Stop after the first failing --repeat iteration or stalled re-run') parser.add_argument('--system-ram-reserved', type=int, default=None, help='GiB reserved RAM') parser.add_argument('--dfuse-dir', default='/tmp', help='parent directory for all dfuse mounts') parser.add_argument('--perf-check', action='store_true') @@ -6988,7 +7359,8 @@ def main(): if fatal_errors.errors: print("Significant errors encountered") - sys.exit(1) + _exit_now(1) + _exit_now(0) if __name__ == '__main__': From e3242d1f685bac00febafa03bac5e841600c87de Mon Sep 17 00:00:00 2001 From: Michael MacDonald Date: Thu, 6 Aug 2026 12:55:34 -0400 Subject: [PATCH 2/2] DAOS-19386 dfuse: move notify calls to an asynchronous queue Most fuse_lowlevel_notify_* calls were already effectively fire-and-forget, but blocked the calling worker while the kernel processed them. In certain scenarios, e.g. mkdir racing with multiple setxattrs in the same directory, this pattern could result in all threads being blocked, and then the mount would hang. Making the notify calls properly asynchronous means that workers will not block on waiting for a response. A dedicated thread now handles queued notifications, with some logic to coalesce notifications. duns_create_path, which does need the invalidation's effect before returning, now polls until the path is confirmed bound to the new container before returning success. Signed-off-by: Michael MacDonald --- src/client/dfs/duns.c | 89 +++++- src/client/dfuse/SConscript | 3 + src/client/dfuse/dfuse.h | 21 ++ src/client/dfuse/dfuse_main.c | 11 +- src/client/dfuse/inval.c | 392 +++++++++++++++++++++++++- src/client/dfuse/ops/ioctl.c | 12 +- src/client/dfuse/ops/lookup.c | 7 +- src/client/dfuse/ops/open.c | 6 +- src/client/dfuse/ops/opendir.c | 9 +- src/client/dfuse/ops/rename.c | 8 +- src/client/dfuse/ops/setxattr.c | 18 +- src/client/dfuse/ops/unlink.c | 11 +- src/client/dfuse/tests/SConscript | 17 ++ src/client/dfuse/tests/notify_tests.c | 341 ++++++++++++++++++++++ utils/node_local_test.py | 164 ++++++++++- utils/utest.yaml | 4 + 16 files changed, 1030 insertions(+), 83 deletions(-) create mode 100644 src/client/dfuse/tests/SConscript create mode 100644 src/client/dfuse/tests/notify_tests.c diff --git a/src/client/dfs/duns.c b/src/client/dfs/duns.c index 8b8ae332da2..a710a6ab5f4 100644 --- a/src/client/dfs/duns.c +++ b/src/client/dfs/duns.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2019-2024 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -899,6 +900,69 @@ duns_link_lustre_path(const char *pool, const char *cont, daos_cont_layout_t typ } #endif +/* Bounds queued-notification delivery latency, normally microseconds; does not wait out + * dentry expiry on mounts with long timeouts. + */ +#define DUNS_RESOLVE_TIMEOUT_MS 10000 +#define DUNS_RESOLVE_BACKOFF_MS_MAX 50 + +/* Poll until dfuse binds the entry point to the new container, or we + * hit the timeout. + */ +static int +duns_wait_for_resolution(const char *path, uuid_t cont_uuid) +{ + struct timespec start, now; + int backoff_ms = 1; + int rc; + + clock_gettime(CLOCK_MONOTONIC, &start); + + while (1) { + struct dfuse_il_reply il_reply = {}; + int fd; + int64_t elapsed_ms; + + fd = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW); + if (fd == -1) { + rc = errno; + /* ENOLINK: dfuse attempted entry-point resolution and the container + * connect failed - expected during rebinding, and terminal here since a + * failed connect will not heal within the poll window. + */ + if (rc != ENOLINK) + D_ERROR("Failed to open %s to verify resolution: %d (%s)\n", path, + rc, strerror(rc)); + return rc; + } + + rc = ioctl(fd, DFUSE_IOCTL_IL, &il_reply); + close(fd); + if (rc == -1) { + rc = errno; + D_ERROR("Dfuse IL ioctl failed on %s: %d (%s)\n", path, rc, strerror(rc)); + return rc; + } + + if (uuid_compare(il_reply.fir_cont, cont_uuid) == 0) + return 0; + + clock_gettime(CLOCK_MONOTONIC, &now); + elapsed_ms = + (now.tv_sec - start.tv_sec) * 1000 + (now.tv_nsec - start.tv_nsec) / 1000000; + if (elapsed_ms >= DUNS_RESOLVE_TIMEOUT_MS) { + D_ERROR("Entry point %s still bound to container " DF_UUIDF " after %dms, " + "expected " DF_UUIDF "\n", + path, DP_UUID(il_reply.fir_cont), DUNS_RESOLVE_TIMEOUT_MS, + DP_UUID(cont_uuid)); + return ENOLINK; + } + + usleep(backoff_ms * 1000); + backoff_ms = min(backoff_ms * 2, DUNS_RESOLVE_BACKOFF_MS_MAX); + } +} + int duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp) { @@ -1099,20 +1163,12 @@ duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp) goto err_cont; } if (backend_dfuse) { - struct stat finfo; - /* - * This next stat will cause dfuse to lookup the entry point and perform a - * container connect, therefore this data will be read from root of the new - * container, not the directory. - * - * TODO: This could call getxattr to verify success. + /* Confirm dfuse has looked up the entry point and connected to the new + * container. */ - rc = stat(path, &finfo); - if (rc) { - rc = errno; - D_ERROR("Failed to access new container: %d (%s)\n", rc, strerror(rc)); - goto err_link; - } + rc = duns_wait_for_resolution(path, attrp->da_cuuid); + if (rc) + goto err_verify; } return rc; @@ -1126,6 +1182,13 @@ duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp) else if (attrp->da_type != DAOS_PROP_CO_LAYOUT_UNKNOWN) unlink(path); return rc; +err_verify: + /* clean up the path before destroying the linked container */ + rmdir(path); + rc2 = daos_cont_destroy(poh, attrp->da_cont, 1, NULL); + if (rc2) + D_ERROR("Failed to cleanup created container %s (%d)\n", attrp->da_cont, rc2); + return rc; } int diff --git a/src/client/dfuse/SConscript b/src/client/dfuse/SConscript index c699dd2acbb..94eea938c69 100644 --- a/src/client/dfuse/SConscript +++ b/src/client/dfuse/SConscript @@ -240,6 +240,9 @@ def scons(): cenv.Install(os.path.join("$PREFIX", 'bin'), dfuse_bin) + if prereqs.test_requested(): + SConscript('tests/SConscript', exports={'denv': cenv}) + if __name__ == "SCons.Script": scons() diff --git a/src/client/dfuse/dfuse.h b/src/client/dfuse/dfuse.h index ab14572fa1f..5fdb1b77480 100644 --- a/src/client/dfuse/dfuse.h +++ b/src/client/dfuse/dfuse.h @@ -13,6 +13,11 @@ #include #include +/* Only inval.c may call these directly. */ +#ifndef DFUSE_NOTIFY_RAW_OK +#pragma GCC poison fuse_lowlevel_notify_inval_entry fuse_lowlevel_notify_expire_entry fuse_lowlevel_notify_delete fuse_lowlevel_notify_inval_inode +#endif + #include #include #include @@ -76,6 +81,11 @@ struct dfuse_info { ATOMIC uint64_t di_fh_count; ATOMIC uint64_t di_pool_count; ATOMIC uint64_t di_container_count; + + ATOMIC uint64_t di_notify_enqueued; + ATOMIC uint64_t di_notify_coalesced; + ATOMIC uint64_t di_notify_delivered; + ATOMIC uint64_t di_notify_dropped; }; struct dfuse_eq { @@ -1201,6 +1211,17 @@ ival_thread_stop(); void ival_fini(); +/* Fire-and-forget reverse notifications, delivered by the notify thread in inval.c */ +void +dfuse_notify_inval_entry(struct dfuse_info *dfuse_info, fuse_ino_t parent, const char *name); + +void +dfuse_notify_delete(struct dfuse_info *dfuse_info, fuse_ino_t parent, fuse_ino_t ino, + const char *name); + +void +dfuse_notify_inval_inode(struct dfuse_info *dfuse_info, fuse_ino_t ino); + /* Data caching functions */ /* Mark the data cache as up-to-date from now */ diff --git a/src/client/dfuse/dfuse_main.c b/src/client/dfuse/dfuse_main.c index 872ae186dbe..bfa56b08567 100644 --- a/src/client/dfuse/dfuse_main.c +++ b/src/client/dfuse/dfuse_main.c @@ -1,6 +1,6 @@ /** * (C) Copyright 2016-2024 Intel Corporation. - * (C) Copyright 2025 Hewlett Packard Enterprise Development LP + * (C) Copyright 2025-2026 Hewlett Packard Enterprise Development LP * (C) Copyright 2025 Google LLC * * SPDX-License-Identifier: BSD-2-Clause-Patent @@ -977,6 +977,15 @@ main(int argc, char **argv) } } + if (dfuse_info) + DFUSE_TRA_INFO(dfuse_info, + "Notify: enqueued=" DF_U64 " coalesced=" DF_U64 " delivered=" DF_U64 + " dropped=" DF_U64, + atomic_load_relaxed(&dfuse_info->di_notify_enqueued), + atomic_load_relaxed(&dfuse_info->di_notify_coalesced), + atomic_load_relaxed(&dfuse_info->di_notify_delivered), + atomic_load_relaxed(&dfuse_info->di_notify_dropped)); + DFUSE_TRA_DOWN(dfuse_info); daos_fini(); out_debug: diff --git a/src/client/dfuse/inval.c b/src/client/dfuse/inval.c index 3ddcc052d86..8f7c5c19e28 100644 --- a/src/client/dfuse/inval.c +++ b/src/client/dfuse/inval.c @@ -1,6 +1,7 @@ /** * (C) Copyright 2016-2024 Intel Corporation. * (C) Copyright 2025 Google LLC + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -10,6 +11,8 @@ #include #include "dfuse_common.h" +/* Allow this file to call the raw fuse_lowlevel_notify_* symbols. */ +#define DFUSE_NOTIFY_RAW_OK #include "dfuse.h" /* Evict inodes based on timeout. @@ -92,7 +95,7 @@ struct dfuse_time_entry { struct dfuse_ival { d_list_t time_entry_list; struct fuse_session *session; - bool session_dead; + ATOMIC bool session_dead; }; /* The core data from struct dfuse_inode_entry. No additional inode references are held on inodes @@ -180,7 +183,7 @@ ival_loop(int *sleep_time) DFUSE_TRA_DEBUG(&ival_data, "Unlocking, allowing to sleep for %d seconds", *sleep_time); D_MUTEX_UNLOCK(&ival_lock); - if (idx == 0 || ival_data.session_dead) + if (idx == 0 || atomic_load_relaxed(&ival_data.session_dead)) return false; for (int i = 0; i < idx; i++) { @@ -194,7 +197,7 @@ ival_loop(int *sleep_time) if (rc && rc != -ENOENT && rc != -EBADF) DHS_ERROR(&ival_data, -rc, "notify_inval_entry() failed"); if (rc == -EBADF) - ival_data.session_dead = true; + atomic_store_relaxed(&ival_data.session_dead, true); } return (idx == EVICT_COUNT); @@ -236,6 +239,315 @@ ival_thread_fn(void *arg) return NULL; } +/* Advisory reverse notifications. + * + * The fuse_lowlevel_notify_*() calls block in the kernel on inode locks that in-flight requests + * hold while waiting on dfuse, so request handlers must never issue one directly; they enqueue + * here and a dedicated thread delivers. Nothing waits on delivery: a stalled or dropped entry + * leaves a dentry cached until its normal timeout, the same as a failed kernel call. Duplicate + * queued entries coalesce; an entry leaves the coalesce index when popped, before delivery, so + * an enqueue racing delivery is never absorbed into a stale notification. + */ +enum notify_kind { + NOTIFY_INVAL_ENTRY, + NOTIFY_DELETE, + NOTIFY_INVAL_INODE, +}; + +/* Queued by copy; the source inode may be released once enqueued */ +struct notify_entry { + d_list_t ne_list; + d_list_t ne_hlist; /* coalesce bucket, valid only while queued */ + enum notify_kind ne_kind; + fuse_ino_t ne_parent; + fuse_ino_t ne_ino; + char ne_name[NAME_MAX + 1]; + size_t ne_namelen; +}; + +/* Drop rather than grow without bound; invalidation is advisory */ +#define NOTIFY_QUEUE_MAX 16384 + +/* Depth at which delivery is judged stalled rather than merely busy */ +#define NOTIFY_CONGEST_HIGH (NOTIFY_QUEUE_MAX / 2) + +#define NOTIFY_COALESCE_BUCKETS 4096 + +static pthread_mutex_t notify_lock = PTHREAD_MUTEX_INITIALIZER; +static d_list_t notify_queue; +static d_list_t notify_coalesce[NOTIFY_COALESCE_BUCKETS]; +static sem_t notify_sem; +static bool notify_stop; +static unsigned int notify_queued; +static pthread_t notify_thread; +static struct d_slab_type *notify_slab; + +/* Count of enqueues seen at/above NOTIFY_CONGEST_HIGH; gates the congestion warning rate. + * Under notify_lock. + */ +static uint64_t notify_congest_total; + +/* Trace identity; carries dfuse_info for the teardown paths, which take no arguments */ +static struct dfuse_notify { + struct dfuse_info *dn_info; +} notify_data; + +/* FNV-1a over the coalesce key; 64-bit inos are folded as two 32-bit halves */ +static uint32_t +notify_hash(enum notify_kind kind, fuse_ino_t parent, fuse_ino_t ino, const char *name, + size_t namelen) +{ + uint32_t h = 2166136261u; + size_t i; + + h = (h ^ (uint32_t)kind) * 16777619u; + h = (h ^ (uint32_t)parent) * 16777619u; + h = (h ^ (uint32_t)(parent >> 32)) * 16777619u; + h = (h ^ (uint32_t)ino) * 16777619u; + h = (h ^ (uint32_t)(ino >> 32)) * 16777619u; + for (i = 0; i < namelen; i++) + h = (h ^ (unsigned char)name[i]) * 16777619u; + + return h % NOTIFY_COALESCE_BUCKETS; +} + +/* Must be called with notify_lock held. */ +static struct notify_entry * +notify_coalesce_find(enum notify_kind kind, fuse_ino_t parent, fuse_ino_t ino, const char *name, + size_t namelen, uint32_t bucket) +{ + struct notify_entry *ne; + + d_list_for_each_entry(ne, ¬ify_coalesce[bucket], ne_hlist) { + if (ne->ne_kind != kind || ne->ne_parent != parent || ne->ne_ino != ino) + continue; + if (ne->ne_namelen != namelen) + continue; + if (namelen != 0 && memcmp(ne->ne_name, name, namelen) != 0) + continue; + return ne; + } + + return NULL; +} + +/* Must be called with notify_lock held. */ +static struct notify_entry * +notify_dequeue_locked(void) +{ + struct notify_entry *ne; + + if (d_list_empty(¬ify_queue)) + return NULL; + + ne = d_list_entry(notify_queue.next, struct notify_entry, ne_list); + d_list_del(&ne->ne_list); + d_list_del(&ne->ne_hlist); + notify_queued--; + + return ne; +} + +/* Discard everything queued, at stop or after EBADF. Must be called with notify_lock held. */ +static void +notify_drain_locked(void) +{ + struct notify_entry *ne, *nep; + unsigned int count = 0; + + d_list_for_each_entry_safe(ne, nep, ¬ify_queue, ne_list) { + d_list_del(&ne->ne_list); + d_list_del(&ne->ne_hlist); + d_slab_release(notify_slab, ne); + count++; + } + notify_queued = 0; + + if (count == 0) + return; + + atomic_fetch_add_relaxed(¬ify_data.dn_info->di_notify_dropped, count); + DFUSE_TRA_DEBUG(¬ify_data, "Drained %u queued notifications", count); +} + +/* Rate-limited: a mass drop must not flood the log */ +static void +notify_drop(struct dfuse_info *dfuse_info, enum notify_kind kind, fuse_ino_t parent) +{ + uint64_t total; + + total = atomic_fetch_add_relaxed(&dfuse_info->di_notify_dropped, 1) + 1; + + if (total == 1 || (total % 1000) == 0) + DFUSE_TRA_WARNING(¬ify_data, + "Dropped advisory notify kind %d parent %#lx, %lu total", kind, + parent, total); +} + +/* Rate-limited: sustained congestion must not flood the log. Called with notify_lock held. */ +static void +notify_congest_warn(unsigned int depth) +{ + notify_congest_total++; + + if (notify_congest_total == 1 || (notify_congest_total % 4096) == 0) + DFUSE_TRA_WARNING(¬ify_data, "Notify queue congested: depth %u", depth); +} + +/* Blocking here is expected; only the notify thread runs this */ +static void +notify_run(struct notify_entry *ne) +{ + int rc = 0; + + switch (ne->ne_kind) { + case NOTIFY_INVAL_ENTRY: + rc = fuse_lowlevel_notify_inval_entry(ival_data.session, ne->ne_parent, ne->ne_name, + ne->ne_namelen); + break; + case NOTIFY_DELETE: + rc = fuse_lowlevel_notify_delete(ival_data.session, ne->ne_parent, ne->ne_ino, + ne->ne_name, ne->ne_namelen); + break; + case NOTIFY_INVAL_INODE: + rc = fuse_lowlevel_notify_inval_inode(ival_data.session, ne->ne_ino, 0, 0); + break; + } + + /* Session is gone; latch (shared with ival_loop) and let the loop drain */ + if (rc == -EBADF) { + atomic_store_relaxed(&ival_data.session_dead, true); + atomic_fetch_add_relaxed(¬ify_data.dn_info->di_notify_dropped, 1); + return; + } + + /* ENOENT/ENOTDIR just mean nothing was cached; any other error is unexpected and does + * not count as delivered. + */ + if (rc != 0 && rc != -ENOENT && rc != -ENOTDIR) { + DHS_ERROR(¬ify_data, -rc, "notify() failed, kind %d parent %#lx", ne->ne_kind, + ne->ne_parent); + return; + } + + atomic_fetch_add_relaxed(¬ify_data.dn_info->di_notify_delivered, 1); +} + +static void * +notify_thread_fn(void *arg) +{ + while (1) { + struct notify_entry *ne = NULL; + bool stop; + bool dead; + + if (sem_wait(¬ify_sem) != 0) { + D_ASSERTF(errno == EINTR, "sem_wait: %d (%s)\n", errno, strerror(errno)); + continue; + } + + D_MUTEX_LOCK(¬ify_lock); + stop = notify_stop; + dead = atomic_load_relaxed(&ival_data.session_dead); + if (stop || dead) + notify_drain_locked(); + else + ne = notify_dequeue_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + + if (stop) + return NULL; + + if (ne == NULL) + continue; + + notify_run(ne); + d_slab_release(notify_slab, ne); + } + return NULL; +} + +/* Best effort; failures result in dropped notifications. */ +static void +notify_enqueue(struct dfuse_info *dfuse_info, enum notify_kind kind, fuse_ino_t parent, + fuse_ino_t ino, const char *name) +{ + struct notify_entry *ne; + size_t namelen = name ? strnlen(name, NAME_MAX) : 0; + uint32_t bucket = notify_hash(kind, parent, ino, name, namelen); + + D_MUTEX_LOCK(¬ify_lock); + + if (notify_stop) { + D_MUTEX_UNLOCK(¬ify_lock); + notify_drop(dfuse_info, kind, parent); + return; + } + + if (notify_coalesce_find(kind, parent, ino, name, namelen, bucket) != NULL) { + D_MUTEX_UNLOCK(¬ify_lock); + atomic_fetch_add_relaxed(&dfuse_info->di_notify_coalesced, 1); + DFUSE_TRA_DEBUG(¬ify_data, "Coalesced kind %d parent %#lx " DF_DE, kind, parent, + DP_DE(name ? name : "")); + return; + } + + if (notify_queued >= NOTIFY_QUEUE_MAX) { + D_MUTEX_UNLOCK(¬ify_lock); + notify_drop(dfuse_info, kind, parent); + return; + } + + ne = d_slab_acquire(notify_slab); + if (ne == NULL) { + D_MUTEX_UNLOCK(¬ify_lock); + notify_drop(dfuse_info, kind, parent); + return; + } + + ne->ne_kind = kind; + ne->ne_parent = parent; + ne->ne_ino = ino; + ne->ne_namelen = namelen; + if (namelen > 0) + memcpy(ne->ne_name, name, namelen); + ne->ne_name[namelen] = '\0'; + + d_list_add_tail(&ne->ne_hlist, ¬ify_coalesce[bucket]); + d_list_add_tail(&ne->ne_list, ¬ify_queue); + notify_queued++; + + if (notify_queued >= NOTIFY_CONGEST_HIGH) + notify_congest_warn(notify_queued); + + D_MUTEX_UNLOCK(¬ify_lock); + + atomic_fetch_add_relaxed(&dfuse_info->di_notify_enqueued, 1); + DFUSE_TRA_DEBUG(¬ify_data, "Enqueued kind %d parent %#lx " DF_DE, kind, parent, + DP_DE(name ? name : "")); + + sem_post(¬ify_sem); +} + +void +dfuse_notify_inval_entry(struct dfuse_info *dfuse_info, fuse_ino_t parent, const char *name) +{ + notify_enqueue(dfuse_info, NOTIFY_INVAL_ENTRY, parent, 0, name); +} + +void +dfuse_notify_delete(struct dfuse_info *dfuse_info, fuse_ino_t parent, fuse_ino_t ino, + const char *name) +{ + notify_enqueue(dfuse_info, NOTIFY_DELETE, parent, ino, name); +} + +void +dfuse_notify_inval_inode(struct dfuse_info *dfuse_info, fuse_ino_t ino) +{ + notify_enqueue(dfuse_info, NOTIFY_INVAL_INODE, 0, ino, NULL); +} + /* Allocate and insert a new time value entry */ static int ival_bucket_add(d_list_t *list, double timeout) @@ -263,28 +575,55 @@ int ival_init(struct dfuse_info *dfuse_info) { int rc; + int i; DFUSE_TRA_UP(&ival_data, dfuse_info, "invalidator"); + DFUSE_TRA_UP(¬ify_data, dfuse_info, "notify"); D_INIT_LIST_HEAD(&ival_data.time_entry_list); + notify_data.dn_info = dfuse_info; + D_INIT_LIST_HEAD(¬ify_queue); + for (i = 0; i < NOTIFY_COALESCE_BUCKETS; i++) + D_INIT_LIST_HEAD(¬ify_coalesce[i]); + rc = sem_init(&ival_sem, 0, 0); if (rc != 0) D_GOTO(out, rc = errno); + rc = sem_init(¬ify_sem, 0, 0); + if (rc != 0) { + rc = errno; + goto ival_sem; + } + rc = ival_bucket_add(&ival_data.time_entry_list, 0); if (rc) - goto sem; + goto notify_sem; out: return rc; -sem: +notify_sem: + sem_destroy(¬ify_sem); +ival_sem: sem_destroy(&ival_sem); + DFUSE_TRA_DOWN(¬ify_data); DFUSE_TRA_DOWN(&ival_data); return rc; } -/* Start the thread. Not called until after fuse is mounted */ +/* Register the notify entry slab type. Split out of ival_thread_start() so tests can prepare + * the queue for notify_enqueue()/notify_dequeue_locked() without starting the delivery thread. + */ +static int +notify_init(struct dfuse_info *dfuse_info) +{ + struct d_slab_reg notify_slab_reg = {POOL_TYPE_INIT(notify_entry, ne_list)}; + + return d_slab_register(&dfuse_info->di_slab, ¬ify_slab_reg, dfuse_info, ¬ify_slab); +} + +/* Start the threads. Not called until after fuse is mounted */ int ival_thread_start(struct dfuse_info *dfuse_info) { @@ -292,16 +631,31 @@ ival_thread_start(struct dfuse_info *dfuse_info) ival_data.session = dfuse_info->di_session; - rc = pthread_create(&ival_thread, NULL, ival_thread_fn, NULL); + rc = notify_init(dfuse_info); + if (rc != -DER_SUCCESS) + return daos_der2errno(rc); + + rc = pthread_create(¬ify_thread, NULL, notify_thread_fn, NULL); if (rc != 0) - goto out; + return rc; + pthread_setname_np(notify_thread, "dfuse notify"); + + rc = pthread_create(&ival_thread, NULL, ival_thread_fn, NULL); + if (rc != 0) { + D_MUTEX_LOCK(¬ify_lock); + notify_stop = true; + D_MUTEX_UNLOCK(¬ify_lock); + sem_post(¬ify_sem); + pthread_join(notify_thread, NULL); + notify_thread = 0; + return rc; + } pthread_setname_np(ival_thread, "dfuse inval"); -out: - return rc; + return 0; } -/* Stop thread, remove all inodes from the invalidation queues and teardown all data structures +/* Stop threads, remove all inodes from the invalidation queues and teardown all data structures. * May be called without thread_start() having been called. */ void @@ -314,6 +668,18 @@ ival_thread_stop() if (ival_thread) pthread_join(ival_thread, NULL); ival_thread = 0; + + /* Latch stop before waking the thread: anything still landing behind an in-flight + * request during unmount is dropped rather than queued behind a thread about to exit. + */ + D_MUTEX_LOCK(¬ify_lock); + notify_stop = true; + D_MUTEX_UNLOCK(¬ify_lock); + sem_post(¬ify_sem); + + if (notify_thread) + pthread_join(notify_thread, NULL); + notify_thread = 0; } void @@ -331,6 +697,10 @@ ival_fini() d_list_del(&dte->dte_list); D_FREE(dte); } + + sem_destroy(¬ify_sem); + sem_destroy(&ival_sem); + DFUSE_TRA_DOWN(¬ify_data); DFUSE_TRA_DOWN(&ival_data); } diff --git a/src/client/dfuse/ops/ioctl.c b/src/client/dfuse/ops/ioctl.c index 8d073b859ff..3526ec47262 100644 --- a/src/client/dfuse/ops/ioctl.c +++ b/src/client/dfuse/ops/ioctl.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2016-2024 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -44,16 +45,7 @@ handle_il_ioctl(struct dfuse_obj_hdl *oh, fuse_req_t req) il_reply.fir_flags |= DFUSE_IOCTL_FLAGS_MCACHE; if (oh->doh_writeable) { - rc = fuse_lowlevel_notify_inval_inode(dfuse_info->di_session, - oh->doh_ie->ie_stat.st_ino, 0, 0); - - if (rc == 0) { - DFUSE_TRA_DEBUG(oh, "inval inode %#lx rc is %d", oh->doh_ie->ie_stat.st_ino, - rc); - } else { - DFUSE_TRA_ERROR(oh, "inval inode %#lx rc is %d", oh->doh_ie->ie_stat.st_ino, - rc); - } + dfuse_notify_inval_inode(dfuse_info, oh->doh_ie->ie_stat.st_ino); /* Mark this file handle as using the IL or similar, and if this is new then mark * the inode as well diff --git a/src/client/dfuse/ops/lookup.c b/src/client/dfuse/ops/lookup.c index 1f1f25201b9..d7bba8c2979 100644 --- a/src/client/dfuse/ops/lookup.c +++ b/src/client/dfuse/ops/lookup.c @@ -1,6 +1,6 @@ /** * (C) Copyright 2016-2024 Intel Corporation. - * (C) Copyright 2025 Hewlett Packard Enterprise Development LP + * (C) Copyright 2025-2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -163,10 +163,7 @@ dfuse_reply_entry(struct dfuse_info *dfuse_info, struct dfuse_inode_entry *ie, if (wipe_parent == 0) return; - rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, wipe_parent, wipe_name, - strnlen(wipe_name, NAME_MAX)); - if (rc && rc != -ENOENT) - DS_ERROR(-rc, "inval_entry() failed"); + dfuse_notify_inval_entry(dfuse_info, wipe_parent, wipe_name); return; out_err: diff --git a/src/client/dfuse/ops/open.c b/src/client/dfuse/ops/open.c index 5a5f7d3494e..183bdba6783 100644 --- a/src/client/dfuse/ops/open.c +++ b/src/client/dfuse/ops/open.c @@ -242,11 +242,7 @@ dfuse_cb_release(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi) dfuse_inode_decref(dfuse_info, oh->doh_parent_dir); } if (ie) { - rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent, - ie->ie_name, strnlen(ie->ie_name, NAME_MAX)); - - if (rc != 0 && rc != -ENOENT) - DHS_ERROR(ie, -rc, "inval_entry() error"); + dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name); dfuse_inode_decref(dfuse_info, ie); } dfuse_oh_free(dfuse_info, oh); diff --git a/src/client/dfuse/ops/opendir.c b/src/client/dfuse/ops/opendir.c index 60d1e6ab4d5..d6fce8667d0 100644 --- a/src/client/dfuse/ops/opendir.c +++ b/src/client/dfuse/ops/opendir.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2016-2024 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -81,13 +82,7 @@ dfuse_cb_releasedir(fuse_req_t req, struct dfuse_inode_entry *ino, struct fuse_f DFUSE_REPLY_ZERO_OH(oh, req); if (ie) { - int rc; - - rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent, - ie->ie_name, strnlen(ie->ie_name, NAME_MAX)); - - if (rc != 0 && rc != -ENOENT) - DHS_ERROR(ie, -rc, "inval_entry() error"); + dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name); dfuse_inode_decref(dfuse_info, ie); } dfuse_oh_free(dfuse_info, oh); diff --git a/src/client/dfuse/ops/rename.c b/src/client/dfuse/ops/rename.c index fcf95b82466..e74f143c087 100644 --- a/src/client/dfuse/ops/rename.c +++ b/src/client/dfuse/ops/rename.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2016-2023 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -18,7 +19,6 @@ dfuse_oid_moved(struct dfuse_info *dfuse_info, daos_obj_id_t *oid, struct dfuse_ const char *name, struct dfuse_inode_entry *newparent, const char *newname) { struct dfuse_inode_entry *ie; - int rc; ino_t ino; dfuse_compute_inode(parent->ie_dfs, oid, &ino); @@ -34,11 +34,7 @@ dfuse_oid_moved(struct dfuse_info *dfuse_info, daos_obj_id_t *oid, struct dfuse_ (strncmp(ie->ie_name, name, NAME_MAX) != 0)) { DFUSE_TRA_DEBUG(ie, "Invalidating old name"); - rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent, - ie->ie_name, strnlen(ie->ie_name, NAME_MAX)); - - if (rc && rc != -ENOENT) - DFUSE_TRA_ERROR(ie, "inval_entry() returned: %d (%s)", rc, strerror(-rc)); + dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name); } /* Update the inode entry data */ diff --git a/src/client/dfuse/ops/setxattr.c b/src/client/dfuse/ops/setxattr.c index e50c010b377..861a7303e38 100644 --- a/src/client/dfuse/ops/setxattr.c +++ b/src/client/dfuse/ops/setxattr.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2019-2022 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -46,22 +47,15 @@ dfuse_cb_setxattr(fuse_req_t req, struct dfuse_inode_entry *inode, rc = dfs_setxattr(inode->ie_dfs->dfs_ns, inode->ie_obj, name, value, size, flags); if (rc == 0) { - /* Optionally remove the dentry to force a new lookup on access. - * If the xattr is to set a UNS entry point, and dentry_dir - * caching is enabled then invalidate the dentry here, to force - * a lookup which will check the xattr and return the linked - * container. The fuse header says this potentially deadlocks - * however it does appear to work, and calling this after the - * reply will introduce a race condition that future lookups - * will be skipped. + /* Setting a UNS entry point renumbers the inode, so invalidate the dentry to + * force a fresh lookup. Safe out of order: a racing lookup holds the parent + * shared while it instantiates, the invalidation takes it exclusive, so a stale + * dentry is always in place before the queued invalidation kills it. */ if (duns_attr && inode->ie_dfs->dfc_dentry_dir_timeout > 0) { struct dfuse_info *dfuse_info = fuse_req_userdata(req); - rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, - inode->ie_parent, inode->ie_name, - strnlen(inode->ie_name, NAME_MAX)); - DFUSE_TRA_INFO(inode, "inval_entry() rc is %d", rc); + dfuse_notify_inval_entry(dfuse_info, inode->ie_parent, inode->ie_name); } DFUSE_REPLY_ZERO(inode, req); return; diff --git a/src/client/dfuse/ops/unlink.c b/src/client/dfuse/ops/unlink.c index 778d44e0bed..9af5a68ba28 100644 --- a/src/client/dfuse/ops/unlink.c +++ b/src/client/dfuse/ops/unlink.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2016-2023 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -18,7 +19,6 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t struct dfuse_inode_entry *parent, const char *name) { struct dfuse_inode_entry *ie; - int rc; fuse_ino_t ino; ino_t parent_ino; @@ -46,9 +46,7 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t * unlinked so will destroy it anyway, but there is a race here so try and destroy it * even though most of the time we expect this to fail. */ - rc = fuse_lowlevel_notify_inval_inode(dfuse_info->di_session, ino, 0, 0); - if (rc && rc != -ENOENT) - DHS_ERROR(ie, -rc, "inval_inode() error"); + dfuse_notify_inval_inode(dfuse_info, ino); /* If the kernel was aware of this inode at an old location then remove that which should * trigger a forget call. Checking the test logs shows that we do see the forget anyway @@ -58,10 +56,7 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t DFUSE_TRA_DEBUG(ie, "Telling kernel to forget %#lx " DF_DE, ie->ie_parent, DP_DE(ie->ie_name)); - rc = fuse_lowlevel_notify_delete(dfuse_info->di_session, ie->ie_parent, ino, - ie->ie_name, strnlen(ie->ie_name, NAME_MAX)); - if (rc && rc != -ENOENT) - DHS_ERROR(ie, -rc, "notify_delete() error"); + dfuse_notify_delete(dfuse_info, ie->ie_parent, ino, ie->ie_name); } /* Drop the ref again */ diff --git a/src/client/dfuse/tests/SConscript b/src/client/dfuse/tests/SConscript new file mode 100644 index 00000000000..7883d06abec --- /dev/null +++ b/src/client/dfuse/tests/SConscript @@ -0,0 +1,17 @@ +"""Build DFuse unit tests""" + + +def scons(): + """Execute build""" + Import('denv') + + tenv = denv.Clone() + tenv.AppendUnique(LIBS=['cmocka']) + + notify_tests = tenv.d_test_program('notify_tests', 'notify_tests.c') + + tenv.Install('$PREFIX/bin/', [notify_tests]) + + +if __name__ == "SCons.Script": + scons() diff --git a/src/client/dfuse/tests/notify_tests.c b/src/client/dfuse/tests/notify_tests.c new file mode 100644 index 00000000000..86510c64631 --- /dev/null +++ b/src/client/dfuse/tests/notify_tests.c @@ -0,0 +1,341 @@ +/** + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP + * + * SPDX-License-Identifier: BSD-2-Clause-Patent + */ +#include +#include +#include +#include +#include +#include + +#include +#include + +/* DFUSE_NOTIFY_RAW_OK is defined inside inval.c before it pulls in dfuse.h, so the poison + * pragma there never fires for this TU; these stand in for the real fuse session calls. + */ +static int stub_notify_rc; +static int stub_inval_entry_calls; +static int stub_delete_calls; +static int stub_inval_inode_calls; + +#include "../inval.c" + +int +fuse_lowlevel_notify_inval_entry(struct fuse_session *se, fuse_ino_t parent, const char *name, + size_t namelen) +{ + stub_inval_entry_calls++; + return stub_notify_rc; +} + +int +fuse_lowlevel_notify_expire_entry(struct fuse_session *se, fuse_ino_t parent, const char *name, + size_t namelen) +{ + return stub_notify_rc; +} + +int +fuse_lowlevel_notify_delete(struct fuse_session *se, fuse_ino_t parent, fuse_ino_t child, + const char *name, size_t namelen) +{ + stub_delete_calls++; + return stub_notify_rc; +} + +int +fuse_lowlevel_notify_inval_inode(struct fuse_session *se, fuse_ino_t ino, off_t off, off_t len) +{ + stub_inval_inode_calls++; + return stub_notify_rc; +} + +/* Eviction sweep is out of scope here; ival_loop() is compiled but never invoked. */ +bool +dfuse_dentry_get_valid(struct dfuse_inode_entry *ie, double max_age, double *timeout) +{ + return false; +} + +static struct dfuse_info tnotify_info; + +static void +tnotify_reset(void) +{ + memset(&tnotify_info, 0, sizeof(tnotify_info)); + memset(&ival_data, 0, sizeof(ival_data)); + memset(¬ify_data, 0, sizeof(notify_data)); + notify_queued = 0; + notify_stop = false; + notify_congest_total = 0; + stub_notify_rc = 0; + stub_inval_entry_calls = 0; + stub_delete_calls = 0; + stub_inval_inode_calls = 0; + + assert_int_equal(d_slab_init(&tnotify_info.di_slab, &tnotify_info), -DER_SUCCESS); + assert_int_equal(ival_init(&tnotify_info), 0); + assert_int_equal(notify_init(&tnotify_info), -DER_SUCCESS); +} + +static int +tnotify_setup(void **state) +{ + tnotify_reset(); + return 0; +} + +static int +tnotify_teardown(void **state) +{ + D_MUTEX_LOCK(¬ify_lock); + notify_drain_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + + ival_fini(); + d_slab_destroy(&tnotify_info.di_slab); + return 0; +} + +static void +test_coalesce_duplicate(void **state) +{ + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + assert_int_equal(notify_queued, 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 1); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + assert_int_equal(notify_queued, 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 1); +} + +static void +test_coalesce_key_fields_differ(void **state) +{ + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 10, 0, "foo"); + assert_int_equal(notify_queued, 2); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 11, 0, "foo"); + assert_int_equal(notify_queued, 3); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "bar"); + assert_int_equal(notify_queued, 4); + + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0); +} + +/* Same parent+name but different child ino: the kernel verifies child identity on delete, so + * collapsing these would silently drop one of the two deletions (design-review finding). + */ +static void +test_coalesce_delete_distinct_child_ino(void **state) +{ + notify_enqueue(&tnotify_info, NOTIFY_DELETE, 10, 100, "foo"); + notify_enqueue(&tnotify_info, NOTIFY_DELETE, 10, 200, "foo"); + + assert_int_equal(notify_queued, 2); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0); +} + +static void +test_pop_before_delivery_not_coalesced(void **state) +{ + struct notify_entry *ne; + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + + D_MUTEX_LOCK(¬ify_lock); + ne = notify_dequeue_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + assert_non_null(ne); + d_slab_release(notify_slab, ne); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + assert_int_equal(notify_queued, 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 2); +} + +static void +test_backstop_drop_at_max(void **state) +{ + int i; + + for (i = 0; i < NOTIFY_QUEUE_MAX; i++) + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, i + 1, NULL); + + assert_int_equal(notify_queued, NOTIFY_QUEUE_MAX); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 0); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, NOTIFY_QUEUE_MAX + 1, NULL); + + assert_int_equal(notify_queued, NOTIFY_QUEUE_MAX); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1); +} + +static void +test_stop_latch_drops(void **state) +{ + notify_stop = true; + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + + assert_int_equal(notify_queued, 0); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 0); +} + +static void +test_congest_warn_silent_below_threshold(void **state) +{ + fuse_ino_t ino = 1; + + for (; notify_queued < NOTIFY_CONGEST_HIGH - 1; ino++) + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino, NULL); + + assert_int_equal(notify_congest_total, 0); +} + +static void +test_congest_warn_at_and_above_threshold(void **state) +{ + fuse_ino_t ino = 1; + + for (; notify_queued < NOTIFY_CONGEST_HIGH; ino++) + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino, NULL); + assert_int_equal(notify_congest_total, 1); + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino++, NULL); + assert_int_equal(notify_congest_total, 2); +} + +static void +test_delivery_rc_enoent_ignored(void **state) +{ + struct notify_entry *ne; + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + + D_MUTEX_LOCK(¬ify_lock); + ne = notify_dequeue_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + assert_non_null(ne); + + stub_notify_rc = -ENOENT; + notify_run(ne); + d_slab_release(notify_slab, ne); + + assert_false(atomic_load_relaxed(&ival_data.session_dead)); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_delivered), 1); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 0); +} + +static void +test_delivery_rc_ebadf_latches_and_drains(void **state) +{ + struct notify_entry *ne; + + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo"); + notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 11, 0, "bar"); + + D_MUTEX_LOCK(¬ify_lock); + ne = notify_dequeue_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + assert_non_null(ne); + + stub_notify_rc = -EBADF; + notify_run(ne); + d_slab_release(notify_slab, ne); + + assert_true(atomic_load_relaxed(&ival_data.session_dead)); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1); + + /* Session dead: the delivery thread drains whatever remains queued */ + D_MUTEX_LOCK(¬ify_lock); + notify_drain_locked(); + D_MUTEX_UNLOCK(¬ify_lock); + + assert_int_equal(notify_queued, 0); + assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 2); +} + +static void +test_hash_same_key_same_bucket(void **state) +{ + uint32_t b1 = notify_hash(NOTIFY_DELETE, 42, 7, "somefile", 8); + uint32_t b2 = notify_hash(NOTIFY_DELETE, 42, 7, "somefile", 8); + + assert_int_equal(b1, b2); +} + +#define HASH_SAMPLE_COUNT 4096 + +static void +test_hash_distribution_smoke(void **state) +{ + static unsigned int counts[NOTIFY_COALESCE_BUCKETS]; + unsigned int max_count = 0; + int i; + + memset(counts, 0, sizeof(counts)); + + for (i = 0; i < HASH_SAMPLE_COUNT; i++) { + char name[32]; + uint32_t bucket; + + snprintf(name, sizeof(name), "file-%d", i); + bucket = notify_hash(NOTIFY_INVAL_ENTRY, i, 0, name, strlen(name)); + counts[bucket]++; + if (counts[bucket] > max_count) + max_count = counts[bucket]; + } + + /* Loose FNV-1a smoke check: average is 1/bucket here, no bucket should be swamped */ + assert_true(max_count < 20); +} + +int +main(void) +{ + const struct CMUnitTest tests[] = { + cmocka_unit_test_setup_teardown(test_coalesce_duplicate, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_coalesce_key_fields_differ, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_coalesce_delete_distinct_child_ino, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_pop_before_delivery_not_coalesced, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_backstop_drop_at_max, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_stop_latch_drops, tnotify_setup, tnotify_teardown), + cmocka_unit_test_setup_teardown(test_congest_warn_silent_below_threshold, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_congest_warn_at_and_above_threshold, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_delivery_rc_enoent_ignored, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_delivery_rc_ebadf_latches_and_drains, + tnotify_setup, tnotify_teardown), + cmocka_unit_test_setup_teardown(test_hash_same_key_same_bucket, tnotify_setup, + tnotify_teardown), + cmocka_unit_test_setup_teardown(test_hash_distribution_smoke, tnotify_setup, + tnotify_teardown), + }; + int rc; + + rc = daos_debug_init(DAOS_LOG_DEFAULT); + if (rc != 0) + return rc; + + rc = cmocka_run_group_tests_name("dfuse notify queue", tests, NULL, NULL); + + daos_debug_fini(); + + return rc; +} diff --git a/utils/node_local_test.py b/utils/node_local_test.py index a8813982122..49a94197e27 100755 --- a/utils/node_local_test.py +++ b/utils/node_local_test.py @@ -1375,7 +1375,7 @@ class DFuse(): # pylint: disable-next=too-many-arguments def __init__(self, daos, conf, pool=None, container=None, mount_path=None, uns_path=None, caching=True, wbcache=True, multi_user=False, ro=False, dump_h=False, - read_h=False, file_h=None): + read_h=False, file_h=None, thread_count=None): if mount_path: self.dir = mount_path else: @@ -1391,6 +1391,7 @@ def __init__(self, daos, conf, pool=None, container=None, mount_path=None, uns_p self.conf = conf self.multi_user = multi_user self.cores = 0 + self.thread_count = thread_count self._daos = daos self.caching = caching self.wbcache = wbcache @@ -1465,7 +1466,9 @@ def start(self, v_hint=None, use_oopt=False): if self.multi_user: cmd.append('--multi-user') - if not self.cores: + if self.thread_count: + cmd.extend(['--thread-count', str(self.thread_count)]) + elif not self.cores: # Use a lower default thread-count for NLT due to running tests in parallel. cmd.extend(['--thread-count', '4']) @@ -4721,6 +4724,142 @@ def import_torch(self, server): return importlib.import_module('pydaos.torch') +class StressTests(PosixTests): + """Stress and race reproducers, run sequentially against a dedicated server""" + + @staticmethod + def generate_test_list(): + """Generate list of stress tests""" + return [x for x in dir(StressTests) + if x.startswith('test') and x not in dir(PosixTests)] + + def _run_uns_wedge_test(self, v_hint, churn_sh, create_path_fn, create_must_succeed): + """Race UNS container-create against a churn script; detect a wedged mount. + + Fails by timing out - a wedged mount answers nothing further and blocked + processes cannot be killed. churn_sh is a shell template with {dfuse_dir}/ + {stop_file} placeholders; create_path_fn(dfuse_dir, idx) builds the create path. + """ + # --thread-count is reduced by the event-queue count, so this leaves exactly one + # worker; more would need enough concurrency to consume every worker to wedge. + dfuse = DFuse(self.server, self.conf, caching=True, container=self.container, + thread_count=2) + dfuse.use_valgrind = False + # Churn generates high op volume; debug logs would blow the NLT log budget. + dfuse.log_mask = 'WARN' + dfuse.start(v_hint=v_hint) + + churn = [] + stop_file = join(dfuse.dir, 'stop_churn') + cmd = churn_sh.format(dfuse_dir=dfuse.dir, stop_file=stop_file) + + cmd_env = get_base_env() + cmd_env['D_LOG_MASK'] = 'WARN' + cmd_env['DAOS_AGENT_DRPC_DIR'] = self.conf.agent_dir + + wedged = False + aborted = False + stuck = False + successes = 0 + proc = None + try: + for _ in range(4): + # pylint: disable-next=consider-using-with + churn.append(subprocess.Popen(['sh', '-c', cmd])) + + for idx in range(8): + create_cmd = [join(self.conf['PREFIX'], 'bin', 'daos'), 'container', 'create', + '--type', 'POSIX', '--path', create_path_fn(dfuse.dir, idx)] + # Popen rather than run(timeout=): a wedged mount leaves this process in + # uninterruptible sleep, so run() would hang trying to kill it. The + # connection has to be aborted before the process can be reaped. + # pylint: disable-next=consider-using-with + proc = subprocess.Popen(create_cmd, env=cmd_env, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT) + try: + stdout, _ = proc.communicate(timeout=60) + except subprocess.TimeoutExpired: + print(f'container create {idx} did not return; mount is wedged') + wedged = True + break + if proc.returncode == 0: + successes += 1 + else: + print(f'container create {idx} failed rc={proc.returncode}') + print(stdout.decode('utf-8', errors='replace')) + if create_must_succeed: + self.fail() + + if wedged: + # Nothing below can complete against a wedged mount and the processes cannot + # be signalled, so break the connection before trying to clean up. + _, lines = _abort_fuse_connections() + for line in lines: + print(line) + aborted = True + proc.wait(timeout=60) + self.fail() + + if not create_must_succeed and successes == 0: + print(f'None of 8 creates against {v_hint} succeeded') + self.fail() + finally: + try: + with open(stop_file, 'w'): + pass + except OSError: + pass + for cproc in churn: + try: + cproc.wait(timeout=30) + except subprocess.TimeoutExpired: + cproc.kill() + try: + cproc.wait(timeout=10) + except subprocess.TimeoutExpired: + print('Churn process is unkillable; mount is still wedged') + stuck = True + if stuck and not aborted: + # The create loop never saw the wedge, so it was never aborted above; abort + # now or dfuse.stop() below hangs the same way the churn children did. + _, lines = _abort_fuse_connections() + for line in lines: + print(line) + if dfuse.stop(): + self.fatal_errors = True + if stuck and not aborted: + self.fail() + + def test_uns_create_vs_dir_lock(self): + """Race UNS container-create against mkdir churn holding the mount root locked. + + Exercises the setxattr notification path. Fails by timing out. + """ + churn_sh = ('i=0; while [ ! -e {stop_file} ]; do ' + 'mkdir {dfuse_dir}/churn_$$_$i 2>/dev/null; i=$((i+1)); done') + self._run_uns_wedge_test( + 'uns_create_vs_dir_lock', churn_sh, + lambda dfuse_dir, idx: join(dfuse_dir, f'uns_{idx}'), + create_must_succeed=True) + + def test_uns_create_vs_rmdir_churn(self): + """Race UNS container-create against rmdir/rename churn on the same paths. + + Exercises the unlink and rename notification paths. Fails by timing out. + """ + num_paths = 4 + churn_sh = ('i=0; while [ ! -e {stop_file} ]; do ' + f'n=$((i % {num_paths})); ' + 'mkdir {dfuse_dir}/uns_$n 2>/dev/null; ' + 'mv {dfuse_dir}/uns_$n {dfuse_dir}/uns_${{n}}_old 2>/dev/null; ' + 'rmdir {dfuse_dir}/uns_${{n}}_old 2>/dev/null; ' + 'i=$((i+1)); done') + self._run_uns_wedge_test( + 'uns_create_vs_rmdir_churn', churn_sh, + lambda dfuse_dir, idx: join(dfuse_dir, f'uns_{idx % num_paths}'), + create_must_succeed=False) + + class NltStdoutWrapper(): """Class for capturing stdout from threads""" @@ -4798,12 +4937,14 @@ def __del__(self): sys.stderr = self._stderr -def run_posix_tests(server, conf, test_list): +def run_posix_tests(server, conf, test_list, klass=None, sequential=False): """Run one or all posix tests Create a new container per test, to ensure that every test is isolated from others. """ + if klass is None: + klass = PosixTests def _run_test(ptl=None, function=None, test_cb=None): ptl.call_index = 0 @@ -4860,7 +5001,7 @@ def _run_test(ptl=None, function=None, test_cb=None): out_wrapper = NltStdoutWrapper() err_wrapper = NltStderrWrapper() - pto = PosixTests(server, conf, pool=pool) + pto = klass(server, conf, pool=pool) if len(test_list) == 1: obj = getattr(pto, test_list[0]) @@ -4874,11 +5015,15 @@ def _run_test(ptl=None, function=None, test_cb=None): test_list.sort(key=lambda x: x not in slow_tests) for function in test_list: - ptl = PosixTests(server, conf, pool=pool) + ptl = klass(server, conf, pool=pool) obj = getattr(ptl, function) if not callable(obj): continue + if sequential: + _run_test(ptl=ptl, test_cb=obj, function=function) + continue + thread = threading.Thread(None, target=_run_test, name=f'test {function}', @@ -7103,6 +7248,8 @@ def run(wf, args): if args.mode == 'fi': fi_test = True + elif args.mode == 'stress': + pass else: for rep in range(args.repeat): if args.repeat > 1: @@ -7128,6 +7275,13 @@ def run(wf, args): print(f'--failfast set; stopping after iteration {rep + 1}/{args.repeat}') break + if args.mode in ('all', 'stress'): + with DaosServer(conf, test_class='stress', wf=wf_server, + fatal_errors=fatal_errors) as server: + fatal_errors.add_result( + run_posix_tests(server, conf, StressTests.generate_test_list(), + klass=StressTests, sequential=True)) + if args.mode == 'all': with DaosServer(conf, test_class='restart', wf=wf_server, fatal_errors=fatal_errors) as server: diff --git a/utils/utest.yaml b/utils/utest.yaml index e5077e57202..64ccd82526e 100644 --- a/utils/utest.yaml +++ b/utils/utest.yaml @@ -165,6 +165,10 @@ - cmd: ["bin/eq_tests"] - cmd: ["bin/agent_tests"] - cmd: ["bin/job_tests"] +- name: dfuse + base: "PREFIX" + tests: + - cmd: ["bin/notify_tests"] - name: cart base: "BUILD_DIR" tests: