mirror of
https://github.com/capstone-engine/llvm-capstone.git
synced 2024-11-26 23:21:11 +00:00
0d3c40b82b
VSCode's Pylance extension informed me, and text searching confirmed, that these imports are unused. I believe we should be able to remove them harmlessly.
257 lines
10 KiB
Python
Executable File
257 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
#===----------------------------------------------------------------------===##
|
|
#
|
|
# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
|
|
# See https://llvm.org/LICENSE.txt for license information.
|
|
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
|
|
#
|
|
#===----------------------------------------------------------------------===##
|
|
|
|
"""adb_run.py is a utility for running a libc++ test program via adb.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
from typing import List, Tuple
|
|
|
|
|
|
# Sync a host file /path/to/dir/file to ${REMOTE_BASE_DIR}/run-${HASH}/dir/file.
|
|
REMOTE_BASE_DIR = "/data/local/tmp/adb_run"
|
|
|
|
g_job_limit_socket = None
|
|
g_verbose = False
|
|
|
|
|
|
def run_adb_sync_command(command: List[str]) -> None:
|
|
"""Run an adb command and discard the output, unless the command fails. If
|
|
the command fails, dump the output instead, and exit the script with
|
|
failure.
|
|
"""
|
|
if g_verbose:
|
|
sys.stderr.write(f"running: {shlex.join(command)}\n")
|
|
proc = subprocess.run(command, universal_newlines=True,
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, encoding="utf-8")
|
|
if proc.returncode != 0:
|
|
# adb's stdout (e.g. for adb push) should normally be discarded, but
|
|
# on failure, it should be shown. Print it to stderr because it's
|
|
# unrelated to the test program's stdout output. A common error caught
|
|
# here is "No space left on device".
|
|
sys.stderr.write(f"{proc.stdout}\n"
|
|
f"error: adb command exited with {proc.returncode}: "
|
|
f"{shlex.join(command)}\n")
|
|
sys.exit(proc.returncode)
|
|
|
|
|
|
def sync_test_dir(local_dir: str, remote_dir: str) -> None:
|
|
"""Sync the libc++ test directory on the host to the remote device."""
|
|
|
|
# Optimization: The typical libc++ test directory has only a single
|
|
# *.tmp.exe file in it. In that case, skip the `mkdir` command, which is
|
|
# normally necessary because we don't know if the target directory already
|
|
# exists on the device.
|
|
local_files = os.listdir(local_dir)
|
|
if len(local_files) == 1:
|
|
local_file = os.path.join(local_dir, local_files[0])
|
|
remote_file = os.path.join(remote_dir, local_files[0])
|
|
if not os.path.islink(local_file) and os.path.isfile(local_file):
|
|
run_adb_sync_command(["adb", "push", "--sync", local_file,
|
|
remote_file])
|
|
return
|
|
|
|
assert os.path.basename(local_dir) == os.path.basename(remote_dir)
|
|
run_adb_sync_command(["adb", "shell", "mkdir", "-p", remote_dir])
|
|
run_adb_sync_command(["adb", "push", "--sync", local_dir,
|
|
os.path.dirname(remote_dir)])
|
|
|
|
|
|
def build_env_arg(env_args: List[str], prepend_path_args: List[Tuple[str, str]]) -> str:
|
|
components = []
|
|
for arg in env_args:
|
|
k, v = arg.split("=", 1)
|
|
components.append(f"export {k}={shlex.quote(v)}; ")
|
|
for k, v in prepend_path_args:
|
|
components.append(f"export {k}={shlex.quote(v)}${{{k}:+:${k}}}; ")
|
|
return "".join(components)
|
|
|
|
|
|
def run_command(args: argparse.Namespace) -> int:
|
|
local_dir = args.execdir
|
|
assert local_dir.startswith("/")
|
|
assert not local_dir.endswith("/")
|
|
|
|
# Copy each execdir to a subdir of REMOTE_BASE_DIR. Name the directory using
|
|
# a hash of local_dir so that concurrent adb_run invocations don't create
|
|
# the same intermediate parent directory. At least `adb push` has trouble
|
|
# with concurrent mkdir syscalls on common parent directories. (Somehow
|
|
# mkdir fails with EAGAIN/EWOULDBLOCK, see internal Google bug,
|
|
# b/289311228.)
|
|
local_dir_hash = hashlib.sha1(local_dir.encode()).hexdigest()
|
|
remote_dir = f"{REMOTE_BASE_DIR}/run-{local_dir_hash}/{os.path.basename(local_dir)}"
|
|
sync_test_dir(local_dir, remote_dir)
|
|
|
|
adb_shell_command = (
|
|
# Set the environment early so that PATH can be overridden. Overriding
|
|
# PATH is useful for:
|
|
# - Replacing older shell utilities with toybox (e.g. on old devices).
|
|
# - Adding a `bash` command that delegates to `sh` (mksh).
|
|
f"{build_env_arg(args.env, args.prepend_path_env)}"
|
|
|
|
# Set a high oom_score_adj so that, if the test program uses too much
|
|
# memory, it is killed before anything else on the device. The default
|
|
# oom_score_adj is -1000, so a test using too much memory typically
|
|
# crashes the device.
|
|
"echo 1000 >/proc/self/oom_score_adj; "
|
|
|
|
# If we're running as root, switch to the shell user. The libc++
|
|
# filesystem tests require running without root permissions. Some x86
|
|
# emulator devices (before Android N) do not have a working `adb unroot`
|
|
# and always run as root. Non-debug builds typically lack `su` and only
|
|
# run as the shell user.
|
|
#
|
|
# Some libc++ tests create temporary files in the working directory,
|
|
# which might be owned by root. Before switching to shell, make the
|
|
# cwd writable (and readable+executable) to every user.
|
|
#
|
|
# N.B.:
|
|
# - Avoid "id -u" because it wasn't supported until Android M.
|
|
# - The `env` and `which` commands were also added in Android M.
|
|
# - Starting in Android M, su from root->shell resets PATH, so we need
|
|
# to modify it again in the new environment.
|
|
# - Avoid chmod's "a+rwx" syntax because it's not supported until
|
|
# Android N.
|
|
# - Defining this function allows specifying the arguments to the test
|
|
# program (i.e. "$@") only once.
|
|
"run_without_root() {"
|
|
" chmod 777 .;"
|
|
" case \"$(id)\" in"
|
|
" *\"uid=0(root)\"*)"
|
|
" if command -v env >/dev/null; then"
|
|
" su shell \"$(command -v env)\" PATH=\"$PATH\" \"$@\";"
|
|
" else"
|
|
" su shell \"$@\";"
|
|
" fi;;"
|
|
" *) \"$@\";;"
|
|
" esac;"
|
|
"}; "
|
|
)
|
|
|
|
# Older versions of Bionic limit the length of argv[0] to 127 bytes
|
|
# (SOINFO_NAME_LEN-1), and the path to libc++ tests tend to exceed this
|
|
# limit. Changing the working directory works around this limit. The limit
|
|
# is increased to 4095 (PATH_MAX-1) in Android M (API 23).
|
|
command_line = [arg.replace(local_dir + "/", "./") for arg in args.command]
|
|
|
|
# Prior to the adb feature "shell_v2" (added in Android N), `adb shell`
|
|
# always created a pty:
|
|
# - This merged stdout and stderr together.
|
|
# - The pty converts LF to CRLF.
|
|
# - The exit code of the shell command wasn't propagated.
|
|
# Work around all three limitations, unless "shell_v2" is present.
|
|
proc = subprocess.run(["adb", "features"], check=True,
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
|
|
encoding="utf-8")
|
|
adb_features = set(proc.stdout.strip().split())
|
|
has_shell_v2 = "shell_v2" in adb_features
|
|
if has_shell_v2:
|
|
adb_shell_command += (
|
|
f"cd {remote_dir} && run_without_root {shlex.join(command_line)}"
|
|
)
|
|
else:
|
|
adb_shell_command += (
|
|
f"{{"
|
|
f" stdout=$("
|
|
f" cd {remote_dir} && run_without_root {shlex.join(command_line)};"
|
|
f" echo -n __libcxx_adb_exit__=$?"
|
|
f" ); "
|
|
f"}} 2>&1; "
|
|
f"echo -n __libcxx_adb_stdout__\"$stdout\""
|
|
)
|
|
|
|
adb_command_line = ["adb", "shell", adb_shell_command]
|
|
if g_verbose:
|
|
sys.stderr.write(f"running: {shlex.join(adb_command_line)}\n")
|
|
|
|
if has_shell_v2:
|
|
proc = subprocess.run(adb_command_line, shell=False, check=False,
|
|
encoding="utf-8")
|
|
return proc.returncode
|
|
else:
|
|
proc = subprocess.run(adb_command_line, shell=False, check=False,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
encoding="utf-8")
|
|
# The old `adb shell` mode used a pty, which converted LF to CRLF.
|
|
# Convert it back.
|
|
output = proc.stdout.replace("\r\n", "\n")
|
|
|
|
if proc.returncode:
|
|
sys.stderr.write(f"error: adb failed:\n"
|
|
f" command: {shlex.join(adb_command_line)}\n"
|
|
f" output: {output}\n")
|
|
return proc.returncode
|
|
|
|
match = re.match(r"(.*)__libcxx_adb_stdout__(.*)__libcxx_adb_exit__=(\d+)$",
|
|
output, re.DOTALL)
|
|
if not match:
|
|
sys.stderr.write(f"error: could not parse adb output:\n"
|
|
f" command: {shlex.join(adb_command_line)}\n"
|
|
f" output: {output}\n")
|
|
return 1
|
|
|
|
sys.stderr.write(match.group(1))
|
|
sys.stdout.write(match.group(2))
|
|
return int(match.group(3))
|
|
|
|
|
|
def connect_to_job_limiter_server(sock_addr: str) -> None:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
|
|
try:
|
|
sock.connect(sock_addr)
|
|
except (FileNotFoundError, ConnectionRefusedError) as e:
|
|
# Copying-and-pasting an adb_run.py command-line from a lit test failure
|
|
# is likely to fail because the socket no longer exists (or is
|
|
# inactive), so just give a warning.
|
|
sys.stderr.write(f"warning: could not connect to {sock_addr}: {e}\n")
|
|
return
|
|
|
|
# The connect call can succeed before the server has called accept, because
|
|
# of the listen backlog, so wait for the server to send a byte.
|
|
sock.recv(1)
|
|
|
|
# Keep the socket open until this process ends, then let the OS close the
|
|
# connection automatically.
|
|
global g_job_limit_socket
|
|
g_job_limit_socket = sock
|
|
|
|
|
|
def main() -> int:
|
|
"""Main function (pylint wants this docstring)."""
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--execdir", type=str, required=True)
|
|
parser.add_argument("--env", type=str, required=False, action="append",
|
|
default=[], metavar="NAME=VALUE")
|
|
parser.add_argument("--prepend-path-env", type=str, nargs=2, required=False,
|
|
action="append", default=[],
|
|
metavar=("NAME", "PATH"))
|
|
parser.add_argument("--job-limit-socket")
|
|
parser.add_argument("--verbose", "-v", default=False, action="store_true")
|
|
parser.add_argument("command", nargs=argparse.ONE_OR_MORE)
|
|
args = parser.parse_args()
|
|
|
|
global g_verbose
|
|
g_verbose = args.verbose
|
|
if args.job_limit_socket is not None:
|
|
connect_to_job_limiter_server(args.job_limit_socket)
|
|
return run_command(args)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|