Java Concurrency: The AbstractQueuedSynchronizer

Previously we worked with one of the building blocks of Java concurrency the LockSupport class. On this blog we will have a look at the AbstractQueuedSynchronizer.
AbstractQueuedSynchronizer is one of the building blocks of Java synchronization. Lock implementations such as ReentrantLock and Semaphore are based on the AbstractQueuedSynchronizer.
An integer represents the state. Based on the use case what a state represent can vary for example it can represent how many permits are available. Operations that change the state are guarded by a FIFO queue and tries to do so in a fair manner, and provided access to the lock in the order which the thread arrived.

 

The FIFO queue is implemented using a nested Node class is. The Node represents the Thread waiting in the queue. The operations are atomic and the variables are volatile. It contains references to the previous and the next node.

    abstract static class Node {
        volatile Node prev;       // initially attached via casTail
        volatile Node next;       // visibly nonnull when signallable
        Thread waiter;            // visibly nonnull when enqueued
        volatile int status;      // written by owner, atomic bit ops by others

...
    }

An instance of the AbstractQueuedSynchronizer, since it provides a FIFO, will have a reference of the head and tail in the class:

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
...
    private transient volatile Node head;
    private transient volatile Node tail;
...
}

As mentioned we have the state variable¨

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
...
    private volatile int state;
...
}

Τhe state variable can be used to represent the resources available or allocated.

When it comes to ReentrantLock it represent the number of holds on the lock:

public class ReentrantLock implements Lock, java.io.Serializable {
...
    static final class FairSync extends Sync {
...
        final boolean initialTryLock() {
...
            } else if (getExclusiveOwnerThread() == current) {
                if (++c < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(c);
                return true;
            }
...
    }
...
}

When it comes to the Semaphore it represents the permits:

public class Semaphore implements java.io.Serializable {
...
    static final class FairSync extends Sync {
...
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
...
   }
...
}

Since the AbstractQueuedSynchronizer supports a shared mode as well as an exclusive mode, there are two Node implementations to represent it.

    static final class ExclusiveNode extends Node { }
    static final class SharedNode extends Node { }

For example ReentrantLock is on an Exclusive mode while Semaphore is on a shared mode.

The acquire method is where the logic lies.

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

The implementation of acquiring the lock can vary from jdk implementations but the objectives should be the same.
The first step is the thread to try and acquire the lock if it is available without blocking.
This can vary on the Shared on Exclusive mode.
If the lock is held the Thread is blocked and aded to the waiting queue.

    final boolean acquireQueued(final Node node, int arg) {
        boolean interrupted = false;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node))
                    interrupted |= parkAndCheckInterrupt();
            }
        } catch (Throwable t) {
...
        }
    }

As the lock gets released by the thread holding it the threads in the queue are signaled to proceed and acquire the lock.
The mechanism varies whether shared or exclusive.

Now that we know more on the AbstractQueuedSynchronizer we can check its implementations and find more about its usage.