Fork/Join: Chia để trị song song với work-stealing

TL;DR: Fork/Join là framework chuyên cho bài toán divide-and-conquer CPU-bound: task tự chẻ mình thành task con đệ quy rồi gộp kết quả. Sức mạnh nằm ở work-stealing — mỗi worker giữ một deque riêng, owner push/pop ở đầu (LIFO), thread đói việc trộm từ đuôi (task to nhất) — nên tải tự cân ra các core mà không có hàng đợi trung tâm làm cổ chai. Viết task qua RecursiveTask/RecursiveAction với sequential cutoff: ngưỡng quá nhỏ thì chi phí điều phối nuốt hết lợi ích, quá lớn thì không phủ kín core. Parallel stream chạy trên chính commonPool này. Tuyệt đối không block I/O trong compute() — khi buộc phải chặn, dùng ManagedBlocker hoặc pool riêng.

1. Bài toán chia để trị, và vì sao thread pool thường không vừa

Lấy một việc cụ thể, đủ đơn giản để soi rõ cấu trúc: tính tổng một mảng long rất lớn. Tuần tự thì chỉ là một vòng lặp - nhưng nếu mảng có hàng chục triệu phần tử và máy tám core, để bảy core ngồi không là lãng phí.

Ý tưởng chia để trị rất tự nhiên: tổng cả mảng bằng tổng nửa trái cộng nửa phải, mà mỗi nửa lại chẻ tiếp như thế, tới khi mỗi mảnh đủ nhỏ để tính trực tiếp rồi gộp ngược lên. Cái hay là hai nửa độc lập hoàn toàn - không nửa nào đọc hay ghi dữ liệu của nửa kia - nên chạy song song mà không cần một cái khóa nào.

Câu hỏi là chạy song song bằng gì. Phản xạ đầu tiên là ném mỗi mảnh vào một ExecutorService rồi get kết quả về - nhưng đó là một cái bẫy nghiêm trọng (ta gặp lại ở mục 6). Một task chia để trị sau khi fork hai con phải chờ chúng xong mới gộp được; nếu cha và con cùng nằm trong một thread pool cố định, các task cha chiếm hết thread để ngồi chờ, task con không còn thread nào chạy: cả pool đứng hình - deadlock do cạn thread, càng chắc xảy ra khi cây càng sâu.

Fork/Join được thiết kế để giải đúng nút thắt này: nó biết task có dạng cây, biết task cha sẽ chờ con, và tổ chức lại cách thread tiêu thụ công việc sao cho "cha chờ con" không bao giờ làm chết pool.

2. ForkJoinPool và work-stealing

Một thread pool thông thường như Executors.newFixedThreadPool có một hàng đợi trung tâm duy nhất: mọi thread đói việc đều lấy task từ đó. Mô hình này tốt cho các task độc lập, kích thước tương đương, đến từ bên ngoài. Nhưng với cây divide-and-conquer — hàng triệu task con bé tí sinh ra liên tục cùng chen vào một hàng đợi chung — nó thành điểm tranh chấp nóng.

ForkJoinPool làm khác: mỗi worker có một hàng đợi riêng, gọi là work-stealing deque - hàng đợi hai đầu. Khi một task fork ra con, con được đẩy vào đầu local của deque của chính thread đó; khi cần việc tiếp theo, thread cũng lấy từ đầu local. Bình thường mỗi thread làm với hàng đợi riêng theo kiểu stack, gần như không tranh chấp.

Phần thú vị xảy ra khi một thread làm hết việc: thay vì ngồi không, nó trở thành kẻ trộm, nhìn sang deque của thread khác và lấy trộm một task - nhưng từ đầu kia, đuôi của deque nạn nhân. Chủ deque lấy ở một đầu, kẻ trộm lấy ở đầu đối diện, nên hai bên hiếm khi giành cùng một task.

Hình dung một bếp ăn tám đầu bếp, mỗi người một chồng phiếu order úp trước mặt: ai lo chồng nấy, lấy phiếu trên cùng mà làm. Đầu bếp nào làm hết chồng của mình thì rút một phiếu từ đáy chồng của người đang bận nhất - phiếu đáy thường là order lớn chưa ai đụng. Chủ chồng rút từ trên, kẻ giúp việc rút từ dưới, nên hiếm khi với cùng một tờ; tải tự cân, không ai ngồi không khi bếp còn việc.

Sơ đồ dưới chụp lại đúng khoảnh khắc đó — Worker 1 đang bận với deque đầy task, Worker 2 vừa hết việc và đi trộm:

Hai chiều mũi tên là toàn bộ linh hồn thiết kế. Owner làm việc kiểu LIFO ở đầu deque: task con vừa fork còn nóng trong cache, lấy ra làm ngay là rẻ nhất. Thief trộm kiểu FIFO ở đuôi: task nằm đó lâu nhất được fork sớm nhất — nhánh to nhất của cây — nên một lần trộm ôm về cả một cây con đáng kể, đỡ trộm vặt. Nhờ công việc phân tán ra các deque cục bộ, không có hàng đợi trung tâm làm cổ chai, Fork/Join chịu được cây đệ quy hàng triệu node mà không nghẹt.

2.1 commonPool

Ta không nhất thiết phải tự tạo ForkJoinPool: JVM duy trì sẵn một pool dùng chung cho toàn ứng dụng, gọi là common pool, lấy qua ForkJoinPool.commonPool(). Đây cũng là pool mặc định của parallel stream và của CompletableFuture.supplyAsync khi không truyền executor.

Kích thước mặc định của common pool là số core trừ một (cộng chính thread đang gọi, nên hiệu quả bằng số core): với CPU-bound, nhiều thread hơn số core không làm nhanh hơn, chỉ thêm chi phí context switching. Trên JDK 25 vẫn chỉnh được qua property java.util.concurrent.ForkJoinPool.common.parallelism, nhưng mặc định thường đúng.

Cái tiện của common pool cũng là cái bẫy (ta quay lại ở mục 6): vì dùng chung cho cả ứng dụng, một task cư xử xấu - đặc biệt task blocking - có thể bỏ đói mọi thứ khác đang dựa vào nó.

3. RecursiveTask và RecursiveAction

Để mô tả việc cho Fork/Join, ta kế thừa một trong hai lớp: RecursiveTask<V> cho task trả về kết quả, RecursiveAction cho task không trả gì. Cả hai bắt ta cài đúng một phương thức compute(), nơi chứa toàn bộ logic chẻ-và-gộp.

Khung sườn của compute() luôn cùng một hình dạng: mảnh đủ nhỏ thì làm trực tiếp; còn to thì chẻ đôi thành hai con rồi điều phối chúng chạy song song (chi tiết thứ tự ở ngay dưới).

Viết lại bài toán tính tổng mảng long ở mục 1 thành RecursiveTask:

public final class SumTask extends RecursiveTask<Long> {

    private static final int THRESHOLD = 10_000;   // nguong sequential cutoff
    private final long[] data;
    private final int lo, hi;

    SumTask(long[] data, int lo, int hi) { this.data = data; this.lo = lo; this.hi = hi; }

    @Override
    protected Long compute() {
        int length = hi - lo;
        if (length <= THRESHOLD) {              // du nho: lam thang
            long sum = 0;
            for (int i = lo; i < hi; i++) sum += data[i];
            return sum;
        }
        int mid = lo + length / 2;              // con to: che doi
        SumTask left  = new SumTask(data, lo, mid);
        SumTask right = new SumTask(data, mid, hi);
        left.fork();                            // giao nua trai cho pool chay song song
        long rightSum = right.compute();        // tu minh lam nua phai
        long leftSum  = left.join();            // cho va lay ket qua nua trai
        return leftSum + rightSum;
    }

    public static long sum(long[] data) {
        return ForkJoinPool.commonPool().invoke(new SumTask(data, 0, data.length));
    }
}

Cây chia việc đoạn code dựng lên trông như sau — mảng 40 nghìn phần tử, THRESHOLD = 10_000: node trên ngưỡng chẻ đôi, lá dưới ngưỡng tính thẳng bằng vòng for:

flowchart TD
    R["sum [0 .. 40k)"] --> L1["sum [0 .. 20k)"]
    R --> R1["sum [20k .. 40k)"]
    L1 --> L2["sum [0 .. 10k)<br/>la: tinh thang"]
    L1 --> L3["sum [10k .. 20k)<br/>la: tinh thang"]
    R1 --> R2["sum [20k .. 30k)<br/>la: tinh thang"]
    R1 --> R3["sum [30k .. 40k)<br/>la: tinh thang"]

    style R fill:#93C5FD
    style L2 fill:#6EE7B7
    style R3 fill:#6EE7B7

Kết quả chảy ngược từ lá lên gốc qua các lần join: mỗi node cộng hai con, gốc trả về tổng cả mảng. Bốn lá độc lập nhau hoàn toàn nên bốn core có thể cùng tính một lúc.

Thứ tự fork → compute → join không tùy tiện, và nó quyết định nhanh hay chậm: ta fork nửa trái (đẩy vào deque cho thread khác có thể trộm), tự mình chạy nửa phải bằng compute(), rồi mới join nửa trái. Nếu nửa trái đã bị trộm thì lúc join nó đã xong; nếu chưa, join tự thực thi nó ngay trên thread hiện tại thay vì chờ suông. Nhờ thread hiện tại luôn bận với một nửa, ta giảm số task thừa và không để thread chết dí.

Hai biến thể sai hay gặp. Một là fork cả hai con rồi join cả hai: chạy đúng nhưng phí - thread hiện tại đẩy hết việc đi rồi ngồi chờ, tạo thêm một task thừa. Tệ hơn là join con vừa fork trước rồi mới compute con kia: join chặn cho tới khi nửa trái xong mới đụng nửa phải, hai con bị tuần tự hóa, song song biến mất hoàn toàn mà kết quả vẫn đúng nên rất khó phát hiện. Quy tắc an toàn: luôn join theo thứ tự ngược với thứ tự fork.

Nếu bộ quy tắc thứ tự này dễ trượt tay, JDK cho sẵn một idiom gọn hơn: ForkJoinTask.invokeAll(left, right) — tự lo đúng vũ đạo hộ bạn (fork các task sau, chạy task đầu ngay trên thread hiện tại, chờ tất cả hoàn tất):

// Trong compute(), thay cho cap fork/compute/join thu cong (base case giu nguyen):
ForkJoinTask.invokeAll(left, right);    // fork right, compute left, wait for both
return left.join() + right.join();      // both done here -- join returns immediately

Sau khi invokeAll trả về, cả hai con đều đã xong, nên hai lần join phía sau chỉ lấy kết quả, không chặn. Phiên bản này không thể mắc lỗi đảo thứ tự fork/join và đọc rõ ý đồ hơn; nó còn nhận varargs/collection cho cây nhiều nhánh — đáng làm lựa chọn mặc định khi viết compute() mới.

Và compute() không cần một cái khóa nào vì mỗi SumTask chỉ đọc đoạn [lo, hi) của riêng nó, không ghi vào data — task con độc lập, không chia sẻ mutable state là điều kiện tiên quyết để Fork/Join phát huy. Nếu các nhánh tranh nhau ghi vào một biến đếm chung, ta rơi vào shared mutable state của bài Thread Safety, lợi ích song song bị nuốt bởi contention.

4. Sequential cutoff nên đặt bao nhiêu?

Con số THRESHOLD = 10_000 là linh hồn của cả việc - sequential cutoff, ranh giới mà dưới đó ta thôi chẻ và làm thẳng. Đặt sai, framework vẫn chạy đúng nhưng có thể chậm hơn cả vòng lặp tuần tự.

Lý do nằm ở chi phí. Mỗi lần chẻ là tạo hai object task mới, đẩy vào deque, có thể bị trộm sang thread khác, rồi join lại - tất cả đều tốn CPU và bộ nhớ. Nếu ngưỡng quá nhỏ - chẻ tới khi mỗi mảnh chỉ còn mười phần tử - ta tạo hàng triệu task để cộng vài chục con số mỗi cái; chi phí quản lý task lấn át công việc thực, chương trình bò còn chậm hơn một vòng for. Đây là cái bẫy "tách quá nhỏ": tưởng càng song song càng nhanh, hóa ra ngược lại.

Ở thái cực kia, ngưỡng quá lớn thì không chẻ đủ mảnh để phủ kín các core: mảng mười triệu phần tử với ngưỡng năm triệu chỉ tạo hai mảnh - chạy trên hai core, sáu core còn lại đứng nhìn. Tệ hơn, nếu hai mảnh không cân nhau về thời gian, core làm xong trước chẳng có gì để trộm.

Điểm cân bằng là tạo nhiều mảnh hơn số core một chút: khi một core gặp mảnh nặng bất thường, các core khác đã có sẵn một rổ task chờ bị trộm - nhưng mỗi mảnh vẫn phải đủ to để lượng việc của nó đáng kể so với chi phí một lần fork/join. Không có con số vàng - nó phụ thuộc mỗi phần tử tốn bao nhiêu tính toán - nên phải đo: đặt ngưỡng sao cho tổng số task khoảng vài lần tới vài chục lần số core, rồi benchmark để tinh chỉnh.

Và đừng quên phép thử quan trọng nhất: so với phiên bản tuần tự. Song song hóa chỉ đáng khi dữ liệu đủ lớn và phép tính đủ nặng để bù chi phí điều phối. Trên mảng vài nghìn phần tử, một vòng for thường thắng cả Fork/Join lẫn parallel stream - dẫn ta thẳng sang mục tiếp theo.

5. Fork/Join là nền của parallel streams

Phần lớn lập trình viên Java dùng Fork/Join mỗi ngày mà không gọi tên nó: khi viết stream.parallel() hay list.parallelStream(), cái máy bên dưới chính là ForkJoinPool.commonPool(). Stream framework tự lo phần ta vừa viết tay ở mục 3 - tách nguồn dữ liệu qua Spliterator, gói thành task Fork/Join, gộp kết quả.

Nghĩa là viết lại bài tính tổng bằng parallel stream gọn còn một dòng:

long total = java.util.Arrays.stream(data).parallel().sum();

Bên dưới vẫn là cơ chế chẻ-fork-trộm-gộp đó. Đây gần như luôn là cách nên dùng trước: ngắn, ít chỗ sai, JDK đã chỉnh sequential cutoff khá hợp lý qua Spliterator. Tự viết RecursiveTask chỉ đáng khi cấu trúc chẻ không khớp mô hình stream - ví dụ duyệt cây bất đối xứng, hay cần kiểm soát ngưỡng và chiến lược gộp tinh vi hơn.

Một điều cần tỉnh táo: parallel stream chia chung common pool, nên nhét blocking vào một map sẽ ghim thread dùng chung, làm chậm cả CompletableFuture khác. Nó sinh ra cho tính toán CPU-bound trên dữ liệu đã trong bộ nhớ, không phải I/O - đúng điểm khởi đầu cho mục cạm bẫy.

6. Cạm bẫy: blocking trong Fork/Join và chuyện chia sẻ common pool

Cạm bẫy lớn nhất lặp lại cảnh báo từ bài Future & CompletableFuture, nhưng ở đây nghiêm trọng hơn. ForkJoinPool được kích thước theo số core với giả định ngầm: mỗi worker gần như lúc nào cũng đang tính, không ngồi chờ. Nhét một thao tác blocking - đọc file, gọi HTTP, chờ khóa, sleep - vào compute() phá vỡ giả định đó: một thread chờ I/O không tính gì nhưng vẫn chiếm một trong số ít ỏi slot của pool. Vài task blocking là đủ bỏ đói cả pool - không còn thread nào chạy task thực, dù CPU rảnh.

Khi bắt buộc phải chặn, framework cho một lối thoát có kiểm soát: ForkJoinPool.ManagedBlocker. Ta gói thao tác blocking vào một ManagedBlocker và chờ qua ForkJoinPool.managedBlock(...); pool nhận biết một thread sắp chặn và tạm bù thêm worker để duy trì mức song song. Đây là van an toàn cho một thao tác chặn không thể tránh, không phải lời mời chạy I/O ồ ạt trên Fork/Join.

// Goi mot thao tac blocking de ForkJoinPool co the bu them worker trong luc cho.
ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    private Object result;
    private boolean done;
    @Override public boolean block() throws InterruptedException {
        result = slowBlockingCall();   // thao tac chan khong tranh duoc
        done = true;
        return true;                   // true = da xong, khong can chan nua
    }
    @Override public boolean isReleasable() { return done; }
});

Cạm bẫy thứ hai là dùng chung common pool. Vì parallel stream, CompletableFuture.supplyAsync mặc định và mọi commonPool().invoke đều đổ vào một pool, một tác vụ tham lam hay blocking ở một góc có thể âm thầm làm chậm một góc khác: một báo cáo nặng chạy parallel stream làm tăng độ trễ của một luồng CompletableFuture chẳng liên quan, chỉ vì hai bên vô tình chia nhau common pool.

Cách phòng là cô lập: khi một workload đáng kể cần Fork/Join - nhất là nếu có nguy cơ chặn hoặc chạy lâu - hãy tạo một ForkJoinPool riêng và invoke/submit vào đó thay vì mượn common pool:

try (ForkJoinPool pool = new ForkJoinPool(4)) {   // pool rieng, co lap khoi common pool
    long total = pool.invoke(new SumTask(data, 0, data.length));
    // ... dung total
}                                                 // try-with-resources tu shutdown pool

Từ Java 19, ForkJoinPool cài AutoCloseable, nên trên JDK 25 ta đặt được nó trong try-with-resources cho tự đóng. Một pool riêng tách bạch sự cố: nếu nó bị bỏ đói hay chạy chậm, thiệt hại khoanh trong workload đó, không lan ra phần còn lại.

7. Tie-in capstone: đối soát báo cáo bán vé theo divide-and-conquer

TicketFlow v3 chủ yếu sống bằng Executor và CompletableFuture. Fork/Join xuất hiện như một mảnh độc lập, đúng chỗ nó thuộc về: một tác vụ tổng hợp CPU-bound chạy ngoài đường đi của request.

Hình dung cuối ngày hệ thống cần đối soát: quét một mảng rất lớn bản ghi giao dịch trong bộ nhớ, cộng tổng doanh thu và đếm vé theo từng sự kiện. Đây là divide-and-conquer thuần khiết - bản ghi độc lập, phép gộp có tính kết hợp, không I/O - nên hợp Fork/Join hơn hẳn trải lên CompletableFuture. Cấu trúc giống hệt SumTask: chẻ đôi khoảng bản ghi tới một ngưỡng, mỗi lá cộng cục bộ một map doanh thu-theo-sự-kiện, mỗi lần join gộp hai map con - mỗi task chỉ đọc đoạn của riêng nó và trả về map mới nên không có shared mutable state, không cần khóa (đúng tinh thần mục 3).

Quan trọng không kém là biết khi nào không lôi Fork/Join vào: nếu đối soát cần đọc từng dòng từ database thì không còn CPU-bound, nhét truy vấn blocking vào compute() sẽ vấp đúng cạm bẫy mục 6. Ranh giới rạch ròi: I/O và điều phối nghiệp vụ thuộc về Executor và CompletableFuture; chỉ phần tính toán thuần trên dữ liệu trong bộ nhớ mới giao cho Fork/Join, chạy trên một pool riêng để khỏi tranh chấp common pool.

8. 📚 Deep Dive Oracle

9. Tóm tắt

• Fork/Join dành cho divide-and-conquer trên dữ liệu CPU-bound đã nằm sẵn trong bộ nhớ. Sức mạnh là work-stealing: mỗi worker giữ một deque riêng nên gần như không tranh chấp, và tự trộm việc từ đuôi deque kẻ khác khi đói — tải tự cân ra các core mà không cần hàng đợi trung tâm làm cổ chai.
• RecursiveTask/RecursiveAction mô tả việc qua compute(): nhỏ thì làm thẳng, to thì chẻ; fork một con, tự compute con kia, rồi join ngược thứ tự fork — hoặc dùng invokeAll cho chắc. Thứ tự này quyết định hiệu năng.
• Sequential cutoff là tham số sống còn: quá nhỏ thì chi phí fork/join nuốt hết lợi ích, quá lớn thì không đủ mảnh phủ kín core. Tạo nhiều mảnh hơn số core một chút, luôn benchmark so với tuần tự.
• Parallel stream chạy trên commonPool, nên là cách nên thử trước; tự viết RecursiveTask chỉ đáng khi cấu trúc bài toán không khớp mô hình stream.
• Đừng chặn trong Fork/Join — pool sizing theo số core giả định mọi thread đều đang tính, nên vài task blocking là đủ bỏ đói cả pool. Khi buộc phải chặn, dùng ManagedBlocker; workload nặng thì tạo ForkJoinPool riêng.

Cả Phần B — từ Executor, qua Future/CompletableFuture, tới Fork/Join — đứng trên một giả định: mỗi Thread của Java là tài nguyên đắt, gắn với một OS thread suốt vòng đời, nên phải gói vào pool và tránh để nó ngồi chờ. Phần C lật ngược chính giả định ấy — virtual thread, bài tiếp theo.

10. Tự kiểm tra

Bài tiếp theo: Virtual Threads: Thread-per-request trở lại