09 · Locks & Advanced Synchronization

import java.util.concurrent.locks.ReentrantLock;

class Counter {
    private int count = 0;
    private final ReentrantLock lock = new ReentrantLock();

    void increment() {
        lock.lock();
        try {
            count++;
            incrementInternal();  // Same thread can acquire again
        } finally {
            lock.unlock();
        }
    }

    void incrementInternal() {
        lock.lock();  // Reentrant: same thread can acquire again
        try {
            count++;
        } finally {
            lock.unlock();  // Must unlock twice
        }
    }
}

ReentrantLock lock = new ReentrantLock();

try {
    // Try to acquire, wait up to 2 seconds
    if (lock.tryLock(2, TimeUnit.SECONDS)) {
        try {
            // Critical section
        } finally {
            lock.unlock();
        }
    } else {
        System.out.println("Couldn't acquire lock");
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

// Guarantees FIFO ordering of waiting threads
ReentrantLock fairLock = new ReentrantLock(true);

// vs (default)
ReentrantLock unfairLock = new ReentrantLock(false);

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

class CachedData {
    private String data;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    // Many threads can read concurrently
    String getData() {
        lock.readLock().lock();
        try {
            return data;
        } finally {
            lock.readLock().unlock();
        }
    }

    // Only one thread can write (exclusive)
    void setData(String newData) {
        lock.writeLock().lock();
        try {
            data = newData;
        } finally {
            lock.writeLock().unlock();
        }
    }
}

Multiple readers: | R1 read |---| R2 read |---| R3 read |--- (concurrent)
Exclusive writer: |---- W1 write ----|  (blocks all readers)

import java.util.concurrent.locks.StampedLock;

class OptimizedCache {
    private String data;
    private final StampedLock lock = new StampedLock();

    String getData() {
        // Optimistic read (no lock)
        long stamp = lock.tryOptimisticRead();
        String result = data;

        // Validate
        if (!lock.validate(stamp)) {
            // If validation failed, acquire read lock
            stamp = lock.readLock();
            try {
                result = data;
            } finally {
                lock.unlockRead(stamp);
            }
        }
        return result;
    }
}

import java.util.concurrent.Semaphore;

class ConnectionPool {
    private final Semaphore semaphore;

    ConnectionPool(int maxConnections) {
        this.semaphore = new Semaphore(maxConnections);
    }

    Connection getConnection() throws InterruptedException {
        semaphore.acquire();  // Decrement, block if 0
        return new Connection();
    }

    void releaseConnection(Connection conn) {
        conn.close();
        semaphore.release();   // Increment, wake waiting threads
    }
}

// Usage
ConnectionPool pool = new ConnectionPool(5);

// At most 5 threads can acquire simultaneously
for (int i = 0; i < 100; i++) {
    new Thread(() -> {
        try {
            Connection conn = pool.getConnection();
            useConnection(conn);
            pool.releaseConnection(conn);
        } catch (InterruptedException e) {}
    }).start();
}

Semaphore mutex = new Semaphore(1);  // Acts like a mutex

mutex.acquire();   // Lock
try {
    // Critical section
} finally {
    mutex.release();  // Unlock
}

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

class BoundedBuffer<T> {
    private final T[] buffer;
    private int count = 0;

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();

    public BoundedBuffer(int size) {
        buffer = new Object[size];
    }

    void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (count == buffer.length) {  // Wait if full
                notFull.await();
            }
            buffer[count++] = item;
            notEmpty.signal();  // Signal waiting take()
        } finally {
            lock.unlock();
        }
    }

    T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {  // Wait if empty
                notEmpty.await();
            }
            T item = buffer[--count];
            notFull.signal();  // Signal waiting put()
            return item;
        } finally {
            lock.unlock();
        }
    }
}

class Account {
    private int balance = 100;

    synchronized void transfer(Account other, int amount) {
        this.balance -= amount;
        other.balance += amount;
    }
}

// Deadlock scenario
Account acc1 = new Account();
Account acc2 = new Account();

Thread t1 = new Thread(() -> acc1.transfer(acc2, 50));  // Locks acc1, waits for acc2
Thread t2 = new Thread(() -> acc2.transfer(acc1, 50));  // Locks acc2, waits for acc1
// DEADLOCK!

class Account {
    private static final Object globalOrder = new Object();
    private int balance = 100;
    private int id;

    void transfer(Account other, int amount) {
        // Acquire locks in consistent order
        Account first = this.id < other.id ? this : other;
        Account second = this.id < other.id ? other : this;

        synchronized (first) {
            synchronized (second) {
                first.balance -= amount;
                second.balance += amount;
            }
        }
    }
}

class Account {
    private int balance = 100;
    private ReentrantLock lock = new ReentrantLock();

    void transfer(Account other, int amount) throws InterruptedException {
        if (!this.lock.tryLock(1, TimeUnit.SECONDS)) {
            throw new RuntimeException("Couldn't acquire lock");
        }
        try {
            if (!other.lock.tryLock(1, TimeUnit.SECONDS)) {
                throw new RuntimeException("Couldn't acquire lock on other");
            }
            try {
                this.balance -= amount;
                other.balance += amount;
            } finally {
                other.lock.unlock();
            }
        } finally {
            this.lock.unlock();
        }
    }
}