1from __future__ import annotations 2 3import argparse 4import concurrent.futures 5import json 6import logging 7import os 8import re 9import shutil 10import subprocess 11import sys 12import time 13from enum import Enum 14from pathlib import Path 15from sysconfig import get_paths as gp 16from typing import Any, NamedTuple 17 18 19# PyTorch directory root 20def scm_root() -> str: 21 path = os.path.abspath(os.getcwd()) 22 while True: 23 if os.path.exists(os.path.join(path, ".git")): 24 return path 25 if os.path.isdir(os.path.join(path, ".hg")): 26 return path 27 n = len(path) 28 path = os.path.dirname(path) 29 if len(path) == n: 30 raise RuntimeError("Unable to find SCM root") 31 32 33PYTORCH_ROOT = scm_root() 34IS_WINDOWS: bool = os.name == "nt" 35 36 37# Returns '/usr/local/include/python<version number>' 38def get_python_include_dir() -> str: 39 return gp()["include"] 40 41 42def eprint(*args: Any, **kwargs: Any) -> None: 43 print(*args, file=sys.stderr, flush=True, **kwargs) 44 45 46class LintSeverity(str, Enum): 47 ERROR = "error" 48 WARNING = "warning" 49 ADVICE = "advice" 50 DISABLED = "disabled" 51 52 53class LintMessage(NamedTuple): 54 path: str | None 55 line: int | None 56 char: int | None 57 code: str 58 severity: LintSeverity 59 name: str 60 original: str | None 61 replacement: str | None 62 description: str | None 63 64 65def as_posix(name: str) -> str: 66 return name.replace("\\", "/") if IS_WINDOWS else name 67 68 69# c10/core/DispatchKey.cpp:281:26: error: 'k' used after it was moved [bugprone-use-after-move] 70RESULTS_RE: re.Pattern[str] = re.compile( 71 r"""(?mx) 72 ^ 73 (?P<file>.*?): 74 (?P<line>\d+): 75 (?:(?P<column>-?\d+):)? 76 \s(?P<severity>\S+?):? 77 \s(?P<message>.*) 78 \s(?P<code>\[.*\]) 79 $ 80 """ 81) 82 83 84def run_command( 85 args: list[str], 86) -> subprocess.CompletedProcess[bytes]: 87 logging.debug("$ %s", " ".join(args)) 88 start_time = time.monotonic() 89 try: 90 return subprocess.run( 91 args, 92 capture_output=True, 93 check=False, 94 ) 95 finally: 96 end_time = time.monotonic() 97 logging.debug("took %dms", (end_time - start_time) * 1000) 98 99 100# Severity is either "error" or "note": 101# https://github.com/python/mypy/blob/8b47a032e1317fb8e3f9a818005a6b63e9bf0311/mypy/errors.py#L46-L47 102severities = { 103 "error": LintSeverity.ERROR, 104 "warning": LintSeverity.WARNING, 105} 106 107 108def clang_search_dirs() -> list[str]: 109 # Compilers are ordered based on fallback preference 110 # We pick the first one that is available on the system 111 compilers = ["clang", "gcc", "cpp", "cc"] 112 compilers = [c for c in compilers if shutil.which(c) is not None] 113 if len(compilers) == 0: 114 raise RuntimeError(f"None of {compilers} were found") 115 compiler = compilers[0] 116 117 result = subprocess.run( 118 [compiler, "-E", "-x", "c++", "-", "-v"], 119 stdin=subprocess.DEVNULL, 120 capture_output=True, 121 check=True, 122 ) 123 stderr = result.stderr.decode().strip().split("\n") 124 search_start = r"#include.*search starts here:" 125 search_end = r"End of search list." 126 127 append_path = False 128 search_paths = [] 129 for line in stderr: 130 if re.match(search_start, line): 131 if append_path: 132 continue 133 else: 134 append_path = True 135 elif re.match(search_end, line): 136 break 137 elif append_path: 138 search_paths.append(line.strip()) 139 140 return search_paths 141 142 143include_args = [] 144include_dir = [ 145 "/usr/lib/llvm-11/include/openmp", 146 get_python_include_dir(), 147 os.path.join(PYTORCH_ROOT, "third_party/pybind11/include"), 148] + clang_search_dirs() 149for dir in include_dir: 150 include_args += ["--extra-arg", f"-I{dir}"] 151 152 153def check_file( 154 filename: str, 155 binary: str, 156 build_dir: Path, 157) -> list[LintMessage]: 158 try: 159 proc = run_command( 160 [binary, f"-p={build_dir}", *include_args, filename], 161 ) 162 except OSError as err: 163 return [ 164 LintMessage( 165 path=filename, 166 line=None, 167 char=None, 168 code="CLANGTIDY", 169 severity=LintSeverity.ERROR, 170 name="command-failed", 171 original=None, 172 replacement=None, 173 description=(f"Failed due to {err.__class__.__name__}:\n{err}"), 174 ) 175 ] 176 lint_messages = [] 177 try: 178 # Change the current working directory to the build directory, since 179 # clang-tidy will report files relative to the build directory. 180 saved_cwd = os.getcwd() 181 os.chdir(build_dir) 182 183 for match in RESULTS_RE.finditer(proc.stdout.decode()): 184 # Convert the reported path to an absolute path. 185 abs_path = str(Path(match["file"]).resolve()) 186 message = LintMessage( 187 path=abs_path, 188 name=match["code"], 189 description=match["message"], 190 line=int(match["line"]), 191 char=int(match["column"]) 192 if match["column"] is not None and not match["column"].startswith("-") 193 else None, 194 code="CLANGTIDY", 195 severity=severities.get(match["severity"], LintSeverity.ERROR), 196 original=None, 197 replacement=None, 198 ) 199 lint_messages.append(message) 200 finally: 201 os.chdir(saved_cwd) 202 203 return lint_messages 204 205 206def main() -> None: 207 parser = argparse.ArgumentParser( 208 description="clang-tidy wrapper linter.", 209 fromfile_prefix_chars="@", 210 ) 211 parser.add_argument( 212 "--binary", 213 required=True, 214 help="clang-tidy binary path", 215 ) 216 parser.add_argument( 217 "--build-dir", 218 "--build_dir", 219 required=True, 220 help=( 221 "Where the compile_commands.json file is located. " 222 "Gets passed to clang-tidy -p" 223 ), 224 ) 225 parser.add_argument( 226 "--verbose", 227 action="store_true", 228 help="verbose logging", 229 ) 230 parser.add_argument( 231 "filenames", 232 nargs="+", 233 help="paths to lint", 234 ) 235 args = parser.parse_args() 236 237 logging.basicConfig( 238 format="<%(threadName)s:%(levelname)s> %(message)s", 239 level=logging.NOTSET 240 if args.verbose 241 else logging.DEBUG 242 if len(args.filenames) < 1000 243 else logging.INFO, 244 stream=sys.stderr, 245 ) 246 247 if not os.path.exists(args.binary): 248 err_msg = LintMessage( 249 path="<none>", 250 line=None, 251 char=None, 252 code="CLANGTIDY", 253 severity=LintSeverity.ERROR, 254 name="command-failed", 255 original=None, 256 replacement=None, 257 description=( 258 f"Could not find clang-tidy binary at {args.binary}," 259 " you may need to run `lintrunner init`." 260 ), 261 ) 262 print(json.dumps(err_msg._asdict()), flush=True) 263 sys.exit(0) 264 265 abs_build_dir = Path(args.build_dir).resolve() 266 267 # Get the absolute path to clang-tidy and use this instead of the relative 268 # path such as .lintbin/clang-tidy. The problem here is that os.chdir is 269 # per process, and the linter uses it to move between the current directory 270 # and the build folder. And there is no .lintbin directory in the latter. 271 # When it happens in a race condition, the linter command will fails with 272 # the following no such file or directory error: '.lintbin/clang-tidy' 273 binary_path = os.path.abspath(args.binary) 274 275 with concurrent.futures.ThreadPoolExecutor( 276 max_workers=os.cpu_count(), 277 thread_name_prefix="Thread", 278 ) as executor: 279 futures = { 280 executor.submit( 281 check_file, 282 filename, 283 binary_path, 284 abs_build_dir, 285 ): filename 286 for filename in args.filenames 287 } 288 for future in concurrent.futures.as_completed(futures): 289 try: 290 for lint_message in future.result(): 291 print(json.dumps(lint_message._asdict()), flush=True) 292 except Exception: 293 logging.critical('Failed at "%s".', futures[future]) 294 raise 295 296 297if __name__ == "__main__": 298 main() 299