diff --git a/.claude/hooks/posttooluse_format.py b/.claude/hooks/posttooluse_format.py index bb637f4..6f932bc 100644 --- a/.claude/hooks/posttooluse_format.py +++ b/.claude/hooks/posttooluse_format.py @@ -13,6 +13,34 @@ import os import subprocess import sys +from pathlib import Path + + +def project_python() -> str: + """Interpreter that actually has black and ruff installed. + + The hook itself may be run by any interpreter -- typically the system + Python, which has no dev tooling. Prefer the project virtualenv so the + formatters are found; fall back to whatever is running us. + """ + root = Path(__file__).resolve().parents[2] + for candidate in ( + root / ".venv" / "Scripts" / "python.exe", # Windows + root / ".venv" / "bin" / "python", # POSIX + ): + if candidate.exists(): + return str(candidate) + return sys.executable + + +def missing_module(result: subprocess.CompletedProcess) -> bool: + """True when the interpreter ran but the tool is not installed. + + `python -m ruff` with ruff absent exits non-zero with a normal message + rather than raising, so this case is indistinguishable from a lint failure + unless the output is inspected. + """ + return "No module named" in ((result.stderr or "") + (result.stdout or "")) def run(args: list[str]) -> subprocess.CompletedProcess | None: @@ -43,18 +71,20 @@ def main() -> None: if not os.path.exists(file_path): sys.exit(0) - black_result = run([sys.executable, "-m", "black", file_path]) - if black_result is None: + python = project_python() + + black_result = run([python, "-m", "black", file_path]) + if black_result is None or missing_module(black_result): # black not installed / not runnable - tolerate silently. sys.exit(0) - ruff_fix_result = run([sys.executable, "-m", "ruff", "check", "--fix", file_path]) - if ruff_fix_result is None: + ruff_fix_result = run([python, "-m", "ruff", "check", "--fix", file_path]) + if ruff_fix_result is None or missing_module(ruff_fix_result): # ruff not installed / not runnable - tolerate silently. sys.exit(0) - ruff_check_result = run([sys.executable, "-m", "ruff", "check", file_path]) - if ruff_check_result is None: + ruff_check_result = run([python, "-m", "ruff", "check", file_path]) + if ruff_check_result is None or missing_module(ruff_check_result): sys.exit(0) if ruff_check_result.returncode != 0: diff --git a/CHANGELOG.md b/CHANGELOG.md index 1863bfe..1d8568b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,62 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`cpu_affinity` config option.** Pins the run, its worker processes and the + EddyPro executables to a chosen set of logical CPUs. Accepts `performance` + for auto-detection, an explicit list of CPU indices, or null (the default, + unchanged behaviour). + + On Intel hybrid CPUs the OS parks long-running background work on the + efficiency cores. `eddypro_rp` is single-threaded, so measured on an i7-12700K + one year of 10 Hz data took 241 s pinned to the performance cores against + 474-501 s unpinned -- roughly a factor of two, invisible to CPU monitoring + because a saturated efficiency core reports the same ~100% of a core as a + saturated performance one. Pinning also removed the run-to-run variance + entirely. Auto-detection leaves affinity alone on CPUs with no such split, and + an unrecognised value is logged and ignored rather than failing the run. + +- **Throughput signal in the performance monitor.** The monitor can now count + completed work items and record `work_items` and `work_items_per_s` alongside + the CPU and disk series; the analyser derives `mean_seconds_per_work_item` and + the HTML report shows it. `run` points this at the binned-cospectra directory, + which EddyPro fills one file per flux averaging period. + + This exists because CPU utilisation cannot distinguish a saturated fast core + from a saturated slow one. On Intel hybrid CPUs (12th gen and later) Windows + will park a long-running background process on the efficiency cores; measured + on a 12700K, the same workload took 241 s pinned to P-cores and 474-501 s + unpinned, while `cpu_percent_of_core` read 96-103% in *every* case. Without a + work-rate series there is no signal at all that half the machine's performance + is missing. Opt-in via `progress_dir`; runs without one are unaffected. + ### Fixed +- **The bottleneck analyser could not detect a CPU bottleneck.** `monitor.py` + divides process-tree CPU by the logical core count before writing + `cpu_percent`, so a single-threaded EddyPro saturating one core of twenty + recorded ~5% -- far below the 70/90% thresholds the classifier compared it + against. Every run was therefore reported as `NONE: no clear bottleneck ... + headroom to increase max_processes`, which was the right advice only by + coincidence and would have read identically on a fully saturated machine. + Confirmed against a real 7-day run whose `eddypro_rp` phase sustained 100.3% + of one core (p95) while `cpu_percent` read 5.0%. The classifier now: + - judges machine-level saturation on `system_cpu_percent`, because each + parallel worker's monitor only ever sees its own process tree and so can + never observe that the machine as a whole is full; + - adds a `CPU_SINGLE_CORE` verdict driven by `cpu_percent_of_core`, naming + the case where the workload is pinned by single-thread speed while cores + sit idle -- the normal state for `eddypro_rp`, and the one that actually + justifies raising `max_processes`; + - falls back to the old normalised figure when `system_cpu_percent` is + absent, so older metrics files stay classifiable. + + The new `single_core_bound_percent` threshold (default 95) is tunable via + `performance_thresholds`. The HTML report now shows machine-wide CPU and + percent-of-one-core as separate columns, instead of a single normalised + figure that reads as near-zero for any single-threaded run. + - **Monitor reported 0.0% CPU for every child process.** psutil records the CPU-times baseline on the `Process` *instance*, and `children(recursive=True)` builds fresh objects on every sample, so each descendant's first (always-zero) @@ -18,6 +72,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 EddyPro spawning workers — which produced exactly the all-zero CPU column the process-tree fix was meant to eliminate. +- **EddyPro output was written to the console twice.** `setup_logging` always + attaches a stdout `StreamHandler`, so with both `stream_output: true` and + `log_eddypro_output: true` -- the shipped defaults -- the streaming loop's + `print()` duplicated every line the logger had already emitted. This doubled + console volume and log-file size and halved the interval between log + rotations on multi-hour runs. The direct echo is now used only when the + logger is not already mirroring the output, so live progress still works with + `log_eddypro_output: false`. + - **`eddypro-batch` with no subcommand printed a config error instead of help.** `main()` validated the config file before dispatching, so on a fresh clone -- where `config/config.yaml` does not exist yet -- the first command diff --git a/config/config.yaml.example b/config/config.yaml.example index 87ae8be..84f7898 100644 --- a/config/config.yaml.example +++ b/config/config.yaml.example @@ -33,6 +33,17 @@ multiprocessing: False # Set to True to process years in parallel # Rule of thumb: one worker per year, bounded by physical cores and disk throughput max_processes: 16 +# CPU affinity for the run and every process it launches. +# null -> leave scheduling to the OS (default) +# performance -> auto-detect and pin to the performance cores +# [0, 1, 2, ...] -> pin to these logical CPU indices +# On Intel hybrid CPUs (12th gen and later) the OS will park a long-running +# background job on the efficiency cores. EddyPro is single-threaded, so this +# costs roughly a factor of two -- and CPU monitoring cannot see it, because a +# saturated efficiency core reports the same ~100% as a saturated performance +# core. Pinning also makes run times reproducible. +cpu_affinity: null + # Control output streaming (EddyPro subprocess outputs) stream_output: True # Set to False to keep EddyPro output quiet diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 1ffde30..345a453 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -76,6 +76,7 @@ The following keys **must** be present in your configuration file: | `log_backup_count` | int or null | Number of rotated log files to keep | | `log_eddypro_output` | bool | Write EddyPro stdout/stderr to logs | | `monitoring_enabled` | bool | Enable/disable performance monitoring (default: `true`) | +| `cpu_affinity` | str, list[int], or null | Pin the run to specific logical CPUs (default: `null`, no pinning) | ## Configuration Details @@ -411,6 +412,51 @@ monitoring_enabled: false --- +### cpu_affinity + +**Type:** String, list of integers, or null + +**Default:** `null` (leave scheduling to the OS) + +**Description:** Restricts the run -- and every process it launches, since child +processes inherit affinity -- to a set of logical CPUs. + +**Values:** + +| Value | Meaning | +|-------|---------| +| `null` | No pinning | +| `performance` | Auto-detect the performance cores and pin to them | +| `[0, 1, 2, ...]` | Pin to these logical CPU indices | + +**Why:** Intel hybrid CPUs (12th generation and later) mix performance cores +with efficiency cores, and the OS will park a long-running background job on the +efficiency ones. `eddypro_rp` is single-threaded, so this costs roughly a factor +of two. Measured on an i7-12700K, one year of 10 Hz data: + +| | Wall time | CPU % of one core | +|---|---|---| +| Pinned to performance cores | 241 s, 241 s | 102.6% | +| Unpinned | 474 s, 501 s | 96.5-97.8% | + +Note the second column: **CPU utilisation is identical**. A demoted run looks +healthy in the bottleneck report. Pinning also removes the variance -- the +pinned runs agreed exactly, the unpinned pair differed by 5.7%. + +**Auto-detection:** `performance` derives the split from the core counts. +Performance cores carry two hardware threads and efficiency cores carry one, so +`p_cores = logical - physical`, occupying logical indices `0 .. 2*p_cores-1`. +Where that cannot be determined -- no SMT, or no efficiency cores -- affinity is +left unchanged and a message is logged. On a CPU without efficiency cores there +is nothing to gain, so this is the correct outcome rather than a failure. + +**Never fatal:** an unrecognised value is logged and ignored. A failed +optimisation must not stop a multi-hour run. + +**CLI Override:** none -- config only. + +--- + ### performance_thresholds **Type:** Mapping (optional) @@ -426,15 +472,22 @@ SATA default and a genuinely saturated disk is never flagged at all. | Key | Default | Meaning | |-----|---------|---------| -| `cpu_high_percent` | 90 | At or above this sustained (p95) CPU, status is RED | +| `cpu_high_percent` | 90 | At or above this sustained (p95) **machine-wide** CPU, status is RED | | `cpu_moderate_percent` | 70 | At or above this, status is YELLOW | | `cpu_idle_percent` | 40 | Below this, a busy disk is read as the limiting factor | +| `single_core_bound_percent` | 95 | Applied to `cpu_percent_of_core`, where 100 = one core fully busy. At or above this, the run is reported `CPU_SINGLE_CORE` | | `memory_high_percent` | 85 | System memory use that counts as RED | | `memory_moderate_percent` | 70 | System memory use that counts as YELLOW | | `disk_high_mb_per_s` | 450 | Combined read+write throughput counting as RED | | `disk_moderate_mb_per_s` | 250 | Throughput counting as YELLOW | | `disk_high_iops` | 20000 | Combined IOPS above which latency is the suspect | +The CPU verdict is judged on `system_cpu_percent` (machine-wide). Each parallel +worker runs its own monitor and sees only its own process tree, so per-process +CPU can never reveal that the machine as a whole is full. Files written by a +monitor that did not record the system column fall back to the normalised +process figure. + Unknown keys are ignored, so a config written for a newer version still loads. Suggested disk limits by medium: diff --git a/docs/MULTI_YEAR_RUNS.md b/docs/MULTI_YEAR_RUNS.md index 36e5f54..4432624 100644 --- a/docs/MULTI_YEAR_RUNS.md +++ b/docs/MULTI_YEAR_RUNS.md @@ -153,16 +153,66 @@ decide whether `max_processes` is set well: | Classification | What it means | What to do | |---|---|---| -| CPU | Workers are compute-bound; cores are the limit | Lower `max_processes` toward the physical core count if you see thrashing; otherwise this is healthy utilization | +| CPU | The **machine** is compute-bound; cores are the limit | Lower `max_processes` toward the physical core count if you see thrashing; otherwise this is healthy utilization | +| CPU_SINGLE_CORE | EddyPro is pegging one core while the machine sits idle. This is the normal state for a single-year run, because `eddypro_rp` is largely single-threaded | Raise `max_processes` and run more years concurrently. A faster disk will not help | | MEMORY | Workers are approaching available RAM | Lower `max_processes`, or process fewer years concurrently | | DISK_THROUGHPUT | Aggregate read/write MB/s is saturating the disk | Lower `max_processes`, or move input/output to faster storage (SSD/NVMe) | | DISK_IOPS | Many small reads/writes are the limit (common with many small raw files) | Move data to faster storage; concatenating raw files can help more than adding workers | -With N years running in parallel, remember that each worker's own CPU% is a -share of the *whole machine* — a healthy multi-year run with `max_processes: -6` on an 8-core machine will show each worker around 60-80% CPU, not 100%. -Read the aggregate/system columns in the metrics CSVs, not just one worker's -process columns, when deciding whether you're actually CPU-bound. +The metrics CSV carries three different CPU columns and they answer different +questions: + +| Column | Meaning | Use it for | +|---|---|---| +| `cpu_percent_of_core` | 100 = one core fully busy, 200 = two | Is the workload itself pinned by single-thread speed? | +| `system_cpu_percent` | Machine-wide utilisation | Is the *machine* full? This drives the CPU verdict | +| `cpu_percent` | The process tree's share of the whole machine | Comparing one worker's footprint against the box | + +`cpu_percent` is divided by the logical core count, so on a 20-thread machine a +single-threaded EddyPro saturating one core reads as ~5%. That is not idleness -- +check `cpu_percent_of_core` before concluding a run was cheap. + +## Hybrid CPUs: pin EddyPro to the performance cores + +On Intel hybrid CPUs (12th generation and later) the cores are not all equal: +an i7-12700K has 8 performance cores (logical 0-15) and 4 efficiency cores +(16-19). Windows will park a long-running background process on the efficiency +cores, especially while you are using the machine for something else. + +`eddypro_rp` is single-threaded, so this costs about a factor of two. Measured +on a 12700K, one year of 10 Hz data: + +| | Wall time | s per flux period | `cpu_percent_of_core` p95 | +|---|---|---|---| +| Pinned to P-cores | 241 s, 241 s | 0.70 | 102.6% | +| Unpinned | 474 s, 501 s | 1.39, 1.47 | 96.5-97.8% | + +Note the last column: **CPU utilisation is identical**. A demoted run looks +perfectly healthy in the bottleneck report -- it is saturating a core, it is +just a slower core. Only the `work_items_per_s` throughput series reveals it. +System CPU sat at 17-21% in all four runs, so this is core placement, not +contention. + +Pinning also makes runs reproducible: the two pinned runs agreed exactly, while +the unpinned pair differed by 5.7%. + +Set `cpu_affinity` in the config and the pipeline handles it -- child +processes inherit affinity, so pinning the launching process covers the workers +and the EddyPro executables they run: + +```yaml +cpu_affinity: performance # or an explicit list, e.g. [0, 1, 2, ..., 15] +``` + +Auto-detection derives the split from the core counts and logs what it chose. +On a CPU without efficiency cores there is nothing to gain and affinity is left +alone. See [CONFIG.md](CONFIG.md#cpu_affinity) for the details. + +Check your own topology with: + +```powershell +Get-CimInstance Win32_Processor | Select-Object Name, NumberOfCores, NumberOfLogicalProcessors +``` ## Choosing `max_processes` diff --git a/docs/REPORTING.md b/docs/REPORTING.md index 892d406..11947e1 100644 --- a/docs/REPORTING.md +++ b/docs/REPORTING.md @@ -226,6 +226,11 @@ timestamp,relative_time,system_cpu_percent,system_memory_total,system_memory_ava - `system_memory_total`, `system_memory_available`, `system_memory_percent` - `system_disk_read_bytes`, `system_disk_write_bytes`, `system_disk_read_count`, `system_disk_write_count` - `process_cpu_percent`, `process_memory_rss`, `process_memory_vms`, `process_memory_percent` +- `work_items`, `work_items_per_s`: completed work items and their rate, when a + progress directory is configured. For `run` these count binned-cospectra + files, i.e. flux averaging periods finished. This is the only column that + distinguishes a saturated *fast* core from a saturated *slow* one -- see + [MULTI_YEAR_RUNS.md](MULTI_YEAR_RUNS.md) on hybrid-CPU core placement - `process_io_read_bytes`, `process_io_write_bytes`, `process_io_read_count`, `process_io_write_count` Column presence can vary by platform and psutil capabilities. The CSV is a raw diff --git a/src/eddypro_batch_processor/analysis.py b/src/eddypro_batch_processor/analysis.py index a2ce91f..9a92fad 100644 --- a/src/eddypro_batch_processor/analysis.py +++ b/src/eddypro_batch_processor/analysis.py @@ -22,7 +22,15 @@ logger = logging.getLogger(__name__) Status = Literal["GREEN", "YELLOW", "RED", "UNKNOWN"] -Bottleneck = Literal["CPU", "MEMORY", "DISK_THROUGHPUT", "DISK_IOPS", "NONE", "UNKNOWN"] +Bottleneck = Literal[ + "CPU", + "CPU_SINGLE_CORE", + "MEMORY", + "DISK_THROUGHPUT", + "DISK_IOPS", + "NONE", + "UNKNOWN", +] class PerformanceThresholds(TypedDict, total=False): @@ -31,6 +39,7 @@ class PerformanceThresholds(TypedDict, total=False): cpu_high_percent: float cpu_moderate_percent: float cpu_idle_percent: float + single_core_bound_percent: float memory_high_percent: float memory_moderate_percent: float disk_high_mb_per_s: float @@ -44,6 +53,10 @@ class PerformanceThresholds(TypedDict, total=False): # Below this, the CPU is considered idle enough that a high I/O rate is # evidence the run was waiting on the disk rather than computing. "cpu_idle_percent": 40.0, + # Applied to cpu_percent_of_core, where 100 means "one core fully busy". + # EddyPro is largely single-threaded, so a run pegged at ~100 is limited by + # single-thread speed no matter how many cores the machine has. + "single_core_bound_percent": 95.0, "memory_high_percent": 85.0, "memory_moderate_percent": 70.0, # Calibrated for a SATA SSD (~550 MB/s sequential), which is the common @@ -85,15 +98,26 @@ class ScenarioAnalysis: duration_seconds: float = 0.0 cpu: MetricStats = field(default_factory=MetricStats) + # Un-normalised: 100 means one core fully busy, 200 two, and so on. This is + # the column that reveals single-thread saturation on a many-core machine, + # which `cpu` (divided by the core count) hides. + cpu_percent_of_core: MetricStats = field(default_factory=MetricStats) + system_cpu_percent: MetricStats = field(default_factory=MetricStats) memory_mb: MetricStats = field(default_factory=MetricStats) system_memory_percent: MetricStats = field(default_factory=MetricStats) read_mb_per_s: MetricStats = field(default_factory=MetricStats) write_mb_per_s: MetricStats = field(default_factory=MetricStats) read_iops: MetricStats = field(default_factory=MetricStats) write_iops: MetricStats = field(default_factory=MetricStats) + # Work completed per second. CPU cannot distinguish a saturated fast core + # from a saturated slow one, so throughput is what exposes a run that was + # demoted onto an efficiency core: identical CPU, half the work done. + work_items_per_s: MetricStats = field(default_factory=MetricStats) total_read_mb: float = 0.0 total_write_mb: float = 0.0 + total_work_items: float = 0.0 + mean_seconds_per_work_item: float = 0.0 peak_memory_mb: float = 0.0 cpu_status: Status = "UNKNOWN" @@ -204,12 +228,16 @@ def series(key: str) -> list[float]: return [v for v in (_to_float(r.get(key)) for r in rows) if v is not None] cpu = series("cpu_percent") + cpu_of_core = series("cpu_percent_of_core") + sys_cpu = series("system_cpu_percent") memory = series("memory_mb") sys_mem = series("system_memory_percent") read_rate = series("read_mb_per_s") write_rate = series("write_mb_per_s") read_iops = series("read_iops") write_iops = series("write_iops") + work_rate = series("work_items_per_s") + work_total = series("work_items") read_total = series("read_mb") write_total = series("write_mb") rel_time = series("relative_time") @@ -219,17 +247,26 @@ def series(key: str) -> list[float]: sample_count=len(rows), duration_seconds=round(rel_time[-1], 2) if rel_time else 0.0, cpu=_stats(cpu), + cpu_percent_of_core=_stats(cpu_of_core), + system_cpu_percent=_stats(sys_cpu), memory_mb=_stats(memory), system_memory_percent=_stats(sys_mem), read_mb_per_s=_stats(read_rate), write_mb_per_s=_stats(write_rate), read_iops=_stats(read_iops), write_iops=_stats(write_iops), + work_items_per_s=_stats(work_rate), + total_work_items=round(work_total[-1], 0) if work_total else 0.0, total_read_mb=round(read_total[-1], 3) if read_total else 0.0, total_write_mb=round(write_total[-1], 3) if write_total else 0.0, peak_memory_mb=round(max(memory), 3) if memory else 0.0, ) + if analysis.total_work_items > 0 and analysis.duration_seconds > 0: + analysis.mean_seconds_per_work_item = round( + analysis.duration_seconds / analysis.total_work_items, 3 + ) + self._classify(analysis) return analysis @@ -252,8 +289,17 @@ def _classify(self, a: ScenarioAnalysis) -> None: t = self.thresholds # CPU: judged on sustained load, so p95 rather than the peak. + # + # Machine-level saturation must come from the system-wide column. Each + # parallel worker runs its own monitor and sees only its own process + # tree, so per-process CPU can never reveal that the *machine* is full. + # Fall back to the normalised process figure when the system column is + # absent, which keeps files from older monitors classifiable. + machine_cpu = ( + a.system_cpu_percent.p95 if a.system_cpu_percent.max > 0 else a.cpu.p95 + ) a.cpu_status = self._level( - a.cpu.p95, t["cpu_moderate_percent"], t["cpu_high_percent"] + machine_cpu, t["cpu_moderate_percent"], t["cpu_high_percent"] ) # Memory: system-wide pressure matters more than the process footprint, @@ -274,16 +320,16 @@ def _classify(self, a: ScenarioAnalysis) -> None: disk_by_iops = "RED" if total_iops > t["disk_high_iops"] else "GREEN" a.disk_status = "RED" if "RED" in (disk_by_rate, disk_by_iops) else disk_by_rate - cpu_idle = a.cpu.p95 < t["cpu_idle_percent"] + cpu_idle = machine_cpu < t["cpu_idle_percent"] # Priority: saturated CPU is the clearest signal. Otherwise a busy disk # paired with an idle CPU means the run was waiting on I/O. if a.cpu_status == "RED": a.primary_bottleneck = "CPU" a.explanation = ( - f"CPU saturated: sustained (p95) utilisation {a.cpu.p95:.1f}%. " - f"Processing is compute-bound; more parallel years will not help " - f"unless spare cores are available." + f"CPU saturated: sustained (p95) machine-wide utilisation " + f"{machine_cpu:.1f}%. Processing is compute-bound; more parallel " + f"years will not help unless spare cores are available." ) elif a.memory_status == "RED": a.primary_bottleneck = "MEMORY" @@ -296,26 +342,40 @@ def _classify(self, a: ScenarioAnalysis) -> None: a.primary_bottleneck = "DISK_THROUGHPUT" a.explanation = ( f"Disk-bound: {disk_rate:.1f} MB/s sustained while the CPU sat at " - f"{a.cpu.p95:.1f}%. The run is waiting on storage -- move the data " + f"{machine_cpu:.1f}%. The run is waiting on storage -- move the data " f"to a faster disk before adding parallelism." ) elif cpu_idle and disk_by_iops == "RED": a.primary_bottleneck = "DISK_IOPS" a.explanation = ( f"Disk-bound on latency: {total_iops:.0f} IOPS (p95) with the CPU at " - f"{a.cpu.p95:.1f}%. Many small reads -- typical of raw files split " + f"{machine_cpu:.1f}%. Many small reads -- typical of raw files split " f"into short intervals." ) + elif a.cpu_percent_of_core.p95 >= t["single_core_bound_percent"]: + # The machine has spare capacity but the workload itself is pinned to + # a core. Adding workers is the fix; a faster disk is not. + a.primary_bottleneck = "CPU_SINGLE_CORE" + cores_busy = a.cpu_percent_of_core.p95 / 100.0 + a.explanation = ( + f"Single-core bound: the EddyPro process tree sustained " + f"{a.cpu_percent_of_core.p95:.0f}% of one core " + f"({cores_busy:.1f} core(s) busy) while the machine as a whole sat " + f"at {machine_cpu:.1f}%. EddyPro is largely single-threaded, so a " + f"faster disk will not help -- run more years concurrently " + f"(raise max_processes) to use the idle cores." + ) elif a.cpu_status == "YELLOW": a.primary_bottleneck = "CPU" a.explanation = ( - f"Moderately CPU-bound: sustained utilisation {a.cpu.p95:.1f}%. " - f"Some headroom remains." + f"Moderately CPU-bound: sustained machine-wide utilisation " + f"{machine_cpu:.1f}%. Some headroom remains." ) else: a.primary_bottleneck = "NONE" a.explanation = ( - f"No clear bottleneck: CPU {a.cpu.p95:.1f}% (p95), disk " + f"No clear bottleneck: CPU {machine_cpu:.1f}% (p95) machine-wide, " + f"{a.cpu_percent_of_core.p95:.0f}% of one core, disk " f"{disk_rate:.1f} MB/s, peak memory {a.peak_memory_mb:.0f} MB. " f"There is headroom to increase max_processes." ) diff --git a/src/eddypro_batch_processor/cli.py b/src/eddypro_batch_processor/cli.py index 78e416b..29da686 100644 --- a/src/eddypro_batch_processor/cli.py +++ b/src/eddypro_batch_processor/cli.py @@ -556,6 +556,10 @@ def cmd_run(args: argparse.Namespace) -> int: # noqa: PLR0912, PLR0915 config.get("log_backup_count"), ) + # Pin before any work starts; ProcessPoolExecutor workers and the EddyPro + # executables they launch all inherit this process's affinity. + core.apply_cpu_affinity(config) + # Collect INI parameter overrides ini_parameters = {} if args.rot_meth is not None: @@ -832,6 +836,10 @@ def cmd_scenarios(args: argparse.Namespace) -> int: # noqa: PLR0911 config.get("log_backup_count"), ) + # Pin before any work starts; ProcessPoolExecutor workers and the EddyPro + # executables they launch all inherit this process's affinity. + core.apply_cpu_affinity(config) + # Apply CLI overrides site_id = args.site if args.site else config.get("site_id") years = args.years if args.years else config.get("years_to_process", []) diff --git a/src/eddypro_batch_processor/core.py b/src/eddypro_batch_processor/core.py index 2cf68ab..9bb63f6 100644 --- a/src/eddypro_batch_processor/core.py +++ b/src/eddypro_batch_processor/core.py @@ -16,6 +16,7 @@ from pathlib import Path from typing import Any, NoReturn +import psutil import yaml from . import analysis, ecmd, ini_tools, report @@ -137,6 +138,92 @@ def validate_config(self, config: dict[str, Any] | None = None) -> None: logging.info("Configuration validation passed.") +def resolve_performance_cores() -> list[int] | None: + """Best-effort logical CPU indices of the performance cores. + + Intel hybrid CPUs (12th generation and later) mix performance cores, which + carry two hardware threads each, with efficiency cores, which carry one. So + given L logical and P physical cores:: + + p_cores * 2 + e_cores == L + p_cores + e_cores == P + + which solves to ``p_cores = L - P``. Their logical indices are ``0 .. + 2*p_cores - 1`` under the standard Windows enumeration, where P-core threads + are numbered first. + + Returns None when the split cannot be determined -- no hyper-threading, no + efficiency cores, or an unexpected topology -- in which case the caller + should leave affinity alone rather than guess. + """ + try: + logical = psutil.cpu_count(logical=True) + physical = psutil.cpu_count(logical=False) + except Exception: + return None + if not logical or not physical or logical <= physical: + return None # no SMT, so P-cores and E-cores are indistinguishable here + + p_cores = logical - physical + e_cores = physical - p_cores + if p_cores <= 0 or e_cores <= 0: + return None # uniform SMT topology: every core is a performance core + return list(range(2 * p_cores)) + + +def apply_cpu_affinity(config: dict[str, Any]) -> None: + """Pin this process (and therefore its children) per ``cpu_affinity``. + + Child processes inherit affinity, so setting it once here covers the + ProcessPoolExecutor workers and the EddyPro executables they launch. + + Why this exists: ``eddypro_rp`` is single-threaded, and Windows will park a + long-running background process on the efficiency cores. Measured on a + 12700K, the same year of data took 241 s pinned to the performance cores and + 474-501 s unpinned -- while CPU utilisation read 96-103% of a core either + way, so nothing in the metrics revealed the loss. + + Accepts ``"performance"`` for auto-detection, an explicit list of logical CPU + indices, or null/absent to leave affinity untouched. Never fatal: a bad value + is logged and ignored, because a failed optimisation must not stop a run. + """ + setting = config.get("cpu_affinity") + if setting is None or setting is False: + return + + if isinstance(setting, str): + if setting.lower() != "performance": + logging.warning( + f"Unknown cpu_affinity value {setting!r}; expected 'performance', " + f"a list of CPU indices, or null. Leaving affinity unchanged." + ) + return + cores = resolve_performance_cores() + if not cores: + logging.info( + "cpu_affinity: 'performance' requested but this CPU has no " + "detectable performance/efficiency split; leaving affinity unchanged." + ) + return + elif isinstance(setting, list) and all(isinstance(c, int) for c in setting): + cores = setting + else: + logging.warning( + f"Invalid cpu_affinity value {setting!r}; leaving affinity unchanged." + ) + return + + try: + proc = psutil.Process() + proc.cpu_affinity(cores) + logging.info( + f"Pinned to {len(cores)} logical CPU(s): {cores[0]}-{cores[-1]}. " + f"Child processes inherit this." + ) + except Exception as e: + logging.warning(f"Could not set CPU affinity to {cores}: {e}") + + def run_subprocess_with_monitoring( command: list[str], working_dir: Path, @@ -146,6 +233,7 @@ def run_subprocess_with_monitoring( scenario_suffix: str = "", log_output: bool = True, monitoring_enabled: bool = True, + progress_dir: Path | None = None, ) -> int: """ Execute a subprocess command with performance monitoring. @@ -165,6 +253,8 @@ def run_subprocess_with_monitoring( log_output: Whether to mirror subprocess output into the log monitoring_enabled: When False, no monitor is started and no metrics files are written + progress_dir: Optional directory whose file count proxies work completed, + recorded as a throughput series alongside CPU Returns: Subprocess return code, or -1 if an exception occurs @@ -178,6 +268,7 @@ def run_subprocess_with_monitoring( output_dir=metrics_output_dir, scenario_suffix=scenario_suffix, enabled=monitoring_enabled, + progress_dir=progress_dir, ) as monitor: process = subprocess.Popen( # nosec B603 command, @@ -191,12 +282,20 @@ def run_subprocess_with_monitoring( if monitor and process.pid: monitor.attach_process(process.pid) - # Handle output streaming + # Handle output streaming. + # + # `setup_logging` always attaches a stdout StreamHandler, so when + # log_output is on the logger already puts every line on the + # console. Printing as well emitted each EddyPro line twice, which + # doubled console volume and log-file size and halved the interval + # between log rotations. Echo directly only when the logger is not + # already doing it. if stream_output and process.stdout: for line in process.stdout: - print(line, end="") if log_output: output_logger.info(line.rstrip("\n")) + else: + print(line, end="") process.wait() else: stdout_data, _ = process.communicate() @@ -289,6 +388,10 @@ def run_eddypro_with_monitoring( scenario_suffix=f"{scenario_suffix}_rp" if scenario_suffix else "rp", log_output=log_output, monitoring_enabled=monitoring_enabled, + # One binned-cospectra file per flux averaging period, so counting them + # measures periods completed. Throughput is the only signal that tells a + # saturated fast core from a saturated slow one. + progress_dir=output_dir / "eddypro_binned_cospectra", ) if rp_return_code != 0: logging.error(f"eddypro_rp failed with return code {rp_return_code}") diff --git a/src/eddypro_batch_processor/monitor.py b/src/eddypro_batch_processor/monitor.py index b03a1a8..ce7a8d6 100644 --- a/src/eddypro_batch_processor/monitor.py +++ b/src/eddypro_batch_processor/monitor.py @@ -55,6 +55,8 @@ "read_iops", "write_iops", "num_processes", + "work_items", + "work_items_per_s", "system_cpu_percent", "system_memory_percent", "system_memory_used_mb", @@ -79,6 +81,8 @@ def __init__( interval_seconds: float = 0.5, output_dir: str | Path | None = None, scenario_suffix: str = "", + progress_dir: str | Path | None = None, + progress_glob: str = "*", ): """ Initialize the performance monitor. @@ -87,6 +91,12 @@ def __init__( interval_seconds: Sampling interval in seconds (default: 0.5) output_dir: Directory to write metrics files (default: current directory) scenario_suffix: Suffix to append to output filenames for scenario runs + progress_dir: Optional directory whose file count is a proxy for work + completed. CPU alone cannot tell a saturated fast core from a + saturated slow one -- on a hybrid P-core/E-core CPU a demoted run + reports the same ~100% of a core while delivering roughly half the + throughput. Counting finished work makes that visible. + progress_glob: Pattern selecting the files to count in progress_dir Raises: ImportError: If psutil is not available @@ -100,6 +110,8 @@ def __init__( self.interval_seconds = max(0.1, interval_seconds) # Minimum 0.1s self.output_dir = Path(output_dir) if output_dir else Path.cwd() self.scenario_suffix = scenario_suffix + self.progress_dir = Path(progress_dir) if progress_dir else None + self.progress_glob = progress_glob # Number of logical CPUs, used to normalise process CPU onto a 0-100 scale # so it is directly comparable with psutil.cpu_percent(). @@ -133,6 +145,7 @@ def __init__( self._prev_time: float | None = None self._prev_io: tuple[float, float, float, float] | None = None self._prev_system_io: tuple[float, float] | None = None + self._prev_work_items: int | None = None # Output file paths self._metrics_csv_path = self._get_output_path("metrics.csv") @@ -176,6 +189,7 @@ def start_monitoring(self, process_pid: int | None = None) -> None: self._prev_time = None self._prev_io = None self._prev_system_io = None + self._prev_work_items = None # Prime the system-wide CPU counter. The first call after import always # returns 0.0 because there is no previous measurement to diff against. @@ -330,6 +344,7 @@ def _collect_sample(self) -> dict[str, Any] | None: } sample.update(self._collect_system_metrics(elapsed)) sample.update(self._collect_process_metrics(elapsed)) + sample.update(self._collect_progress_metrics(elapsed)) self._prev_time = timestamp except Exception as e: @@ -338,6 +353,38 @@ def _collect_sample(self) -> dict[str, Any] | None: else: return sample + def _collect_progress_metrics(self, elapsed: float | None) -> dict[str, Any]: + """Count completed work items and derive a rate. + + Throughput is the only signal that distinguishes a saturated fast core + from a saturated slow one. CPU utilisation reads ~100% of a core either + way, so without this a run demoted onto an efficiency core looks + perfectly healthy while taking twice as long. + """ + metrics: dict[str, Any] = {"work_items": 0, "work_items_per_s": 0.0} + if self.progress_dir is None: + return metrics + + try: + # The directory is created by the workload, so it legitimately does + # not exist for the first samples of a run. + count = ( + sum(1 for _ in self.progress_dir.glob(self.progress_glob)) + if self.progress_dir.is_dir() + else 0 + ) + except OSError as e: + logger.debug(f"Failed to count progress items: {e}") + return metrics + + metrics["work_items"] = count + if self._prev_work_items is not None and elapsed and elapsed > 0: + metrics["work_items_per_s"] = round( + max(0, count - self._prev_work_items) / elapsed, 4 + ) + self._prev_work_items = count + return metrics + def _collect_system_metrics(self, elapsed: float | None) -> dict[str, Any]: """Collect system-wide metrics, converting disk counters into rates.""" metrics: dict[str, Any] = { @@ -679,6 +726,8 @@ def create_monitor( interval_seconds: float = 0.5, output_dir: str | Path | None = None, scenario_suffix: str = "", + progress_dir: str | Path | None = None, + progress_glob: str = "*", ) -> PerformanceMonitor | None: """ Create a performance monitor instance with error handling. @@ -687,6 +736,8 @@ def create_monitor( interval_seconds: Sampling interval in seconds (default: 0.5) output_dir: Directory to write metrics files scenario_suffix: Suffix for scenario-specific output files + progress_dir: Optional directory whose file count proxies work completed + progress_glob: Pattern selecting the files to count in progress_dir Returns: PerformanceMonitor instance, or None if psutil is not available @@ -703,6 +754,8 @@ def create_monitor( interval_seconds=interval_seconds, output_dir=output_dir, scenario_suffix=scenario_suffix, + progress_dir=progress_dir, + progress_glob=progress_glob, ) except ImportError as e: logger.warning(f"Failed to create performance monitor: {e}") @@ -731,11 +784,19 @@ def __init__( scenario_suffix: str = "", process_pid: int | None = None, enabled: bool = True, + progress_dir: str | Path | None = None, + progress_glob: str = "*", ): """Initialize monitored operation context.""" self.enabled = enabled self.monitor = ( - create_monitor(interval_seconds, output_dir, scenario_suffix) + create_monitor( + interval_seconds, + output_dir, + scenario_suffix, + progress_dir, + progress_glob, + ) if enabled else None ) diff --git a/src/eddypro_batch_processor/report.py b/src/eddypro_batch_processor/report.py index 2d251a2..22393ec 100644 --- a/src/eddypro_batch_processor/report.py +++ b/src/eddypro_batch_processor/report.py @@ -718,19 +718,28 @@ def _dot(status: str) -> str: - + + + """) for entry in perf_entries: cpu = entry.get("cpu", {}) or {} + # Machine-wide where recorded; the normalised process figure is a + # fraction of the whole machine and reads as near-zero on its own. + sys_cpu = entry.get("system_cpu_percent", {}) or {} + core_cpu = entry.get("cpu_percent_of_core", {}) or {} + machine_p95 = sys_cpu.get("p95") or cpu.get("p95", 0) html_parts.append(f""" - + + + diff --git a/tests/test_analysis.py b/tests/test_analysis.py index e67e707..ec4a76f 100644 --- a/tests/test_analysis.py +++ b/tests/test_analysis.py @@ -34,6 +34,8 @@ "read_iops", "write_iops", "num_processes", + "work_items", + "work_items_per_s", "system_cpu_percent", "system_memory_percent", "system_memory_used_mb", @@ -136,6 +138,140 @@ def test_no_bottleneck(self, tmp_path): assert result.cpu_status == "GREEN" assert "headroom" in result.explanation.lower() + def test_single_core_bound_is_detected(self, tmp_path): + """The signature of this pipeline: one core pegged, machine mostly idle. + + ``cpu_percent`` is divided by the logical core count, so a single-threaded + EddyPro saturating one core of twenty reports ~5% -- far below every CPU + threshold. Classifying that as "no bottleneck" hides the one finding that + matters, namely that the cores are there but unused. + """ + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=5.0, + cpu_percent_of_core=100.0, + system_cpu_percent=23.0, + system_memory_percent=40.0, + read_mb_per_s=3.0, + ) + result = BottleneckAnalyzer().analyze(csv_path) + assert result.primary_bottleneck == "CPU_SINGLE_CORE" + assert "max_processes" in result.explanation + assert result.cpu_percent_of_core.p95 == pytest.approx(100.0) + assert result.system_cpu_percent.p95 == pytest.approx(23.0) + + def test_machine_saturation_uses_the_system_column(self, tmp_path): + """With N workers, each monitor sees only its own tree. + + Per-process CPU therefore cannot reveal a full machine; the system-wide + column must drive the RED verdict, and it must outrank single-core. + """ + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=5.0, + cpu_percent_of_core=100.0, + system_cpu_percent=95.0, + system_memory_percent=40.0, + ) + result = BottleneckAnalyzer().analyze(csv_path) + assert result.primary_bottleneck == "CPU" + assert result.cpu_status == "RED" + + def test_single_core_does_not_mask_a_disk_bottleneck(self, tmp_path): + """A saturated disk with an idle machine still outranks single-core.""" + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=5.0, + cpu_percent_of_core=100.0, + system_cpu_percent=10.0, + system_memory_percent=40.0, + read_mb_per_s=400.0, + write_mb_per_s=200.0, + ) + assert ( + BottleneckAnalyzer().analyze(csv_path).primary_bottleneck + == "DISK_THROUGHPUT" + ) + + def test_idle_single_threaded_run_is_not_single_core_bound(self, tmp_path): + """A process using half a core is not pinned by single-thread speed.""" + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=2.5, + cpu_percent_of_core=50.0, + system_cpu_percent=12.0, + system_memory_percent=40.0, + ) + assert BottleneckAnalyzer().analyze(csv_path).primary_bottleneck == "NONE" + + def test_single_core_threshold_is_tunable(self, tmp_path): + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=5.0, + cpu_percent_of_core=80.0, + system_cpu_percent=20.0, + system_memory_percent=40.0, + ) + assert BottleneckAnalyzer().analyze(csv_path).primary_bottleneck == "NONE" + relaxed = BottleneckAnalyzer({"single_core_bound_percent": 75.0}) + assert relaxed.analyze(csv_path).primary_bottleneck == "CPU_SINGLE_CORE" + + def test_throughput_is_recorded_alongside_cpu(self, tmp_path): + """Two runs can look identical on CPU yet differ in work done. + + On a hybrid P-core/E-core CPU, a run demoted onto an efficiency core + still reports ~100% of a core while delivering roughly half the + throughput. Without a work-rate series there is no signal at all that + this happened. + """ + fast = write_metrics( + tmp_path / "fast.csv", + [ + { + "cpu_percent_of_core": 100.0, + "system_cpu_percent": 20.0, + "work_items": i * 2, + "work_items_per_s": 2.0, + } + for i in range(20) + ], + ) + slow = write_metrics( + tmp_path / "slow.csv", + [ + { + "cpu_percent_of_core": 100.0, + "system_cpu_percent": 20.0, + "work_items": i, + "work_items_per_s": 1.0, + } + for i in range(20) + ], + ) + a = BottleneckAnalyzer() + f, s_ = a.analyze(fast), a.analyze(slow) + + # Indistinguishable on CPU ... + assert f.cpu_percent_of_core.p95 == s_.cpu_percent_of_core.p95 + assert f.primary_bottleneck == s_.primary_bottleneck == "CPU_SINGLE_CORE" + # ... but the throughput series separates them. + assert f.work_items_per_s.p95 == pytest.approx(2.0) + assert s_.work_items_per_s.p95 == pytest.approx(1.0) + assert f.mean_seconds_per_work_item < s_.mean_seconds_per_work_item + + def test_throughput_absent_is_not_an_error(self, tmp_path): + """Runs with no progress directory must still classify cleanly.""" + csv_path = make_series( + tmp_path / "metrics.csv", + cpu_percent=5.0, + cpu_percent_of_core=100.0, + system_cpu_percent=20.0, + ) + result = BottleneckAnalyzer().analyze(csv_path) + assert result.primary_bottleneck == "CPU_SINGLE_CORE" + assert result.total_work_items == 0.0 + assert result.mean_seconds_per_work_item == 0.0 + def test_cpu_takes_priority_over_disk(self, tmp_path): # A saturated CPU outranks a busy disk: the CPU is the binding limit. csv_path = make_series( diff --git a/tests/test_core.py b/tests/test_core.py index 6ec906a..0a72c11 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,5 +1,7 @@ """Tests for core module functionality.""" +import logging +import sys import tempfile from pathlib import Path from unittest.mock import patch @@ -7,6 +9,7 @@ import pytest import yaml +from eddypro_batch_processor import core from eddypro_batch_processor.core import ( EddyProBatchProcessor, load_config, @@ -317,3 +320,118 @@ def _mock_copytree(src: Path, dst: Path, dirs_exist_ok: bool = True) -> Path: if __name__ == "__main__": pytest.main([__file__]) + + +class TestSubprocessOutputEchoing: + """EddyPro output must reach the console exactly once. + + `setup_logging` always attaches a stdout StreamHandler, so when `log_output` + is on the logger already puts each line on the console. Printing as well + emitted everything twice, doubling log-file size and halving the interval + between rotations on multi-hour runs. + + These assert on `print` rather than on captured log records: the logger is + called exactly once either way, so a caplog-based test passes against the + bug it is meant to catch. + """ + + MARKER = "EDDYPRO_MARKER_LINE" + + def _run(self, tmp_path, *, log_output): + return core.run_subprocess_with_monitoring( + command=[sys.executable, "-c", f"print('{self.MARKER}')"], + working_dir=tmp_path, + stream_output=True, + log_output=log_output, + monitoring_enabled=False, + output_dir=tmp_path, + ) + + def test_no_direct_echo_when_the_logger_mirrors_output(self, tmp_path, caplog): + """With log_output on, the logger is the single emitter.""" + with ( + patch("builtins.print") as mock_print, + caplog.at_level(logging.INFO, logger="eddypro_batch_processor.eddypro"), + ): + assert self._run(tmp_path, log_output=True) == 0 + + assert not any( + self.MARKER in str(call) for call in mock_print.call_args_list + ), "output was printed as well as logged, so each line appears twice" + logged = [r for r in caplog.records if self.MARKER in r.getMessage()] + assert len(logged) == 1, "the line must still reach the log exactly once" + + def test_direct_echo_survives_when_the_logger_is_silent(self, tmp_path): + """With log_output off, printing is the only route to the console.""" + with patch("builtins.print") as mock_print: + assert self._run(tmp_path, log_output=False) == 0 + + assert any( + self.MARKER in str(call) for call in mock_print.call_args_list + ), "live progress was dropped entirely" + + +class TestCpuAffinity: + """Pinning is an optimisation, so a bad value must never stop a run.""" + + def test_performance_cores_derived_from_core_counts(self): + """8 P-cores (2 threads each) + 4 E-cores = 20 logical / 12 physical.""" + with (patch("eddypro_batch_processor.core.psutil.cpu_count") as cpu_count,): + cpu_count.side_effect = lambda logical=True: 20 if logical else 12 + assert core.resolve_performance_cores() == list(range(16)) + + def test_no_hybrid_split_returns_none(self): + """A uniform SMT CPU (8 cores / 16 threads) has no E-cores to avoid.""" + with patch("eddypro_batch_processor.core.psutil.cpu_count") as cpu_count: + cpu_count.side_effect = lambda logical=True: 16 if logical else 8 + assert core.resolve_performance_cores() is None + + def test_no_smt_returns_none(self): + """Without SMT the thread-count trick cannot identify core types.""" + with patch("eddypro_batch_processor.core.psutil.cpu_count") as cpu_count: + cpu_count.side_effect = lambda logical=True: 8 + assert core.resolve_performance_cores() is None + + def test_absent_setting_leaves_affinity_untouched(self): + with patch("eddypro_batch_processor.core.psutil.Process") as proc: + core.apply_cpu_affinity({}) + proc.assert_not_called() + + def test_explicit_list_is_applied(self): + with patch("eddypro_batch_processor.core.psutil.Process") as proc: + core.apply_cpu_affinity({"cpu_affinity": [0, 1, 2, 3]}) + proc.return_value.cpu_affinity.assert_called_once_with([0, 1, 2, 3]) + + def test_performance_keyword_uses_detection(self): + with ( + patch("eddypro_batch_processor.core.psutil.Process") as proc, + patch( + "eddypro_batch_processor.core.resolve_performance_cores", + return_value=[0, 1], + ), + ): + core.apply_cpu_affinity({"cpu_affinity": "performance"}) + proc.return_value.cpu_affinity.assert_called_once_with([0, 1]) + + def test_performance_keyword_no_split_is_a_no_op(self): + with ( + patch("eddypro_batch_processor.core.psutil.Process") as proc, + patch( + "eddypro_batch_processor.core.resolve_performance_cores", + return_value=None, + ), + ): + core.apply_cpu_affinity({"cpu_affinity": "performance"}) + proc.return_value.cpu_affinity.assert_not_called() + + def test_garbage_value_is_ignored_not_raised(self): + with patch("eddypro_batch_processor.core.psutil.Process") as proc: + core.apply_cpu_affinity({"cpu_affinity": "wibble"}) + core.apply_cpu_affinity({"cpu_affinity": {"a": 1}}) + proc.return_value.cpu_affinity.assert_not_called() + + def test_failure_to_pin_does_not_abort_the_run(self): + """A multi-hour run must not die because affinity could not be set.""" + with patch("eddypro_batch_processor.core.psutil.Process") as proc: + proc.return_value.cpu_affinity.side_effect = OSError("denied") + core.apply_cpu_affinity({"cpu_affinity": [0, 1]}) # must not raise diff --git a/tests/test_monitor.py b/tests/test_monitor.py index e51ed95..3430732 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -5,6 +5,7 @@ dependencies to ensure deterministic behavior. """ +import csv import itertools import json import tempfile @@ -20,6 +21,7 @@ except ImportError: psutil_module = None +from eddypro_batch_processor import monitor as monitor_mod from eddypro_batch_processor.monitor import ( MonitoredOperation, PerformanceMonitor, @@ -537,3 +539,67 @@ def check_samples(): # All threads should have gotten valid results assert len(results) == 10 assert all(isinstance(count, int) for count in results) + + +class TestProgressThroughput: + """The work-rate series must survive the whole wiring path. + + These deliberately go through `create_monitor` and `MonitoredOperation` + rather than constructing `PerformanceMonitor` directly: the first version of + this feature accepted `progress_dir` at every layer but forgot to forward it + to the constructor, so every run silently recorded zero work items. A test + that built the monitor directly would have passed. + """ + + def test_create_monitor_forwards_progress_args(self, tmp_path): + mon = monitor_mod.create_monitor( + output_dir=tmp_path, progress_dir=tmp_path / "work", progress_glob="*.bin" + ) + assert mon is not None + assert mon.progress_dir == tmp_path / "work" + assert mon.progress_glob == "*.bin" + + def test_monitored_operation_forwards_progress_args(self, tmp_path): + op = monitor_mod.MonitoredOperation( + output_dir=tmp_path, progress_dir=tmp_path / "work" + ) + assert op.monitor is not None + assert op.monitor.progress_dir == tmp_path / "work" + + def test_work_items_counted_as_files_appear(self, tmp_path): + """A growing directory must show up as a non-zero work count.""" + work = tmp_path / "work" + work.mkdir() + mon = monitor_mod.create_monitor( + interval_seconds=0.1, output_dir=tmp_path, progress_dir=work + ) + assert mon is not None + mon.start_monitoring() + try: + for i in range(5): + (work / f"item_{i}.dat").write_text("x", encoding="utf-8") + time.sleep(0.15) + finally: + mon.stop_monitoring() + + rows = list( + csv.DictReader( + (tmp_path / "metrics.csv").open(newline="", encoding="utf-8") + ) + ) + counts = [int(float(r["work_items"])) for r in rows if r.get("work_items")] + assert counts, "work_items column was never populated" + assert max(counts) >= 3, f"expected files to be counted, saw {counts}" + + def test_missing_progress_dir_is_not_an_error(self, tmp_path): + """The workload creates the directory, so early samples precede it.""" + mon = monitor_mod.create_monitor( + interval_seconds=0.1, + output_dir=tmp_path, + progress_dir=tmp_path / "never_created", + ) + assert mon is not None + mon.start_monitoring() + time.sleep(0.25) + summary = mon.stop_monitoring() + assert summary # completed without raising
RunCPUMemoryDiskBottleneckCPU p95Peak RAM (MB)BottleneckCPU p95 (machine)CPU p95 (% of 1 core)s / work itemPeak RAM (MB) Read (MB)Write (MB)
{entry.get("scenario_name", "?")} {_dot(entry.get("cpu_status", "UNKNOWN"))} {_dot(entry.get("memory_status", "UNKNOWN"))} {_dot(entry.get("disk_status", "UNKNOWN"))} {entry.get("primary_bottleneck", "?")}{cpu.get("p95", 0):.1f}%{machine_p95:.1f}%{core_cpu.get("p95", 0):.0f}%{entry.get("mean_seconds_per_work_item", 0) or 0:.2f} {entry.get("peak_memory_mb", 0):.0f} {entry.get("total_read_mb", 0):.0f} {entry.get("total_write_mb", 0):.0f}