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

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.

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:

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.

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.

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.

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.

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):

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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}