from __future__ import annotations
import argparse
import shutil
from pathlib import Path
from time import perf_counter
FILE_COUNT = 50000
WRITE_SIZE = 1024
BENCH_DIR_NAME = "___benti"
def build_payload(write_size: int, index: int) -> str:
"""指定サイズの書き込みデータを作る。"""
base_text = f"benchmark-{index}-"
repeat_count = (write_size // len(base_text)) + 1
payload = (base_text * repeat_count)[:write_size]
return payload
def create_files(target_dir: Path, file_count: int, write_size: int) -> None:
"""指定数のファイルを作成する。"""
target_dir.mkdir(parents=True, exist_ok=True)
for index in range(file_count):
file_path = target_dir / f"file_{index:04d}.txt"
file_path.write_text(build_payload(write_size, index), encoding="utf-8")
def read_files(target_dir: Path) -> None:
"""ディレクトリ内の全ファイルを読み込む。"""
for file_path in sorted(target_dir.iterdir()):
if file_path.is_file():
file_path.read_text(encoding="utf-8")
def update_files(target_dir: Path, write_size: int, file_count: int) -> None:
"""ディレクトリ内の全ファイルを書き換える。"""
for index, file_path in enumerate(sorted(target_dir.iterdir())):
if file_path.is_file():
file_path.write_text(
build_payload(write_size, index + file_count), encoding="utf-8"
)
def delete_files(target_dir: Path) -> None:
"""ディレクトリ内の全ファイルを削除する。"""
for file_path in sorted(target_dir.iterdir()):
if file_path.is_file():
file_path.unlink()
def cleanup_directory(target_dir: Path) -> None:
"""ベンチ用ディレクトリを片付ける。"""
if target_dir.exists():
shutil.rmtree(target_dir)
def run_benchmark(base_dir: Path, file_count: int, write_size: int) -> None:
"""ファイル作成、読み込み、削除の各処理時間を計測する。"""
target_dir = base_dir / BENCH_DIR_NAME
cleanup_directory(target_dir)
measurements: list[tuple[str, float]] = []
start = perf_counter()
create_files(target_dir, file_count, write_size)
measurements.append(("作成", perf_counter() - start))
start = perf_counter()
read_files(target_dir)
measurements.append(("読み込み", perf_counter() - start))
start = perf_counter()
update_files(target_dir, write_size, file_count)
measurements.append(("更新", perf_counter() - start))
start = perf_counter()
delete_files(target_dir)
measurements.append(("削除", perf_counter() - start))
cleanup_directory(target_dir)
total_elapsed = sum(elapsed for _, elapsed in measurements)
for label, elapsed in measurements:
print(f"{label}: {elapsed:.6f} 秒")
print(f"合計: {total_elapsed:.6f} 秒")
def parse_args() -> argparse.Namespace:
"""コマンドライン引数を解析する。"""
parser = argparse.ArgumentParser(description="ファイル操作の簡易ベンチマーク")
parser.add_argument(
"--count",
type=int,
default=FILE_COUNT,
help="作成・読み込み・削除するファイル数",
)
parser.add_argument(
"--write-size",
type=int,
default=WRITE_SIZE,
help="1ファイルあたりの書き込み容量（バイト）",
)
parser.add_argument(
"--base-dir",
type=Path,
default=Path(__file__).resolve().parent.parent,
help="___benti ディレクトリを作成する基準ディレクトリ",
)
return parser.parse_args()
def main() -> None:
"""ベンチマークを実行する。"""
args = parse_args()
run_benchmark(args.base_dir, args.count, args.write_size)
if __name__ == "__main__":
main()
