Virtual Threads

// ── Creating virtual threads ──────────────────────────────────────────
// Method 1: Thread.ofVirtual().start(Runnable)
Thread vt1 = Thread.ofVirtual().name("virtual-1").start(() -> {
    System.out.println("Hello from: " + Thread.currentThread());
    System.out.println("Is virtual: " + Thread.currentThread().isVirtual());
});
vt1.join();

// Method 2: Thread.ofVirtual().unstarted(Runnable)
Thread vt2 = Thread.ofVirtual().name("virtual-2").unstarted(() -> {
    System.out.println("Running on carrier: " + Thread.currentThread());
});
vt2.start();
vt2.join();

// Method 3: Executors.newVirtualThreadPerTaskExecutor()
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    // Submits each task as a NEW virtual thread — no pooling:
    for (int i = 0; i < 10; i++) {
        final int id = i;
        executor.submit(() -> {
            System.out.println("Task " + id + " on " + Thread.currentThread());
            Thread.sleep(100);   // blocks virtual thread, not carrier thread
            return "Result " + id;
        });
    }
}   // executor.close() waits for all tasks to complete — AutoCloseable

// ── Blocking a virtual thread — carrier is NOT blocked ────────────────
try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) {
    // Submit 100,000 tasks that each sleep for 1 second:
    long start = System.nanoTime();
    List<Future<?>> futures = new ArrayList<>();
    for (int i = 0; i < 100_000; i++) {
        futures.add(exec.submit(() -> {
            Thread.sleep(1000);   // blocks virtual thread, frees carrier for others
            return null;
        }));
    }
    for (Future<?> f : futures) f.get();
    long elapsed = (System.nanoTime() - start) / 1_000_000;
    System.out.printf("100,000 tasks x 1s sleep completed in %dms%n", elapsed);
    // With platform threads: ~12,500 seconds (100000/8 threads * 1s)
    // With virtual threads: ~1,000ms (all 100000 sleep concurrently)
}

// ── isVirtual, carrier detection ─────────────────────────────────────
Thread platformThread = Thread.ofPlatform().start(() -> {
    System.out.println("Platform: " + Thread.currentThread().isVirtual()); // false
});
Thread virtualThread = Thread.ofVirtual().start(() -> {
    System.out.println("Virtual:  " + Thread.currentThread().isVirtual()); // true
    // Thread.currentThread() returns the virtual thread, NOT the carrier
});
platformThread.join(); virtualThread.join();

// ── Pinning: synchronized blocks prevent unmounting ───────────────────
Object monitor = new Object();

// This virtual thread is PINNED while inside synchronized:
Thread pinned = Thread.ofVirtual().start(() -> {
    synchronized (monitor) {
        System.out.println("Inside synchronized — virtual thread is PINNED");
        try {
            Thread.sleep(1000);  // carrier thread BLOCKED — pinning defeats scalability
        } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        System.out.println("Leaving synchronized — carrier free again");
    }
});
pinned.join();

// ── Fix: replace synchronized with ReentrantLock ─────────────────────
ReentrantLock lock = new ReentrantLock();

Thread unpinned = Thread.ofVirtual().start(() -> {
    lock.lock();
    try {
        System.out.println("Inside ReentrantLock — virtual thread CAN be unmounted");
        Thread.sleep(1000);   // virtual thread parks, carrier is FREE for other work
        System.out.println("Done — no pinning occurred");
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        lock.unlock();
    }
});
unpinned.join();

// ── Detecting pinning: JVM flag ───────────────────────────────────────
// Run with: -Djdk.tracePinnedThreads=full
// Output when pinning occurs:
// Thread[#21,ForkJoinPool-1-worker-1,5,CarrierThreads]
//     java.base/java.lang.VirtualThread$PinnedScope.run(VirtualThread.java:...)
//     ...
//     PinningExample.lambda$0(PinningExample.java:12)  ← synchronized block

// ── ThreadLocal vs virtual threads ────────────────────────────────────
ThreadLocal<String> localValue = new ThreadLocal<>();

// ThreadLocal IS per-virtual-thread (each virtual thread = separate Thread object):
Thread v1 = Thread.ofVirtual().start(() -> {
    localValue.set("Value from virtual thread 1");
    Thread.sleep(100);
    System.out.println(Thread.currentThread() + ": " + localValue.get());
    // "Value from virtual thread 1" — isolated correctly
});
Thread v2 = Thread.ofVirtual().start(() -> {
    localValue.set("Value from virtual thread 2");
    Thread.sleep(100);
    System.out.println(Thread.currentThread() + ": " + localValue.get());
    // "Value from virtual thread 2" — isolated from v1
});
v1.join(); v2.join();

// ── Scoped values: virtual-thread-friendly immutable inheritance ───────
// Java 21 (preview) / Java 23+ (standard):
ScopedValue<String> REQUEST_ID = ScopedValue.newInstance();

// Bind a value for the scope of a task and all virtual threads it spawns:
ScopedValue.where(REQUEST_ID, "req-12345").run(() -> {
    System.out.println("Main task: " + REQUEST_ID.get());   // req-12345
    Thread.ofVirtual().start(() -> {
        System.out.println("Child virtual thread: " + REQUEST_ID.get()); // req-12345 — inherited
    }).join();
});

// Outside the scope:
System.out.println("Outside scope: " + REQUEST_ID.isBound());   // false

// ── Structured Concurrency with StructuredTaskScope ───────────────────
// Java 21+
public record UserData(String name, String email) {}
public record OrderData(List<String> orders) {}

public UserProfile loadUserProfile(long userId) throws InterruptedException, ExecutionException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        // Fork two concurrent virtual-thread tasks:
        StructuredTaskScope.Subtask<UserData>  userTask  = scope.fork(() -> fetchUserData(userId));
        StructuredTaskScope.Subtask<OrderData> orderTask = scope.fork(() -> fetchOrderData(userId));

        scope.join();           // wait for both to complete
        scope.throwIfFailed();  // propagate any exception — cancels all if one fails

        // Both succeeded — combine results:
        return new UserProfile(userTask.get(), orderTask.get());
    }
    // scope.close() automatically cancels/joins any remaining tasks
}

// ── ShutdownOnSuccess: race multiple computations, take the fastest ────
public String fetchWithFallback(String primaryUrl, String backupUrl)
        throws InterruptedException, ExecutionException {

    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
        scope.fork(() -> httpGet(primaryUrl));   // try primary
        scope.fork(() -> httpGet(backupUrl));    // try backup concurrently

        scope.join();   // wait for first success — cancels the other
        return scope.result();  // result of the winning task
    }
}

// ── Migration: thread pool → virtual threads ──────────────────────────
// BEFORE: bounded thread pool to limit concurrency for I/O-bound work
ExecutorService oldPool = Executors.newFixedThreadPool(200);  // 200 platform threads

Future<String> future1 = oldPool.submit(() -> fetchFromDatabase(1));
Future<String> future2 = oldPool.submit(() -> fetchFromDatabase(2));
// ... only 200 tasks can run concurrently — rest queue

// AFTER: virtual thread executor — no arbitrary limit needed
try (ExecutorService vtExec = Executors.newVirtualThreadPerTaskExecutor()) {
    Future<String> vt1 = vtExec.submit(() -> fetchFromDatabase(1));
    Future<String> vt2 = vtExec.submit(() -> fetchFromDatabase(2));
    // ... 100,000 tasks can run concurrently if needed
    // I/O blocking unmounts virtual threads — carrier threads never starve
}

// ── When NOT to use virtual threads ───────────────────────────────────
// CPU-bound: virtual threads don't help — still limited by carrier count
// Sequential map/reduce on 10M numbers — use parallel streams, not virtual threads:
long sum = LongStream.range(0, 10_000_000).parallel().sum();   // ✓ correct tool

// WRONG: virtual threads for CPU work — no speedup, just overhead:
try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<Long>> futures = new ArrayList<>();
    for (int i = 0; i < 10_000_000; i++) {
        final int n = i;
        futures.add(exec.submit(() -> (long)(n * n)));  // 10M virtual threads — wasteful
    }
    long s = 0;
    for (Future<Long> f : futures) s += f.get();
}   // far slower than parallel streams for CPU work

// ── Correct use case: concurrent I/O with simple blocking code ────────
public List<String> fetchAll(List<String> urls) throws InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        List<StructuredTaskScope.Subtask<String>> tasks = urls.stream()
            .map(url -> scope.fork(() -> httpGet(url)))  // each URL = one virtual thread
            .toList();
        scope.join().throwIfFailed();
        return tasks.stream().map(StructuredTaskScope.Subtask::get).toList();
    }
}
// All URLs are fetched concurrently — each virtual thread parks during I/O
// Code is as simple as a sequential for loop but executes in parallel