Thread Pool

// ── Full ThreadPoolExecutor construction ─────────────────────────────
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    4,                          // corePoolSize: always keep 4 threads
    16,                         // maximumPoolSize: allow bursts up to 16
    60L,                        // keepAliveTime: idle non-core threads live 60...
    TimeUnit.SECONDS,           // ...seconds before termination
    new ArrayBlockingQueue<>(200),  // work queue: bounded, holds up to 200 tasks
    new NamedThreadFactory("api-worker", false),  // custom thread factory
    new ThreadPoolExecutor.CallerRunsPolicy()     // backpressure on rejection
);

// ── Thread creation algorithm — step by step ─────────────────────────
// Pool starts: 0 threads, corePoolSize=4, maxPoolSize=16, queue capacity=200

// Submit task 1: threads(0) < corePoolSize(4) → CREATE thread-1, run task-1
// Submit task 2: threads(1) < corePoolSize(4) → CREATE thread-2, run task-2
// Submit task 3: threads(2) < corePoolSize(4) → CREATE thread-3, run task-3
// Submit task 4: threads(3) < corePoolSize(4) → CREATE thread-4, run task-4
// Submit task 5: threads(4) >= corePoolSize(4) → QUEUE task-5 (queue size: 1)
// ...
// Submit task 204: threads(4) >= core, queue(200) FULL → threads(4) < max(16) → CREATE thread-5
// Submit task 205: threads(5) >= core, queue FULL, threads < max → CREATE thread-6
// ...
// Submit task 216: threads(16) = max, queue FULL → REJECT (CallerRunsPolicy: run in caller)

// Verify this behavior empirically:
ThreadPoolExecutor observable = new ThreadPoolExecutor(
    2, 6, 30L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(4)
);

CountDownLatch taskLatch = new CountDownLatch(1);

// Submit 10 long-running tasks: 2 core + 4 queue + 4 extra threads:
for (int i = 1; i <= 10; i++) {
    int id = i;
    try {
        observable.execute(() -> {
            System.out.printf("Task %d on %s (pool=%d active=%d queue=%d)%n",
                id, Thread.currentThread().getName(),
                observable.getPoolSize(), observable.getActiveCount(),
                observable.getQueue().size());
            try { taskLatch.await(); } catch (InterruptedException e) {}
        });
    } catch (RejectedExecutionException e) {
        System.out.println("Task " + id + " REJECTED");
    }
}

Thread.sleep(200);
System.out.printf("Final: pool=%d active=%d queued=%d%n",
    observable.getPoolSize(), observable.getActiveCount(), observable.getQueue().size());
// pool=6, active=6, queued=4  (tasks 7-10 rejected by CallerRunsPolicy, so caller ran them)

taskLatch.countDown();
observable.shutdown();

// ── allowCoreThreadTimeOut — let pool shrink to zero ─────────────────
ThreadPoolExecutor shrinkable = new ThreadPoolExecutor(
    4, 4, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>()
);
shrinkable.allowCoreThreadTimeOut(true);   // core threads also time out when idle
shrinkable.submit(() -> System.out.println("Quick task"));
shrinkable.shutdown();
// After 10 seconds of idleness, pool shrinks from 4 threads to 0

// ── prestartAllCoreThreads — eager thread creation ────────────────────
ThreadPoolExecutor eager = new ThreadPoolExecutor(
    4, 4, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()
);
System.out.println("Before prestart: " + eager.getPoolSize());  // 0
eager.prestartAllCoreThreads();
System.out.println("After prestart:  " + eager.getPoolSize());  // 4
// Useful when first-task latency matters and thread creation delay is unacceptable
eager.shutdown();

// ── SynchronousQueue — no buffering, immediate handoff ───────────────
// newCachedThreadPool() uses SynchronousQueue with Integer.MAX_VALUE max threads:
ThreadPoolExecutor syncPool = new ThreadPoolExecutor(
    0, Integer.MAX_VALUE,
    60L, TimeUnit.SECONDS,
    new SynchronousQueue<>()    // no buffering — task must find a thread immediately
);

// Under load: each submitted task either finds an idle thread or creates a new one:
for (int i = 0; i < 5; i++) {
    syncPool.execute(() -> {
        try { Thread.sleep(100); } catch (InterruptedException e) {}
    });
}
Thread.sleep(50);
System.out.println("Cached pool threads: " + syncPool.getPoolSize());  // 5 (one per task)
syncPool.shutdown();

// ── LinkedBlockingQueue bounded — production-safe ─────────────────────
ThreadPoolExecutor linkedBounded = new ThreadPoolExecutor(
    4, 4,                           // fixed size — max is irrelevant with unbounded queue
    0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<>(500)  // ALWAYS specify capacity in production
);
// If 500+ tasks queue up, 501st is rejected — explicit backpressure instead of OOM

// ── LinkedBlockingQueue unbounded — DANGEROUS in production ───────────
// DO NOT use in production without queue depth monitoring and circuit breakers:
ThreadPoolExecutor risky = new ThreadPoolExecutor(
    4, 4, 0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<>()   // Integer.MAX_VALUE capacity — grows forever
);
// Under sustained overload: risky.getQueue().size() → millions → OutOfMemoryError

// ── ArrayBlockingQueue — strict bound, good for backpressure ──────────
ThreadPoolExecutor arrayPool = new ThreadPoolExecutor(
    2, 8,                            // 2 core, burst to 8
    30L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(50),    // max 50 queued tasks (after max threads created)
    Executors.defaultThreadFactory(),
    new ThreadPoolExecutor.CallerRunsPolicy()
);
// Queue fills at 50 tasks → new threads created up to 8 → then CallerRunsPolicy kicks in

// ── PriorityBlockingQueue — priority-ordered execution ────────────────
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    final int priority;
    final String name;
    PrioritizedTask(int priority, String name) {
        this.priority = priority;
        this.name = name;
    }
    @Override public void run() { System.out.println("Running: " + name + " (p=" + priority + ")"); }
    @Override public int compareTo(PrioritizedTask o) {
        return Integer.compare(o.priority, this.priority);  // higher priority = dequeued first
    }
}

ThreadPoolExecutor priorityPool = new ThreadPoolExecutor(
    1, 1, 0L, TimeUnit.MILLISECONDS,
    new PriorityBlockingQueue<>()
);

// Submit low-priority first, then high-priority:
priorityPool.execute(new PrioritizedTask(1, "background-job"));
priorityPool.execute(new PrioritizedTask(5, "user-request"));
priorityPool.execute(new PrioritizedTask(3, "analytics"));
priorityPool.execute(new PrioritizedTask(5, "another-user-request"));

// Execution order: user-request (5), another-user-request (5), analytics (3), background-job (1)
priorityPool.shutdown();
priorityPool.awaitTermination(5, TimeUnit.SECONDS);

// ── Queue depth monitoring — essential for production ─────────────────
ThreadPoolExecutor prod = new ThreadPoolExecutor(
    8, 8, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1000)
);

// Export queue depth as a metric every 10 seconds:
ScheduledExecutorService metrics = Executors.newSingleThreadScheduledExecutor();
metrics.scheduleAtFixedRate(() -> {
    int queueDepth = prod.getQueue().size();
    double utilization = (double) prod.getActiveCount() / prod.getPoolSize();
    System.out.printf("[METRICS] queue=%d utilization=%.1f%%%n",
        queueDepth, utilization * 100);
    if (queueDepth > 800) {
        System.err.println("[ALERT] Queue depth critical: " + queueDepth);
    }
}, 0, 10, TimeUnit.SECONDS);

// ── CPU-bound sizing — cores + 1 ─────────────────────────────────────
int cores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuPool = new ThreadPoolExecutor(
    cores + 1, cores + 1,        // fixed at cores+1 — no burst beyond cores
    0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<>(10_000),
    new NamedThreadFactory("cpu-worker", false),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

// Parallel computation — each task gets a full core:
List<Future<Long>> cpuFutures = new ArrayList<>();
for (int i = 0; i < cores * 2; i++) {
    final int from = i * 500_000;
    cpuFutures.add(cpuPool.submit(() -> {
        long sum = 0;
        for (int j = from; j < from + 500_000; j++) sum += j;
        return sum;
    }));
}
long total = 0;
for (Future<Long> f : cpuFutures) total += f.get();
System.out.println("Sum: " + total);
cpuPool.shutdown();

// ── I/O-bound sizing — cores × (1 + wait/compute) ────────────────────
// Measured: tasks spend ~95% blocking on DB (wait=950ms, compute=50ms)
// wait/compute = 19; cores = 8; optimal = 8 × 20 = 160 threads
// Start with formula, tune empirically:
ExecutorService ioPool = new ThreadPoolExecutor(
    160, 160,
    60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(5000),
    new NamedThreadFactory("db-worker", false),
    new ThreadPoolExecutor.AbortPolicy()
);

// ── Thread pool starvation — the deadlock pattern ─────────────────────
ExecutorService tinyPool = Executors.newFixedThreadPool(2);

// Outer task submits inner task to SAME pool and blocks waiting:
Future<?> outer = tinyPool.submit(() -> {
    System.out.println("Outer task started — submitting inner");
    Future<?> inner = tinyPool.submit(() -> {
        System.out.println("Inner task running");
        return "inner done";
    });
    try {
        String result = (String) inner.get();   // DEADLOCK: both pool threads blocked here
        System.out.println("Outer got: " + result);
    } catch (Exception e) { Thread.currentThread().interrupt(); }
});

// Second outer task fills the pool:
tinyPool.submit(() -> {
    System.out.println("Second outer — also blocked");
    try { outer.get(); } catch (Exception e) {}
});

// inner task is queued but can NEVER run — all 2 threads are blocked waiting for it
// SOLUTION: use a separate pool for inner tasks, or use CompletableFuture.thenApply()

// ── CompletableFuture avoids starvation (no blocking pool threads) ────
CompletableFuture.supplyAsync(() -> "outer result", tinyPool)
    .thenComposeAsync(r ->
        CompletableFuture.supplyAsync(() -> r + " + inner result", tinyPool),
        tinyPool   // thenComposeAsync doesn't block — thread released while inner runs
    )
    .thenAccept(System.out::println);

tinyPool.shutdown();

// ── Virtual threads — Java 21, for I/O-bound workloads ───────────────
// No thread pool needed for I/O-bound work with virtual threads:
try (ExecutorService vThreadExecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<String>> vFutures = new ArrayList<>();
    for (int i = 0; i < 10_000; i++) {
        int id = i;
        vFutures.add(vThreadExecutor.submit(() -> {
            Thread.sleep(100);   // blocks I/O — virtual thread unmounts, carrier free
            return "result-" + id;
        }));
    }
    // All 10,000 tasks run concurrently on a handful of carrier threads:
    long done = vFutures.stream().filter(f -> {
        try { f.get(); return true; } catch (Exception e) { return false; }
    }).count();
    System.out.println("Completed: " + done);  // 10,000
}
// Vs OS threads: 10,000 threads × 1MB stack = 10GB — infeasible
// Virtual threads: 10,000 virtual threads on ~8 carriers — trivial