6.32 Nonfair Semaphore

1. What It Is

Rate limiting using a nonfair strategy.

2. Usage

public class SemaphoreTest
{
    private final static int THREAD_COUNT = 100;
    private final static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT);

    public static void main(String[] args) throws InterruptedException
    {
        Semaphore semaphore = new Semaphore(10);

        for (int i = 0; i < THREAD_COUNT; i++)
        {
            new Thread(()->{
                try
                {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + " is accessing the resource...");
                    TimeUnit.SECONDS.sleep(3);

                }
                catch (Exception e)
                {
                    e.printStackTrace();
                }
                finally
                {
                    semaphore.release();
                    countDownLatch.countDown();
                }
            }).start();
        }

        countDownLatch.await();
    }
}

3. Principle Analysis

3.1. Constructor

public Semaphore(int permits) {
    // Create a nonfair Sync
    sync = new NonfairSync(permits);
}

3.1.1. Nonfair Sync

• NonfairSync

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;

    NonfairSync(int permits) {
        // Call Sync
        super(permits);
    }

    protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }
}

• Sync

abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;

    Sync(int permits) {
        // Ultimately, set permits permits
        setState(permits);
    }
}

3.2. acquire

public void acquire() throws InterruptedException {
    // AQS.acquireSharedInterruptibly
    sync.acquireSharedInterruptibly(1);
}

3.2.1. Call AQS to Acquire a Shared Lock

• AQS.acquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // Semaphore.NonfairSync overrides tryAcquireShared.
    // If there are not enough permits, it returns a negative value.
    // doAcquireSharedInterruptibly then joins the AQS queue and blocks while waiting to be woken.
    // If there are enough permits, it returns a value >= 0, and the caller of acquire can continue executing business logic.
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

3.2.1.1. Try to Acquire a Shared Lock [Nonfair: Try Directly Regardless of Whether Someone Is Queued Ahead]

• Semaphore.NonfairSync#tryAcquireShared

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

• Semaphore.Sync#nonfairTryAcquireShared

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        // Current number of permits
        int available = getState();
        // Are there enough permits for me to acquire?
        int remaining = available - acquires;
        // If < 0, there are not enough permits, so return this value
        if (remaining < 0 ||
        // >= 0 means there are enough permits, so CAS-update the remaining permits
            compareAndSetState(available, remaining))
            return remaining;
    }
}

3.3. release

• Semaphore#release()

public void release() {
    // AQS.releaseShared
    sync.releaseShared(1);
}

3.3.1. Call AQS to Release a Shared Lock

• AQS#releaseShared

public final boolean releaseShared(int arg) {
    // Semaphore.Sync overrides tryReleaseShared
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

3.3.1.1. Try to Release the Lock

• Semaphore.Sync#tryReleaseShared

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        // Get the current number of permits
        int current = getState();
        // Add the permits back
        int next = current + releases;
        // Throw an exception on overflow
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        // CAS-update the number of permits
        if (compareAndSetState(current, next))
            return true;
    }
}