Parallel File Processor with ThreadPoolExecutor

Parallel File Processor with ThreadPoolExecutor

Problem Statement

Whether you're processing log files, transforming JSON exports, or generating checksums for a release bundle, applying the same operation to a directory full of files is a common task. Doing this serially wastes time — most file operations (reading, parsing, hashing, writing) are I/O-bound and spend the vast majority of their wall-clock time waiting on the disk or network, not burning CPU. On a machine with dozens of files, a serial loop can take minutes when it should take seconds.

This script solves that by walking a directory tree and dispatching a user-selected operation (hash, line count, or JSON validation) across a ThreadPoolExecutor. Because threads release the GIL during blocking I/O, ThreadPoolExecutor is the right tool here — no multiprocessing overhead, no pickling, no inter-process communication. You get parallelism, progress reporting, and structured error handling in a single self-contained script with zero third-party dependencies.

How It Works

The script has three layered components. First, a file discovery step walks the target directory using os.walk, filtering by extension and skipping files above a configurable size limit (so a single giant file can't stall the queue). Second, a worker function is submitted for each file to a ThreadPoolExecutor. Each worker runs the selected operation — SHA-256 hashing, line counting, or JSON parsing — and returns a small result dataclass rather than printing directly, which keeps output ordered and thread-safe. Third, as futures complete via as_completed, the main thread prints progress and tallies successes/failures.

Design decisions worth noting: results are returned as frozen dataclasses so they're immutable and easy to test; the executor is created with a context manager to guarantee shutdown even on KeyboardInterrupt; and a --dry-run flag lets you preview the file list before committing to the work. Errors inside workers are caught and returned as failed results rather than propagating — this means one bad file doesn't kill the whole batch, and the summary at the end tells you exactly what went wrong.

Full Script

#!/usr/bin/env python3
"""
Parallel File Processor with ThreadPoolExecutor.

Walks a directory tree and applies an operation (hash, line-count, or JSON
validation) to every matching file, using a thread pool for parallelism.
Standard library only.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Iterable

# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_WORKERS = int(os.environ.get("PFP_WORKERS", "4"))
DEFAULT_CHUNK_SIZE = int(os.environ.get("PFP_CHUNK_SIZE", str(64 * 1024)))  # 64 KiB
DEFAULT_MAX_MB = int(os.environ.get("PFP_MAX_FILE_MB", "256"))
DEFAULT_EXT = os.environ.get("PFP_EXT", "")


# ---------------------------------------------------------------------------
# Result container
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Result:
    """Immutable outcome of processing a single file."""
    path: Path
    operation: str
    ok: bool
    value: str = ""
    error: str = ""
    elapsed: float = 0.0


# ---------------------------------------------------------------------------
# Operations
# ---------------------------------------------------------------------------
def op_hash(path: Path, chunk_size: int = DEFAULT_CHUNK_SIZE) -> str:
    """Return the SHA-256 hex digest of a file, streamed in chunks."""
    h = hashlib.sha256()
    with path.open("rb") as fh:
        while chunk := fh.read(chunk_size):
            h.update(chunk)
    return h.hexdigest()


def op_lines(path: Path, chunk_size: int = DEFAULT_CHUNK_SIZE) -> str:
    """Count newline characters without loading the whole file."""
    count = 0
    with path.open("rb") as fh:
        while chunk := fh.read(chunk_size):
            count += chunk.count(b"\n")
    return str(count)


def op_json(path: Path, **_kwargs) -> str:
    """Validate that a file is well-formed JSON; return top-level type."""
    with path.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    return type(data).__name__


OPERATIONS: dict[str, Callable[..., str]] = {
    "hash": op_hash,
    "lines": op_lines,
    "json": op_json,
}


# ---------------------------------------------------------------------------
# Discovery
# ---------------------------------------------------------------------------
def discover_files(
    root: Path,
    extension: str,
    max_bytes: int,
) -> list[Path]:
    """Recursively find files under `root` matching filters."""
    if not root.exists():
        raise FileNotFoundError(f"Directory does not exist: {root}")
    if not root.is_dir():
        raise NotADirectoryError(f"Not a directory: {root}")

    ext = extension.lower()
    found: list[Path] = []
    for dirpath, _dirnames, filenames in os.walk(root):
        for name in filenames:
            p = Path(dirpath) / name
            if ext and p.suffix.lower() != ext:
                continue
            try:
                if p.stat().st_size > max_bytes:
                    continue
            except OSError:
                # Broken symlink or permission issue — skip silently at discovery.
                continue
            found.append(p)
    return found


# ---------------------------------------------------------------------------
# Worker
# ---------------------------------------------------------------------------
def process_one(path: Path, op_name: str, chunk_size: int) -> Result:
    """Run a single operation, capturing any exception as a failed Result."""
    start = time.perf_counter()
    try:
        func = OPERATIONS[op_name]
        value = func(path, chunk_size=chunk_size) if op_name != "json" else func(path)
        return Result(
            path=path,
            operation=op_name,
            ok=True,
            value=value,
            elapsed=time.perf_counter() - start,
        )
    except Exception as exc:  # noqa: BLE001 — deliberate: report, don't crash
        return Result(
            path=path,
            operation=op_name,
            ok=False,
            error=f"{type(exc).__name__}: {exc}",
            elapsed=time.perf_counter() - start,
        )


# ---------------------------------------------------------------------------
# Driver
# ---------------------------------------------------------------------------
def run(
    files: Iterable[Path],
    op_name: str,
    workers: int,
    chunk_size: int,
    verbose: bool,
) -> tuple[list[Result], float]:
    """Dispatch files to a thread pool and gather results."""
    files = list(files)
    total = len(files)
    results: list[Result] = []
    wall_start = time.perf_counter()

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {
            pool.submit(process_one, p, op_name, chunk_size): p for p in files
        }
        try:
            for i, fut in enumerate(as_completed(futures), start=1):
                res = fut.result()
                results.append(res)
                if verbose:
                    status = "OK " if res.ok else "ERR"
                    detail = res.value if res.ok else res.error
                    print(f"[{i:>4}/{total}] {status} {res.path} -> {detail}")
                elif i % 25 == 0 or i == total:
                    print(f"  ... {i}/{total} processed", file=sys.stderr)
        except KeyboardInterrupt:
            print("\nInterrupted — cancelling pending work...", file=sys.stderr)
            for fut in futures:
                fut.cancel()
            raise

    return results, time.perf_counter() - wall_start


def summarize(results: list[Result], wall: float) -> int:
    """Print a final summary. Returns process exit code."""
    ok = [r for r in results if r.ok]
    bad = [r for r in results if not r.ok]
    total_time = sum(r.elapsed for r in results)
    speedup = (total_time / wall) if wall > 0 else 0.0

    print("\n" + "=" * 60)
    print(f"Processed : {len(results)} file(s)")
    print(f"Succeeded : {len(ok)}")
    print(f"Failed    : {len(bad)}")
    print(f"Wall time : {wall:.3f}s   (sum of worker time: {total_time:.3f}s)")
    print(f"Parallelism factor: {speedup:.2f}x")

    if bad:
        print("\nFailures:")
        for r in bad:
            print(f"  - {r.path}: {r.error}")
    return 0 if not bad else 1


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Parallel file processor using ThreadPoolExecutor.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--dir", default=".", help="Root directory to scan")
    p.add_argument(
        "--op",
        choices=sorted(OPERATIONS.keys()),
        default="hash",
        help="Operation to apply to each file",
    )
    p.add_argument(
        "--workers",
        type=int,
        default=DEFAULT_WORKERS,
        help="Number of worker threads",
    )
    p.add_argument(
        "--ext",
        default=DEFAULT_EXT,
        help="Only process files with this extension (e.g. .json)",
    )
    p.add_argument(
        "--chunk-size",
        type=int,
        default=DEFAULT_CHUNK_SIZE,
        help="Read chunk size in bytes for streaming ops",
    )
    p.add_argument(
        "--max-mb",
        type=int,
        default=DEFAULT_MAX_MB,
        help="Skip files larger than this many megabytes",
    )
    p.add_argument(
        "--dry-run",
        action="store_true",
        help="List matching files and exit without processing",
    )
    p.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Print one line per file",
    )
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    if args.workers < 1:
        print("error: --workers must be >= 1", file=sys.stderr)
        return 2

    root = Path(args.dir).expanduser().resolve()
    max_bytes = args.max_mb * 1024 * 1024

    try:
        files = discover_files(root, args.ext, max_bytes)
    except (FileNotFoundError, NotADirectoryError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    print(
        f"Discovered {len(files)} file(s) under {root} "
        f"(op={args.op}, workers={args.workers}, ext={args.ext or '*'})"
    )

    if not files:
        return 0

    if args.dry_run:
        for f in files:
            print(f)
        return 0

    try:
        results, wall = run(
            files,
            op_name=args.op,
            workers=args.workers,
            chunk_size=args.chunk_size,
            verbose=args.verbose,
        )
    except KeyboardInterrupt:
        return 130

    return summarize(results, wall)


if __name__ == "__main__":
    raise SystemExit(main())

Configuration

Environment variables are read at import time, so you can set them in a shell profile or a systemd unit without editing the script.

Usage Examples

1. Hash every JSON file with 8 workers:

$ python3 parallel_files.py --dir ./data --workers 8 --ext .json --op hash -v
Discovered 3 file(s) under /home/me/data (op=hash, workers=8, ext=.json)
[   1/3] OK  /home/me/data/a.json -> 3f786850e387550fdab836ed7e6dc881de23001b...
[   2/3] OK  /home/me/data/b.json -> 89e6c98d92887913cadf06b2adb97f26cde4849b...
[   3/3] OK  /home/me/data/c.json -> 2c26b46b68ffc68ff99b453c1d30413413422d70...

============================================================
Processed : 3 file(s)
Succeeded : 3
Failed    : 0
Wall time : 0.021s   (sum of worker time: 0.047s)
Parallelism factor: 2.24x

2. Validate a directory of JSON files:

$ python3 parallel_files.py --dir ./exports --ext .json --op json
Discovered 128 file(s) under /home/me/exports (op=json, workers=4, ext=.json)
  ... 25/128 processed
  ... 50/128 processed
  ... 75/128 processed
  ... 100/128 processed
  ... 128/128 processed

============================================================
Processed : 128 file(s)
Succeeded : 126
Failed    : 2
Wall time : 0.847s   (sum of worker time: 2.913s)
Parallelism factor: 3.44x

Failures:
  - /home/me/exports/broken.json: JSONDecodeError: Expecting ',' delimiter: line 42
  - /home/me/exports/empty.json: JSONDecodeError: Expecting value: line 1 column 1

3. Dry-run to preview the workload:

$ python3 parallel_files.py --dir /var/log --ext .log --dry-run
Discovered 47 file(s) under /var/log (op=hash, workers=4, ext=.log)
/var/log/app.log
/var/log/db.log
/var/log/nginx/access.log
...

Error Handling

• Missing / non-directory root: discover_files raises FileNotFoundError or NotADirectoryError; main catches these and exits with code 2 and a clear message.
• Per-file failures (permission denied, malformed JSON, encoding errors): caught inside process_one and returned as a failed Result. The batch continues; failures are listed in the summary and cause exit code 1.
• Broken symlinks or unreadable stat: skipped silently during discovery (they'd fail anyway and would just add noise).
• Oversized files: filtered out at discovery time via --max-mb, so a multi-GB file can't monopolize a worker.
• Interrupt (Ctrl-C): run catches KeyboardInterrupt, cancels pending futures, and the executor context manager shuts down the pool cleanly. Exit code 130.
• Invalid worker count: rejected before any work begins with exit code 2.

Extending Ideas

• Add new operations: register a function in OPERATIONS with signature (path: Path, **kwargs) -> str — the CLI automatically exposes it via --op. Good candidates: MIME detection, image dimension probing, CSV row counting, or text encoding sniffing.
• Switch to ProcessPoolExecutor for CPU-bound work: if you add operations like image resizing or compression, swap the executor class — the rest of the driver code is unchanged because both expose the same submit/as_completed interface.
• Emit machine-readable output: replace the summarize print block with json.dumps([asdict(r) for r in results]) and add a --format json flag for pipeline consumption (e.g., piping into jq or a monitoring system).
• Stream results to a sink: instead of accumulating in a list, write each Result to a SQLite table or CSV as it completes — useful for very large directories where holding everything in memory is undesirable.