Coverage for /home/jenkins/.local/lib/python3.10/site-packages/hyper_multicore/_build/command

3# Licensed under the Apache License, Version 2.0 (the "License");

4# you may not use this file except in compliance with the License.

5# You may obtain a copy of the License at

7# http://www.apache.org/licenses/LICENSE-2.0

9# Unless required by applicable law or agreed to in writing, software

10# distributed under the License is distributed on an "AS IS" BASIS,

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

12# See the License for the specific language governing permissions and

13# limitations under the License.

14# ============================================================================

15"""Execute tools with live console output, transcripts and bounded stage logs."""

17import codecs

18import os

19import re

20import shlex

21import shutil

22import signal

23import subprocess

24import sys

25import tempfile

26import threading

27import time

28from contextlib import contextmanager, ExitStack, redirect_stderr, redirect_stdout

29from pathlib import Path

31from .build_utils import atomic_json as write_json

33ROOT = Path(__file__).resolve().parents[2]

34CONSOLE_LOCK = threading.RLock()

37def format_message(text: str, level="INFO") -> str:

38 """Keep controller feedback consistent in terminals, redirected output and CI."""

39 return f"{time.strftime('%H:%M:%S')} {level:<5} {text}\n"

42def message(text: str, level="INFO"):

43 """Write one complete message so tool-reader threads cannot split its prefix."""

44 stream = sys.stderr if level == "ERROR" else sys.stdout

45 with CONSOLE_LOCK:

46 stream.write(format_message(text, level))

47 stream.flush()

50class Transcript:

51 """Mirror an existing stream without taking ownership of its terminal descriptor."""

53 def __init__(self, stream, log, lock):

54 self.stream, self.log, self.lock = stream, log, lock

56 def write(self, text):

57 with self.lock:

58 self.log.write(text)

59 self.log.flush()

60 return self.stream.write(text)

62 def flush(self):

63 with self.lock:

64 self.log.flush()

65 self.stream.flush()

67 def __getattr__(self, name):

68 return getattr(self.stream, name)

71@contextmanager

72def capture_log(path: Path):

73 """Retain controller, stdout and stderr output together, including failure guidance."""

74 lock = threading.RLock()

75 with path.open("w", encoding="utf-8") as log:

76 with redirect_stdout(Transcript(sys.stdout, log, lock)), redirect_stderr(Transcript(sys.stderr, log, lock)):

77 yield

80def stage_title(label: str) -> str:

81 """Translate internal task identifiers into readable progress titles."""

82 label = re.sub(r"cp\d+-[^ ]+-torch[^ ]+-pta([^-]+)-[0-9a-f]{20}", r"PTA \1", label)

83 label = re.sub(r"pt([\d.]+(?:post\d+)?)-[0-9a-f]{12}", r"PTA \1", label)

84 titles = {

85 "package-wheel": "Packaging wheel", "common-resources": "Building shared native resources",

86 "install-adapter-": "Installing adapter: ",

87 "venv-": "Creating framework environment: ", "install-": "Installing framework packages: ",

88 "download-": "Downloading framework wheels: ",

89 "configure-": "Configuring adapter: ",

90 "compile-": "Building adapter: ", "ctest-": "Testing SHMEM: ",

91 }

92 for prefix, title in titles.items():

93 if label.startswith(prefix):

94 return title + label[len(prefix):]

95 return label.replace("-", " ").capitalize()

98def visible_stage(label: str, capture: bool) -> bool:

99 """Internal identity/hash probes remain in stage records rather than terminal progress."""

100 return not capture and not label.startswith("identity-")

101

102

103def compiler_metrics(output: str) -> dict:

104 """Count logged compiler commands and separately expose observable CMake/OPC actions."""

105 commands = 0

106 for line in output.splitlines():

107 if not re.search(r"(?:^|\s)-c(?:\s|$)", line):

108 continue

109 try:

110 tokens = shlex.split(line)

111 except ValueError:

112 continue

113 if "-c" in tokens and any(

114 re.search(r"(?:^|-)(?:g\+\+|c\+\+|clang\+\+|bisheng)(?:-\d+(?:\.\d+)*)?$", Path(token).name)

115 for token in tokens

116 ):

117 commands += 1

118 object_actions = len(re.findall(r"Building (?:C|CXX) object ", output))

119 return {

120 "compiler_invocations": max(commands, object_actions),

121 "compiler_commands_logged": commands,

122 "host_object_compile_actions": object_actions,

123 "kernel_compile_jobs": output.count("Opc tool start working now"),

124 }

125

126

127def stage_failure(label: str, status: int, log: Path, start=0, end=None) -> str:

128 """Surface the first tool diagnostic while keeping full output in its log."""

129 detail = f"Stage {label} failed ({status}); inspect {log}"

130 lines = read_log_range(log, start, end).splitlines()

131 errors = [

132 (i, line)

133 for i, line in enumerate(lines)

135 ]

136 if errors:

137 index = errors[0][0]

138 detail += "\nDiagnostic:\n" + "\n".join(line[:600] for line in lines[index : index + 3])

139 return detail

140

141

142class StageOutput:

143 """Mirror tool logs live and report quiet stages without piping or blocking the child."""

144

145 def __init__(self, log: Path, label: str, started: float, *, heartbeat=10.0, context=""):

146 self.log, self.label, self.started = log, label, started

147 self.heartbeat, self.context = heartbeat, context

148 self.pending = ""

149 self.stopped = threading.Event()

150 self.worker = threading.Thread(target=self.follow, name="native-build-output", daemon=True)

151

152 def __enter__(self):

153 self.worker.start()

154 return self

155

156 def __exit__(self, *_):

157 self.stopped.set()

158 self.worker.join()

159

160 def emit(self, text: str) -> bool:

161 """A closed terminal must not leave the log reader alive or lose disk evidence."""

162 try:

163 if text:

164 with CONSOLE_LOCK:

165 sys.stdout.write(text)

166 sys.stdout.flush()

167 return True

168 except (OSError, ValueError):

169 return False

170

171 def follow(self):

172 """Drain final bytes before exiting; captured machine stdout is never echoed."""

173 decoder = codecs.getincrementaldecoder("utf-8")("replace")

174 last_output = time.monotonic()

175 with self.log.open("rb") as source:

176 while True:

177 data = source.read(8192)

178 if data:

179 if not self.emit_lines(decoder.decode(data)):

180 return

181 last_output = time.monotonic()

182 continue

183 if self.stopped.is_set():

184 break

185 now = time.monotonic()

186 if now - last_output >= self.heartbeat:

187 if not self.emit(

188 format_message(

189 f"{self.context + ': ' if self.context else ''}"

190 f"{stage_title(self.label)}: still running ({now - self.started:.0f}s elapsed)"

191 )

192 ):

193 return

194 last_output = now

195 self.stopped.wait(0.1)

196 self.emit_lines(decoder.decode(b"", final=True), final=True)

197

198 def emit_lines(self, text: str, *, final=False) -> bool:

199 """Keep each contextual line intact even when several children write partial chunks."""

200 if not self.context:

201 return self.emit(text)

202 self.pending += text.replace("\r\n", "\n").replace("\r", "\n")

203 while "\n" in self.pending:

204 line, self.pending = self.pending.split("\n", 1)

205 if not self.emit(f"[{self.context}] {line}\n"):

206 return False

207 if final and self.pending:

208 return self.emit(f"[{self.context}] {self.pending}\n")

209 return True

210

211

212def signal_groups(processes, number):

213 """Signal every group before waiting for any individual child."""

214 for process in processes:

215 try:

216 os.killpg(process.pid, number)

217 except ProcessLookupError:

218 pass

219

220

221def wait_for_groups(processes):

222 """Bound descendant cleanup by one shared deadline, rather than one per target."""

223 groups = {process.pid for process in processes}

224 deadline = time.monotonic() + 10

225 while time.monotonic() < deadline:

226 remaining = False

227 for stat in Path("/proc").glob("[0-9]*/stat"):

228 try:

229 fields = stat.read_text().rsplit(")", 1)[1].split()

230 remaining |= fields[0] != "Z" and int(fields[2]) in groups

231 except (OSError, ValueError, IndexError):

232 continue

233 if not remaining:

234 return

235 time.sleep(0.05)

236

237

238def stop_stages(processes) -> None:

239 """Terminate all active groups under common grace periods, including descendants."""

240 signal_groups(processes, signal.SIGTERM)

241 deadline = time.monotonic() + 15

242 while any(process.poll() is None for process in processes) and time.monotonic() < deadline:

243 time.sleep(0.05)

244 signal_groups(processes, signal.SIGKILL)

245 for process in processes:

246 process.wait()

247 wait_for_groups(processes)

248

249

250def read_log_range(path: Path, start=0, end=None) -> str:

251 """Decode only this command's byte interval, never preceding tool output."""

252 with path.open("rb") as source:

253 source.seek(start)

254 return source.read(-1 if end is None else end - start).decode("utf-8", errors="replace")

255

256

257class CommandCancelled(RuntimeError):

258 """An otherwise valid command was stopped by another failure or user cancellation."""

259

260

261class Commands:

262 """Execute concurrent tools while serializing raw-log blocks and stage records."""

263

264 def __init__(self, directory: Path, *, root: Path = ROOT, events=None, scratch=None):

265 self.directory = directory

266 self.directory.mkdir(parents=True, exist_ok=True)

267 self.log = directory / "tools.log"

268 self.log.touch(exist_ok=True)

269 self.root = root

270 self.events = [] if events is None else events

271 self.scratch = scratch or directory

272 self.cancelled = threading.Event()

273 self.failure = None

274 self._processes = {}

275 self._process_lock = threading.RLock()

276 self._output_lock = threading.RLock()

277 self._local = threading.local()

278

279 @contextmanager

280 def scope(self, name):

281 """Attach an opaque output label to one calling thread, without shared environment mutation."""

282 previous = getattr(self._local, "context", "")

283 self._local.context = f"{previous} / {name}" if previous else name

284 try:

285 yield

286 finally:

287 self._local.context = previous

288

289 def add_event(self, record):

290 """Append complete stage metadata and replace JSON under the same writer lock."""

291 with self._output_lock:

292 self.events.append(record)

293 write_json(self.directory / "stages.json", self.events)

294

295 def cancel(self, error):

296 """The first cause wins; no process may start after the cancellation snapshot."""

297 with self._process_lock:

298 if self.cancelled.is_set():

299 return

300 self.failure = error

301 self.cancelled.set()

302 processes = tuple(self._processes.values())

303 if processes:

304 message(f"Stopping {len(processes)} active tool process groups", "WARN")

305 stop_stages(processes)

306

307 @contextmanager

308 def start(self, command, output, env, capture):

309 """Register starts atomically and scope process resources to execution."""

310 with ExitStack() as resources:

311 with self._process_lock:

312 if self.cancelled.is_set():

313 raise CommandCancelled("Build cancelled; no further commands will start")

314 process = resources.enter_context(subprocess.Popen(

315 command, cwd=self.root, env=env, stdout=subprocess.PIPE if capture else output,

316 stderr=output, text=True, start_new_session=True,

317 ))

318 self._processes[process.pid] = process

319 yield process

320

321 def record(self, command, label, started, wall_started, spool, status, error=None):

322 """Merge one complete raw block and give it an exclusive byte interval."""

323 text = spool.read_text(encoding="utf-8", errors="replace")

324 record = {

325 "stage": label, "context": getattr(self._local, "context", ""),

326 "started_at": wall_started, "ended_at": time.time(),

327 "elapsed_seconds": time.monotonic() - started, "exit_code": status,

328 "status": command_status(status, error), "log": str(self.log), "command": list(command),

329 "disk_free_bytes": shutil.disk_usage(self.directory).free,

330 **compiler_metrics(text),

331 }

332 if error:

333 record["error"] = str(error) or type(error).__name__

334 with self._output_lock:

335 with self.log.open("ab") as output, spool.open("rb") as source:

336 record["log_start"] = output.tell()

337 shutil.copyfileobj(source, output)

338 record["log_end"] = output.tell()

339 self.add_event(record)

340 return record

341

342 def execute(self, command, label, spool, started, env, capture):

343 """Each child owns its output file; only terminal rendering is shared while running."""

344 stdout = None

345 with spool.open("wb", buffering=0) as output, self.start(command, output, env, capture) as process:

346 try:

347 with StageOutput(spool, label, started,

348 context=getattr(self._local, "context", "")):

349 stdout, _ = process.communicate()

350 except BaseException as error:

351 self.cancel(error)

352 if capture:

353 stdout, _ = process.communicate()

354 raise

355 finally:

356 if process.returncode and not self.cancelled.is_set():

357 stop_stages((process,))

358 with self._process_lock:

359 self._processes.pop(process.pid, None)

360 if capture and stdout:

361 output.write(bytes(stdout, "utf-8"))

362 return stdout, process.returncode

363

364 def __call__(self, command, *, label: str, env=None, capture=False):

365 if self.cancelled.is_set():

366 raise CommandCancelled("Build cancelled; no further commands will start")

367 self.scratch.mkdir(parents=True, exist_ok=True)

368 with tempfile.NamedTemporaryFile(dir=self.scratch, prefix="tool-", suffix=".tmp", delete=False) as file:

369 spool = Path(file.name)

370 started, wall_started = time.monotonic(), time.time()

371 stdout, status, error = None, None, None

372 title = stage_title(label)

373 context = getattr(self._local, "context", "")

374 if context:

375 title = f"{context}: {title}"

376 visible = visible_stage(label, capture)

377 if visible:

378 message(title)

379 try:

380 try:

381 stdout, status = self.execute(command, label, spool, started, env, capture)

382 if self.cancelled.is_set():

383 error = CommandCancelled("Stage cancelled after another failure or interruption")

384 except BaseException as caught:

385 error = caught

386 raise

387 finally:

388 record = self.record(command, label, started, wall_started, spool, status, error)

389 finally:

390 spool.unlink(missing_ok=True)

391 if error is not None:

392 raise error

393 if status or visible:

394 outcome = "failed" if status else "completed"

395 message(f"{title}: {outcome} in {record['elapsed_seconds']:.1f}s (exit {status})",

396 "ERROR" if status else "INFO")

397 if status:

398 diagnostic = stage_failure(label, status, self.log, record["log_start"], record["log_end"])

399 raise RuntimeError(f"{diagnostic}\nCommand executable: {command[0]}")

400 return stdout

401

402

403def command_status(status, error):

404 """Keep collateral cancellation distinct from the initiating command failure."""

405 if isinstance(error, KeyboardInterrupt):

406 return "interrupted"

407 if isinstance(error, CommandCancelled):

408 return "cancelled"

409 return "failed" if error or status else "succeeded"

Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _build / command_runner.py: 0%

276 statements