diff --git a/CLAUDE.md b/CLAUDE.md index 6a762e61aaab..1bf530b02ae0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,6 +80,21 @@ against a scratch tree and assert on the hashfile and on-disk sharing. Dedupe cases need a reflink fs (`DUPEREMOVE_TEST_DIR`, set by `devenv.sh`). Keep tests in `tests/`; no shell tests. +- **The suite runs in parallel by default** (`make integration TEST_JOBS=…`, + `tests/run.py -j`; the why is in that file's docstring). 26.4→4.3s on 4 cores; + the real payoff is the sanitizer builds, which re-run all of it. **A new test + must not share state outside `setUp`'s per-test `mkdtemp` + hashfile** or it + will flake in parallel; `TEST_JOBS=1` is the sequential fallback for pinning + such a flake down. The suite is I/O-bound, so `auto` deliberately + over-subscribes (`2 × nproc`, capped) — don't "fix" it back to `nproc`. +- **A test asserting on the *physical* extent layout must set `serial = True`** + (`DuperemoveTest.serial`), which holds it back to a one-at-a-time pass after + the pool drains. Per-test scratch isolation doesn't help here: the + fsync-forced-extent-boundary trick and fiemap counts depend on btrfs + writeback, which concurrent I/O perturbs — CI caught exactly this on btrfs + (`test_extent_order_independent`, `test_streaming_dedupe`) while xfs passed. + The four `fsync`-boundary files are already marked. + - **Never scan/benchmark out of `/tmp` — it's tmpfs**, not reflink-capable and rejected by `is_fs_supported()`, so a scan there stores **0 files silently** and dedupe is a no-op. Use real btrfs/xfs and verify a non-zero file count diff --git a/Makefile b/Makefile index b521bb695889..e96769ac0916 100644 --- a/Makefile +++ b/Makefile @@ -140,9 +140,14 @@ test: # End-to-end suite (Python stdlib unittest). Dedupe cases need a reflink fs; # override the scratch dir with DUPEREMOVE_TEST_DIR=/path. +# +# Worker processes for the suite - tests/run.py's -j, see `tests/run.py --help`. +# Not make's own -j; TEST_JOBS=1 is the sequential fallback. The same value +# sizes the valgrind leg below, where each worker is far heavier. +TEST_JOBS ?= auto .PHONY: integration integration: oans - $(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py + $(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py -j $(TEST_JOBS) # Same end-to-end suite, but every oans invocation runs under valgrind memcheck # (via tests/valgrind-wrap.sh). Findings go to per-pid logs; a non-empty log @@ -153,7 +158,7 @@ VGLOGDIR = $(CURDIR)/.vglogs integration-valgrind: oans @command -v valgrind >/dev/null 2>&1 || { echo "valgrind not installed"; exit 1; } rm -rf $(VGLOGDIR) && mkdir -p $(VGLOGDIR) - OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py + OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py -j $(TEST_JOBS) @if find $(VGLOGDIR) -type f -size +0c | grep -q .; then \ echo "=== valgrind reported errors/leaks ==="; \ find $(VGLOGDIR) -type f -size +0c -exec cat {} +; \ diff --git a/tests/README.md b/tests/README.md index 2c8da626e231..3c07108fe991 100644 --- a/tests/README.md +++ b/tests/README.md @@ -21,11 +21,17 @@ Run both with `make check`. ```sh make integration # build oans and run the suite +make integration TEST_JOBS=1 # ... sequentially (see -j below) python3 tests/run.py # run directly (binary must be built) python3 tests/run.py hardlink dedupe # only tests whose id matches a pattern +python3 tests/run.py -j 8 # 8 workers ('auto' by default) python3 -m unittest discover -s tests/integration -v # plain unittest, no banner ``` +The suite runs across worker processes by default, so tests must keep to the +per-test scratch `setUp` hands them. `-j 1` runs them one at a time, which is +where to start if a test only fails in company. + Environment: - `DUPEREMOVE=/path/to/oans` — test a specific binary (defaults to the one @@ -52,6 +58,12 @@ Subclass `DuperemoveTest` (from `harness`) and add `test_*` methods. Each test gets a fresh scratch directory in `self.work` and a per-test hashfile in `self.hf`; both are cleaned up automatically. +That isolation is what lets the suite run in parallel, so keep to it — a test +reaching outside its own `self.work` will flake. If a test asserts on the +*physical* extent layout (the fsync-forced extent boundary trick, or fiemap +extent counts), set `serial = True` on the class: no amount of file isolation +helps there, because concurrent I/O changes how the kernel lays extents out. + ```python from harness import DuperemoveTest, requires_reflink diff --git a/tests/integration/harness.py b/tests/integration/harness.py index c95aa7cc1ace..1193f9feffe7 100644 --- a/tests/integration/harness.py +++ b/tests/integration/harness.py @@ -183,9 +183,41 @@ def scratch_fstype(directory=TEST_ROOT): # Base test case # -------------------------------------------------------------------------- +_libc = ctypes.CDLL("libc.so.6", use_errno=True) + + +def _settle_scratch(): + """syncfs() the scratch filesystem so FIEMAP sees real extents. + + A test builds files and hands them straight to a scan that asserts on their + extents. Under delayed allocation those extents may not exist yet when oans + maps the file - it correctly records none, and the test reads the shortfall + as a scanner bug. Running the suite in parallel pushes writeback far enough + behind to lose this routinely: test_sparse_file_scans failed 4 runs in 12, + and test_hardlink_pair_does_not_empty_hashfile saw 381 of 401 extents. + + Once per oans invocation, not once per file: fsync()ing each created file + costs a journal commit apiece and took the suite from 5.4s to 18s, where one + syncfs() of the whole scratch is a single syscall for the same guarantee. + """ + fd = os.open(TEST_ROOT, os.O_RDONLY) + try: + _libc.syncfs(fd) + finally: + os.close(fd) + + class DuperemoveTest(unittest.TestCase): """Base class: a fresh scratch dir + hashfile per test, plus helpers.""" + # Set True on a class whose assertions depend on the *physical* extent + # layout the kernel happens to produce - the fsync-forced extent boundary + # trick, or fiemap counts. Concurrent I/O perturbs btrfs writeback enough + # that the layout the setup intends is not the one it gets, so tests/run.py + # holds these back and runs them one at a time after the pool drains. + # Per-test scratch isolation is *not* what this is for; that already works. + serial = False + @classmethod def setUpClass(cls): if not (os.path.isfile(DUPEREMOVE) and os.access(DUPEREMOVE, os.X_OK)): @@ -211,6 +243,7 @@ def dm(self, *args, hashfile=True, stdin=None, env=None, quiet=True): Runs with -q by default (terse output); pass quiet=False to get the full human summary block (e.g. to assert on the 'Reclaimed' line). """ + _settle_scratch() # the tree must be on disk before oans maps it cmd = [DUPEREMOVE, "--io-threads=4"] if quiet: cmd.insert(1, "-q") @@ -382,8 +415,10 @@ def make_sparse(self, relpath, head, hole, tail): def make_trailing_hole(self, relpath, data, size): """Write `data`, then extend the file to `size` so it ends in a hole.""" - p = self.write(relpath, data) - os.truncate(p, size) + p = self.path(relpath) + with open(p, "wb") as f: + f.write(data) + f.truncate(size) return p def reflink(self, src_rel, dst_rel): diff --git a/tests/integration/test_extent_dedupe.py b/tests/integration/test_extent_dedupe.py index 78abeaf70bdb..5ccc7355750f 100644 --- a/tests/integration/test_extent_dedupe.py +++ b/tests/integration/test_extent_dedupe.py @@ -23,6 +23,9 @@ @requires_btrfs class ExtentDedupeTest(DuperemoveTest): + # Extent-layout sensitive: see DuperemoveTest.serial. + serial = True + def _mkfile(self, rel, head, tail): """head, an fsync to force an extent boundary, then the shared tail.""" p = self.path(rel) diff --git a/tests/integration/test_extent_order_independent.py b/tests/integration/test_extent_order_independent.py index 27f51712e827..c40e09a29661 100644 --- a/tests/integration/test_extent_order_independent.py +++ b/tests/integration/test_extent_order_independent.py @@ -20,6 +20,9 @@ @requires_btrfs class ExtentOrderIndependentTest(DuperemoveTest): + # Extent-layout sensitive: see DuperemoveTest.serial. + serial = True + def _mkfile(self, rel, head, tail): """head, an fsync to force an extent boundary, then the shared tail.""" p = self.path(rel) diff --git a/tests/integration/test_least_fragmented_target.py b/tests/integration/test_least_fragmented_target.py index 027141e015f8..8ff6b43c0a8e 100644 --- a/tests/integration/test_least_fragmented_target.py +++ b/tests/integration/test_least_fragmented_target.py @@ -13,6 +13,9 @@ @requires_btrfs class LeastFragmentedTargetTest(DuperemoveTest): + # Extent-layout sensitive: see DuperemoveTest.serial. + serial = True + def _fragment(self, rel, content): """Write content, then rewrite alternate 4K blocks in place (COW) so the file ends up split across many physical extents.""" diff --git a/tests/integration/test_streaming_dedupe.py b/tests/integration/test_streaming_dedupe.py index af7a0298dc5e..1bb6384728b4 100644 --- a/tests/integration/test_streaming_dedupe.py +++ b/tests/integration/test_streaming_dedupe.py @@ -18,6 +18,9 @@ @requires_reflink class StreamingDedupeTest(DuperemoveTest): + # Extent-layout sensitive: see DuperemoveTest.serial. + serial = True + # tiny passes + small batchsize => work spans many generation batches, so # the producer/watermark and cross-batch anchor paths are all exercised. ENV = {"DUPEREMOVE_FILES_PER_PASS": "8"} diff --git a/tests/run.py b/tests/run.py index c313339b4eca..e7af220603e7 100755 --- a/tests/run.py +++ b/tests/run.py @@ -2,21 +2,38 @@ """Run the oans integration tests. Thin wrapper over unittest that prints a short banner (which binary, where the -scratch lives, whether reflink works) and lets you filter tests by substring. +scratch lives, whether reflink works), lets you filter tests by substring, and +spreads the suite over worker processes. Usage: tests/run.py run everything tests/run.py hardlink dedupe only tests whose id contains a given string + tests/run.py -j 8 run on 8 workers + tests/run.py -j 1 one worker, i.e. strictly sequential DUPEREMOVE=/path tests/run.py test a specific binary DUPEREMOVE_TEST_DIR=/mnt/btrfs tests/run.py choose the scratch filesystem -Equivalent to `python3 -m unittest discover -s tests/integration`, minus the -banner and filtering. Exit status is non-zero if any test fails. +Workers are *processes*, not threads: test_long_path chdir()s and cwd is +process-global. Tests need no cooperation to be split up - harness.setUp already +gives each one its own mkdtemp scratch and its own hashfile inside it - but a +test that reaches outside that will flake in parallel. The exception is tests +asserting on the *physical* extent layout, which concurrent I/O perturbs no +matter how isolated their files are; those set DuperemoveTest.serial and run one +at a time once the pool has drained. + +Results are collected as they land and replayed into a real unittest result, so +the output is unittest's own in every mode, `-j 1` included. + +Exit status is non-zero if any test fails. """ +import argparse +import concurrent.futures as futures +import multiprocessing import os import subprocess import sys +import time import unittest HERE = os.path.dirname(os.path.abspath(__file__)) @@ -25,6 +42,25 @@ import harness # noqa: E402 (needs the sys.path insert above) +# The suite is I/O-bound, not CPU-bound: workers sit in subprocess.run(), sync() +# and autotune's cache drop, so total CPU is flat across job counts and nproc +# alone under-subscribes badly. Measured on a 4-core box over XFS: j=4 6.9s, +# j=8 4.4s, j=12 3.6s, j=16 3.6s. Hence 2x nproc, and a ceiling because a worker +# is not always a plain oans - under valgrind or ASAN each one costs far more +# memory, and the scratch filesystem is shared. 8 is a deliberate compromise +# short of the ~12 plateau, not a measured optimum. +MAX_AUTO_JOBS = 8 + +# What a worker ships back per test, and the result method that replays it. +# addUnexpectedSuccess takes no detail argument; the rest take one. +_REPLAY = { + "failures": "addFailure", + "errors": "addError", + "skipped": "addSkip", + "expectedFailures": "addExpectedFailure", + "unexpectedSuccesses": "addUnexpectedSuccess", +} + def _matches(test_id, patterns): return not patterns or any(p in test_id for p in patterns) @@ -58,17 +94,136 @@ def _tsan_note(): "sets them.") +def _jobs(value): + """Parse -j: a positive count, or 'auto'. argparse reports int()'s ValueError.""" + if value == "auto": + return min(2 * (os.cpu_count() or 1), MAX_AUTO_JOBS) + jobs = int(value) + if jobs < 1: + raise argparse.ArgumentTypeError("must be >= 1 or 'auto'") + return jobs + + +class _Stream: + """The writeln()-capable stream unittest's result object expects.""" + + def write(self, text): + sys.stdout.write(text) + + def writeln(self, text=""): + sys.stdout.write(text + "\n") + + def flush(self): + sys.stdout.flush() + + +class _WorkerTest: + """Stands in for a test that ran in a worker, for display purposes only.""" + + def __init__(self, test_id): + self.test_id = test_id + + def __str__(self): + return f"{self.test_id.rsplit('.', 1)[-1]} ({self.test_id})" + + def shortDescription(self): + return None + + +class _ReplayResult(unittest.TextTestResult): + """Collects worker outcomes; tracebacks arrive already formatted.""" + + def _exc_info_to_string(self, err, test): + return err + + +def _run_one(test_id): + """Run one test in this worker and return a picklable summary of it. + + Dispatch is per test rather than per class, so setUpClass runs once per test + (120x rather than 33x). That is free today - harness's is a stat and a + makedirs - but a real class fixture would have to be dispatched as a unit. + """ + result = unittest.TestResult() + started = time.monotonic() + unittest.defaultTestLoader.loadTestsFromName(test_id).run(result) + elapsed = time.monotonic() - started + + outcomes = [(kind, detail) + for kind in ("failures", "errors", "skipped", "expectedFailures") + for _test, detail in getattr(result, kind)] + outcomes += [("unexpectedSuccesses", None) + for _test in result.unexpectedSuccesses] + return elapsed, outcomes + + +def _dispatch(test_ids, jobs, result): + """Run test_ids across `jobs` workers, recording outcomes into `result`.""" + # Pin fork: workers inherit the parent's already-imported test modules, so + # loadTestsFromName costs ~0.2ms. Python 3.14 defaults Linux to forkserver, + # under which every worker would re-import harness - whose module level + # probes reflink support and the scratch fstype - at ~0.15s each. + context = multiprocessing.get_context("fork") + with futures.ProcessPoolExecutor(max_workers=jobs, mp_context=context) as pool: + pending = {pool.submit(_run_one, tid): tid for tid in test_ids} + for future in futures.as_completed(pending): + test = _WorkerTest(pending[future]) + result.startTest(test) + try: + elapsed, outcomes = future.result() + except Exception as exc: # worker died: segfault, OOM, ... + outcomes = [("errors", f"worker died: {exc}\n")] + if not outcomes: + result.addSuccess(test) + for kind, detail in outcomes: + add = getattr(result, _REPLAY[kind]) + add(test) if detail is None else add(test, detail) + result.stopTest(test) + + +def _run_suite(parallel_ids, serial_ids, jobs, verbosity=2): + result = _ReplayResult(_Stream(), True, verbosity) + started = time.monotonic() + + _dispatch(parallel_ids, jobs, result) + # DuperemoveTest.serial tests want the box to themselves: their setup builds + # a specific physical extent layout, which concurrent I/O disturbs. Held to + # the end so they don't serialise the rest of the run either. + if serial_ids: + _dispatch(serial_ids, 1, result) + + result.printErrors() + tail = f" (+{len(serial_ids)} serial)" if serial_ids else "" + result.stream.writeln(result.separator2) + result.stream.writeln(f"Ran {result.testsRun} tests in " + f"{time.monotonic() - started:.2f}s on {jobs} workers{tail}") + result.stream.writeln() + if result.wasSuccessful(): + result.stream.writeln(f"OK (skipped={len(result.skipped)})") + return 0 + result.stream.writeln(f"FAILED (failures={len(result.failures)}, " + f"errors={len(result.errors)}, " + f"skipped={len(result.skipped)})") + return 1 + + def main(argv): - patterns = argv[1:] + parser = argparse.ArgumentParser( + description="Run the oans integration tests.", + formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__) + parser.add_argument("-j", "--jobs", type=_jobs, default="auto", + help="worker processes, or 'auto' (default: auto)") + parser.add_argument("patterns", nargs="*", + help="only run tests whose id contains one of these") + args = parser.parse_args(argv[1:]) version = subprocess.run([harness.DUPEREMOVE, "--version"], capture_output=True, text=True).stdout.strip() - fstype = subprocess.run(["stat", "-f", "-c", "%T", harness.TEST_ROOT], - capture_output=True, text=True).stdout.strip() print("oans integration tests") print(f" binary : {harness.DUPEREMOVE} ({version})") - print(f" scratch: {harness.TEST_ROOT} ({fstype})") + print(f" scratch: {harness.TEST_ROOT} ({harness.scratch_fstype()})") print(f" reflink: {'yes' if harness.REFLINK else 'no (dedupe tests will skip)'}") + print(f" jobs : {args.jobs}") tsan = _tsan_note() if tsan: print(f" sanitize: {tsan}") @@ -76,15 +231,13 @@ def main(argv): loader = unittest.TestLoader() discovered = loader.discover(start_dir=INTEGRATION_DIR, pattern="test_*.py") - - # Flatten and apply the substring filter. - suite = unittest.TestSuite() - for test in _iter_tests(discovered): - if _matches(test.id(), patterns): - suite.addTest(test) - - result = unittest.TextTestRunner(verbosity=2).run(suite) - return 0 if result.wasSuccessful() else 1 + tests = [t for t in _iter_tests(discovered) if _matches(t.id(), args.patterns)] + serial = [t.id() for t in tests if getattr(t, "serial", False)] + parallel = [t.id() for t in tests if not getattr(t, "serial", False)] + if serial: + print(f" serial : {len(serial)} extent-layout tests held to the end\n", + flush=True) + return _run_suite(parallel, serial, args.jobs) def _iter_tests(suite):