Skip to content

Commit 2e129ca

Browse files
Stream pull request diffs to disk, and prune CI's uploaded runs []
The forge service held a pull request's tests/ diff in memory several times over: as git's output, decoded, stripped, and split into lines. The diff is now processed in a streamed fashion. Since the diff now costs disk rather than memory, its default limit rises to 512 MiB. A failing CI leg's upload held every test's .rt.tar.gz (edgy creates a .rt.tar.gz for every test, this can be useful for analysis but is not useful for AcknowlEDG Forge and results in larger amounts of dating needing to be stored and processed. Now, before uploading, the leg now runs the new edg-prune-test-run (edgtest.prune_run_archives), which removes .rt.tar.gz files not associated with a changes.elist file.
1 parent 66607c4 commit 2e129ca

11 files changed

Lines changed: 460 additions & 87 deletions

File tree

‎.github/workflows/ci-toolchain.yml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -267,6 +267,17 @@ jobs:
267267
268268
edg-prefix-env-loader .tmp/extensive-ci.edgivk
269269
270+
# Reviews (acknowledg-forge, edgy-review) read only the .rt.tar.gz
271+
# files changes.elist names; edgy -A archives every test's. Pruning
272+
# must never keep the results from being uploaded, hence
273+
# continue-on-error.
274+
- name: Prune .rt.tar.gz results changes.elist doesn't name
275+
if: ${{ failure() }}
276+
continue-on-error: true
277+
env:
278+
RUN_TIMESTAMP: ${{ inputs.run_timestamp }}
279+
run: edg-prune-test-run "tests/runs/${RUN_TIMESTAMP}"
280+
270281
- name: Upload partial elist, elog, and .rt.tar.gz results
271282
if: ${{ failure() }}
272283
uses: actions/upload-artifact@v7

‎dev_tools/bin/edg-prune-test-run‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
#!/usr/bin/env python3
2+
3+
# Part of the EDG Compiler Project, under the Apache License v2.0 with LLVM
4+
# Exceptions.
5+
# See https://edgcpp.org/LICENSE.txt for license information.
6+
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
7+
8+
'''Prune a test run down to what a review of it reads, e.g., before CI
9+
uploads the run:
10+
11+
edg-prune-test-run tests/runs/<run timestamp>
12+
13+
Reviews of a run read only the .rt.tar files its changes.elist files name.
14+
The replays directory is removed whole. In each configuration, a suite
15+
with no test named there is removed whole (its directory, and its
16+
suite-<name> logs), as is every suite of a configuration without a
17+
changes.elist; of the other suites, the .rt.tar files not named there are
18+
removed.
19+
'''
20+
21+
import argparse
22+
import sys
23+
24+
import edgtest
25+
import edgutil
26+
27+
from pathlib import Path
28+
29+
def _parse_args() -> argparse.Namespace:
30+
parser = argparse.ArgumentParser(
31+
description = (
32+
"Remove the .rt.tar files of a test run that no configuration's "
33+
'changes.elist names.'
34+
)
35+
)
36+
parser.add_argument(
37+
'run_dir',
38+
type = Path,
39+
help = 'the run directory (e.g., tests/runs/<run timestamp>)'
40+
)
41+
return parser.parse_args()
42+
43+
def main() -> int:
44+
args = _parse_args()
45+
if not args.run_dir.is_dir():
46+
edgutil.eprint(f"error: {args.run_dir} is not a directory")
47+
return 1
48+
49+
# Counts: of .rt.tar files kept and removed, and of suites removed.
50+
kept, removed, suites = edgtest.prune_run_archives(args.run_dir)
51+
print(
52+
f"Kept {kept} .rt.tar files named by changes.elist files; removed "
53+
f"{removed} others, and {suites} suites with no changes."
54+
)
55+
return 0
56+
57+
if __name__ == '__main__':
58+
sys.exit(main())

‎dev_tools/pylibs/edgacknowledg/forge/config.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,9 @@
3232
# suite's name, is short.
3333
DEFAULT_MAX_PATTERNS = 20_000
3434
# The largest diff (of tests/, in bytes) a pull request may have: it is
35-
# held in memory, and copied and split into lines more than once.
36-
DEFAULT_MAX_DIFF_BYTES = 64 * 1024 * 1024
35+
# kept with the review's rematerialized run (on disk, never in memory
36+
# whole), and the builder applies all of it.
37+
DEFAULT_MAX_DIFF_BYTES = 512 * 1024 * 1024
3738
# The most entries an artifact may have (its zip's, and its tarballs'
3839
# together), and the largest its zip's central directory may be: zipfile
3940
# reads all of that directory into memory, before we can count. CI's test

‎dev_tools/pylibs/edgacknowledg/forge/processor.py‎

Lines changed: 39 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -397,42 +397,54 @@ async def _changes_section(self,
397397
if reused is not None:
398398
return reused
399399

400-
diff = await self._workspaces.diff(
401-
info.repository,
402-
merge_base,
403-
head_sha,
404-
['tests/']
405-
)
406-
if diff.strip() == '':
407-
return None
408-
409-
worktree = await self._workspaces.checkout(
410-
info,
411-
record.review_id,
412-
'base',
413-
merge_base,
414-
changes_patterns(
415-
diff,
416-
max_patterns = self._config.max_sparse_patterns
417-
)
418-
)
419-
runs_dir = ensure_real_directory(worktree, PurePosixPath('tests', 'runs'))
420-
run_dir = runs_dir / f"remat-{_sha256(source_key.encode())[:16]}"
421-
remove_tree(run_dir)
422-
423400
# The builder sees the worktree read-only, so it writes elsewhere (on
424401
# the same file system, for the rename into place).
402+
run_name = f"remat-{_sha256(source_key.encode())[:16]}"
425403
staging_root = ensure_real_directory(
426404
self._workspaces.build_staging_dir.parent,
427405
PurePosixPath(self._workspaces.build_staging_dir.name)
428406
)
429407
staging = Path(tempfile.mkdtemp(
430-
prefix = f"{record.review_id}-{run_dir.name}-",
408+
prefix = f"{record.review_id}-{run_name}-",
431409
dir = staging_root
432410
))
433411
try:
412+
# Only ever in this file, never held in memory whole: it may be large.
434413
diff_file = staging / 'pull-request.diff'
435-
diff_file.write_text(diff, encoding = 'utf-8')
414+
await self._workspaces.diff(
415+
info.repository,
416+
merge_base,
417+
head_sha,
418+
['tests/'],
419+
diff_file
420+
)
421+
if diff_file.stat().st_size == 0:
422+
remove_tree(staging)
423+
return None
424+
with open(diff_file, encoding = 'utf-8') as handle:
425+
patterns = changes_patterns(
426+
handle,
427+
max_patterns = self._config.max_sparse_patterns
428+
)
429+
430+
worktree = await self._workspaces.checkout(
431+
info,
432+
record.review_id,
433+
'base',
434+
merge_base,
435+
patterns
436+
)
437+
# Prepare where the run goes once built: tests/runs/<run name> in the
438+
# worktree (through real directories only, not links the pull request
439+
# committed), cleared of any earlier attempt so that the staging
440+
# directory can be renamed into place.
441+
runs_dir = ensure_real_directory(
442+
worktree,
443+
PurePosixPath('tests', 'runs')
444+
)
445+
run_dir = runs_dir / run_name
446+
remove_tree(run_dir)
447+
436448
timestamp = self._clock().strftime(edgutil.RUN_TIMESTAMP_FORMAT)
437449
result = await self._sandbox.run(
438450
[
@@ -477,6 +489,7 @@ async def _changes_section(self,
477489
)
478490
remove_tree(staging)
479491
return None
492+
fingerprint = await asyncio.to_thread(_file_sha256, diff_file)
480493
staging.rename(run_dir)
481494
except BaseException:
482495
remove_tree(staging)
@@ -487,7 +500,7 @@ async def _changes_section(self,
487500
name = CHANGES_NAME,
488501
slug = section_slug(CHANGES_NAME),
489502
source_key = source_key,
490-
content_fingerprint = _sha256(diff.encode()),
503+
content_fingerprint = fingerprint,
491504
root = run_dir / timestamp,
492505
home = worktree / 'tests',
493506
details = {'run_dir': str(run_dir)},

‎dev_tools/pylibs/edgacknowledg/forge/workspace.py‎

Lines changed: 72 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626

2727
import asyncio
2828
import base64
29+
import codecs
2930
import logging
3031
import os
3132
import re
@@ -39,7 +40,7 @@
3940

4041
from dataclasses import dataclass
4142
from pathlib import Path, PurePosixPath
42-
from typing import Iterable, Optional, Sequence
43+
from typing import Callable, Iterable, Optional, Sequence
4344

4445
import edgtest
4546
import edgutil
@@ -81,6 +82,10 @@ class UnsafeArchiveError(WorkspaceError):
8182
'''Raised for an artifact that is too large or has unsafe entries.'''
8283

8384

85+
class _OutputTooLarge(Exception):
86+
'''Raised when a process's output is over its limit (it is stopped).'''
87+
88+
8489
@dataclass(frozen = True)
8590
class GitResult:
8691
returncode: int
@@ -140,16 +145,18 @@ def changed_paths(diff: str) -> list[str]:
140145
return paths
141146

142147

143-
def changes_patterns(diff: str,
148+
def changes_patterns(lines_iter: Iterable[str],
144149
*,
145150
max_patterns: int = DEFAULT_MAX_PATTERNS) -> list[str]:
146-
'''Sparse-checkout patterns for rematerializing ``diff`` (of tests/) on
147-
the base tree: the test home and every affected test; at most
151+
'''Sparse-checkout patterns for rematerializing the diff (of tests/) of
152+
``lines_iter`` (with or without their line ends, e.g. a file's) on the
153+
base tree: the test home and every affected test; at most
148154
``max_patterns``.
149155
'''
150156
patterns = dict.fromkeys(_TEST_HOME_PATTERNS)
151157
current_suite: Optional[str] = None
152-
for line in diff.splitlines():
158+
for line in lines_iter:
159+
line = line.rstrip('\n')
153160
match = _DIFF_HEADER.match(line)
154161
if match is not None:
155162
path = PurePosixPath(match.group(1))
@@ -478,9 +485,13 @@ async def git(self,
478485
cwd: Optional[Path] = None,
479486
stdin: Optional[str] = None,
480487
check: bool = True,
481-
max_output: Optional[int] = None) -> GitResult:
482-
'''Run git; with ``max_output``, its output may be at most that many
483-
bytes (it is stopped, and ``WorkspaceError`` raised, beyond).
488+
max_output: Optional[int] = None,
489+
stdout_sink: Optional[Callable[[bytes], None]] = None
490+
) -> GitResult:
491+
'''Run git; with ``stdout_sink``, its output goes there as it comes,
492+
rather than into the result, and may be at most ``max_output`` bytes
493+
(it is stopped, and ``_OutputTooLarge`` raised, beyond: the caller
494+
knows which limit that is).
484495
'''
485496
command = ['git']
486497
if git_dir is not None:
@@ -497,18 +508,15 @@ async def git(self,
497508
cwd = cwd,
498509
env = self._git_env()
499510
)
500-
if max_output is None:
511+
if stdout_sink is None:
512+
assert max_output is None
501513
stdout, stderr = await process.communicate(
502514
stdin.encode() if stdin is not None else None
503515
)
504516
else:
505-
assert stdin is None
506-
bounded = await _communicate_bounded(process, max_output)
507-
if bounded is None:
508-
raise WorkspaceError(
509-
f"git {args[0]}'s output is over {max_output} bytes"
510-
)
511-
stdout, stderr = bounded
517+
assert stdin is None and max_output is not None
518+
stderr = await _communicate_bounded(process, max_output, stdout_sink)
519+
stdout = b''
512520
assert process.returncode is not None
513521
result = GitResult(
514522
process.returncode,
@@ -609,15 +617,42 @@ async def diff(self,
609617
repository: str,
610618
base: str,
611619
head: str,
612-
paths: Sequence[str]) -> str:
620+
paths: Sequence[str],
621+
destination: Path) -> None:
622+
'''Write the diff of ``paths`` from ``base`` to ``head`` to (the new
623+
file) ``destination``, as it comes (it may be large): UTF-8, with what
624+
isn't replaced (U+FFFD).
625+
'''
626+
decoder = codecs.getincrementaldecoder('utf-8')(errors = 'replace')
613627
async with self.lock(repository):
614-
result = await self.git(
615-
'diff', '--no-renames', '--no-color', '--no-ext-diff', '--no-textconv',
616-
base, head, '--', *paths,
617-
git_dir = self.clone_dir(repository),
618-
max_output = self._max_diff_bytes
619-
)
620-
return result.stdout
628+
with open(destination, 'xb') as handle:
629+
630+
# Decoded and encoded again so that what isn't UTF-8 is replaced:
631+
# the builder's edg-rematerialize-test-run (and changes_patterns)
632+
# read it as UTF-8 text, and the Changes section's fingerprint
633+
# (the file's) stays what it was when the diff was decoded whole.
634+
# Incrementally, so a character split between chunks survives.
635+
636+
def write(chunk: bytes) -> None:
637+
handle.write(decoder.decode(chunk).encode())
638+
639+
try:
640+
await self.git(
641+
'diff', '--no-renames', '--no-color', '--no-ext-diff',
642+
'--no-textconv', base, head, '--', *paths,
643+
git_dir = self.clone_dir(repository),
644+
max_output = self._max_diff_bytes,
645+
stdout_sink = write
646+
)
647+
except _OutputTooLarge:
648+
raise WorkspaceError(
649+
f"the diff is over {self._max_diff_bytes} bytes "
650+
f"({ENV_PREFIX}MAX_DIFF_BYTES)"
651+
) from None
652+
653+
# Finish decoding: bytes the decoder still holds (a character the
654+
# output ended in the middle of) are written too, replaced.
655+
handle.write(decoder.decode(b'', final = True).encode())
621656

622657
async def checkout(self,
623658
info: PullRequestInfo,
@@ -702,31 +737,31 @@ async def collect_garbage(self, repository: str, open_count: int) -> None:
702737

703738

704739
async def _communicate_bounded(process: asyncio.subprocess.Process,
705-
limit: int) -> Optional[tuple[bytes, bytes]]:
706-
'''``process``'s output (stdout, stderr), once it exits; ``None``, having
707-
stopped it, should its stdout be over ``limit`` bytes. Should this be
708-
cancelled (or fail), it is stopped too, rather than left writing to a
709-
pipe no one reads.
740+
limit: int,
741+
sink: Callable[[bytes], None]) -> bytes:
742+
'''Pass ``process``'s stdout to ``sink`` as it comes, in chunks (none held
743+
here); return its stderr once it exits. Should its stdout be over
744+
``limit`` bytes, it is stopped, and ``_OutputTooLarge`` raised. Should
745+
this be cancelled (or fail), it is stopped too, rather than left writing
746+
to a pipe no one reads.
710747
'''
711748
assert process.stdout is not None and process.stderr is not None
712-
errors = asyncio.create_task(process.stderr.read())
713-
chunks: list[bytes] = []
749+
reading_stderr = asyncio.create_task(process.stderr.read())
714750
size = 0
715751
try:
716752
while chunk := await process.stdout.read(64 * 1024):
717753
size += len(chunk)
718754
if size > limit:
719-
await edgutil.async_terminate_within(process, _TERMINATE_ALLOWANCE)
720-
return None
721-
chunks.append(chunk)
722-
output = b''.join(chunks), await errors
755+
raise _OutputTooLarge()
756+
sink(chunk)
757+
stderr = await reading_stderr
723758
await process.wait()
724-
return output
759+
return stderr
725760
except BaseException:
726761
await edgutil.async_terminate_within(process, _TERMINATE_ALLOWANCE)
727762
raise
728763
finally:
729-
errors.cancel()
764+
reading_stderr.cancel()
730765

731766

732767
def basic_auth_header(user: str, password: str) -> str:

0 commit comments

Comments
 (0)