1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
# Copyright 2024 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import contextlib
import os
import sys
import signal
import subprocess
import gclient_paths
# Keeps track of whether our SIGINT handler was called.
_is_canceled = False
def _register_build_id(local_dev_server_path, build_id, out_dir):
subprocess.run([
local_dev_server_path, '--register-build-id', build_id, '--builder-pid',
str(os.getpid()), '--output-directory',
os.path.abspath(out_dir)
])
def _print_status(local_dev_server_path, build_id):
subprocess.run([local_dev_server_path, '--print-status', build_id])
def _get_server_path():
src_dir = gclient_paths.GetPrimarySolutionPath()
return os.path.join(src_dir, 'build/android/fast_local_dev_server.py')
def _set_signal_handler(local_dev_server_path, build_id):
original_sigint_handler = signal.getsignal(signal.SIGINT)
def _kill_handler(signum, frame):
global _is_canceled
_is_canceled = True
# Cancel the pending build tasks if user CTRL+c early.
print('🛑 Canceling pending build_server tasks', file=sys.stderr)
subprocess.run([local_dev_server_path, '--cancel-build', build_id])
original_sigint_handler(signum, frame)
signal.signal(signal.SIGINT, _kill_handler)
def _start_server(local_dev_server_path):
subprocess.Popen([local_dev_server_path, '--exit-on-idle', '--quiet'],
start_new_session=True)
def _set_tty_env():
# If the caller explicitly set this environment variable, just use it
# instead of failing.
if "AUTONINJA_STDOUT_NAME" in os.environ:
return True
stdout_name = os.readlink('/proc/self/fd/1')
# Anonymous pipes can't be opened. These look like "pipe:[394765110]".
ret = os.path.exists(stdout_name)
if ret:
os.environ.setdefault("AUTONINJA_STDOUT_NAME", stdout_name)
return ret
@contextlib.contextmanager
def build_server_context(build_id, out_dir, use_android_build_server=False):
if not use_android_build_server or not _set_tty_env():
yield
return
server_path = _get_server_path()
_start_server(server_path)
# Tell the build server about us.
_register_build_id(server_path, build_id, out_dir)
_set_signal_handler(server_path, build_id)
yield
# No need to print status if we CTRL+Ced out.
if (not _is_canceled):
_print_status(server_path, build_id)
|
