diff --git a/mlpstorage_py/benchmarks/kvcache.py b/mlpstorage_py/benchmarks/kvcache.py index b6bca948..70964743 100755 --- a/mlpstorage_py/benchmarks/kvcache.py +++ b/mlpstorage_py/benchmarks/kvcache.py @@ -858,11 +858,19 @@ def _aggregate_option_results( except Exception as e: self.logger.warning(f"Failed to parse {result_file}: {e}") missing_files.append(str(result_file)) - all_read_bw.append(sum(trial_read_bw)) - all_write_bw.append(sum(trial_write_bw)) - all_avg_throughput.append(sum(trial_avg_throughput)) - all_storage_throughput.append(sum(trial_storage_throughput)) - all_p95_latency.append(max(trial_p95_latency) if trial_p95_latency else 0.0) + # A trial that produced no rank data (every rank file missing or + # unparseable) must not contribute a sum([])==0.0 into the + # across-trial aggregate — that 0.0 is a total-loss sentinel, not + # a measured value, and reportgen surfaces it verbatim (P5). The + # five per-rank lists are populated together, so any one being + # non-empty means the trial had at least one parsed rank; a + # present-rank 0.0 (CPU tier) is legitimate and still counted. + if trial_read_bw: + all_read_bw.append(sum(trial_read_bw)) + all_write_bw.append(sum(trial_write_bw)) + all_avg_throughput.append(sum(trial_avg_throughput)) + all_storage_throughput.append(sum(trial_storage_throughput)) + all_p95_latency.append(max(trial_p95_latency)) if missing_files: hosts = getattr(self.args, 'hosts', None) or [] multi_host = any(not _is_localhost(h.split(':')[0]) for h in hosts) @@ -879,10 +887,12 @@ def _aggregate_option_results( ) return { 'option': option, - 'aggregated_read_bandwidth_gbps': fmean(all_read_bw) if all_read_bw else 0.0, - 'aggregated_write_bandwidth_gbps': fmean(all_write_bw) if all_write_bw else 0.0, - 'aggregated_avg_throughput_tokens_per_sec': fmean(all_avg_throughput) if all_avg_throughput else 0.0, - 'aggregated_storage_throughput_tokens_per_sec': fmean(all_storage_throughput) if all_storage_throughput else 0.0, + # Total loss across every trial -> None (blank), never a fake 0.0 + # (P5). aggregated_p95_latency_ms already did this. + 'aggregated_read_bandwidth_gbps': fmean(all_read_bw) if all_read_bw else None, + 'aggregated_write_bandwidth_gbps': fmean(all_write_bw) if all_write_bw else None, + 'aggregated_avg_throughput_tokens_per_sec': fmean(all_avg_throughput) if all_avg_throughput else None, + 'aggregated_storage_throughput_tokens_per_sec': fmean(all_storage_throughput) if all_storage_throughput else None, 'aggregated_p95_latency_ms': max(all_p95_latency) if all_p95_latency else None, 'rank_count': expected_rank_count, 'trial_count': len(trial_dirs), diff --git a/mlpstorage_py/report_generator.py b/mlpstorage_py/report_generator.py index 78e5b685..11a70a9f 100755 --- a/mlpstorage_py/report_generator.py +++ b/mlpstorage_py/report_generator.py @@ -1014,10 +1014,10 @@ def _process_single_run(self, benchmark_run: BenchmarkRun) -> None: # ------------------------------------------------------------------ # # ``_aggregate_workload_metrics`` is the single point where per-workload - # aggregation math lives. It is added here (Plan 06-02) but not yet - # wired into ``_process_workload_groups``; wiring happens in Plan 06-04 - # (the TODO at ``metrics={}`` in this file's ``_process_workload_groups`` - # is the eventual call site). + # aggregation math lives. It is wired into the reporting pass at the + # ``aggregated_metrics = self._aggregate_workload_metrics(...)`` call in + # ``accumulate_results`` (the ``metrics={}`` in ``_model_group_folder``'s + # proxy Result is a throwaway for folder derivation, not this call site). # # Dispatch on ``runs[0].benchmark_type``: # - training -> ``_aggregate_training`` (D-19, 5-run mean) @@ -1040,8 +1040,8 @@ def _aggregate_workload_metrics( Compute aggregated metrics for one workload (per D-19..D-22). Dispatches on ``runs[0].benchmark_type`` and returns the aggregated - metric dict that eventually populates the ``metrics=`` slot on the - workload ``Result`` (see the TODO at ``_process_workload_groups``). + metric dict that populates the ``metrics=`` slot on the workload + ``Result`` (wired in ``accumulate_results``). Args: runs: The workload's ``BenchmarkRun`` invocations. For training, @@ -1330,9 +1330,10 @@ def _aggregate_vdb( VDB-branch aggregation (D-21) — pass-through, NOT math. vdb's internal ``vdb-aggregate`` tool (see - ``benchmarks/vectordbbench.py:508``) owns the math contract per - the D-22 boundary; this helper copies pre-computed values from - the workload's ``summary.json`` into ``vdb_*`` columns. + ``VectorDBBenchmark._run_aggregate`` in ``benchmarks/vectordbbench.py``) + owns the math contract per the D-22 boundary; this helper copies + pre-computed values from the workload's ``summary.json`` into + ``vdb_*`` columns. Read fields (per ``submission_checker/checks/vdb_checks.py:44-51`` ``_REQUIRED_METRIC_FIELDS``): ``throughput_qps``, diff --git a/mlpstorage_py/rules/models.py b/mlpstorage_py/rules/models.py index cba80028..6bff60fe 100755 --- a/mlpstorage_py/rules/models.py +++ b/mlpstorage_py/rules/models.py @@ -768,6 +768,42 @@ def extract(benchmark) -> BenchmarkRunData: ) +def _filter_list_metrics(metric_block, logger=None, context: str = ""): + """Filter a DLIO summary ``metric`` block to list-valued keys only. + + The report-generator aggregation paths call ``fmean()`` over each metric + value, so scalar/string entries (``train_au_mean_percentage``, + ``train_au_meet_expectation``, ...) must be dropped or ``fmean`` raises + ``TypeError`` on a non-iterable. Returns the filtered dict, or ``None`` if + ``metric_block`` is not a dict or nothing survives the filter. + + A diagnostic is logged (never silent) both when non-list keys are dropped + and when the block collapses to nothing, so a blank ``metrics`` column is + always traceable to a logged reason rather than disappearing quietly. + """ + if not isinstance(metric_block, dict): + return None + + filtered = {k: v for k, v in metric_block.items() if isinstance(v, list)} + dropped = [k for k, v in metric_block.items() if not isinstance(v, list)] + + if dropped and logger: + logger.warning( + f"{context}dropped {len(dropped)} non-list metric key(s) not " + f"usable for aggregation: {sorted(dropped)}" + ) + + if not filtered: + if logger: + logger.warning( + f"{context}no list-valued metrics remain after filtering; " + f"metrics will be blank for this run" + ) + return None + + return filtered + + class DLIOResultParser: """Parses DLIO benchmark result files into BenchmarkRunData.""" @@ -843,7 +879,12 @@ def parse(self, result_dir: str, metadata: Optional[Dict] = None) -> BenchmarkRu parameters=hydra_workload_config, override_parameters=override_parameters, system_info=system_info, - metrics=summary.get("metric"), + # Filter to list-valued keys (matching the metadata-complete + # path) so report_generator's fmean(metric_list) never receives + # a scalar; dropped/blanked keys are logged, not silent. + metrics=_filter_list_metrics( + summary.get("metric"), self.logger, context=f"{result_dir}: " + ), result_dir=result_dir, accelerator=accelerator, run_args=(metadata or {}).get('args', {}) or {}, @@ -891,7 +932,7 @@ def extract(self, result_dir: str, logger=None) -> BenchmarkRunData: metadata = self._load_metadata(result_dir, logger) if metadata and self._is_complete_metadata(metadata): - return self._from_metadata(metadata, result_dir) + return self._from_metadata(metadata, result_dir, logger=logger) if self.result_parser is None: self.result_parser = DLIOResultParser(logger=logger) @@ -916,7 +957,7 @@ def _is_complete_metadata(self, metadata: Dict) -> bool: required_fields = ['benchmark_type', 'run_datetime', 'num_processes', 'parameters'] return all(field in metadata for field in required_fields) - def _from_metadata(self, metadata: Dict, result_dir: str) -> BenchmarkRunData: + def _from_metadata(self, metadata: Dict, result_dir: str, logger=None) -> BenchmarkRunData: """Create BenchmarkRunData from a complete metadata dict.""" benchmark_type_str = metadata.get('benchmark_type', '') benchmark_type = None @@ -944,16 +985,14 @@ def _from_metadata(self, metadata: Dict, result_dir: str) -> BenchmarkRunData: if not end_datetime: end_datetime = summary.get('end_time') or summary.get('end', '') if not run_metrics: - metric_block = summary.get('metric') - if isinstance(metric_block, dict): - # Filter to list-valued keys so _aggregate_training's - # fmean(metric_list) doesn't blow up on scalars like - # train_au_mean_percentage or strings like - # train_au_meet_expectation. - run_metrics = { - k: v for k, v in metric_block.items() - if isinstance(v, list) - } or None + # Filter to list-valued keys so _aggregate_training's + # fmean(metric_list) doesn't blow up on scalars like + # train_au_mean_percentage or strings like + # train_au_meet_expectation. Dropped/blanked keys are + # logged rather than silently discarded. + run_metrics = _filter_list_metrics( + summary.get('metric'), logger, context=f"{result_dir}: " + ) return BenchmarkRunData( benchmark_type=benchmark_type, diff --git a/mlpstorage_py/submission_checker/checks/checkpointing_checks.py b/mlpstorage_py/submission_checker/checks/checkpointing_checks.py index 04959c20..9824213f 100644 --- a/mlpstorage_py/submission_checker/checks/checkpointing_checks.py +++ b/mlpstorage_py/submission_checker/checks/checkpointing_checks.py @@ -291,15 +291,19 @@ def closed_accelerators_per_host(self): @rule("4.3.4", "checkpointAggregateAcceleratorMemory") def aggregate_accelerator_memory(self): """ - Verify total accelerator memory >= checkpoint size. - H100 has 80GB per accelerator. + Verify total accelerator memory >= checkpoint size (warnings-only). + + The per-accelerator memory baseline is the H100's 80 GiB; checkpointing + metadata carries no accelerator-type field, so this is advisory. A + shortfall is surfaced to reviewers via ``warn_violation`` and never + flips a submission INVALID (late-window doctrine). Always returns True. (Rules.md 4.3.4) """ valid = True if self.mode != "checkpointing": return valid - ACCELERATOR_MEMORY_GB = 80 # H100 + ACCELERATOR_MEMORY_GB = 80 # H100 baseline (advisory) for summary, metadata, _ in self._iter_valid_files(): checkpoint_size_gb = summary.get("metric", {}).get("checkpoint_size_GB", 0) @@ -308,13 +312,15 @@ def aggregate_accelerator_memory(self): total_accelerator_memory = num_accelerators * ACCELERATOR_MEMORY_GB if total_accelerator_memory < checkpoint_size_gb: - self.log_violation( + self.warn_violation( "4.3.4", "checkpointAggregateAcceleratorMemory", self.path, - "aggregate accelerator memory %.2fGiB < checkpoint size %.2fGiB", + "aggregate accelerator memory %.2fGiB (%d accelerators x 80 GiB " + "H100 baseline) < checkpoint size %.2fGiB; verify against the " + "actual accelerator's memory (Rules.md 4.3.4).", total_accelerator_memory, + num_accelerators, checkpoint_size_gb, ) - valid = False return valid diff --git a/mlpstorage_py/submission_checker/parsers/json_parser.py b/mlpstorage_py/submission_checker/parsers/json_parser.py index 466c208c..0e89c0fd 100644 --- a/mlpstorage_py/submission_checker/parsers/json_parser.py +++ b/mlpstorage_py/submission_checker/parsers/json_parser.py @@ -49,7 +49,7 @@ def get_keys(self): return self.keys def __contains__(self, key): - return key in self.messages + return key in self.d def __repr__(self): return f"" diff --git a/mlpstorage_py/tests/conftest.py b/mlpstorage_py/tests/conftest.py index 327862ca..72202774 100644 --- a/mlpstorage_py/tests/conftest.py +++ b/mlpstorage_py/tests/conftest.py @@ -475,6 +475,7 @@ def build_submission(tmp_path, **overrides) -> Path: chkpt_checkpoint_folder = overrides.pop("chkpt_checkpoint_folder", None) chkpt_results_dir = overrides.pop("chkpt_results_dir", None) chkpt_summary_checkpoint_size_GB = overrides.pop("chkpt_summary_checkpoint_size_GB", None) + chkpt_summary_num_accelerators = overrides.pop("chkpt_summary_num_accelerators", None) run_data_dir = overrides.pop("run_data_dir", None) run_results_dir = overrides.pop("run_results_dir", None) @@ -818,6 +819,8 @@ def build_submission(tmp_path, **overrides) -> Path: metric = dict(chkpt_summary.get("metric", {})) metric["checkpoint_size_GB"] = chkpt_summary_checkpoint_size_GB chkpt_summary["metric"] = metric + if chkpt_summary_num_accelerators is not None: + chkpt_summary["num_accelerators"] = chkpt_summary_num_accelerators (ts_dir / "summary.json").write_text( json.dumps(chkpt_summary), encoding="utf-8" diff --git a/mlpstorage_py/tests/test_checkpointing_check_retrofit.py b/mlpstorage_py/tests/test_checkpointing_check_retrofit.py index 791be72a..bcf8892c 100644 --- a/mlpstorage_py/tests/test_checkpointing_check_retrofit.py +++ b/mlpstorage_py/tests/test_checkpointing_check_retrofit.py @@ -136,6 +136,52 @@ def test_4_3_1_undersized_checkpoint_emits_warn_prefix_not_error(tmp_path, mock_ ) +# --------------------------------------------------------------------------- +# 4.3.4 aggregate_accelerator_memory — warnings-only (C2) +# --------------------------------------------------------------------------- + +def test_4_3_4_insufficient_memory_warns_not_errors(tmp_path, mock_logger): + """4.3.4: aggregate accelerator memory < checkpoint size → WARNING, never fail. + + Late-stage submission-window doctrine: this check must not flip a + submission INVALID. With 8 accelerators (× the 80 GiB baseline = 640 GiB) + and a 1000 GiB checkpoint, the aggregate is insufficient, so the rule + warns via warn_violation and still returns True. + """ + from mlpstorage_py.tests.conftest import build_submission + root = build_submission( + tmp_path, + chkpt_summary_num_accelerators=8, + chkpt_summary_checkpoint_size_GB=1000, + ) + check = _run_checkpointing_check(root, mock_logger) + result = check.aggregate_accelerator_memory() + assert result is True + assert any( + m.startswith("[4.3.4 checkpointAggregateAcceleratorMemory]") + for m in mock_logger.warnings + ), f"expected [4.3.4 ...] warning; got warnings={mock_logger.warnings}" + assert not any( + m.startswith("[4.3.4 ") for m in mock_logger.errors + ), f"4.3.4 must warn, not error; got errors={mock_logger.errors}" + + +def test_4_3_4_sufficient_memory_no_warning(tmp_path, mock_logger): + """Aggregate accelerator memory >= checkpoint size → no warning, returns True.""" + from mlpstorage_py.tests.conftest import build_submission + root = build_submission( + tmp_path, + chkpt_summary_num_accelerators=8, + chkpt_summary_checkpoint_size_GB=100, # 8*80=640 GiB >= 100 + ) + check = _run_checkpointing_check(root, mock_logger) + result = check.aggregate_accelerator_memory() + assert result is True + assert not any( + m.startswith("[4.3.4 ") for m in mock_logger.warnings + mock_logger.errors + ), f"expected no 4.3.4 violation; got {mock_logger.warnings + mock_logger.errors}" + + # --------------------------------------------------------------------------- # Behavior-preservation: 4.4.1 checkpointPathArgs # --------------------------------------------------------------------------- diff --git a/mlpstorage_py/tests/test_json_parser.py b/mlpstorage_py/tests/test_json_parser.py new file mode 100644 index 00000000..5291026c --- /dev/null +++ b/mlpstorage_py/tests/test_json_parser.py @@ -0,0 +1,31 @@ +"""Unit tests for the submission-checker JSONParser. + +Covers the ``in`` membership operator (``__contains__``), which referenced a +non-existent ``self.messages`` attribute and raised ``AttributeError`` for any +``key in parser`` test — a latent bug (callers use ``[]`` / ``.get()``). +""" + +import json + +from mlpstorage_py.submission_checker.parsers.json_parser import JSONParser + + +def _parser(tmp_path, payload): + path = tmp_path / "summary.json" + path.write_text(json.dumps(payload), encoding="utf-8") + return JSONParser(str(path)) + + +def test_contains_top_level_key(tmp_path): + """``key in parser`` is True for a present top-level key, False otherwise.""" + parser = _parser(tmp_path, {"alpha": 1, "beta": 2}) + assert "alpha" in parser + assert "beta" in parser + assert "missing" not in parser + + +def test_contains_on_normalized_non_dict(tmp_path): + """A non-dict JSON body is normalized under ``summary`` and is found via ``in``.""" + parser = _parser(tmp_path, [1, 2, 3]) + assert "summary" in parser + assert "alpha" not in parser diff --git a/tests/unit/test_benchmarks_kvcache.py b/tests/unit/test_benchmarks_kvcache.py index ece26586..7b191edb 100755 --- a/tests/unit/test_benchmarks_kvcache.py +++ b/tests/unit/test_benchmarks_kvcache.py @@ -620,7 +620,9 @@ def test_uses_glob_not_constructed_filename(self, tmp_path): assert result['partial_failure'] is False def test_none_p95_when_no_successful_reads(self, tmp_path): - """aggregated_p95_latency_ms is 0.0 when all rank files are missing.""" + """P5: aggregated_p95_latency_ms is None when all rank files are + missing (total data loss) — a trial that produced no rank data must + not contribute a fake 0.0.""" bm = _make_run_benchmark(tmp_path) trial_dir = tmp_path / 'trial_0' # Both rank dirs exist but have no json files @@ -629,8 +631,48 @@ def test_none_p95_when_no_successful_reads(self, tmp_path): result = bm._aggregate_option_results(1, [str(trial_dir)], expected_rank_count=2) - # Empty trial contributes 0.0; fmean([0.0]) = 0.0 - assert result['aggregated_p95_latency_ms'] == pytest.approx(0.0) + # Total loss -> None (blank), not a misleading measured 0.0. + assert result['aggregated_p95_latency_ms'] is None + assert result['partial_failure'] is True + + def test_total_data_loss_emits_none_not_zero(self, tmp_path): + """P5: when NO rank produced data across all trials (total loss), + every aggregated bandwidth/throughput field must be None (blank), + not a misleading measured 0.0 that reportgen surfaces verbatim.""" + bm = _make_run_benchmark(tmp_path) + trial_dir = tmp_path / 'trial_0' + # Rank dirs exist but carry no result JSON -> nothing parsed. + (trial_dir / 'rank_0').mkdir(parents=True) + (trial_dir / 'rank_1').mkdir(parents=True) + + result = bm._aggregate_option_results(1, [str(trial_dir)], expected_rank_count=2) + + assert result['aggregated_read_bandwidth_gbps'] is None + assert result['aggregated_write_bandwidth_gbps'] is None + assert result['aggregated_avg_throughput_tokens_per_sec'] is None + assert result['aggregated_storage_throughput_tokens_per_sec'] is None + assert result['aggregated_p95_latency_ms'] is None + assert result['partial_failure'] is True + + def test_empty_trial_not_diluted_into_mean(self, tmp_path): + """P5: a trial where every rank is missing must not fold a 0.0 into + the across-trial mean — the mean is taken over the trials that + actually produced data.""" + bm = _make_run_benchmark(tmp_path) + good = tmp_path / 'trial_0' + self._make_rank_file(good / 'rank_0', bw=3.0, p95=10.0) + self._make_rank_file(good / 'rank_1', bw=1.0, p95=12.0) # trial sum = 4.0 + empty = tmp_path / 'trial_1' + (empty / 'rank_0').mkdir(parents=True) + (empty / 'rank_1').mkdir(parents=True) + + result = bm._aggregate_option_results( + 1, [str(good), str(empty)], expected_rank_count=2 + ) + + # Old behavior folded the empty trial's 0.0 in: fmean([4.0, 0.0]) = 2.0. + # Correct: mean over the single trial that had data = 4.0. + assert result['aggregated_read_bandwidth_gbps'] == pytest.approx(4.0) assert result['partial_failure'] is True diff --git a/tests/unit/test_rules_extractors.py b/tests/unit/test_rules_extractors.py index 717fa429..689af100 100755 --- a/tests/unit/test_rules_extractors.py +++ b/tests/unit/test_rules_extractors.py @@ -346,6 +346,59 @@ def test_parse_extracts_accelerator(self, mock_logger, training_result_dir): # From workload=unet3d_h100, should extract h100 assert result.accelerator == "h100" + def test_parse_filters_non_list_metrics_with_diagnostic( + self, mock_logger, tmp_path, + ): + """P6b: the DLIOResultParser fallback path stored the summary + `metric` block UNFILTERED, so scalar/string entries could reach + report_generator's `fmean(metric_list)` and raise TypeError. The + parse path must filter to list-valued keys — like the + metadata-complete path — and log a diagnostic rather than dropping + the scalars silently. + """ + import yaml + + result_dir = tmp_path / "training_run" + result_dir.mkdir() + summary = { + "start": "2026-07-01 10:00:00", + "end": "2026-07-01 10:15:00", + "num_accelerators": 8, + "workload": "unet3d_h100", + "metric": { + "train_au_percentage": [96.1, 95.8, 96.4], + "train_throughput_samples_per_second": [128.0, 130.0], + "train_au_mean_percentage": 96.1, # scalar + "train_au_meet_expectation": "success", # string + }, + } + with open(result_dir / "summary.json", 'w') as f: + json.dump(summary, f) + + hydra_dir = result_dir / "dlio_config" + hydra_dir.mkdir() + with open(hydra_dir / "config.yaml", 'w') as f: + yaml.dump({"workload": { + "model": {"name": "unet3d"}, + "workflow": {"generate_data": False, "train": True, "checkpoint": False}, + }}, f) + with open(hydra_dir / "overrides.yaml", 'w') as f: + yaml.dump(["workload=unet3d_h100"], f) + + parser = DLIOResultParser(logger=mock_logger) + result = parser.parse(str(result_dir)) + + assert result.metrics is not None + assert "train_au_percentage" in result.metrics + assert "train_throughput_samples_per_second" in result.metrics + # Non-list entries must be filtered so fmean() never sees a scalar. + assert "train_au_mean_percentage" not in result.metrics + assert "train_au_meet_expectation" not in result.metrics + # Dropped keys are surfaced, not silently discarded. + assert mock_logger.warning.called, ( + "dropped non-list metric keys must emit a diagnostic" + ) + class TestResultFilesExtractor: """Tests for ResultFilesExtractor class.""" @@ -628,6 +681,48 @@ def test_from_metadata_prefers_explicit_metrics_over_summary( assert result.metrics == {"checkpoint_save_time": [1.2, 1.3, 1.1]} + def test_extract_logs_when_metrics_collapse_to_none( + self, mock_logger, tmp_path, + ): + """P6a: when the summary `metric` fallback contains NO list-valued + keys, metrics collapses to None. That must not happen silently — a + diagnostic is logged so a blank metrics column is traceable to a + reason. Exercised via the public `extract()`, which threads the + logger down to the metadata-complete path. + """ + result_dir = tmp_path / "training_run" + result_dir.mkdir() + + metadata = { + "benchmark_type": "training", + "model": "unet3d", + "command": "run", + "run_datetime": "20260701_100000", + "num_processes": 8, + "parameters": {}, + } + with open(result_dir / "training_20260701_100000_metadata.json", 'w') as f: + json.dump(metadata, f) + + # All-scalar / all-string metric block: nothing survives the + # list-only filter, so metrics becomes None. + summary = { + "metric": { + "train_au_mean_percentage": 96.1, + "train_au_meet_expectation": "success", + }, + } + with open(result_dir / "summary.json", 'w') as f: + json.dump(summary, f) + + extractor = ResultFilesExtractor() + result = extractor.extract(str(result_dir), logger=mock_logger) + + assert result.metrics is None + assert mock_logger.warning.called, ( + "collapsing metrics to None must emit a diagnostic, not blank silently" + ) + class TestExtractorIntegration: """Integration tests using fixture data from tests/fixtures."""