diff --git a/Rules.md b/Rules.md index 23d535da..0990de88 100644 --- a/Rules.md +++ b/Rules.md @@ -516,11 +516,11 @@ root_folder (or any name you prefer) **NOTE: In the ``--ppn`` syntax above, the ``slotcount`` value means the number of processes per node to run.** -**\*\* NOTE: In CLOSED submissions, ``--num-checkpoints-write`` and ``--num-checkpoints-read`` may be set to ``0`` only as part of the two-invocation cache-flush workflow described in §4.7.1: one invocation runs the write phase with ``--num-checkpoints-read=0`` and the next runs the read phase with ``--num-checkpoints-write=0``. The default for both flags is 10 and the total work performed across both invocations must still be 10 writes followed by 10 reads.** +**\*\* NOTE: In CLOSED submissions, ``--num-checkpoints-write`` and ``--num-checkpoints-read`` may be set to ``0`` only as part of the two-invocation failover workflow described in §4.7.1: one invocation runs the write phase with ``--num-checkpoints-read=0`` and the next runs the read phase with ``--num-checkpoints-write=0``. The default for both flags is 10 and the total work performed across both invocations must still be 10 writes followed by 10 reads.** ## 4.7. Storage System Must Be Simultaneously R/W or _Remappable_ -4.7.1. **checkpointCacheFlushValidation** -- A cache flush between the write and read phases is only required when the client node has enough memory to cache all of the checkpoints written by that client during the run. The benchmark writes 10 sequential checkpoints specifically to overfill typical filesystem caches; on most submission configurations the early checkpoints have already been evicted by the time the read phase begins, so no flush is required. As a rule of thumb (see `checkpointing/README.md`), a flush is required when the total checkpoint size written per client is less than 3× the client node's memory capacity. When a flush is required, the submitter must execute the run in two invocations: the write phase with ``--num-checkpoints-read=0``, followed by the cache flush during a pause of no more than 30 seconds, then the read phase with ``--num-checkpoints-write=0``. The validator must confirm this split occurred and that the inter-phase gap did not exceed 30 seconds. +4.7.1. **checkpointCacheFlushValidation** -- Checkpointing models the failure of a client node followed by another client picking up the last checkpoint file written by the failed node for the read phase. In every submission the write phase (10 checkpoint files written) runs first, followed by the read phase (10 checkpoint files read). When the storage system supports the client-to-client handoff transparently — i.e., the read phase can proceed immediately after the write phase without external orchestration — the write and read phases may be executed as a single combined invocation, and no gap check applies. Storage system architectures that require an external callout (e.g., a submitter-provided script) to complete the failover between the writing and reading clients must instead execute the write and read phases as two separate invocations, with the submitter's failover callout occurring between them. A common in-callout activity is clearing a client-side filesystem cache when the total checkpoint size written per client is less than 3× the client node's memory capacity (see ``checkpointing/README.md``); the callout is not limited to that activity. To ensure the callout is a lightweight programmatic step rather than a long-running manual procedure, the validator confirms that the read-phase invocation was launched — i.e., its ``mlpstorage`` process reached the entry point of ``main.py`` — within 30 seconds of the write-phase invocation ending. The gap is measured as ``read.invocation_start_time − write.summary.end_time``, where ``invocation_start_time`` is captured at ``mlpstorage`` process start (before framework startup, MPI spawn, and other unavoidable per-invocation overhead) and ``end_time`` is recorded in the write invocation's ``summary.json``. A negative gap indicates clock skew between the write and read nodes and is reported as such rather than as a causality violation. 4.7.2. **checkpointTotalTestDuration** -- The validator must verify that the total test duration starts from the timestamp of the first checkpoint written and ends at the ending timestamp of the last checkpoint read, notably including the "remapping" time. diff --git a/mlpstorage_py/_invocation.py b/mlpstorage_py/_invocation.py new file mode 100644 index 00000000..bc181b3d --- /dev/null +++ b/mlpstorage_py/_invocation.py @@ -0,0 +1,14 @@ +"""Earliest wall-clock instant of the mlpstorage process. + +Captured at first import of this module. ``main.py`` imports this before +any heavier ``mlpstorage_py`` submodule so the timestamp is recorded +before framework startup (Python imports, MPI spawn, CAP env-validation, +etc.). ``benchmarks/base.py`` reads the value when populating +``invocation_start_time`` in the run's metadata; the submission checker +uses that field as the read-side origin of the §4.7.1 gap check so +per-invocation framework overhead is not charged against the 30-second +failover-callout budget. +""" +import time + +INVOCATION_START: float = time.time() diff --git a/mlpstorage_py/benchmarks/base.py b/mlpstorage_py/benchmarks/base.py index 88fa57cb..356cdd60 100755 --- a/mlpstorage_py/benchmarks/base.py +++ b/mlpstorage_py/benchmarks/base.py @@ -42,10 +42,12 @@ def _run(self): import types import uuid from argparse import Namespace +from datetime import datetime from typing import Tuple, Dict, Any, List, Optional, Callable, Set, TYPE_CHECKING from functools import wraps +from mlpstorage_py._invocation import INVOCATION_START from mlpstorage_py.config import PARAM_VALIDATION, DATETIME_STR, MLPS_DEBUG, EXEC_TYPE from mlpstorage_py.errors import ConfigurationError, ErrorCode, FileSystemError from mlpstorage_py.run_directory import ( @@ -422,6 +424,9 @@ def metadata(self) -> Dict[str, Any]: # Additional context (not part of BenchmarkRunData but useful) metadata['runtime'] = self.runtime + # §4.7.1 gap origin — ISO local naive to match DLIO summary.json's + # start/end format so _parse_iso_gap can consume both fields uniformly. + metadata['invocation_start_time'] = datetime.fromtimestamp(INVOCATION_START).isoformat() metadata['verification'] = self.verification.name if self.verification else None metadata['executed_command'] = getattr(self, 'executed_command', None) metadata['command_output_files'] = self.command_output_files diff --git a/mlpstorage_py/main.py b/mlpstorage_py/main.py index 67f7b50c..df0190f9 100755 --- a/mlpstorage_py/main.py +++ b/mlpstorage_py/main.py @@ -14,6 +14,11 @@ import sys import traceback +# §4.7.1 gap origin — must precede all heavier mlpstorage_py imports so the +# timestamp is captured before framework startup (Python imports, MPI spawn, +# CAP env-validation, etc.). See mlpstorage_py/_invocation.py. +from mlpstorage_py._invocation import INVOCATION_START # noqa: F401 + from mlpstorage_py.cli_parser import parse_arguments, validate_args, update_args from mlpstorage_py.config import HISTFILE, DATETIME_STR, EXIT_CODE, DEFAULT_RESULTS_DIR, get_datetime_string, HYDRA_OUTPUT_SUBDIR from mlpstorage_py.debug import debugger_hook, MLPS_DEBUG diff --git a/mlpstorage_py/submission_checker/checks/checkpointing_checks.py b/mlpstorage_py/submission_checker/checks/checkpointing_checks.py index 9ef5f26b..480fe707 100644 --- a/mlpstorage_py/submission_checker/checks/checkpointing_checks.py +++ b/mlpstorage_py/submission_checker/checks/checkpointing_checks.py @@ -575,18 +575,36 @@ def open_mpi_processes(self): @rule("4.7.1", "checkpointCacheFlushValidation") def cache_flush_validation(self): - """Detect cache-flush pattern in split-mode runs (write then read with <=30s gap). + """Verify §4.7.1 failover-callout gap in split-mode runs (<=30s). (Rules.md 4.7.1) - Combined-mode submissions (write and read in the same run) are valid per - Pitfall 15 — return True with no violation when no split-mode pairs exist. - Reads timestamps from summary.json (tries 'end_time'/'start_time' first, - then 'end'/'start' fallback per RESEARCH.md Gray Area 3). - - 30-second threshold is strict per Rules.md 4.7.1; no slop applied (Gemini - review concern #3 noted; revisit with CACHE_FLUSH_SLOP_SECONDS constant - in constants.py if filesystem-metadata latency becomes empirically - problematic). + Combined-mode submissions (single invocation with both writes and + reads) are valid — return True with no violation when no split-mode + pairs exist. + + The gap is measured as + ``read.metadata.invocation_start_time − write.summary.end_time`` so + that per-invocation framework startup on the read side (Python + imports, MPI spawn, CAP env-validation — up to ~50s in the wild) is + not charged against the 30-second callout budget. See Rules.md 4.7.1 + for the semantic and mlpstorage_py/_invocation.py for the capture. + + **Backward compat** — results directories produced by an mlpstorage + version predating this rule do not carry ``invocation_start_time`` + in the read-phase metadata. In that case the check falls back to the + legacy origin (``read.summary.start_time`` / ``start``), which + conflates the failover callout with per-invocation framework startup + and so is not authoritative. To honor the rule that was in force + when those results were produced, a gap breach measured against the + legacy origin is emitted as a warning (``warn_violation``) rather + than a hard failure. The submitter can regenerate with a current + mlpstorage to get the honest measurement. + + A negative gap indicates NTP skew between the write and read nodes + (the metadata is timestamped on the read-invocation host, the summary + on the write-invocation host); it is reported as such rather than as + an ordering / causality violation (which is the domain of + ``checkpoint_invocation_structure``). """ valid = True if self.mode != "checkpointing": @@ -596,14 +614,33 @@ def cache_flush_validation(self): return valid for write_entry, read_entry in pairs: write_summary, _, write_ts = write_entry - read_summary, _, read_ts = read_entry + read_summary, read_metadata, read_ts = read_entry write_end = (write_summary or {}).get("end_time") or (write_summary or {}).get("end") - read_start = (read_summary or {}).get("start_time") or (read_summary or {}).get("start") - if write_end is None or read_start is None: + read_start = (read_metadata or {}).get("invocation_start_time") + used_legacy_origin = False + if read_start is None: + # Backward-compat fallback for results dirs that predate the + # §4.7.1 fix and don't carry invocation_start_time. Measure + # against read.summary.start_time and downgrade any breach + # to warn_violation below. + read_start = (read_summary or {}).get("start_time") or (read_summary or {}).get("start") + used_legacy_origin = True + if write_end is None: + self.log_violation( + "4.7.1", "checkpointCacheFlushValidation", self.path, + "cannot compute failover-callout gap: missing end_time in " + "write-phase summary.json (write_ts=%s, read_ts=%s)", + write_ts, read_ts, + ) + valid = False + continue + if read_start is None: self.log_violation( "4.7.1", "checkpointCacheFlushValidation", self.path, - "cannot compute cache-flush gap: missing end_time/start_time in summary.json " - "(write_ts=%s, read_ts=%s)", write_ts, read_ts, + "cannot compute failover-callout gap: read-phase metadata " + "has no invocation_start_time and read-phase summary.json " + "has no start_time/start (write_ts=%s, read_ts=%s)", + write_ts, read_ts, ) valid = False continue @@ -617,11 +654,43 @@ def cache_flush_validation(self): ) valid = False continue + if gap_seconds < 0: + # Per Rules.md 4.7.1: negative gap → clock-skew warning, not + # a causality violation. Ordering / no-overlap is enforced + # separately by checkpoint_invocation_structure. + self.warn_violation( + "4.7.1", "checkpointCacheFlushValidation", self.path, + "failover-callout gap is negative (%.1fs); likely NTP " + "clock skew between write host and read host " + "(write end=%s, read start=%s)", + gap_seconds, write_end, read_start, + ) + continue if gap_seconds > 30: + if used_legacy_origin: + # Pre-fix results dir: honor the rule that was in force + # when the run was produced by warning rather than + # failing. The gap measurement here includes per- + # invocation framework startup and is not authoritative + # against the new (invocation_start_time-based) rule. + self.warn_violation( + "4.7.1", "checkpointCacheFlushValidation", self.path, + "failover-callout gap %.1f seconds exceeds 30-second " + "limit, measured against read.summary.start_time " + "because invocation_start_time is not present in " + "read-phase metadata (results dir predates the " + "§4.7.1 gap-origin fix). Gap includes per-invocation " + "framework startup so this reading is not " + "authoritative — re-run with a current mlpstorage " + "for the honest measurement. " + "(write end=%s, read start=%s)", + gap_seconds, write_end, read_start, + ) + continue self.log_violation( "4.7.1", "checkpointCacheFlushValidation", self.path, - "cache-flush gap %.1f seconds exceeds 30-second limit " - "(write end=%s, read start=%s)", + "failover-callout gap %.1f seconds exceeds 30-second limit " + "(write end=%s, read invocation_start=%s)", gap_seconds, write_end, read_start, ) valid = False diff --git a/mlpstorage_py/tests/conftest.py b/mlpstorage_py/tests/conftest.py index 662f6903..9d45c05f 100644 --- a/mlpstorage_py/tests/conftest.py +++ b/mlpstorage_py/tests/conftest.py @@ -331,7 +331,19 @@ def build_submission(tmp_path, **overrides) -> Path: * ``chkpt_cache_flush_gap_seconds`` (int, default 25) — CHKPT-02/04: when ``chkpt_split_mode=True`` AND ``chkpt_summary_timestamps=True``, sets the gap between a write run's ``end`` and the following read run's ``start`` to - this many seconds. + this many seconds. Also sets the read metadata's ``invocation_start_time`` + to write ``end`` + this many seconds — since Rules.md §4.7.1 §4.7.1 measures + the failover-callout gap from write.summary.end_time to + read.metadata.invocation_start_time, this is the value that + ``cache_flush_validation`` sees. May be negative to simulate NTP clock skew + between the write and read hosts. + * ``chkpt_omit_invocation_start_time`` (bool, default False) — CHKPT-02: + when True, do NOT emit ``invocation_start_time`` in read-side metadata + (simulates a results dir produced by an mlpstorage version predating + the §4.7.1 gap-origin fix). Exercises the backward-compat fallback + path in ``cache_flush_validation`` — the check falls back to + ``read.summary.start_time`` and downgrades a 30-second breach from + hard failure to warning. * ``chkpt_model`` (str, default "llama3-8b") — CHKPT-01: model directory name under ``checkpointing/``. * ``chkpt_open_num_processes`` (int | None) — CHKPT-01: when non-None, sets @@ -400,6 +412,7 @@ def build_submission(tmp_path, **overrides) -> Path: chkpt_summary_timestamps = overrides.pop("chkpt_summary_timestamps", False) chkpt_split_mode = overrides.pop("chkpt_split_mode", False) chkpt_cache_flush_gap_seconds = overrides.pop("chkpt_cache_flush_gap_seconds", 25) + chkpt_omit_invocation_start_time = overrides.pop("chkpt_omit_invocation_start_time", False) chkpt_model = overrides.pop("chkpt_model", "llama3-8b") chkpt_open_num_processes = overrides.pop("chkpt_open_num_processes", None) chkpt_closed_num_processes = overrides.pop("chkpt_closed_num_processes", None) @@ -743,6 +756,37 @@ def build_submission(tmp_path, **overrides) -> Path: chkpt_meta["args"]["num_checkpoints_write"] = 0 chkpt_meta["args"]["num_checkpoints_read"] = 10 + # §4.7.1: metadata.invocation_start_time is the read-side origin + # of the failover-callout gap check (see checkpointing_checks.py + # cache_flush_validation). Emit unconditionally to mirror real + # mlpstorage runs, except when the test explicitly opts out to + # exercise the missing-field branch. + if chkpt_summary_timestamps and not ( + chkpt_omit_invocation_start_time + and chkpt_split_mode + and ts in read_timestamps + ): + if chkpt_split_mode: + if ts in write_timestamps: + # Write invocation entered mlpstorage right before write_start. + wi = write_timestamps.index(ts) + _write_start = _BASE_DT + datetime.timedelta(minutes=10 * wi) + chkpt_meta["invocation_start_time"] = _write_start.isoformat() + else: + # Read invocation entered mlpstorage `gap_seconds` after + # the write's summary.end_time — this is what §4.7.1 + # measures. Negative gap models NTP clock skew. + ri = read_timestamps.index(ts) + _write_start = _BASE_DT + datetime.timedelta(minutes=10 * ri) + _write_end = _write_start + datetime.timedelta(minutes=5) + _inv = _write_end + datetime.timedelta( + seconds=chkpt_cache_flush_gap_seconds + ) + chkpt_meta["invocation_start_time"] = _inv.isoformat() + else: + _ts_start = _BASE_DT + datetime.timedelta(minutes=10 * i) + chkpt_meta["invocation_start_time"] = _ts_start.isoformat() + (ts_dir / "metadata.json").write_text( json.dumps(chkpt_meta), encoding="utf-8" ) diff --git a/mlpstorage_py/tests/test_checkpointing_check_phase2.py b/mlpstorage_py/tests/test_checkpointing_check_phase2.py index c0349ffe..792199cf 100644 --- a/mlpstorage_py/tests/test_checkpointing_check_phase2.py +++ b/mlpstorage_py/tests/test_checkpointing_check_phase2.py @@ -184,7 +184,7 @@ def test_split_mode_with_45s_gap_emits_4_7_1(self, tmp_path, mock_logger): assert "30-second limit" in mock_logger.errors[0] def test_split_mode_missing_timestamps_emits_4_7_1(self, tmp_path, mock_logger): - """Split-mode without timestamps → [4.7.1] 'missing end_time/start_time'.""" + """Split-mode without write summary end_time → [4.7.1] 'missing end_time'.""" from mlpstorage_py.tests.conftest import build_submission root = build_submission( tmp_path, @@ -197,7 +197,67 @@ def test_split_mode_missing_timestamps_emits_4_7_1(self, tmp_path, mock_logger): assert len(mock_logger.errors) >= 1 assert mock_logger.errors[0].startswith("[4.7.1 checkpointCacheFlushValidation]"), \ f"Expected [4.7.1 checkpointCacheFlushValidation]; got {mock_logger.errors[0]!r}" - assert "missing end_time/start_time" in mock_logger.errors[0] + assert "missing end_time in write-phase summary.json" in mock_logger.errors[0] + + def test_split_mode_missing_invocation_start_time_falls_back_to_legacy_origin_and_passes(self, tmp_path, mock_logger): + """Backward-compat: pre-fix results dir (no invocation_start_time) falls back + to read.summary.start_time. Fixture default gap is 25s → passes silently.""" + from mlpstorage_py.tests.conftest import build_submission + root = build_submission( + tmp_path, + chkpt_split_mode=True, + chkpt_summary_timestamps=True, + chkpt_omit_invocation_start_time=True, + # default chkpt_cache_flush_gap_seconds=25 → legacy gap is 25s → under 30 + ) + check = _run_checkpointing_check(root, mock_logger) + result = check.cache_flush_validation() + assert result is True + assert mock_logger.errors == [] + assert mock_logger.warnings == [] + + def test_split_mode_legacy_origin_gap_over_30_emits_warning_not_error(self, tmp_path, mock_logger): + """Backward-compat: pre-fix results dir with 45s legacy gap → warn_violation, + not log_violation. Honors the rule that was in force when the run was produced.""" + from mlpstorage_py.tests.conftest import build_submission + root = build_submission( + tmp_path, + chkpt_split_mode=True, + chkpt_summary_timestamps=True, + chkpt_omit_invocation_start_time=True, + chkpt_cache_flush_gap_seconds=45, + ) + check = _run_checkpointing_check(root, mock_logger) + result = check.cache_flush_validation() + # Legacy origin + gap > 30 → warning, not a hard failure. + assert result is True + assert mock_logger.errors == [] + assert any( + w.startswith("[4.7.1 checkpointCacheFlushValidation]") + and "exceeds 30-second limit" in w + and "predates the §4.7.1 gap-origin fix" in w + for w in mock_logger.warnings + ), f"Expected legacy-origin over-limit warning; got warnings={mock_logger.warnings!r}" + + def test_split_mode_negative_gap_emits_clock_skew_warning(self, tmp_path, mock_logger): + """Split-mode with negative gap → warning (clock skew), not a hard violation.""" + from mlpstorage_py.tests.conftest import build_submission + root = build_submission( + tmp_path, + chkpt_split_mode=True, + chkpt_summary_timestamps=True, + chkpt_cache_flush_gap_seconds=-3, + ) + check = _run_checkpointing_check(root, mock_logger) + result = check.cache_flush_validation() + # Negative gap → warning-level, not an error. cache_flush_validation + # returns True (no hard violation) but emits a warning. + assert result is True + assert mock_logger.errors == [] + assert any( + "gap is negative" in w and "clock skew" in w + for w in mock_logger.warnings + ), f"Expected clock-skew warning; got warnings={mock_logger.warnings!r}" # ---------------------------------------------------------------------------