Structured Concurrency — StructuredTaskScope Java 25

TL;DR: Structured Concurrency (GA Java 25, JEP 505) áp dụng nguyên tắc structured programming vào concurrency: task con không được sống lâu hơn task cha. StructuredTaskScope đảm bảo nếu scope thoát (kể cả do exception), tất cả subtask bị cancel và join. ShutdownOnFailure dừng toàn bộ khi 1 task fail. ShutdownOnSuccess trả kết quả nhanh nhất. So với CompletableFuture.allOf(), Structured Concurrency tự động cancel, error propagation rõ ràng, và code sequential-style đọc dễ hơn nhiều. Thiết kế tối ưu cho Virtual Thread (bài 05) — scale tới triệu subtask không tốn OS thread.

Bài này build-on bài 05 (Virtual Thread), bài 07 (CompletableFuture), và bài 06 (ReentrantLock) — đọc 3 bài đó trước. Đặc biệt bài 07 mục 7 về allOf — những giới hạn ở đó là motivation trực tiếp cho Structured Concurrency.

1. Scenario — resource leak với CompletableFuture

Service cần lấy 3 piece of data song song: user profile, active orders, personalized recommendations. Dùng CompletableFuture.allOf():

public ProfileView getProfile(long userId) {
    ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();

    CompletableFuture<User>        userCF  = CompletableFuture.supplyAsync(() -> fetchUser(userId), pool);
    CompletableFuture<List<Order>> orderCF = CompletableFuture.supplyAsync(() -> fetchOrders(userId), pool);
    CompletableFuture<List<Item>>  recCF   = CompletableFuture.supplyAsync(() -> recommend(userId), pool);

    CompletableFuture.allOf(userCF, orderCF, recCF).join();  // wait all

    return new ProfileView(userCF.join(), orderCF.join(), recCF.join());
}

Bài toán: nếu fetchUser() ném exception sau 50ms, allOf fail ngay — nhưng fetchOrders() và recommend() tiếp tục chạy đến khi kết thúc, tốn resource vô ích.

Hơn nữa: nếu method caller timeout và cancel, làm sao propagate cancel xuống 3 CF con? Không có API tự động — phải track và cancel() thủ công từng CF. Và như đã thấy ở bài 07, CompletableFuture.cancel() không thực sự interrupt computation.

Structured Concurrency giải quyết: scope fail → tất cả subtask bị interrupt và join trước khi scope đóng. Không resource leak. Không manual cancel tracking.

2. Định nghĩa — structured concurrency là gì và vì sao tồn tại

Structured concurrency (thuật ngữ do Nathaniel J. Smith giới thiệu 2016, sau đó Ron Pressler của Oracle áp vào Project Loom) áp nguyên tắc structured programming vào concurrent code:

Nếu task A khởi động task B, A không được kết thúc trước khi B kết thúc.

Giống như trong structured programming: nếu function A gọi function B, A không return trước khi B return. Call stack đảm bảo điều này tự động. Structured Concurrency mang đảm bảo tương tự vào concurrent task.

Lịch sử JEP:

• JEP 428 (Java 19 incubator): StructuredTaskScope trong jdk.incubator.concurrent.
• JEP 437 (Java 20 incubator): tiếp tục incubation, ít thay đổi.
• JEP 453 (Java 21 preview): chuyển sang java.util.concurrent, thêm Subtask<T>.
• JEP 462 (Java 21 second preview): refinement, Subtask.state().
• JEP 480 (Java 23 preview): API ổn định hơn.
• JEP 499 (Java 24 preview): cleanup cuối.
• JEP 505 (Java 25): GA — finalized. Không cần --enable-preview.

Hôm nay (2026-05-08), Java 25 đã GA (tháng 9/2025) — Structured Concurrency production-ready.

3. Unstructured vs Structured — so sánh cơ chế

Unstructured (CompletableFuture / Future):

Main task
├── CF A (submit to pool) ──────────────────── runs independently
├── CF B (submit to pool) ──────────────────── runs independently
└── CF C (submit to pool) ──────────────────── runs independently
                                                ^
                                                Nếu main fail, A/B/C vẫn chạy

Task con tồn tại độc lập với main. Lifecycle không liên kết. Cancellation không tự động propagate. Exception từ 1 task không tự cancel tasks còn lại.

Structured (StructuredTaskScope):

Main task (scope owner)
├── scope.fork(A) ─── Subtask A
├── scope.fork(B) ─── Subtask B
└── scope.fork(C) ─── Subtask C
scope.join()  ← wait all A, B, C
// scope.close() → nếu scope thoát, cancel bất kỳ subtask còn chạy

Tất cả subtask được join trước khi scope đóng. Nếu scope thoát do exception hoặc timeout, JVM interrupt tất cả subtask chưa xong. Task con không thể "sống lâu hơn" task cha.

4. API cơ bản — StructuredTaskScope

StructuredTaskScope<T> là abstract class trong java.util.concurrent. Lifecycle qua try-with-resources:

import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Subtask;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {

    Subtask<User>        userTask  = scope.fork(() -> fetchUser(id));
    Subtask<List<Order>> orderTask = scope.fork(() -> fetchOrders(id));
    Subtask<List<Item>>  recTask   = scope.fork(() -> recommend(id));

    scope.join();            // wait for all subtasks to complete
    scope.throwIfFailed();   // rethrow if any subtask failed

    // All succeeded -- access results
    return new ProfileView(
        userTask.get(),
        orderTask.get(),
        recTask.get()
    );

} // scope.close() auto-cancels any still-running subtasks

Giải thích từng bước:

1. new StructuredTaskScope.ShutdownOnFailure() — tạo scope với policy "nếu 1 subtask fail, cancel tất cả còn lại".
2. scope.fork(callable) — start subtask trên 1 Virtual Thread mới trong scope. Trả về Subtask<T> (handle). Subtask bắt đầu chạy ngay.
3. scope.join() — block scope owner thread cho đến khi tất cả subtask hoàn thành (hoặc scope shutdown do policy).
4. scope.throwIfFailed() — nếu bất kỳ subtask nào ném exception, rethrow ở đây (wrapped trong exception phù hợp). Nếu không có failure, no-op.
5. subtask.get() — lấy kết quả subtask đã hoàn thành. Chỉ gọi sau join(). Nếu subtask fail, throw exception. Nếu subtask bị cancel, throw CancellationException.
6. scope.close() (tự động qua try-with-resources) — cancel bất kỳ subtask chưa xong, đợi chúng kết thúc, đóng scope.

Subtask state

Subtask.State state = userTask.state();
// RUNNING     -- subtask dang chay
// SUCCESS     -- hoan thanh, userTask.get() tra ket qua
// FAILED      -- nen exception, userTask.exception() tra throwable
// UNAVAILABLE -- chua fork hoac scope chua join

5. Policy ShutdownOnFailure — fail fast

ShutdownOnFailure (inner class của StructuredTaskScope): khi bất kỳ subtask nào fail (ném exception), scope shutdown — cancel interrupt tất cả subtask còn đang chạy. join() trả ngay khi tất cả hoàn thành (cancelled hoặc thành công).

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<String> a = scope.fork(() -> callServiceA());
    Subtask<String> b = scope.fork(() -> callServiceB());
    Subtask<String> c = scope.fork(() -> callServiceC()); // throws after 100ms

    scope.join();          // returns after C fails + A, B cancelled
    scope.throwIfFailed(); // rethrows C's exception

    return a.get() + b.get();  // never reached if C failed
}

Kịch bản: A cần 500ms, B cần 300ms, C fail sau 100ms. Với ShutdownOnFailure:

• 100ms: C fail → scope shutdown → A và B bị interrupt.
• A và B nhận InterruptedException (hoặc IO interrupted) và kết thúc.
• join() trả khi A, B đã kết thúc (nhanh vì interrupt).
• throwIfFailed() rethrow exception của C.
• Tổng thời gian: ~100ms + overhead. Không phải 500ms.

Use case: tất cả subtask phải thành công. Fail nhanh khi có lỗi.

6. Policy ShutdownOnSuccess — first result wins

ShutdownOnSuccess<T>: khi bất kỳ subtask nào thành công, scope shutdown — cancel interrupt subtasks còn lại.

try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> primaryDb.query(id));    // 200ms nếu primary up
    scope.fork(() -> replicaDb.query(id));    // 250ms nếu replica up
    scope.fork(() -> cacheLayer.get(id));     // 50ms nếu cache hit

    scope.join();
    return scope.result();   // result of the first successful subtask
}

Kịch bản: cache hit sau 50ms → scope shutdown → primary và replica query bị cancel. scope.result() trả giá trị từ cache subtask. Total latency: ~50ms thay vì 200ms.

Use case: redundant lookup — nhiều source trả cùng data, lấy kết quả nhanh nhất. Hedged request pattern.

Nếu tất cả subtask fail, scope.result() throw ExecutionException wrapping exception của subtask cuối fail.

// Xu ly khi tat ca fail
try {
    return scope.result();
} catch (ExecutionException ex) {
    throw new ServiceUnavailableException("All sources failed", ex.getCause());
}

7. Custom policy — extend StructuredTaskScope

Custom policy bằng cách override handleComplete(Subtask<? extends T> subtask):

// Policy: majority vote -- ket thuc khi co nhieu hon N/2 ket qua giong nhau
class MajorityVoteScope<T> extends StructuredTaskScope<T> {
    private final List<T> results = new CopyOnWriteArrayList<>();
    private final int threshold;
    private volatile T majorityResult;

    MajorityVoteScope(int numTasks) {
        this.threshold = numTasks / 2 + 1;
    }

    @Override
    protected void handleComplete(Subtask<? extends T> subtask) {
        if (subtask.state() == Subtask.State.SUCCESS) {
            T val = subtask.get();
            results.add(val);
            // Count occurrences
            long count = results.stream().filter(val::equals).count();
            if (count >= threshold) {
                majorityResult = val;
                shutdown();   // cancel remaining subtasks
            }
        }
    }

    public T majorityResult() {
        if (majorityResult == null) throw new IllegalStateException("No majority");
        return majorityResult;
    }
}

Sử dụng:

try (var scope = new MajorityVoteScope<String>(3)) {
    scope.fork(() -> node1.vote(proposal));
    scope.fork(() -> node2.vote(proposal));
    scope.fork(() -> node3.vote(proposal));
    scope.join();
    return scope.majorityResult();
}

handleComplete được gọi mỗi khi 1 subtask hoàn thành (success, failure, hoặc cancelled) — trên subtask thread. Thread-safe access bắt buộc. shutdown() method bảo scope cancel và kết thúc join.

8. Integration với Virtual Thread

Structured Concurrency được thiết kế cùng với Virtual Thread (bài 05) — chúng hoàn toàn bổ sung nhau.

scope.fork() mặc định tạo Virtual Thread cho mỗi subtask. Không phải platform thread, không pool, không size limit:

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    // 10000 HTTP calls -- moi call la 1 virtual thread
    List<Subtask<String>> tasks = new ArrayList<>();
    for (String url : urls) {
        tasks.add(scope.fork(() -> httpGet(url)));
    }
    scope.join();
    scope.throwIfFailed();
    return tasks.stream().map(Subtask::get).toList();
}

10000 subtask = 10000 Virtual Thread. Khi mỗi thread block I/O (HTTP), JVM unmount khỏi carrier — carrier free xử lý thread khác. Effective concurrency cao mà không block OS thread. Carrier pool vẫn nhỏ (= số CPU core).

Cancellation qua interrupt: khi scope shutdown (do policy hoặc exception), JVM gọi Thread.interrupt() trên tất cả virtual thread đang chạy trong scope. Virtual Thread respond nhanh vì:

• Thread.sleep(), socket.read(), Lock.lockInterruptibly() — tất cả check interrupt và throw InterruptedException.
• Không tốn OS thread khi sleep → interrupt đến ngay không cần wake OS thread.

Kết quả: cancel propagate nhanh, resource giải phóng sớm.

9. Pitfall — lỗi thường gặp

Pitfall 1 — fork sau join

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<String> a = scope.fork(() -> taskA());
    scope.join();   // join truoc khi fork B
    Subtask<String> b = scope.fork(() -> taskB());  // IllegalStateException!
}

fork() chỉ hợp lệ trước join(). Sau khi join() được gọi, scope không nhận fork mới.

Pitfall 2 — quên try-with-resources

// BAD -- scope khong bao gio close
StructuredTaskScope.ShutdownOnFailure scope = new StructuredTaskScope.ShutdownOnFailure();
scope.fork(() -> taskA());
scope.join();
// neu exception thrown, scope.close() khong duoc goi
// subtask A co the con running -- resource leak

Luôn dùng try (var scope = ...). AutoCloseable đảm bảo close() được gọi ngay cả khi exception.

Pitfall 3 — subtask.get() trước join

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<String> a = scope.fork(() -> taskA());
    String result = a.get();   // IllegalStateException -- subtask may not be done
    scope.join();
}

subtask.get() chỉ được gọi sau scope.join() hoặc sau khi subtask.state() là SUCCESS.

Pitfall 4 — scope tái sử dụng

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    // Round 1
    scope.fork(() -> taskA());
    scope.join();

    // Round 2 -- KHONG duoc dung lai cung scope
    scope.fork(() -> taskB());  // may throw IllegalStateException
}

Mỗi scope dùng 1 lần. Nếu cần nhiều round, tạo scope mới.

10. So sánh với CompletableFuture

Khi nào dùng cái nào:

• StructuredTaskScope: orchestrate finite set task biết trước, cần cancel propagation, code dễ đọc, Java 25+.
• CompletableFuture: fluent async pipeline (chain, transform, compose), reactive-like data flow, cần tích hợp với library trả CF, Java 8+.
• Cả hai có thể cùng tồn tại: dùng Structured Concurrency cho orchestration scope, CF cho async I/O adapter.

11. Mermaid — lifecycle của StructuredTaskScope

sequenceDiagram
    participant Owner as Scope Owner Thread
    participant S as StructuredTaskScope
    participant A as Subtask A (VT)
    participant B as Subtask B (VT)
    participant C as Subtask C (VT)

    Owner->>S: new ShutdownOnFailure()
    Owner->>S: fork(taskA) -- start A
    S-->>A: spawn virtual thread
    Owner->>S: fork(taskB)
    S-->>B: spawn virtual thread
    Owner->>S: fork(taskC)
    S-->>C: spawn virtual thread
    Owner->>S: join() -- wait
    Note over A: taskA running (200ms)
    Note over B: taskB running (300ms)
    C->>S: FAILED after 100ms
    S->>A: interrupt (ShutdownOnFailure policy)
    S->>B: interrupt
    A-->>S: cancelled
    B-->>S: cancelled
    S-->>Owner: join() returns
    Owner->>S: throwIfFailed() -- rethrows C's exception
    Owner->>S: close() (try-with-resources)

12. Deep Dive

13. Self-check

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
  Subtask<String> t = scope.fork(() -> fetchData());
  scope.join();
  scope.throwIfFailed();
  return t.get();
}