Synchronous class based AbstractQueuedSynchronizer (AQS)

We introduced many synchronization class before, such as already, Semaphore, CountDownLatch, ReentrantReadWriteLock, FutureTask, etc.

AQS encapsulates many of the details of the design when implementing synchronizers. It provides the wait queues for FIFO and provides an int state to represent the current state.

According to the JDK, it is not recommended that we use AQS directly. We usually need to build an inner class that inherits AQS and override the following methods as needed:

  • tryAcquire
  • tryRelease
  • tryAcquireShared
  • tryReleaseShared
  • isHeldExclusively

In these methods, we need to call getState, setState, or compareAndSetState to change the state value.

The above method refers to two types of operations, exclusive (e.g., ReentrantLock) and shared (e.g., Semaphore,CountdownLatch).

The difference is that more than one thread can obtain the synchronization state at the same time.

For example, if we run multiple threads to read at the same time, but the notification allows only one thread to write, then the read lock is a shared operation and the write lock is an exclusive operation.

In synchronous classes built on QAS, the most basic operations are the get and release operations. The state represents the values on which the get and release operations depend.

State is an int value that you can use to indicate any State, such as ReentrantLock to indicate how many times the owner thread has repeatedly acquired the lock. Semaphore uses it to indicate the remaining permissions, while FutureTask uses it to indicate the status of the task (start, run, complete, or cancel). Of course, you can also customize additional state variables to represent other information.

The pseudo-code below represents the form of the get and release operation in AQS:

   Acquire:
       while(! tryAcquire(arg)) { enqueue threadif it is not already queued;
          possibly block current thread;
       }
  
   Release:
       if (tryRelease(arg))
          unblock the first queued thread;
Copy the code

The fetch operation first determines whether the current state allows the fetch operation, and if not, the current thread is queued and may block.

If yes, remove the thread from the queue and run.

Let’s look at a concrete implementation:

public class AQSUsage {

    private final Sync sync= new Sync();

    private class Sync extends AbstractQueuedSynchronizer{
        protected int tryAcquireShared(int ignored){
            return (getState() ==1 )? 1: -1;
        }
        protected boolean tryReleaseShared(int ignored){
            setState(1);
            return true; }}public void release(a) {
        sync.releaseShared(0);
    }
    public void acquire(a) throws InterruptedException {
        sync.acquireSharedInterruptibly(0); }}Copy the code

In the example above, we define an internal class Sync that implements the tryAcquireShared and tryReleaseShared methods, where we determine and set the value of state.

Sync. ReleaseShared and sync. AcquireSharedInterruptibly will call tryAcquireShared and tryReleaseShared method respectively.

As mentioned earlier, many synchronization classes are implemented using AQS, so let’s look at tryAcquire implementations of other standard synchronization classes.

Let’s start with ReentrantLock:

   final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true; }}else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
Copy the code

ReentrantLock supports only exclusive locks. So it needs to implement the tryAcquire method. It also maintains an owner variable that holds the identifier of the current owner thread to implement reentrant locking.

Let’s look at the implementation of Semaphore and CountDownLatch. Since they are shared operations, we need to implement the tryAcqureShared method:

        final int tryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    returnremaining; }}Copy the code

Examples of this article can be found at github.com/ddean2009/l…

For more, visit flydean’s blog