diff --git a/README.md b/README.md index ead3ec69..a4b64353 100644 --- a/README.md +++ b/README.md @@ -111,4 +111,14 @@ python3 -m venv .venv - [插件开发](docs/development/plugin-development.md) - [更新日志](CHANGELOG.md) -WaveBench 使用 MIT 许可证。感谢 Linux DO 社区提供交流和支持。 +WaveBench 使用 MIT 许可证。 + +## 特别感谢 + +linuxDoLogo + +感谢 [Linux DO 社区](https://linux.do/)提供交流和支持。 + +Krill + +感谢 [Krill AI](https://www.krill-code.com/) 对本项目的赞助。 \ No newline at end of file diff --git a/src/wavebench/services/run_pipeline.py b/src/wavebench/services/run_pipeline.py index de805092..656ee920 100644 --- a/src/wavebench/services/run_pipeline.py +++ b/src/wavebench/services/run_pipeline.py @@ -37,6 +37,7 @@ from wavebench.services.run_analysis import evaluate_expect from wavebench.services.run_artifacts import RunStepRecord from wavebench.services.run_plan import RunPlan, RunStep +from wavebench.services.platform_io import _replace_file ANALYSIS_PIPELINE_SCHEMA = "wavebench.analysis_pipeline.v1" @@ -765,7 +766,7 @@ def blocks(): file.write(encoded) file.flush() os.fsync(file.fileno()) - os.replace(temporary, path) + _replace_file(temporary, path) if budget: budget.committed_file(path.stat().st_size) finally: @@ -796,7 +797,7 @@ def blocks(): budget.pending_file(file.tell()) file.flush() os.fsync(file.fileno()) - os.replace(temporary, path) + _replace_file(temporary, path) if budget: budget.committed_file(path.stat().st_size) finally: @@ -812,7 +813,7 @@ def _atomic_write_bytes(path: Path, data: bytes, *, budget: AnalysisBudget | Non file.write(data) file.flush() os.fsync(file.fileno()) - os.replace(temporary, path) + _replace_file(temporary, path) if budget: budget.committed_file(len(data)) finally: diff --git a/tests/test_analysis_resources.py b/tests/test_analysis_resources.py index 9457461d..b2b9b929 100644 --- a/tests/test_analysis_resources.py +++ b/tests/test_analysis_resources.py @@ -3,6 +3,7 @@ from hashlib import sha256 import io from unittest.mock import patch +from types import SimpleNamespace import numpy as np import pytest @@ -15,7 +16,7 @@ from wavebench.errors import ConfigError, DataError, ExecutionIntentError, error_envelope from wavebench.services.analysis_service import run_analysis, check_analysis from wavebench.services.execution_intent import build_execution_intent, verify_execution_intent -from wavebench.services.run_pipeline import _atomic_write_csv, _export_signal +from wavebench.services.run_pipeline import _atomic_write_csv, _atomic_write_npy, _atomic_write_bytes, _export_signal from wavebench.report.analysis import read_curve, write_analysis_report from test_analysis_service import analysis_input as analysis_input from test_psd_pipeline import PARAMS, plan_for, EXPORT @@ -147,12 +148,64 @@ def test_write_limit_and_io_failure_cleanup(tmp_path): _atomic_write_csv(path, ["x", "y"], np.ones((100, 2)), budget=AnalysisBudget(replace(AnalysisLimits(), max_output_bytes=10))) assert path.read_text() == "keep original" - with patch("wavebench.services.run_pipeline.os.replace", side_effect=OSError("disk full")): + with patch("wavebench.services.run_pipeline._replace_file", side_effect=OSError("disk full")): with pytest.raises(OSError): _atomic_write_csv(path, ["x", "y"], np.ones((2, 2))) assert list(tmp_path.iterdir()) == [path] +@pytest.mark.parametrize("kind", ["bytes", "npy", "csv"]) +@pytest.mark.parametrize("persistent", [False, True]) +def test_analysis_windows_replace_contention(tmp_path, monkeypatch, kind, persistent): + from unittest.mock import Mock + from wavebench.services import platform_io + + path = tmp_path / "result" + path.write_bytes(b"keep original") + real_replace = platform_io.os.replace + attempts = [] + + def move(source, target, flags): + attempts.append((source, target)) + assert path.read_bytes() == b"keep original" + if persistent or len(attempts) == 1: + return 0 + real_replace(source, target) + return 1 + + kernel32 = SimpleNamespace(MoveFileExW=Mock(side_effect=move)) + monkeypatch.setattr(platform_io, "os", SimpleNamespace(name="nt")) + monkeypatch.setattr(platform_io.ctypes, "WinDLL", lambda *a, **kw: kernel32, raising=False) + monkeypatch.setattr(platform_io.ctypes, "get_last_error", lambda: 32, raising=False) + sleep = Mock() + monkeypatch.setattr(platform_io.time, "sleep", sleep) + budget = AnalysisBudget(AnalysisLimits()) + + def write(): + if kind == "bytes": + _atomic_write_bytes(path, b"new contents", budget=budget) + elif kind == "npy": + _atomic_write_npy(path, np.ones((2, 2)), budget=budget) + else: + _atomic_write_csv(path, ["x", "y"], np.ones((2, 2)), budget=budget) + + if persistent: + with pytest.raises(OSError, match="MoveFileExW failed"): + write() + assert path.read_bytes() == b"keep original" + assert budget.output_files == budget.output_bytes == 0 + assert len(attempts) == 3 + else: + write() + assert path.read_bytes() != b"keep original" + assert budget.output_files == 1 + assert budget.output_bytes == path.stat().st_size + assert len(attempts) == 2 + assert sleep.call_count == len(attempts) - 1 + assert len(set(attempts)) == 1 + assert list(tmp_path.iterdir()) == [path] + + def test_runtime_fft_rejected_before_call_and_previous_export_retained(tmp_path, analysis_input): capture, recipe = analysis_input recipe.write_text('schema="wavebench.analysis_recipe.v1"\noperations=[{op="export",name="before",formats=["npy"]},{op="fft"},{op="export",name="after",formats=["npy"]}]\n[resources]\nmax_fft_length=64\n')