Concurrency & Multithreading

// ── Threads & Processes in Java ──────────────────────────────

// ─── Creating threads: 3 ways ─────────────────────────────────

// Way 1: Extend Thread
class CounterThread extends Thread {
    private final String name;
    private final int    upto;

    CounterThread(String name, int upto) {
        super(name);
        this.name = name;
        this.upto = upto;
    }

    @Override
    public void run() {
        for (int i = 1; i <= upto; i++) {
            System.out.printf("[%s] count=%d%n", name, i);
            try { Thread.sleep(50); } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
    }
}

// Way 2: Implement Runnable (preferred — keeps class hierarchy free)
public class App {
    public static void main(String[] args) throws InterruptedException {

        // Runnable lambda
        Runnable task = () -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("[Runnable] step " + i + " on " + Thread.currentThread().getName());
            }
        };

        Thread t1 = new Thread(task, "Worker-1");
        Thread t2 = new CounterThread("Counter", 5);

        t1.start();   // schedules on OS thread pool — NOT t1.run()!
        t2.start();

        // main thread continues here — t1, t2 run concurrently

        t1.join();    // main waits for t1 to finish
        t2.join();    // main waits for t2 to finish

        System.out.println("Both threads done.");

        // ── Thread lifecycle ──────────────────────────────────
        // NEW → RUNNABLE → RUNNING → BLOCKED/WAITING/TIMED_WAITING → TERMINATED
        System.out.println("t1 state: " + t1.getState()); // TERMINATED

        // ── Thread info ───────────────────────────────────────
        Thread current = Thread.currentThread();
        System.out.println("Name:     " + current.getName());
        System.out.println("ID:       " + current.getId());
        System.out.println("Priority: " + current.getPriority()); // 1(MIN)–10(MAX)
        System.out.println("Daemon:   " + current.isDaemon());    // daemon threads die with main
    }
}

// Way 3: ExecutorService (modern — preferred for production)
import java.util.concurrent.*;

ExecutorService pool = Executors.newFixedThreadPool(4);

// Submit a Callable (returns a result)
Future<Integer> future = pool.submit(() -> {
    Thread.sleep(100);
    return 42;
});

System.out.println("Result: " + future.get()); // blocks until done

pool.shutdown();               // stop accepting new tasks
pool.awaitTermination(5, TimeUnit.SECONDS);

// ── Race Conditions — the core danger of shared mutable state ─

// ─── Example: A counter incremented by 2 threads ──────────────
public class RaceConditionDemo {

    // ❌ NOT thread-safe
    static int unsafeCounter = 0;

    public static void main(String[] args) throws InterruptedException {

        Runnable increment = () -> {
            for (int i = 0; i < 100_000; i++) {
                unsafeCounter++;
                // unsafeCounter++ is THREE operations:
                //   1. READ  current value from memory
                //   2. ADD   1
                //   3. WRITE result back to memory
                // If two threads interleave between steps 1 and 3,
                // one increment is LOST.
            }
        };

        Thread t1 = new Thread(increment);
        Thread t2 = new Thread(increment);
        t1.start(); t2.start();
        t1.join();  t2.join();

        // Expected: 200,000
        // Actual:   ~140,000–190,000 — varies every run!
        System.out.println("Unsafe: " + unsafeCounter);

        // ── Fix 1: synchronized method ─────────────────────────
        SafeCounter sc1 = new SafeCounterSync();
        Thread t3 = new Thread(() -> { for (int i=0;i<100_000;i++) sc1.increment(); });
        Thread t4 = new Thread(() -> { for (int i=0;i<100_000;i++) sc1.increment(); });
        t3.start(); t4.start(); t3.join(); t4.join();
        System.out.println("Sync:   " + sc1.get()); // always 200,000 ✓

        // ── Fix 2: AtomicInteger ───────────────────────────────
        java.util.concurrent.atomic.AtomicInteger atomic =
            new java.util.concurrent.atomic.AtomicInteger(0);
        Thread t5 = new Thread(() -> { for (int i=0;i<100_000;i++) atomic.incrementAndGet(); });
        Thread t6 = new Thread(() -> { for (int i=0;i<100_000;i++) atomic.incrementAndGet(); });
        t5.start(); t6.start(); t5.join(); t6.join();
        System.out.println("Atomic: " + atomic.get()); // always 200,000 ✓
    }
}

interface SafeCounter { void increment(); int get(); }

// synchronized keyword: only one thread can hold the monitor at a time
class SafeCounterSync implements SafeCounter {
    private int count = 0;

    // Intrinsic lock on 'this' — entire method is a critical section
    public synchronized void increment() { count++; }
    public synchronized int  get()       { return count; }
}

// ── volatile — guarantees visibility, NOT atomicity ───────────
class StopFlag {
    // Without volatile, a thread's loop may NEVER see the update
    // made by another thread (CPU cache / compiler reordering).
    private volatile boolean running = true;

    public void stop()      { running = false; }
    public boolean isRunning() { return running; }
}

// ── Locks & Synchronization in Java ─────────────────────────

import java.util.concurrent.locks.*;
import java.util.concurrent.*;

// ─── 1. Intrinsic Lock (synchronized) ────────────────────────
public class BankAccount {
    private double balance;

    public BankAccount(double balance) { this.balance = balance; }

    // Method-level lock — holds 'this' monitor
    public synchronized void deposit(double amount) {
        balance += amount;
    }

    // Block-level lock — finer granularity
    public void transfer(BankAccount target, double amount) {
        synchronized (this) {
            synchronized (target) {
                // ⚠️ Potential deadlock if two threads call
                // a.transfer(b, 100) and b.transfer(a, 100) simultaneously
                balance        -= amount;
                target.balance += amount;
            }
        }
    }

    public synchronized double getBalance() { return balance; }
}

// ─── 2. ReentrantLock — explicit, more flexible ──────────────
public class SafeCounter {
    private int           count = 0;
    private final ReentrantLock lock = new ReentrantLock();

    public void increment() {
        lock.lock();               // acquire — must ALWAYS unlock in finally
        try {
            count++;
        } finally {
            lock.unlock();         // guaranteed release
        }
    }

    // TryLock — non-blocking attempt
    public boolean tryIncrement() {
        if (lock.tryLock()) {      // returns false immediately if locked
            try { count++; return true; }
            finally { lock.unlock(); }
        }
        return false;              // didn't acquire — no waiting
    }

    // Timed lock — wait at most 500ms
    public boolean timedIncrement() throws InterruptedException {
        if (lock.tryLock(500, TimeUnit.MILLISECONDS)) {
            try { count++; return true; }
            finally { lock.unlock(); }
        }
        return false;
    }

    public int get() { return count; }
}

// ─── 3. ReadWriteLock — multiple readers OR one writer ────────
public class CachedData {
    private Map<String, String> cache = new HashMap<>();
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final Lock readLock  = rwLock.readLock();
    private final Lock writeLock = rwLock.writeLock();

    // Many threads can read simultaneously
    public String get(String key) {
        readLock.lock();
        try {
            return cache.get(key);
        } finally {
            readLock.unlock();
        }
    }

    // Only one writer at a time — blocks all readers too
    public void put(String key, String value) {
        writeLock.lock();
        try {
            cache.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }
}

// ─── 4. Condition Variables — precise thread coordination ─────
public class BoundedBuffer<T> {
    private final Queue<T> buffer  = new LinkedList<>();
    private final int      maxSize;
    private final Lock     lock    = new ReentrantLock();
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public BoundedBuffer(int maxSize) { this.maxSize = maxSize; }

    public void produce(T item) throws InterruptedException {
        lock.lock();
        try {
            while (buffer.size() == maxSize)
                notFull.await();    // release lock, sleep, re-acquire when signalled
            buffer.add(item);
            System.out.println("Produced: " + item + " | size=" + buffer.size());
            notEmpty.signal();      // wake one waiting consumer
        } finally { lock.unlock(); }
    }

    public T consume() throws InterruptedException {
        lock.lock();
        try {
            while (buffer.isEmpty())
                notEmpty.await();
            T item = buffer.poll();
            System.out.println("Consumed: " + item + " | size=" + buffer.size());
            notFull.signal();
            return item;
        } finally { lock.unlock(); }
    }
}

// ─── 5. Deadlock — how it happens and how to avoid it ─────────
// DEADLOCK: Thread A holds lock1 and waits for lock2.
//           Thread B holds lock2 and waits for lock1. Both wait forever.

// Prevention strategy: always acquire locks in a FIXED global order
public void safeTransfer(BankAccount from, BankAccount to, double amt) {
    // Order by identity hash — consistent global ordering
    BankAccount first  = System.identityHashCode(from) < System.identityHashCode(to) ? from : to;
    BankAccount second = first == from ? to : from;
    synchronized (first) {
        synchronized (second) {
            from.withdraw(amt);
            to.deposit(amt);
        }
    }
}

// ── Concurrent Data Structures in Java ──────────────────────

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.*;

public class ConcurrentDataStructuresDemo {
    public static void main(String[] args) throws Exception {

        // ─── 1. ConcurrentHashMap — thread-safe HashMap ──────
        // Segment-based locking (Java 7) → CAS + bin-level locking (Java 8+)
        // Much better throughput than Collections.synchronizedMap(map)
        ConcurrentHashMap<String, Integer> wordCount = new ConcurrentHashMap<>();

        // Thread-safe merge operation (compute is atomic)
        wordCount.merge("hello", 1, Integer::sum);
        wordCount.merge("hello", 1, Integer::sum);
        wordCount.computeIfAbsent("world", k -> 0);
        wordCount.compute("world", (k, v) -> v == null ? 1 : v + 1);

        System.out.println(wordCount); // {hello=2, world=1}

        // ─── 2. BlockingQueue — producer-consumer backbone ───
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(10); // bounded

        // Producer thread
        Thread producer = new Thread(() -> {
            String[] tasks = {"task-1", "task-2", "task-3", "POISON"};
            for (String t : tasks) {
                try {
                    queue.put(t);     // blocks if full
                    System.out.println("Produced: " + t);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        // Consumer thread
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String task = queue.take(); // blocks if empty
                    if ("POISON".equals(task)) break;
                    System.out.println("Consumed: " + task);
                }
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        });

        producer.start(); consumer.start();
        producer.join();  consumer.join();

        // ─── 3. CopyOnWriteArrayList — read-heavy workloads ──
        // Every write creates a NEW copy of the array.
        // Reads are always lock-free (see the unmodified copy).
        // Best when reads >> writes (event listeners, config lists).
        CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();
        listeners.add("listener-1");
        listeners.add("listener-2");

        // Safe to iterate while another thread modifies
        for (String l : listeners) {
            System.out.println("Notifying: " + l);
        }

        // ─── 4. AtomicInteger / AtomicReference / AtomicLong ─
        AtomicInteger counter = new AtomicInteger(0);
        counter.incrementAndGet();               // i++, thread-safe
        counter.addAndGet(5);                    // += 5, thread-safe
        counter.compareAndSet(6, 10);            // CAS: if val==6, set to 10

        // AtomicReference — CAS on object references
        AtomicReference<String> ref = new AtomicReference<>("initial");
        boolean swapped = ref.compareAndSet("initial", "updated");
        System.out.println(swapped + " " + ref.get()); // true updated

        // ─── 5. ConcurrentLinkedQueue — non-blocking queue ───
        // Uses lock-free CAS operations — no blocking, no contention
        ConcurrentLinkedQueue<Integer> clq = new ConcurrentLinkedQueue<>();
        clq.offer(1); clq.offer(2); clq.offer(3);
        System.out.println(clq.poll()); // 1

        // ─── 6. CountDownLatch — wait for N events ────────────
        CountDownLatch latch = new CountDownLatch(3);

        for (int i = 0; i < 3; i++) {
            int id = i;
            new Thread(() -> {
                System.out.println("Service-" + id + " started");
                latch.countDown(); // decrement counter
            }).start();
        }

        latch.await(); // main blocks until count reaches 0
        System.out.println("All services ready — starting app");

        // ─── 7. CyclicBarrier — all threads reach a point ────
        CyclicBarrier barrier = new CyclicBarrier(3, () ->
            System.out.println("All workers reached barrier — proceeding"));

        for (int i = 0; i < 3; i++) {
            int id = i;
            new Thread(() -> {
                System.out.println("Worker-" + id + " at barrier");
                try { barrier.await(); } catch (Exception e) {}
                System.out.println("Worker-" + id + " proceeding");
            }).start();
        }
    }
}

// ── Concurrency Patterns in Java ────────────────────────────

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

// ─── 1. Producer-Consumer ─────────────────────────────────────
// Already shown with BoundedBuffer. Here: using BlockingQueue directly.
public class OrderProcessingSystem {
    private static final BlockingQueue<Order> orders = new LinkedBlockingQueue<>(50);
    private static final AtomicBoolean running = new AtomicBoolean(true);

    // Multiple producers — web handlers push orders
    static class OrderReceiver implements Runnable {
        public void run() {
            while (running.get()) {
                Order order = receiveFromNetwork(); // blocks on I/O
                try { orders.put(order); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        }
        Order receiveFromNetwork() { return new Order(); }
    }

    // Multiple consumers — worker threads process orders
    static class OrderProcessor implements Runnable {
        public void run() {
            while (running.get() || !orders.isEmpty()) {
                try {
                    Order order = orders.poll(100, TimeUnit.MILLISECONDS);
                    if (order != null) process(order);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        }
        void process(Order o) { System.out.println("Processing " + o); }
    }
}

// ─── 2. Thread Pool ───────────────────────────────────────────
// Creating a thread per request is expensive. Pools reuse threads.
public class WebServer {
    // Executors.newFixedThreadPool(n): n threads, unbounded queue
    // Executors.newCachedThreadPool(): grows/shrinks as needed
    // Executors.newScheduledThreadPool(n): for scheduled tasks

    private final ExecutorService pool = new ThreadPoolExecutor(
        4,            // corePoolSize
        16,           // maximumPoolSize
        60L,          // keepAliveTime
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(100),       // work queue, bounded!
        new ThreadPoolExecutor.CallerRunsPolicy() // rejection policy
    );

    public void handleRequest(Runnable request) {
        pool.submit(request);
    }

    public void shutdown() {
        pool.shutdown();
        try { pool.awaitTermination(30, TimeUnit.SECONDS); }
        catch (InterruptedException e) { pool.shutdownNow(); }
    }
}

// ─── 3. Fork/Join — divide and conquer parallelism ────────────
import java.util.concurrent.RecursiveTask;

public class ParallelSum extends RecursiveTask<Long> {
    private final long[] array;
    private final int    start, end;
    private static final int THRESHOLD = 10_000;

    public ParallelSum(long[] array, int start, int end) {
        this.array = array; this.start = start; this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // Small enough — compute directly
            long sum = 0;
            for (int i = start; i < end; i++) sum += array[i];
            return sum;
        }
        int mid = (start + end) / 2;
        ParallelSum left  = new ParallelSum(array, start, mid);
        ParallelSum right = new ParallelSum(array, mid, end);
        left.fork();                  // submit left to pool
        long rightResult = right.compute(); // compute right directly
        long leftResult  = left.join();     // wait for left
        return leftResult + rightResult;
    }
}

ForkJoinPool pool = ForkJoinPool.commonPool();
long[] data = new long[10_000_000];
Arrays.fill(data, 1L);
long total = pool.invoke(new ParallelSum(data, 0, data.length));
System.out.println("Sum: " + total); // 10,000,000

// ─── 4. CompletableFuture — async pipelines ───────────────────
CompletableFuture<String> pipeline =
    CompletableFuture
        .supplyAsync(() -> fetchUser(1))            // async
        .thenApplyAsync(user -> fetchOrders(user))  // chain
        .thenApplyAsync(orders -> formatReport(orders))
        .exceptionally(e -> "Error: " + e.getMessage());

String result = pipeline.get(5, TimeUnit.SECONDS);

// ── Async & Non-Blocking I/O in Java ────────────────────────

import java.util.concurrent.*;
import java.util.concurrent.CompletableFuture;
import java.net.http.*;
import java.net.URI;

// ─── CompletableFuture — Java's async pipeline ────────────────
public class AsyncDemo {

    public static void main(String[] args) throws Exception {

        // ── Basic async task ─────────────────────────────────
        CompletableFuture<String> future = CompletableFuture
            .supplyAsync(() -> {
                simulateDelay(200);
                return "user-data";
            });                          // runs on ForkJoinPool.commonPool()

        // Non-blocking — do other work while waiting
        System.out.println("Main thread doing other work...");

        String result = future.get();    // blocks only when we need the result
        System.out.println("Got: " + result);

        // ── Chaining: thenApply, thenCompose, thenAccept ──────
        CompletableFuture<String> pipeline = CompletableFuture
            .supplyAsync(() -> fetchUserId())           // returns int
            .thenApplyAsync(id -> fetchUser(id))        // int → User (async)
            .thenApplyAsync(user -> fetchOrders(user))  // User → List<Order>
            .thenApplyAsync(orders -> formatSummary(orders)) // → String
            .exceptionally(ex -> "Error: " + ex.getMessage());

        // ── Parallel composition ──────────────────────────────
        CompletableFuture<String> userFuture   = CompletableFuture.supplyAsync(() -> fetchUser(1));
        CompletableFuture<String> configFuture = CompletableFuture.supplyAsync(() -> fetchConfig());

        // Wait for BOTH
        CompletableFuture<String> combined = userFuture
            .thenCombine(configFuture, (user, cfg) -> user + " + " + cfg);

        // Wait for EITHER (race)
        CompletableFuture<String> either = userFuture.applyToEither(configFuture, s -> s);

        // Wait for ALL N futures
        CompletableFuture.allOf(userFuture, configFuture).join();

        // ── Non-blocking HTTP client (Java 11+) ───────────────
        HttpClient client  = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.example.com/users/1"))
            .header("Accept", "application/json")
            .GET()
            .build();

        // Completely non-blocking — no thread blocked waiting
        CompletableFuture<String> httpFuture = client
            .sendAsync(request, HttpResponse.BodyHandlers.ofString())
            .thenApply(HttpResponse::body);

        System.out.println(httpFuture.get(5, TimeUnit.SECONDS));

        // ── Virtual Threads (Java 21 — Project Loom) ──────────
        // Lightweight threads scheduled by JVM, not OS.
        // Can create MILLIONS of them — no memory per OS thread.
        try (ExecutorService vt = Executors.newVirtualThreadPerTaskExecutor()) {
            IntStream.range(0, 1_000_000).forEach(i ->
                vt.submit(() -> {
                    Thread.sleep(100);   // blocks virtual thread, not OS thread
                    return i * 2;
                })
            );
        } // auto-shutdown — waits for all

        // One virtual thread per request — simple synchronous-style code
        // at the throughput of non-blocking I/O!
    }

    static void simulateDelay(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    static int    fetchUserId()    { simulateDelay(100); return 1; }
    static String fetchUser(int i) { simulateDelay(100); return "user-" + i; }
    static String fetchConfig()    { simulateDelay(50);  return "config"; }
    static String fetchOrders(String u) { return "orders-of-" + u; }
    static String formatSummary(String o) { return "summary: " + o; }
}

// ── Advanced — Java Memory Model & happens-before ────────────

import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.*;

// ─── 1. Java Memory Model (JMM) ──────────────────────────────
// Each thread has its own working memory (CPU cache).
// The JMM defines WHEN a write by one thread becomes VISIBLE to another.

// ─── 2. volatile — guarantees visibility + ordering ───────────
class StopFlag {
    // WITHOUT volatile: JIT may cache 'running' in a register.
    // The worker thread may NEVER see running=false.
    private volatile boolean running = true;  // ← volatile fixes this

    public void stop()         { running = false; }
    public boolean isRunning() { return running; }
}

// volatile DOES NOT guarantee atomicity for compound operations!
class BadVolatileCounter {
    private volatile int counter = 0;
    // counter++ reads, increments, writes — NOT atomic even with volatile!
    public void increment() { counter++; } // ← STILL a race condition
}

// Use AtomicInteger for atomic compound operations
class GoodAtomicCounter {
    private final AtomicInteger counter = new AtomicInteger(0);
    public void increment()    { counter.incrementAndGet(); }  // atomic
    public int  get()          { return counter.get(); }
}

// ─── 3. happens-before relationship ──────────────────────────
// A happens-before B means: all writes in A are visible in B.
//
// Rules that establish happens-before:
//   • Thread start: everything before start() is visible inside run()
//   • Thread join: everything in run() is visible after join() returns
//   • synchronized: unlock of monitor X happens-before lock of X
//   • volatile write happens-before subsequent volatile read of same var
//   • Thread.interrupt() happens-before interrupted exception caught

// ─── 4. Double-Checked Locking — common pattern, subtle bug ──
// WITHOUT volatile — broken in Java!
class SingletonBroken {
    private static SingletonBroken instance;
    public static SingletonBroken getInstance() {
        if (instance == null) {
            synchronized (SingletonBroken.class) {
                if (instance == null)
                    instance = new SingletonBroken();
                    // PROBLEM: object creation is 3 steps:
                    // 1. Allocate memory
                    // 2. Call constructor
                    // 3. Assign to instance
                    // JVM can reorder steps 2 and 3!
                    // Another thread sees instance != null but uninitialized object
            }
        }
        return instance;
    }
}

// CORRECT double-checked locking — volatile prevents reordering
class SingletonCorrect {
    private static volatile SingletonCorrect instance; // ← volatile!
    public static SingletonCorrect getInstance() {
        if (instance == null) {
            synchronized (SingletonCorrect.class) {
                if (instance == null)
                    instance = new SingletonCorrect(); // safe with volatile
            }
        }
        return instance;
    }
}

// BEST: Initialisation-on-Demand Holder — lazy, thread-safe, no sync needed
class SingletonBest {
    private SingletonBest() {}
    private static class Holder {
        // Class loading is atomic — JVM guarantees this is initialised once
        static final SingletonBest INSTANCE = new SingletonBest();
    }
    public static SingletonBest getInstance() { return Holder.INSTANCE; }
}

// ─── 5. StampedLock — optimistic reads ────────────────────────
// Java 8+ — most optimistic read mode: no blocking at all
StampedLock sl = new StampedLock();

// Optimistic read: no lock taken — just a stamp
long stamp = sl.tryOptimisticRead();
double balance = this.balance; // read WITHOUT lock
if (!sl.validate(stamp)) {      // if a write happened since our read...
    stamp   = sl.readLock();    // fall back to a real read lock
    try { balance = this.balance; }
    finally { sl.unlockRead(stamp); }
}