|
3 | 3 | import concurrent.futures |
4 | 4 | import contextlib |
5 | 5 | import datetime |
| 6 | +import io |
6 | 7 | import json |
7 | 8 | import logging |
8 | 9 | import operator |
9 | 10 | import pathlib |
10 | 11 | import shutil |
11 | 12 | import tempfile |
| 13 | +import time |
12 | 14 | import typing |
13 | 15 |
|
14 | 16 | from packaging.requirements import Requirement |
@@ -104,6 +106,16 @@ def __init__( |
104 | 106 | self._stack_filename = self.ctx.work_dir / "bootstrap-stack.json" |
105 | 107 | logger.info("recording bootstrap stack state to %s", self._stack_filename) |
106 | 108 |
|
| 109 | + # Single-threaded pool for background file writes (serialized, never interleave) |
| 110 | + self._write_pool: concurrent.futures.ThreadPoolExecutor | None = ( |
| 111 | + concurrent.futures.ThreadPoolExecutor( |
| 112 | + max_workers=1, thread_name_prefix="fromager-writes" |
| 113 | + ) |
| 114 | + ) |
| 115 | + # Throttle stack-state writes to at most once per interval |
| 116 | + self._last_stack_write: float = 0.0 |
| 117 | + self._STACK_WRITE_INTERVAL: float = 5.0 |
| 118 | + |
107 | 119 | # Track failed packages in test mode (list of typed dicts for JSON export) |
108 | 120 | self.failed_packages: list[FailureRecord] = [] |
109 | 121 |
|
@@ -850,7 +862,7 @@ def add_to_graph( |
850 | 862 | pre_built=pbi.pre_built, |
851 | 863 | constraint=self.ctx.constraints.get_constraint(req.name), |
852 | 864 | ) |
853 | | - self.ctx.write_to_graph_to_file() |
| 865 | + self._write_graph_async() |
854 | 866 |
|
855 | 867 | def _sort_requirements( |
856 | 868 | self, |
@@ -954,21 +966,31 @@ def add_to_build_order( |
954 | 966 | if req.url: |
955 | 967 | info["source_url"] = req.url |
956 | 968 | self._build_stack.append(info) |
957 | | - with open(self._build_order_filename, "w") as f: |
958 | | - # Set default=str because the why value includes |
959 | | - # Requirement and Version instances that can't be |
960 | | - # converted to JSON without help. |
961 | | - json.dump(self._build_stack, f, indent=2, default=str) |
| 969 | + |
| 970 | + def _schedule_write(self, path: pathlib.Path, content: str) -> None: |
| 971 | + """Submit a pre-serialized write to the background write pool.""" |
| 972 | + if self._write_pool is not None: |
| 973 | + self._write_pool.submit(path.write_text, content, "utf-8") |
| 974 | + |
| 975 | + def _write_graph_async(self) -> None: |
| 976 | + """Serialize the dependency graph on the main thread, write it in background.""" |
| 977 | + buf = io.StringIO() |
| 978 | + self.ctx.dependency_graph.serialize(buf) |
| 979 | + self._schedule_write(self.ctx.graph_file, buf.getvalue()) |
962 | 980 |
|
963 | 981 | def _record_stack_state(self, stack: list[Phase]) -> None: |
964 | 982 | """Write the current bootstrap stack to `self._stack_filename`. |
965 | 983 |
|
966 | 984 | Index 0 in the output corresponds to `stack[-1]`, the next item to be |
967 | | - processed. Overwrites the file on each call. |
| 985 | + processed. Throttled to at most once every ``_STACK_WRITE_INTERVAL`` seconds. |
968 | 986 | """ |
| 987 | + now = time.monotonic() |
| 988 | + if now - self._last_stack_write < self._STACK_WRITE_INTERVAL: |
| 989 | + return |
969 | 990 | records = [item.as_json() for item in reversed(stack)] |
970 | 991 | with open(self._stack_filename, "w") as f: |
971 | 992 | json.dump(records, f, indent=2, default=str) |
| 993 | + self._last_stack_write = now |
972 | 994 |
|
973 | 995 | # ---- Iterative bootstrap: phase handlers and helpers ---- |
974 | 996 |
|
@@ -1198,7 +1220,7 @@ def _handle_phase_error( |
1198 | 1220 | self._seen_requirements.discard( |
1199 | 1221 | self._resolved_key(wi.req, wi.resolved_version, "wheel") |
1200 | 1222 | ) |
1201 | | - self.ctx.write_to_graph_to_file() |
| 1223 | + self._write_graph_async() |
1202 | 1224 | return [] |
1203 | 1225 |
|
1204 | 1226 | # Normal mode: fail-fast |
@@ -1246,6 +1268,21 @@ def finalize(self) -> int: |
1246 | 1268 | self._bg_pool.shutdown(wait=True, cancel_futures=True) |
1247 | 1269 | self._bg_pool = None |
1248 | 1270 |
|
| 1271 | + # Write build-order once at the end (data was buffered in _build_stack) |
| 1272 | + with open(self._build_order_filename, "w") as f: |
| 1273 | + # Set default=str because values may include Requirement and Version |
| 1274 | + # instances that can't be converted to JSON without help. |
| 1275 | + json.dump(self._build_stack, f, indent=2, default=str) |
| 1276 | + |
| 1277 | + # Final graph write and stack flush, then drain the write pool |
| 1278 | + self._write_graph_async() |
| 1279 | + records: list[typing.Any] = [] |
| 1280 | + with open(self._stack_filename, "w") as f: |
| 1281 | + json.dump(records, f, indent=2) |
| 1282 | + if self._write_pool is not None: |
| 1283 | + self._write_pool.shutdown(wait=True, cancel_futures=False) |
| 1284 | + self._write_pool = None |
| 1285 | + |
1249 | 1286 | if self.multiple_versions and self._failed_versions: |
1250 | 1287 | self._log_failed_versions_table() |
1251 | 1288 |
|
@@ -1284,3 +1321,6 @@ def __exit__( |
1284 | 1321 | if self._bg_pool is not None: |
1285 | 1322 | self._bg_pool.shutdown(wait=False, cancel_futures=True) |
1286 | 1323 | self._bg_pool = None |
| 1324 | + if self._write_pool is not None: |
| 1325 | + self._write_pool.shutdown(wait=False, cancel_futures=True) |
| 1326 | + self._write_pool = None |
0 commit comments