Python Parallel Batches: Bound Pending Tasks and Report Every File Failure

Submitting one Future per file can create a large backlog when a batch contains thousands of files, even with only four worker threads. Another common failure is to submit work without retrieving results, leaving file errors out of the final report.

This beginner-to-intermediate tutorial builds a read-only UTF-8 text line counter using the Python standard library. It reads a path list incrementally, bounds retained tasks, records each file’s success or failure, and gates the next automation step with an exit status.

1. Method: bound both threads and retained tasks

The existing process pools, thread pools, and coroutines introduction explains concurrency models. Here we connect a thread pool to a runnable batch entry point with these rules:

1. --workers defaults to 4 and accepts 1–16, limiting concurrently running worker threads.
2. pending holds at most 2 × workers Futures, including queued, running, and completed results awaiting collection. At the limit, collect results before reading more paths.
3. Emit one JSON record per completed file, including missing files and invalid UTF-8. Any file failure makes the final exit status 1.

The official ThreadPoolExecutor documentation defines the worker limit. wait(..., FIRST_COMPLETED) waits until at least one task finishes. Our code implements the separate submission bound.

Threads demonstrate independent file reads here; faster execution is not guaranteed. Disk throughput, file sizes, decoding, and counting costs affect results. Start with 2–4 threads and measure your workload.

2. Prepare the environment and input

Use an existing Python 3.9 or newer interpreter; no third-party packages are needed. Examples were verified on Linux with Python 3.11.4. Commands work in bash or zsh.

1. Create a fresh demo directory and keep subsequent commands in the same terminal and directory. Choose a new name when repeating the experiment:

python3 --version
mkdir parallel-batch-demo
cd parallel-batch-demo
printf 'alpha\nbeta\n' > a.txt
printf 'gamma' > 'b space.txt'
printf 'a.txt\nb space.txt\n' > good.txt

2. Save the next section as count_files.py, or download the complete script.
3. good.txt is a BOM-free UTF-8 manifest with one path per line. Relative paths use the manifest’s directory as their base; absolute paths also work. Empty lines are skipped and spaces in filenames are preserved. This format cannot represent filenames containing newlines, and repeated paths are processed repeatedly.

Use this example only with completed ordinary local text files that remain unchanged during reads. Keep FIFOs, device files, and remote mount paths that may block indefinitely out of the manifest.

3. Complete implementation

"""Count UTF-8 text lines with bounded parallel submissions. Python 3.9+."""
import argparse
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
import json
from pathlib import Path
import sys


def worker_count(value):
    number = int(value)
    if not 1 <= number <= 16:
        raise argparse.ArgumentTypeError("workers must be 1..16")
    return number


def count_lines(path):
    with path.open(encoding="utf-8", errors="strict") as source:
        return sum(1 for _ in source)


def collect(pending):
    done, _ = wait(pending, return_when=FIRST_COMPLETED)
    failed = 0
    for future in done:
        path = pending.pop(future)
        record = {"type": "file", "path": str(path)}
        try:
            record.update(ok=True, lines=future.result())
        except (OSError, UnicodeError, ValueError) as exc:
            failed += 1
            record.update(ok=False, error=f"{type(exc).__name__}: {exc}")
        print(json.dumps(record), flush=True)
    return len(done), failed


def run(manifest, workers):
    manifest = manifest.absolute()
    pending = {}
    processed = failed = peak_pending = 0
    with manifest.open(encoding="utf-8", errors="strict") as source:
        with ThreadPoolExecutor(max_workers=workers) as pool:
            for line in source:
                name = line.rstrip("\r\n")
                if not name:
                    continue
                path = manifest.parent / name
                pending[pool.submit(count_lines, path)] = path
                peak_pending = max(peak_pending, len(pending))
                if len(pending) >= 2 * workers:
                    count, errors = collect(pending)
                    processed += count
                    failed += errors
            while pending:
                count, errors = collect(pending)
                processed += count
                failed += errors
    print(json.dumps({
        "type": "summary", "processed": processed, "failed": failed,
        "workers": workers, "peak_pending": peak_pending,
    }), flush=True)
    return 1 if failed else 0


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("manifest", type=Path, help="UTF-8 path list, one per line")
    parser.add_argument("--workers", type=worker_count, default=4)
    args = parser.parse_args()
    try:
        return run(args.manifest, args.workers)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"FATAL: {exc}", file=sys.stderr)
        return 1


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

count_lines() reads files without modifying them. Empty files have zero lines; a final segment without a newline still counts as a line. Text reading handles LF, CRLF, and CR. These are text lines, not CSV records: CSV fields can contain newlines, as explained in splitting CSV by records.

The main thread writes reports; workers only return line counts. Future.result() re-raises worker exceptions, so the collection step catches expected file and encoding errors. Unexpected programming errors still terminate the program with an exception instead of fabricating a successful summary.

Reports follow collection order; tasks collected together have no fixed order. Match records by path, not by their position against the manifest. A summary appears only after all work has been collected. If the manifest itself cannot be opened or decoded, stderr receives FATAL and the exit status is 1. stdout may contain partial file records without a complete summary.

4. Run and verify healthy input

1. View help and process both files. Redirection creates or truncates the report, so choose a fresh report path:

python3 count_files.py --help
python3 count_files.py good.txt --workers 2 > good-report.jsonl
cat good-report.jsonl

The two type: file records have lines values of 2 and 1. The final record is:

{"type": "summary", "processed": 2, "failed": 0, "workers": 2, "peak_pending": 2}

2. Verify by path rather than output order:

python3 - <<'PY_CHECK'
import json
from pathlib import Path
rows = [json.loads(line) for line in
        Path("good-report.jsonl").read_text(encoding="utf-8").splitlines()]
files = rows[:-1]
assert {Path(row["path"]).name: row["lines"] for row in files} == {
    "a.txt": 2, "b space.txt": 1,
}
assert all(row["ok"] for row in files)
assert rows[-1] == {
    "type": "summary", "processed": 2, "failed": 0,
    "workers": 2, "peak_pending": 2,
}
print("PASS: both files counted; pending bound respected")
PY_CHECK

peak_pending measures the peak number of retained Futures, not the number of threads actually running simultaneously. Even a quickly completed file remains in pending until collected.

5. Verify failures and exit status

1. Create invalid UTF-8 and combine healthy, missing, and wrongly encoded files in one manifest:

python3 - <<'PY_DATA'
from pathlib import Path
Path("bad.bin").write_bytes(b"\xff\n")
Path("mixed.txt").write_text(
    "a.txt\nmissing.txt\nbad.bin\nb space.txt\n", encoding="utf-8"
)
PY_DATA
if python3 count_files.py mixed.txt --workers 2 > mixed-report.jsonl; then
  printf 'UNEXPECTED: all inputs accepted\n'
else
  batch_status=$?
  printf 'exit=%s\n' "$batch_status"
fi
python3 - <<'PY_CHECK'
import json
from pathlib import Path
rows = [json.loads(line) for line in
        Path("mixed-report.jsonl").read_text(encoding="utf-8").splitlines()]
assert rows[-1]["processed"] == 4
assert rows[-1]["failed"] == 2
assert rows[-1]["peak_pending"] <= 4
bad = {Path(row["path"]).name: row["error"]
       for row in rows[:-1] if not row["ok"]}
assert bad["missing.txt"].startswith("FileNotFoundError:")
assert bad["bad.bin"].startswith("UnicodeDecodeError:")
print("PASS: two failures recorded; healthy files still completed")
PY_CHECK

Expect exit=1 and PASS. All four entries receive reports, two fail, and healthy files still finish. Failure details appear in each JSONL error field, so empty stderr does not establish success.

2. Gate the next task with the exit status, then check invalid arguments. Run the last two lines interactively; scripts using set -e should put expected failures inside an if condition:

if python3 count_files.py good.txt --workers 2 > next-report.jsonl; then
  python3 -c 'print("NEXT: every listed file passed")'
else
  printf 'STOP: inspect the report and stderr\n' >&2
fi
python3 count_files.py good.txt --workers 0
printf 'exit=%s\n' "$?"

The healthy manifest prints NEXT; replacing it with mixed.txt prints STOP. --workers 0 returns 2, the default error exit for argparse argument validation. An empty manifest returns 0 with processed: 0 and failed: 0; separately require a positive count if your task needs input.

6. Troubleshooting and limits

A Future limit is not a process memory limit: individual manifest lines, text lines, and system buffers still consume memory. This script keeps collecting a batch after individual failures. It does not stop on the first file error, retry tasks, persist progress, or forcibly cancel threads.

Summary

Bound both running worker threads and retained Futures, and retrieve every result. Identify reports by path, aggregate failures, and return a reliable exit status before connecting the checker to another task. Test healthy and broken inputs together, then tune concurrency for your disk and workload.