# How to Optimize `ThreadPoolExecutor` for `多线程分词.py` When Processing Very Large Text Files

> Optimize ThreadPoolExecutor for large text files in 多线程分词.py. Learn to tune max workers, reuse instances, and batch submissions for efficient processing. Boost performance now.

- Repository: [amzxyz/rime-lmdg](https://github.com/amzxyz/rime-lmdg)
- Tags: performance
- Published: 2026-02-24

---

**Optimize the `ThreadPoolExecutor` in `多线程分词.py` by tuning `max_workers` to your I/O profile, reusing global `OpenCC` and `jieba` instances across threads, and batching file submissions to avoid hitting OS file-descriptor limits.**

The `多线程分词.py` script in the `amzxyz/rime-lmdg` repository converts cleaned Chinese text files into tokenized, Simplified Chinese output using Jieba segmentation and OpenCC conversion. When processing gigabytes of corpus data, the default `ThreadPoolExecutor(max_workers=20)` configuration creates significant bottlenecks through excessive I/O blocking, per-thread object overhead, and potential file-descriptor exhaustion.

## Understanding the Processing Pipeline

The script located at `wanxiang/多线程分词.py` performs three core operations per file:

1. **Dictionary Loading**: Loads a user-defined Jieba dictionary once at startup via `load_custom_dict()` (lines 19-30).
2. **Character Conversion**: Converts each line from Traditional to Simplified Chinese using **OpenCC** (lines 16-18).
3. **Tokenization**: Segments text using `jieba.cut` (lines 42-44) and writes results to disk.

The current implementation creates bottlenecks when scaling to large datasets because it opens too many concurrent files, reads line-by-line incurring excessive syscalls, and risks oversubscribing threads relative to available I/O bandwidth.

## Optimization Strategy 1: Tune `max_workers` to the I/O Profile

The default `max_workers=20` is arbitrary and often leads to context-switch thrashing on I/O-bound workloads. Instead, scale the worker count based on your CPU cores and I/O capacity.

**Best practice**: Set `max_workers` to twice the CPU count, capped at 32, to balance parallelism without overwhelming the system.

```python
import os

MAX_WORKERS = min(32, (os.cpu_count() or 1) * 2)

```

This calculation prevents oversubscription while maintaining enough threads to keep the CPU busy during I/O waits. For pure I/O-bound work on NVMe storage, you may increase this further, but start with the CPU-based heuristic.

## Optimization Strategy 2: Reuse Thread-Safe Global Instances

Both `OpenCC` and the loaded Jieba dictionary are **thread-safe for read-only operations**. The original script correctly defines `opencc = OpenCC('t2s')` at the module level, but ensure you never recreate these objects inside `process_file_stream`.

Keep these definitions global:

```python
from opencc import OpenCC
import jieba

# Global, thread-safe objects

opencc = OpenCC('t2s')
jieba.enable_parallel(False)  # Disable internal parallelism; we manage it via ThreadPoolExecutor

def load_custom_dict():
    # Load once at startup

    for path in custom_dict_paths:
        jieba.load_userdict(path)

```

Reusing these instances eliminates per-task object construction overhead, which becomes significant when processing millions of lines across thousands of files.

## Optimization Strategy 3: Batch Submissions to Limit File Handles

Opening dozens of files simultaneously may exceed the OS file-descriptor limit (`ulimit -n`). Instead of submitting all files at once, process them in batches to constrain resource usage.

```python
from concurrent.futures import ThreadPoolExecutor, as_completed

batch_size = 8  # Conservative limit for concurrent open files

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    for i in range(0, len(files), batch_size):
        batch = files[i:i + batch_size]
        futures = {
            executor.submit(process_file_stream, src, dst): (src, dst) 
            for src, dst in batch
        }
        
        for future in as_completed(futures):
            src, dst = futures[future]
            try:
                future.result()
            except Exception as e:
                print(f"Failed processing {src}: {e}")

```

This pattern ensures you never hold more than `batch_size * 2` file handles open simultaneously (input and output per thread), preventing "Too many open files" errors on large corpora.

## Optimization Strategy 4: Read and Write in Larger Blocks

Line-by-line reading generates excessive syscalls. For moderate-size files (under available RAM), read the entire file at once, process in memory, and write once.

```python
def process_file_stream(input_path, output_path):
    with open(input_path, "r", encoding="utf-8") as fin, \
         open(output_path, "w", encoding="utf-8") as fout:
        
        # Single read operation vs. thousands of readline() calls

        text = fin.read()
        
        for line in text.splitlines():
            simplified = opencc.convert(line.strip())
            if simplified:
                segmented = " ".join(jieba.cut(simplified, cut_all=False, HMM=False))
                fout.write(segmented + "\n")

```

For files larger than available memory, implement chunked reading with a buffer size of 4-8 MB instead of line-by-line iteration.

## Optimization Strategy 5: Implement Hash-Based Caching

Avoid re-processing unchanged files during iterative development by caching SHA-256 hashes of input files.

```python
import hashlib
import os

def file_hash(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for block in iter(lambda: f.read(8192), b""):
            h.update(block)
    return h.hexdigest()

def process_file_stream(input_path, output_path):
    # Skip if output exists and input unchanged (simplified check)

    if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
        return
    
    # ... processing logic

```

Store hashes in a sidecar file or database to enable incremental corpus updates without redundant tokenization.

## Optimization Strategy 6: When to Use `ProcessPoolExecutor`

If profiling with `cProfile` reveals that `jieba.cut` dominates CPU usage (>70% of runtime), the **Global Interpreter Lock (GIL)** becomes the bottleneck. Switch to `ProcessPoolExecutor` to bypass the GIL and achieve true parallelism across CPU cores.

```python
from concurrent.futures import ProcessPoolExecutor

# Use processes instead of threads for CPU-bound tokenization

with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
    futures = [executor.submit(process_file_stream, src, dst) 
               for src, dst in files]

```

Note that with processes, you must reload the Jieba dictionary inside each worker function or use `initializer` parameters to set up the worker environment, as global variables are not shared across processes.

## Complete Optimized Implementation

Here is the fully optimized version incorporating all strategies except process-based parallelism (use that only if profiling indicates CPU bottleneck):

```python
import os
import hashlib
from concurrent.futures import ThreadPoolExecutor, as_completed
import jieba
from opencc import OpenCC
from tqdm import tqdm

# Configuration

MAX_WORKERS = min(32, (os.cpu_count() or 1) * 2)
CUSTOM_DICT_DIR = "autodl-tmp/自定义分词词典"
CLEANED_FOLDER = "autodl-tmp/语料清洗后"
SEGMENTED_FOLDER = "autodl-tmp/语料分词后"
FILE_ENCODING = "utf-8"
BATCH_SIZE = 8

# Global thread-safe instances

opencc = OpenCC('t2s')
jieba.enable_parallel(False)

def load_custom_dict():
    if os.path.isdir(CUSTOM_DICT_DIR):
        for fn in os.listdir(CUSTOM_DICT_DIR):
            path = os.path.join(CUSTOM_DICT_DIR, fn)
            if os.path.isfile(path):
                jieba.load_userdict(path)
                print(f"Loaded custom dict: {path}")

def process_file_stream(input_path, output_path):
    try:
        # Cache check

        if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
            return

        with open(input_path, "r", encoding=FILE_ENCODING) as fin, \
             open(output_path, "w", encoding="utf-8") as fout:
            
            # Bulk read for efficiency

            text = fin.read()
            for line in text.splitlines():
                line = opencc.convert(line.strip())
                if line:
                    seg = " ".join(jieba.cut(line, cut_all=False, HMM=False))
                    fout.write(seg + "\n")
                    
    except Exception as exc:
        print(f"Error processing {input_path}: {exc}")

def process_folder_parallel(input_dir, output_dir):
    if not os.path.isdir(input_dir):
        raise FileNotFoundError(f"Input folder '{input_dir}' missing")
    os.makedirs(output_dir, exist_ok=True)

    files = [
        (os.path.join(input_dir, fn), os.path.join(output_dir, f"{fn}_segmented.txt"))
        for fn in os.listdir(input_dir) 
        if os.path.isfile(os.path.join(input_dir, fn))
    ]

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # Batch processing to limit file descriptors

        for i in range(0, len(files), BATCH_SIZE):
            batch = files[i:i + BATCH_SIZE]
            futures = {
                executor.submit(process_file_stream, src, dst): (src, dst) 
                for src, dst in batch
            }
            
            for future in tqdm(as_completed(futures), total=len(batch), desc=f"Batch {i//BATCH_SIZE + 1}"):
                src, dst = futures[future]
                try:
                    future.result()
                except Exception as e:
                    print(f"Failed on {src}: {e}")

if __name__ == "__main__":
    load_custom_dict()
    process_folder_parallel(CLEANED_FOLDER, SEGMENTED_FOLDER)

```

## Summary

- **Scale `max_workers`** using `min(32, os.cpu_count() * 2)` rather than fixed values to match your hardware I/O capacity.
- **Reuse global instances** of `OpenCC` and Jieba's loaded dictionary across all threads to eliminate per-task initialization overhead.
- **Batch file submissions** (e.g., 8 files at a time) to prevent exceeding OS file-descriptor limits when processing thousands of documents.
- **Read files in bulk** when memory permits, or in 4-8 MB chunks, to reduce syscall overhead compared to line-by-line reading.
- **Cache results** using file hashes to skip unchanged files during iterative corpus development.
- **Profile before switching** to `ProcessPoolExecutor`; only use processes if `jieba.cut` dominates CPU time and GIL contention is confirmed.

## Frequently Asked Questions

### What is the optimal `max_workers` value for processing text files on an SSD?

For I/O-bound text processing on fast NVMe storage, start with `min(32, (os.cpu_count() or 1) * 2)`. If CPU usage remains low and disk utilization is high, you can increase this to 64 or 128, but monitor for context-switch degradation. For network-mounted storage or traditional HDDs, reduce to `os.cpu_count()` or lower to prevent thrashing.

### Why does the script fail with "Too many open files" errors?

The default `ThreadPoolExecutor` submits all tasks immediately, opening every input and output file simultaneously. In `多线程分词.py`, this quickly exhausts the `ulimit -n` file-descriptor limit (typically 1024 on Linux). Implement **batch processing** with a batch size of 8-16 to ensure only a subset of files is open at any moment.

### Is `jieba.cut` thread-safe for concurrent use?

Yes, once the dictionary is loaded via `load_custom_dict()`, `jieba.cut` is thread-safe for read-only segmentation. However, **do not** call `jieba.load_userdict` from multiple threads simultaneously. Load dictionaries once at startup before spawning the thread pool, as implemented in the optimized example above.

### When should I switch from `ThreadPoolExecutor` to `ProcessPoolExecutor`?

Switch to `ProcessPoolExecutor` only if profiling (using `cProfile` or `py-spy`) shows that Python spends more than 70% of time inside `jieba.cut` and CPU cores are underutilized due to the GIL. For most corpus processing workflows, the bottleneck is I/O (reading/writing files), making threads more efficient than processes because they share memory and avoid serialization overhead.