Concurrency and Parallelism

public class Counter {
    private int count = 0;  // => Shared mutable state (type: int)
                            // => Without synchronization, race condition guaranteed

    // Synchronized method (locks this object)
    public synchronized void increment() {  // => synchronized keyword acquires intrinsic lock on 'this'
        count++;  // => Atomic operation under lock
                  // => Without lock: read-modify-write not atomic (race condition)
    }

    public synchronized int getCount() {  // => Must synchronize reads too for memory visibility
        return count;  // => Returns current value (type: int)
                      // => Without synchronized: might read stale value from thread cache
    }

    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();  // => counter is shared reference (type: Counter)

        // Create 10 threads incrementing 1000 times each
        Thread[] threads = new Thread[10];  // => threads array holds 10 Thread references (type: Thread[])
        for (int i = 0; i < 10; i++) {  // => i from 0 to 9 (type: int)
            threads[i] = new Thread(() -> {  // => Lambda captures counter reference (final/effectively final)
                                             // => Each thread gets separate lambda instance
                for (int j = 0; j < 1000; j++) {  // => Each thread: 1000 increments
                    counter.increment();  // => Acquires lock, increments, releases lock
                                         // => Threads contend for same lock (serialized access)
                }
            });
            threads[i].start();  // => Starts thread execution (asynchronous)
                                // => main thread continues immediately
        }

        // Wait for all threads
        for (Thread thread : threads) {  // => Iterate all 10 threads (type: Thread)
            thread.join();  // => Blocks main thread until this thread completes
                           // => Ensures all increments finish before reading count
        }

        System.out.println("Final count: " + counter.getCount());  // => Output: Final count: 10000 (correct)
                                                                    // => 10 threads × 1000 increments = 10000
                                                                    // => Synchronized guarantees correctness
    }
}

public class BankAccount {
    private double balance = 0.0;  // => Shared mutable state (type: double)
    private final Object balanceLock = new Object();  // => Dedicated lock object (type: Object)
                                                       // => Better than synchronizing on 'this' (prevents external lock access)

    public void deposit(double amount) {  // => amount is deposit value (type: double)
        // Only synchronize critical section
        synchronized (balanceLock) {  // => Acquires balanceLock monitor
                                      // => Other threads block here if lock held
            balance += amount;  // => Critical section: read balance, add amount, write back
                               // => Atomic under lock protection
        }  // => Releases balanceLock (even if exception thrown)
    }

    public void withdraw(double amount) {  // => amount is withdrawal value (type: double)
        synchronized (balanceLock) {  // => Same lock as deposit (mutual exclusion)
            if (balance >= amount) {  // => Check: sufficient funds?
                balance -= amount;  // => Deduct amount if sufficient
            } else {
                throw new IllegalStateException("Insufficient funds");  // => Throws exception (lock still released)
            }
        }
    }

    public double getBalance() {
        synchronized (balanceLock) {  // => Must lock reads for memory visibility
            return balance;  // => Returns current balance (type: double)
                            // => Without lock: might read stale value from cache
        }
    }
}

public class IdGenerator {
    private static int nextId = 1;

    // Class-level lock (locks IdGenerator.class)
    public static synchronized int generateId() {
        return nextId++;
    }

    // Equivalent to:
    public static int generateIdExplicit() {
        synchronized (IdGenerator.class) {
            return nextId++;
        }
    }
}

import java.util.LinkedList;
import java.util.Queue;

public class ProducerConsumer {
    private final Queue<Integer> queue = new LinkedList<>();  // => Shared buffer (type: LinkedList<Integer>)
    private final int capacity = 5;  // => Maximum queue size (type: int)
    private final Object lock = new Object();  // => Shared lock for coordination (type: Object)

    public void produce(int value) throws InterruptedException {  // => value is item to produce (type: int)
        synchronized (lock) {  // => Acquire lock for exclusive access
            // Wait while queue is full
            while (queue.size() == capacity) {  // => MUST use while (not if) - recheck after wakeup
                System.out.println("Queue full, producer waiting...");
                lock.wait();  // => Releases lock atomically and waits
                             // => Wakes up when notified by consumer
            }

            queue.add(value);  // => Add item to queue (queue now has space)
            System.out.println("Produced: " + value + ", Queue size: " + queue.size());

            // Notify consumers
            lock.notifyAll();  // => Wake all waiting consumers
        }  // => Release lock
    }

    public int consume() throws InterruptedException {
        synchronized (lock) {  // => Acquire same lock as producer
            // Wait while queue is empty
            while (queue.isEmpty()) {  // => MUST use while (not if) for safety
                System.out.println("Queue empty, consumer waiting...");
                lock.wait();  // => Release lock and wait
            }

            int value = queue.poll();  // => Remove item from queue (type: int)
            System.out.println("Consumed: " + value + ", Queue size: " + queue.size());

            // Notify producers
            lock.notifyAll();  // => Wake all waiting producers

            return value;  // => Return consumed value (type: int)
        }
    }

    public static void main(String[] args) {
        ProducerConsumer pc = new ProducerConsumer();  // => Shared ProducerConsumer instance (type: ProducerConsumer)

        // Producer thread
        Thread producer = new Thread(() -> {  // => Lambda creates producer task
            for (int i = 1; i <= 10; i++) {  // => Produce 10 items (i from 1 to 10, type: int)
                try {
                    pc.produce(i);  // => Produce item i
                    Thread.sleep(100);  // => Slow producer: 100ms between items
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();  // => Restore interrupt status
                }
            }
        });

        // Consumer thread
        Thread consumer = new Thread(() -> {  // => Lambda creates consumer task
            for (int i = 1; i <= 10; i++) {  // => Consume 10 items
                try {
                    pc.consume();  // => Consume one item
                    Thread.sleep(200);  // => Slower consumer: 200ms between items
                                       // => Consumer slower than producer → queue fills up
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        producer.start();  // => Start producer thread (asynchronous)
        consumer.start();  // => Start consumer thread (asynchronous)
    }
}

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class BankAccountWithLock {
    private double balance = 0.0;
    private final Lock lock = new ReentrantLock();

    public void deposit(double amount) {
        lock.lock();
        try {
            balance += amount;
        } finally {
            lock.unlock(); // ALWAYS in finally
        }
    }

    public void withdraw(double amount) {
        lock.lock();
        try {
            if (balance >= amount) {
                balance -= amount;
            } else {
                throw new IllegalStateException("Insufficient funds");
            }
        } finally {
            lock.unlock();
        }
    }

    public double getBalance() {
        lock.lock();
        try {
            return balance;
        } finally {
            lock.unlock();
        }
    }
}

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TryLockExample {
    private final Lock lock = new ReentrantLock();

    public void performOperation() {
        boolean acquired = false;
        try {
            // Try to acquire lock for 5 seconds
            acquired = lock.tryLock(5, TimeUnit.SECONDS);

            if (acquired) {
                System.out.println("Lock acquired, performing operation");
                // Critical section
            } else {
                System.out.println("Could not acquire lock, aborting");
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            if (acquired) {
                lock.unlock();
            }
        }
    }
}

public class InterruptibleLockExample {
    private final Lock lock = new ReentrantLock();

    public void performTask() throws InterruptedException {
        lock.lockInterruptibly(); // Can be interrupted while waiting
        try {
            // Critical section
            System.out.println("Performing task");
        } finally {
            lock.unlock();
        }
    }
}

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.HashMap;
import java.util.Map;

public class CachedData {
    private final Map<String, String> cache = new HashMap<>();
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();

    public String get(String key) {
        rwLock.readLock().lock(); // Multiple readers allowed
        try {
            return cache.get(key);
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public void put(String key, String value) {
        rwLock.writeLock().lock(); // Exclusive writer
        try {
            cache.put(key, value);
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    public void clear() {
        rwLock.writeLock().lock();
        try {
            cache.clear();
        } finally {
            rwLock.writeLock().unlock();
        }
    }
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.LinkedList;
import java.util.Queue;

public class BoundedQueue<T> {
    private final Queue<T> queue = new LinkedList<>();
    private final int capacity;
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public BoundedQueue(int capacity) {
        this.capacity = capacity;
    }

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                notFull.await(); // Wait on notFull condition
            }

            queue.add(item);
            notEmpty.signal(); // Signal notEmpty condition

        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await(); // Wait on notEmpty condition
            }

            T item = queue.poll();
            notFull.signal(); // Signal notFull condition

            return item;

        } finally {
            lock.unlock();
        }
    }
}

import java.util.concurrent.*;
import java.util.List;
import java.util.ArrayList;

public class ExecutorExample {
    public static void main(String[] args) throws InterruptedException {
        // Fixed thread pool
        ExecutorService executor = Executors.newFixedThreadPool(4);

        // Submit tasks
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            final int taskId = i;
            Future<Integer> future = executor.submit(() -> {
                System.out.println("Task " + taskId + " on " +
                    Thread.currentThread().getName());
                Thread.sleep(1000);
                return taskId * 2;
            });
            futures.add(future);
        }

        // Get results
        for (Future<Integer> future : futures) {
            try {
                Integer result = future.get(); // Blocking call
                System.out.println("Result: " + result);
            } catch (ExecutionException e) {
                System.err.println("Task failed: " + e.getCause());
            }
        }

        // Shutdown executor
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

import java.util.concurrent.*;

public class CustomThreadPool {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,                              // Core pool size
            4,                              // Maximum pool size
            60,                             // Keep-alive time
            TimeUnit.SECONDS,               // Keep-alive unit
            new LinkedBlockingQueue<>(100), // Work queue
            new ThreadFactory() {           // Thread factory
                private int count = 1;
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "Worker-" + count++);
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // Rejection policy
        );

        // Submit tasks
        for (int i = 1; i <= 20; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " executing");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        executor.shutdown();
    }
}

import java.util.concurrent.*;

public class ScheduledTaskExample {
    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

        // One-time delayed execution
        scheduler.schedule(() -> {
            System.out.println("Delayed task executed");
        }, 3, TimeUnit.SECONDS);

        // Fixed-rate periodic execution (every 2 seconds)
        ScheduledFuture<?> fixedRate = scheduler.scheduleAtFixedRate(() -> {
            System.out.println("Fixed rate task at " + System.currentTimeMillis());
        }, 0, 2, TimeUnit.SECONDS);

        // Fixed-delay periodic execution (2 seconds after previous completion)
        ScheduledFuture<?> fixedDelay = scheduler.scheduleWithFixedDelay(() -> {
            System.out.println("Fixed delay task starting");
            try {
                Thread.sleep(1000); // Task takes 1 second
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, 0, 2, TimeUnit.SECONDS);

        // Run for 10 seconds then cancel
        Thread.sleep(10000);
        fixedRate.cancel(false);
        fixedDelay.cancel(false);

        scheduler.shutdown();
    }
}

public class PoolSizingExample {
    public static void main(String[] args) {
        int cores = Runtime.getRuntime().availableProcessors();

        // CPU-bound tasks
        int cpuBoundPoolSize = cores + 1;
        ExecutorService cpuExecutor = Executors.newFixedThreadPool(cpuBoundPoolSize);

        // I/O-bound tasks (80% wait time, 20% compute time)
        // Wait/Compute ratio = 0.8 / 0.2 = 4
        int ioBoundPoolSize = cores * (1 + 4);
        ExecutorService ioExecutor = Executors.newFixedThreadPool(ioBoundPoolSize);

        System.out.println("CPU cores: " + cores);
        System.out.println("CPU-bound pool size: " + cpuBoundPoolSize);
        System.out.println("I/O-bound pool size: " + ioBoundPoolSize);

        cpuExecutor.shutdown();
        ioExecutor.shutdown();
    }
}

public class ExecutorShutdownExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(4);

        // Submit tasks
        for (int i = 0; i < 10; i++) {
            executor.submit(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // Graceful shutdown
        executor.shutdown(); // No new tasks accepted, existing tasks continue

        try {
            // Wait for tasks to complete
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                // Timeout reached, force shutdown
                List<Runnable> droppedTasks = executor.shutdownNow();
                System.out.println("Dropped tasks: " + droppedTasks.size());

                // Wait a bit for threads to respond to interruption
                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("Executor did not terminate");
                }
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;

public class ConcurrentMapExample {
    private final Map<String, Integer> userScores = new ConcurrentHashMap<>();

    public void updateScore(String userId, int points) {
        // Atomic compute if absent
        userScores.putIfAbsent(userId, 0);

        // Atomic update
        userScores.compute(userId, (key, currentScore) ->
            currentScore == null ? points : currentScore + points);
    }

    public void incrementScore(String userId, int points) {
        // Atomic merge
        userScores.merge(userId, points, Integer::sum);
    }

    public Integer getScore(String userId) {
        return userScores.get(userId);
    }

    public static void main(String[] args) throws InterruptedException {
        ConcurrentMapExample example = new ConcurrentMapExample();

        // Multiple threads updating concurrently
        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    example.incrementScore("user-1", 1);
                }
            });
            threads[i].start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        System.out.println("Final score: " + example.getScore("user-1")); // 1000
    }
}

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.List;

public class EventListenerRegistry {
    private final List<EventListener> listeners = new CopyOnWriteArrayList<>();

    public void addListener(EventListener listener) {
        listeners.add(listener); // Creates copy
    }

    public void removeListener(EventListener listener) {
        listeners.remove(listener); // Creates copy
    }

    public void fireEvent(Event event) {
        // Iteration uses snapshot, no locking needed
        for (EventListener listener : listeners) {
            listener.onEvent(event);
        }
    }
}

interface EventListener {
    void onEvent(Event event);
}

class Event {
    private final String message;
    public Event(String message) { this.message = message; }
    public String getMessage() { return message; }
}

import java.util.concurrent.*;

public class ProducerConsumerQueue {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

        // Producer
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 20; i++) {
                    queue.put(i); // Blocks if queue is full
                    System.out.println("Produced: " + i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // Consumer
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    Integer item = queue.take(); // Blocks if queue is empty
                    System.out.println("Consumed: " + item);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

import java.util.concurrent.*;

public class TaskQueue {
    private final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>(100);

    public void submitTask(Runnable task) throws InterruptedException {
        taskQueue.put(task);
    }

    public Runnable takeTask() throws InterruptedException {
        return taskQueue.take();
    }

    public static void main(String[] args) throws InterruptedException {
        TaskQueue queue = new TaskQueue();

        // Worker threads
        for (int i = 0; i < 4; i++) {
            new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        Runnable task = queue.takeTask();
                        task.run();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }).start();
        }

        // Submit tasks
        for (int i = 1; i <= 20; i++) {
            final int taskId = i;
            queue.submitTask(() -> {
                System.out.println("Executing task " + taskId);
            });
        }
    }
}

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.Queue;

public class NonBlockingQueue {
    private final Queue<String> queue = new ConcurrentLinkedQueue<>();

    public void addMessage(String message) {
        queue.offer(message); // Non-blocking
    }

    public String pollMessage() {
        return queue.poll(); // Non-blocking, returns null if empty
    }

    public static void main(String[] args) throws InterruptedException {
        NonBlockingQueue nbQueue = new NonBlockingQueue();

        // Multiple producers
        for (int i = 0; i < 5; i++) {
            final int producerId = i;
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    nbQueue.addMessage("Producer-" + producerId + "-Msg-" + j);
                }
            }).start();
        }

        // Multiple consumers
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                while (true) {
                    String msg = nbQueue.pollMessage();
                    if (msg != null) {
                        System.out.println("Consumed: " + msg);
                    }
                }
            }).start();
        }

        Thread.sleep(2000);
    }
}

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class CollectionComparison {
    public static void main(String[] args) {
        // Synchronized wrapper (coarse-grained locking)
        Map<String, Integer> syncMap = Collections.synchronizedMap(new HashMap<>());

        // Concurrent collection (fine-grained locking)
        Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();

        // syncMap: Each operation locks entire map
        synchronized (syncMap) {
            syncMap.put("key1", 1);
            syncMap.put("key2", 2);
        }

        // concurrentMap: Operations lock segments independently
        concurrentMap.put("key1", 1);
        concurrentMap.put("key2", 2);
    }
}

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounter {
    private final AtomicInteger count = new AtomicInteger(0);

    public void increment() {
        count.incrementAndGet(); // Atomic increment
    }

    public void decrement() {
        count.decrementAndGet(); // Atomic decrement
    }

    public void add(int value) {
        count.addAndGet(value); // Atomic add
    }

    public int get() {
        return count.get();
    }

    public static void main(String[] args) throws InterruptedException {
        AtomicCounter counter = new AtomicCounter();

        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    counter.increment();
                }
            });
            threads[i].start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        System.out.println("Final count: " + counter.get()); // 10000
    }
}

import java.util.concurrent.atomic.AtomicReference;

public class AtomicReferenceExample {
    private final AtomicReference<UserProfile> currentProfile = new AtomicReference<>();

    public void updateProfile(UserProfile newProfile) {
        currentProfile.set(newProfile);
    }

    public boolean updateIfChanged(UserProfile expected, UserProfile newProfile) {
        // Atomic compare-and-swap
        return currentProfile.compareAndSet(expected, newProfile);
    }

    public UserProfile getProfile() {
        return currentProfile.get();
    }

    public static void main(String[] args) {
        AtomicReferenceExample example = new AtomicReferenceExample();

        UserProfile oldProfile = new UserProfile("Alice", 25);
        UserProfile newProfile = new UserProfile("Alice", 26);

        example.updateProfile(oldProfile);

        // Only updates if current value is oldProfile
        boolean updated = example.updateIfChanged(oldProfile, newProfile);
        System.out.println("Updated: " + updated); // true

        // Second attempt fails (current is now newProfile)
        updated = example.updateIfChanged(oldProfile, new UserProfile("Alice", 27));
        System.out.println("Updated: " + updated); // false
    }
}

class UserProfile {
    private final String name;
    private final int age;

    public UserProfile(String name, int age) {
        this.name = name;
        this.age = age;
    }
}