mirror of
https://github.com/XRPLF/rippled.git
synced 2026-03-13 00:02:33 +00:00
336 lines
12 KiB
Python
336 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
Usage: generate.py
|
|
This script takes no parameters, and can be called from any directory in the file system.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple, Set, Optional
|
|
|
|
# Compile regex patterns once at module level
|
|
INCLUDE_PATTERN = re.compile(r"^\s*#include.*/.*\.h")
|
|
INCLUDE_PATH_PATTERN = re.compile(r'[<"]([^>"]+)[>"]')
|
|
|
|
|
|
def dictionary_sort_key(s: str) -> str:
|
|
"""
|
|
Create a sort key that mimics 'sort -d' (dictionary order).
|
|
Dictionary order only considers blanks and alphanumeric characters.
|
|
This means punctuation like '.' is ignored during sorting.
|
|
"""
|
|
# Keep only alphanumeric characters and spaces
|
|
return "".join(c for c in s if c.isalnum() or c.isspace())
|
|
|
|
|
|
def get_level(file_path: str) -> str:
|
|
"""
|
|
Extract the level from a file path (second and third directory components).
|
|
Equivalent to bash: cut -d/ -f 2,3
|
|
|
|
Examples:
|
|
src/xrpld/app/main.cpp -> xrpld.app
|
|
src/libxrpl/protocol/STObject.cpp -> libxrpl.protocol
|
|
include/xrpl/basics/base_uint.h -> xrpl.basics
|
|
"""
|
|
parts = file_path.split("/")
|
|
|
|
# Get fields 2 and 3 (indices 1 and 2 in 0-based indexing)
|
|
if len(parts) >= 3:
|
|
level = f"{parts[1]}/{parts[2]}"
|
|
elif len(parts) >= 2:
|
|
level = f"{parts[1]}/toplevel"
|
|
else:
|
|
level = file_path
|
|
|
|
# If the "level" indicates a file, cut off the filename
|
|
if "." in level.split("/")[-1]: # Avoid Path object creation
|
|
# Use the "toplevel" label as a workaround for `sort`
|
|
# inconsistencies between different utility versions
|
|
level = level.rsplit("/", 1)[0] + "/toplevel"
|
|
|
|
return level.replace("/", ".")
|
|
|
|
|
|
def extract_include_level(include_line: str) -> Optional[str]:
|
|
"""
|
|
Extract the include path from an #include directive.
|
|
Gets the first two directory components from the include path.
|
|
Equivalent to bash: cut -d/ -f 1,2
|
|
|
|
Examples:
|
|
#include <xrpl/basics/base_uint.h> -> xrpl.basics
|
|
#include "xrpld/app/main/Application.h" -> xrpld.app
|
|
"""
|
|
# Remove everything before the quote or angle bracket
|
|
match = INCLUDE_PATH_PATTERN.search(include_line)
|
|
if not match:
|
|
return None
|
|
|
|
include_path = match.group(1)
|
|
parts = include_path.split("/")
|
|
|
|
# Get first two fields (indices 0 and 1)
|
|
if len(parts) >= 2:
|
|
include_level = f"{parts[0]}/{parts[1]}"
|
|
else:
|
|
include_level = include_path
|
|
|
|
# If the "includelevel" indicates a file, cut off the filename
|
|
if "." in include_level.split("/")[-1]: # Avoid Path object creation
|
|
include_level = include_level.rsplit("/", 1)[0] + "/toplevel"
|
|
|
|
return include_level.replace("/", ".")
|
|
|
|
|
|
def find_repository_directories(
|
|
start_path: Path, depth_limit: int = 10
|
|
) -> Tuple[Path, List[Path]]:
|
|
"""
|
|
Find the repository root by looking for src or include folders.
|
|
Walks up the directory tree from the start path.
|
|
"""
|
|
current = start_path.resolve()
|
|
|
|
# Walk up the directory tree
|
|
for _ in range(depth_limit): # Limit search depth to prevent infinite loops
|
|
src_path = current / "src"
|
|
include_path = current / "include"
|
|
# Check if this directory has src or include folders
|
|
has_src = src_path.exists()
|
|
has_include = include_path.exists()
|
|
|
|
if has_src or has_include:
|
|
return current, [src_path, include_path]
|
|
|
|
# Move up one level
|
|
parent = current.parent
|
|
if parent == current: # Reached filesystem root
|
|
break
|
|
current = parent
|
|
|
|
# If we couldn't find it, raise an error
|
|
raise RuntimeError(
|
|
"Could not find repository root. "
|
|
"Expected to find a directory containing 'src' and/or 'include' folders."
|
|
)
|
|
|
|
|
|
def main():
|
|
# Change to the script's directory
|
|
script_dir = Path(__file__).parent.resolve()
|
|
os.chdir(script_dir)
|
|
|
|
# Clean up and create results directory.
|
|
results_dir = script_dir / "results"
|
|
if results_dir.exists():
|
|
import shutil
|
|
|
|
shutil.rmtree(results_dir)
|
|
results_dir.mkdir()
|
|
|
|
# Find the repository root by searching for src and include directories.
|
|
try:
|
|
repo_root, scan_dirs = find_repository_directories(script_dir)
|
|
|
|
print(f"Found repository root: {repo_root}")
|
|
print(f"Scanning directories:")
|
|
for scan_dir in scan_dirs:
|
|
print(f" - {scan_dir.relative_to(repo_root)}")
|
|
except RuntimeError as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print("\nScanning for raw includes...")
|
|
# Find all #include directives
|
|
raw_includes: List[Tuple[str, str]] = []
|
|
rawincludes_file = results_dir / "rawincludes.txt"
|
|
|
|
# Write to file as we go to avoid storing everything in memory.
|
|
with open(rawincludes_file, "w", buffering=8192) as raw_f:
|
|
for dir_path in scan_dirs:
|
|
print(f" Scanning {dir_path.relative_to(repo_root)}...")
|
|
|
|
for file_path in dir_path.rglob("*"):
|
|
if not file_path.is_file():
|
|
continue
|
|
|
|
try:
|
|
rel_path_str = str(file_path.relative_to(repo_root))
|
|
|
|
# Read file with a large buffer for performance.
|
|
with open(
|
|
file_path,
|
|
"r",
|
|
encoding="utf-8",
|
|
errors="ignore",
|
|
buffering=8192,
|
|
) as f:
|
|
for line in f:
|
|
# Quick check before regex
|
|
if "#include" not in line or "boost" in line:
|
|
continue
|
|
|
|
if INCLUDE_PATTERN.match(line):
|
|
line_stripped = line.strip()
|
|
entry = f"{rel_path_str}:{line_stripped}\n"
|
|
print(entry, end="")
|
|
raw_f.write(entry)
|
|
raw_includes.append((rel_path_str, line_stripped))
|
|
except Exception as e:
|
|
print(f"Error reading {file_path}: {e}", file=sys.stderr)
|
|
|
|
# Build levelization paths and count directly (no need to sort first).
|
|
print("Build levelization paths")
|
|
path_counts: Dict[Tuple[str, str], int] = defaultdict(int)
|
|
|
|
for file_path, include_line in raw_includes:
|
|
include_level = extract_include_level(include_line)
|
|
if not include_level:
|
|
continue
|
|
|
|
level = get_level(file_path)
|
|
if level != include_level:
|
|
path_counts[(level, include_level)] += 1
|
|
|
|
# Sort and deduplicate paths (using dictionary order like bash 'sort -d').
|
|
print("Sort and deduplicate paths")
|
|
|
|
paths_file = results_dir / "paths.txt"
|
|
with open(paths_file, "w") as f:
|
|
# Sort using dictionary order: only alphanumeric and spaces matter
|
|
sorted_items = sorted(
|
|
path_counts.items(),
|
|
key=lambda x: (dictionary_sort_key(x[0][0]), dictionary_sort_key(x[0][1])),
|
|
)
|
|
for (level, include_level), count in sorted_items:
|
|
line = f"{count:7} {level} {include_level}\n"
|
|
print(line.rstrip())
|
|
f.write(line)
|
|
|
|
# Split into flat-file database
|
|
print("Split into flat-file database")
|
|
includes_dir = results_dir / "includes"
|
|
included_by_dir = results_dir / "included_by"
|
|
includes_dir.mkdir()
|
|
included_by_dir.mkdir()
|
|
|
|
# Batch writes by grouping data first to avoid repeated file opens.
|
|
includes_data: Dict[str, List[Tuple[str, int]]] = defaultdict(list)
|
|
included_by_data: Dict[str, List[Tuple[str, int]]] = defaultdict(list)
|
|
|
|
# Process in sorted order to match bash script behaviour (dictionary order).
|
|
sorted_items = sorted(
|
|
path_counts.items(),
|
|
key=lambda x: (dictionary_sort_key(x[0][0]), dictionary_sort_key(x[0][1])),
|
|
)
|
|
for (level, include_level), count in sorted_items:
|
|
includes_data[level].append((include_level, count))
|
|
included_by_data[include_level].append((level, count))
|
|
|
|
# Write all includes files in sorted order (dictionary order).
|
|
for level in sorted(includes_data.keys(), key=dictionary_sort_key):
|
|
entries = includes_data[level]
|
|
with open(includes_dir / level, "w") as f:
|
|
for include_level, count in entries:
|
|
line = f"{include_level} {count}\n"
|
|
print(line.rstrip())
|
|
f.write(line)
|
|
|
|
# Write all included_by files in sorted order (dictionary order).
|
|
for include_level in sorted(included_by_data.keys(), key=dictionary_sort_key):
|
|
entries = included_by_data[include_level]
|
|
with open(included_by_dir / include_level, "w") as f:
|
|
for level, count in entries:
|
|
line = f"{level} {count}\n"
|
|
print(line.rstrip())
|
|
f.write(line)
|
|
|
|
# Search for loops
|
|
print("Search for loops")
|
|
loops_file = results_dir / "loops.txt"
|
|
ordering_file = results_dir / "ordering.txt"
|
|
|
|
loops_found: Set[Tuple[str, str]] = set()
|
|
|
|
# Pre-load all include files into memory to avoid repeated I/O.
|
|
# This is the biggest optimisation - we were reading files repeatedly in nested loops.
|
|
# Use list of tuples to preserve file order.
|
|
includes_cache: Dict[str, List[Tuple[str, int]]] = {}
|
|
includes_lookup: Dict[str, Dict[str, int]] = {} # For fast lookup
|
|
|
|
# Note: bash script uses 'for source in *' which uses standard glob sorting,
|
|
# NOT dictionary order. So we use standard sorted() here, not dictionary_sort_key.
|
|
for include_file in sorted(includes_dir.iterdir(), key=lambda p: p.name):
|
|
if not include_file.is_file():
|
|
continue
|
|
|
|
includes_cache[include_file.name] = []
|
|
includes_lookup[include_file.name] = {}
|
|
with open(include_file, "r") as f:
|
|
for line in f:
|
|
parts = line.strip().split()
|
|
if len(parts) >= 2:
|
|
include_name = parts[0]
|
|
include_count = int(parts[1])
|
|
includes_cache[include_file.name].append(
|
|
(include_name, include_count)
|
|
)
|
|
includes_lookup[include_file.name][include_name] = include_count
|
|
|
|
with open(loops_file, "w", buffering=8192) as loops_f, open(
|
|
ordering_file, "w", buffering=8192
|
|
) as ordering_f:
|
|
|
|
# Use standard sorting to match bash glob expansion 'for source in *'.
|
|
for source in sorted(includes_cache.keys()):
|
|
source_includes = includes_cache[source]
|
|
|
|
for include, include_freq in source_includes:
|
|
# Check if include file exists and references source
|
|
if include not in includes_lookup:
|
|
continue
|
|
|
|
source_freq = includes_lookup[include].get(source)
|
|
|
|
if source_freq is not None:
|
|
# Found a loop
|
|
loop_key = tuple(sorted([source, include]))
|
|
if loop_key in loops_found:
|
|
continue
|
|
loops_found.add(loop_key)
|
|
|
|
loops_f.write(f"Loop: {source} {include}\n")
|
|
|
|
# If the counts are close, indicate that the two modules are
|
|
# on the same level, though they shouldn't be.
|
|
diff = include_freq - source_freq
|
|
if diff > 3:
|
|
loops_f.write(f" {source} > {include}\n\n")
|
|
elif diff < -3:
|
|
loops_f.write(f" {include} > {source}\n\n")
|
|
elif source_freq == include_freq:
|
|
loops_f.write(f" {include} == {source}\n\n")
|
|
else:
|
|
loops_f.write(f" {include} ~= {source}\n\n")
|
|
else:
|
|
ordering_f.write(f"{source} > {include}\n")
|
|
|
|
# Print results
|
|
print("\nOrdering:")
|
|
with open(ordering_file, "r") as f:
|
|
print(f.read(), end="")
|
|
|
|
print("\nLoops:")
|
|
with open(loops_file, "r") as f:
|
|
print(f.read(), end="")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|