Semaphore

// ── Construction and basic acquire/release ────────────────────────────
Semaphore sem = new Semaphore(3);   // 3 permits: at most 3 concurrent holders

// acquire — blocks if no permits available, throws InterruptedException:
sem.acquire();
try {
    doWork();
} finally {
    sem.release();   // always release in finally — any thread can release
}

// tryAcquire — non-blocking:
if (sem.tryAcquire()) {
    try {
        doWork();
    } finally {
        sem.release();
    }
} else {
    System.out.println("No permit available — skipping or queuing");
}

// tryAcquire with timeout:
try {
    if (sem.tryAcquire(500, TimeUnit.MILLISECONDS)) {
        try {
            doWork();
        } finally {
            sem.release();
        }
    } else {
        System.out.println("Timed out — no permit within 500ms");
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

// ── Bulk acquire/release ──────────────────────────────────────────────
Semaphore bulk = new Semaphore(10);
bulk.acquire(3);    // acquires 3 permits atomically — blocks until all 3 available
try {
    doWorkRequiring3Resources();
} finally {
    bulk.release(3);   // releases all 3 at once
}

// ── availablePermits and drainPermits ────────────────────────────────
Semaphore pool = new Semaphore(5);
System.out.println("Available: " + pool.availablePermits());  // 5
pool.acquire(2);
System.out.println("Available: " + pool.availablePermits());  // 3

int drained = pool.drainPermits();   // atomically takes all 3 remaining permits
System.out.println("Drained: "  + drained);                   // 3
System.out.println("Available: "+ pool.availablePermits());   // 0

pool.release(2);  // release the 2 we acquired — does NOT restore drained ones
System.out.println("Available: "+ pool.availablePermits());   // 2

// ── Non-ownership: different threads acquire and release ───────────────
Semaphore signal = new Semaphore(0);   // starts at 0 — producer must signal first

Thread consumer = new Thread(() -> {
    try {
        System.out.println("Consumer: waiting for item");
        signal.acquire();              // blocks at 0 — waits for producer
        System.out.println("Consumer: item received, processing");
    } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
});

Thread producer = new Thread(() -> {
    System.out.println("Producer: producing item");
    produceItem();
    signal.release();                  // different thread releases — no ownership constraint
    System.out.println("Producer: signalled consumer");
});

consumer.start();
Thread.sleep(100);
producer.start();
consumer.join(); producer.join();
// Consumer: waiting for item
// Producer: producing item
// Producer: signalled consumer
// Consumer: item received, processing

// ── Connection pool using Semaphore ──────────────────────────────────
public class ConnectionPool {
    private final Semaphore       available;
    private final Queue<Connection> pool      = new ConcurrentLinkedQueue<>();
    private final int             maxConns;

    public ConnectionPool(int maxConnections) {
        this.maxConns  = maxConnections;
        this.available = new Semaphore(maxConnections, true);  // fair: no starvation
        // Pre-populate the pool:
        for (int i = 0; i < maxConnections; i++) {
            pool.offer(createConnection());
        }
    }

    public Connection acquire() throws InterruptedException {
        available.acquire();           // blocks if all maxConns permits are out
        Connection conn = pool.poll();
        if (conn == null) {
            // Should not happen if pool and semaphore are in sync, but be defensive:
            available.release();
            throw new IllegalStateException("Pool inconsistency — no connection available");
        }
        return conn;
    }

    public Connection acquire(long timeout, TimeUnit unit) throws InterruptedException {
        if (!available.tryAcquire(timeout, unit)) return null;  // timed out
        Connection conn = pool.poll();
        if (conn == null) { available.release(); return null; }
        return conn;
    }

    public void release(Connection conn) {
        if (conn == null) { available.release(); return; }  // broken conn: release permit only
        pool.offer(conn);    // return connection to pool
        available.release(); // release permit — next acquirer can proceed
    }

    public int available() { return available.availablePermits(); }
    private Connection createConnection() { return new Connection(); }
}

// Usage:
ConnectionPool pool = new ConnectionPool(10);
try (Connection conn = pool.acquireAutoCloseable()) {
    conn.executeQuery("SELECT 1");
}   // Connection auto-released — AutoCloseable pattern below

// AutoCloseable wrapper for try-with-resources:
public AutoCloseable acquireAutoCloseable() throws InterruptedException {
    Connection conn = acquire();
    return () -> release(conn);
}

// ── Rate limiter: token bucket via Semaphore ──────────────────────────
public class TokenBucketRateLimiter {
    private final Semaphore           tokens;
    private final int                 capacity;
    private final ScheduledExecutorService refiller = Executors.newSingleThreadScheduledExecutor();

    public TokenBucketRateLimiter(int requestsPerSecond) {
        this.capacity = requestsPerSecond;
        this.tokens   = new Semaphore(requestsPerSecond);
        // Refill at the specified rate — every 1s, release up to capacity tokens:
        refiller.scheduleAtFixedRate(this::refill, 1, 1, TimeUnit.SECONDS);
    }

    private void refill() {
        // drainPermits first to avoid over-issuing beyond capacity:
        int current  = tokens.availablePermits();
        int toRelease = capacity - current;
        if (toRelease > 0) tokens.release(toRelease);
    }

    // Block until a token is available:
    public void acquire() throws InterruptedException {
        tokens.acquire();
    }

    // Return false if no token available now:
    public boolean tryAcquire() {
        return tokens.tryAcquire();
    }

    // Return false if no token available within timeout:
    public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
        return tokens.tryAcquire(timeout, unit);
    }

    public void shutdown() { refiller.shutdown(); }
}

// Usage:
TokenBucketRateLimiter limiter = new TokenBucketRateLimiter(100);  // 100 req/sec
for (int i = 0; i < 200; i++) {
    final int req = i;
    new Thread(() -> {
        try {
            limiter.acquire();   // first 100 proceed immediately, next 100 wait for refill
            System.out.println("Request " + req + " processed at " + System.currentTimeMillis());
        } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }).start();
}

// ── Binary semaphore: mutual exclusion without ownership ──────────────
Semaphore mutex = new Semaphore(1);   // 1 permit = binary semaphore

mutex.acquire();
try {
    criticalSection();
} finally {
    mutex.release();   // can be released by ANY thread — not just the acquirer
}

// ── Binary semaphore: signaling between threads ────────────────────────
Semaphore handoff = new Semaphore(0);   // starts at 0: nobody can acquire yet

Thread receiver = new Thread(() -> {
    System.out.println("Receiver: waiting for handoff");
    try { handoff.acquire(); }              // blocks at 0
    catch (InterruptedException e) { Thread.currentThread().interrupt(); return; }
    System.out.println("Receiver: received handoff — proceeding");
});

Thread sender = new Thread(() -> {
    System.out.println("Sender: preparing handoff");
    prepareData();
    handoff.release();   // sender releases a permit it never acquired — non-ownership at work
    System.out.println("Sender: handoff complete");
});

receiver.start();
Thread.sleep(100);
sender.start();
receiver.join(); sender.join();

// ── Non-reentrancy: binary semaphore deadlocks on double acquire ───────
Semaphore nonReentrant = new Semaphore(1);
nonReentrant.acquire();   // succeeds: 1 permit taken, 0 remaining
try {
    // nonReentrant.acquire();   // DEADLOCK: 0 permits — this thread waits forever for itself
    System.out.println("Would deadlock if acquire called again");
} finally {
    nonReentrant.release();
}

// ReentrantLock does NOT deadlock:
ReentrantLock reentrant = new ReentrantLock();
reentrant.lock();
try {
    reentrant.lock();   // reentrant: hold count goes to 2, NO deadlock
    try {
        doNestedWork();
    } finally { reentrant.unlock(); }  // hold count back to 1
} finally { reentrant.unlock(); }      // hold count to 0, released

// ── Fair vs unfair Semaphore: contention under burst ──────────────────
int PERMITS = 2;
Semaphore unfairSem = new Semaphore(PERMITS, false);  // unfair: barge allowed
Semaphore fairSem   = new Semaphore(PERMITS, true);   // fair:   FIFO

// Under sustained burst, unfair may starve some threads indefinitely.
// Fair guarantees every waiting thread eventually gets a permit.

// ── Comparing Semaphore vs other primitives ───────────────────────────
//
// Pattern                     | Tool
// ----------------------------|-----------------------------------------------
// Limit to N concurrent       | Semaphore(N)
// Mutual exclusion (N=1)      | ReentrantLock or synchronized
// Wait for N events, one-shot | CountDownLatch(N)
// N threads meet at barrier   | CyclicBarrier(N) or Phaser
// Producer-consumer, bounded  | BlockingQueue (ArrayBlockingQueue etc.)
// Read-heavy shared state     | ReentrantReadWriteLock or StampedLock
// Inter-thread signaling      | Semaphore(0) or Condition.await/signal
// Publish immutable result    | Future / CompletableFuture

// ── Semaphore as a bounded channel (manual producer-consumer) ──────────
public class BoundedChannel<T> {
    private final Queue<T>    queue;
    private final Semaphore   slots;     // permits = empty slots
    private final Semaphore   items;     // permits = available items

    public BoundedChannel(int capacity) {
        this.queue = new ConcurrentLinkedQueue<>();
        this.slots = new Semaphore(capacity);  // starts full: capacity empty slots
        this.items = new Semaphore(0);         // starts empty: no items yet
    }

    public void put(T item) throws InterruptedException {
        slots.acquire();    // wait for an empty slot
        queue.offer(item);
        items.release();    // signal that an item is available
    }

    public T take() throws InterruptedException {
        items.acquire();    // wait for an item to be available
        T item = queue.poll();
        slots.release();    // signal that a slot is now free
        return item;
    }
}