Šta je zapravo AQS (AbstractQueuedSynchronizer)?

AQS je skraćenica od AbstractQueuedSynchronizer, odnosno „apstraktni sinhronizator reda". Doslovno se može razumeti ovako:

• apstraktan: apstraktna klasa koja implementira samo deo glavne logike, dok se pojedini metodi implementiraju u podklasama;
• red: koristi FIFO red za skladištenje podataka;
• sinhronizacija: ostvaruje funkciju sinhronizacije.

Čemu onda služi AQS?

AQS je radni okvir za izgradnju brava i sinhronizatora; uz AQS možemo jednostavno i efikasno izgraditi široko primenjive sinhronizatore, kao što su ReentrantLock, Semaphore, ReentrantReadWriteLock, SynchronousQueue, FutureTask itd., koji se svi temelje na AQS-u.

Naravno, možemo iskoristiti AQS i da lako napravimo sopstveni sinhronizator — dovoljno je implementirati njegovih nekoliko protected metoda.

Struktura podataka AQS-a

AQS interno koristi volatile promenljivu state kao oznaku resursa.

/**
 * The synchronization state.
 */
private volatile int state;

Uz to su definisani neki protected metodi za preuzimanje i izmenu stanja state (uz zahvalnost članu zajednice kicheng-u na ukazanoj ispravci):

getState()
setState()
compareAndSetState()

Sve tri operacije su atomične, pri čemu realizacija compareAndSetState zavisi od metoda compareAndSwapInt() klase Unsafe.

AQS interno koristi FIFO duvostruki red, sa dve reference head i tail koje označavaju početak i kraj reda. Struktura podataka je prikazana na slici ispod:

Međutim, AQS ne skladišti direktno niti, već Node čvorove koji sadrže niti.

Node čvor AQS-a

Resursi imaju dva režima deljenja, odnosno dva načina sinhronizacije:

• Ekskluzivni režim (Exclusive): resurs je isključiv — samo jedna nit ga može acquire-ovati. Primer je ReentrantLock (o njemu će detaljno biti reči kasnije).
• Režim deljenja (Share): može ga acquire-ovati više niti istovremeno; konkretan broj resursa se može zadati parametrom. Primer su Semaphore/CountDownLatch.

Uobičajeno je da podklase implementiraju samo jedan od režima, mada postoje i sinhronizatorske klase koje implementiraju oba režima, kao što je ReadWriteLock.

Definicije ova dva režima deljenja resursa u AQS-u nalaze se u internom razredu Node. Pogledajmo strukturu Node-a:

static final class Node {
    // Označava da čvor (i njegova nit) čeka u režimu deljenja
    static final Node SHARED = new Node();
    // Označava da čvor (i njegova nit) čeka u ekskluzivnom režimu
    static final Node EXCLUSIVE = null;

    // Vrednost waitStatus-a koja označava da je čvor (nit) otkazan
    static final int CANCELLED = 1;
    // Vrednost waitStatus-a koja označava da sledeći čvor (nit) treba biti probuđen
    static final int SIGNAL = -1;
    // Vrednost waitStatus-a koja označava da čvor (nit) čeka na neki uslov
    static final int CONDITION = -2;
    /*Vrednost waitStatus-a koja označava da su resursi dostupni i da novi head čvor treba da nastavi sa buđenjem sledećih čvorova (u režimu deljenja, pri konkurentnom oslobađanju resursa; kada head probudi svog naslednika, višak resursa ostavlja za kasnije čvorove; pri postavljanju novog head čvora, on nastavlja da budi sledeće čvorove)*/
    static final int PROPAGATE = -3;

    // Status čekanja; dozvoljene vrednosti: -3, -2, -1, 0, 1
    volatile int waitStatus;
    volatile Node prev; // prethodni čvor
    volatile Node next; // sledeći čvor
    volatile Thread thread; // nit vezana za čvor
    Node nextWaiter; // sledeći čvor koji čeka na uslov u redu čekanja

    // Metod za proveru režima deljenja
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    Node(Thread thread, Node mode) {     // koristi ga addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
    }

    // Ostali metodi su izostavljeni; pogledajte konkretan izvorni kod
}

// Privatni metod addWaiter unutar AQS-a
private Node addWaiter(Node mode) {
    // koristi ovaj konstruktor Node-a
    Node node = new Node(Thread.currentThread(), mode);
    // ostatak koda izostavljen
}

Ovde waitStatus služi da obeleži stanje tekućeg čvora; moguća su sledeća stanja:

• CANCELLED: označava da je tekući čvor (nit) otkazan. Pri čekanju isteka vremena ili prekida ulazi se u ovo stanje; jednom u njemu, stanje čvora se više ne menja;
• SIGNAL: sledeći čvor čeka da tekući čvor ga probudi;
• CONDITION: koristi se uz Condition; trenutna nit je blokirana na Condition; ako druga nit pozove metod signal iz Condition, taj čvor se iz reda čekanja premešta na kraj sinhronizacionog reda, gde čeka na acquire sinhronizacione brave;
• PROPAGATE: režim deljenja — nakon što prethodni čvor probudi sledeći, operacija buđenja se bezuslovno propagira dalje;
• 0: prelazno stanje — sledeći čvor tekućeg čvora je već probuđen, ali nit tekućeg čvora još nije završila izvršenje.

Pomoću Node-a možemo ostvariti dve vrste redova:

1. Prvo, preko prev i next se ostvaruje CLH (Craig, Landin, and Hagersten) red (red za sinhronizaciju niti, dvosmerni red).

U CLH bravi, svaka nit koja čeka ima pridružen Node; svaki Node ima pokazivače prev i next. Kada nit pokuša da acquire-uje bravu i ne uspe, dodaje se na kraj reda i spinuje, čekajući da nit prethodnog čvora oslobodi bravu. Otprilike ovako:

public class CLHLock {
    private volatile Node tail;
    private ThreadLocal<Node> myNode = ThreadLocal.withInitial(Node::new);
    private ThreadLocal<Node> myPred = new ThreadLocal<>();

    public void lock() {
        Node node = myNode.get();
        node.locked = true;
        // sebe stavlja na kraj reda i uzima prethodni čvor
        Node pred = tail;
        myPred.set(pred);
        while (pred.locked) {
            // spinovanje — čeka
        }
    }

    public void unlock() {
        Node node = myNode.get();
        node.locked = false;
        myNode.set(myPred.get());
    }

    private static class Node {
        private volatile boolean locked;
    }
}

2. Drugo, preko nextWaiter se ostvaruje red čekanja niti na Condition (jednosmerni red), koji se prvenstveno koristi u klasi ReentrantLock.

Analiza izvornog koda AQS-a

Dizajn AQS-a se zasniva na obrascu metoda šablona (template method) — on ima nekoliko metoda koje podklase moraju implementirati:

• isHeldExclusively(): da li ta nit trenutno isključivo drži resurs. Implementira se samo ako se koristi Condition.
• tryAcquire(int): ekskluzivni režim. Pokušava da acquire-uje resurs; vraća true pri uspehu, false pri neuspehu.
• tryRelease(int): ekskluzivni režim. Pokušava da oslobodi resurs; vraća true pri uspehu, false pri neuspehu.
• tryAcquireShared(int): režim deljenja. Pokušava da acquire-uje resurs. Negativan broj označava neuspeh; 0 označava uspeh ali bez preostalih resursa; pozitivan broj označava uspeh i preostale resurse.
• tryReleaseShared(int): režim deljenja. Pokušava da oslobodi resurs; vraća true ako nakon oslobađanja treba probuditi naredne čvorove koji čekaju, inače false.

Iako su svi ovi metodi protected, oni nisu konkretno implementirani u AQS-u, već direktno bacaju izuzetak:

protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

Razlog za korišćenje ne-apstraktnih metoda je izbegavanje obaveze da podklase implementiraju sve apstraktne metode, čime se smanjuje uzaludan rad — podklasa implementira samo metode koji je brinu, na primer Semaphore implementira samo tryAcquire, bez ostalih metoda šablona koji mu nisu potrebni:

A AQS implementira niz glavnih delova logike. U nastavku ćemo iz izvornog koda analizirati glavnu logiku acquire-ovanja i oslobađanja resursa:

Acquire resursa

Ulazna tačka za acquire resursa je metod acquire(int arg). arg je broj resursa koje treba acquire-ovati; u ekskluzivnom režimu je uvek 1. Hajde da pogledamo logiku ovog metoda:

// uz zahvalnost članu zajednice 4. na ispravci
public final void acquire(int arg) {
    // tryAcquire ponovo pokušava da acquire-uje resurs brave; vrati true pri uspehu, false pri neuspehu
    if (!tryAcquire(arg) &&
        // ako smo ovde, acquire resursa brave nije uspeo; trenutnu nit treba umotati u Node i dodati na kraj AQS reda
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // prekid niti
        selfInterrupt();
}

Prvo se poziva tryAcquire u pokušaju acquire-ovanja resursa. Gore je rečeno da se ovaj metod konkretno implementira u podklasi; možemo ga pogledati direktno u klasi ReentrantLock.

Ako acquire resursa ne uspe, metodom addWaiter(Node.EXCLUSIVE) ta nit se umeće u red čekanja. Prosleđeni argument označava da se umeće ekskluzivni Node. Konkretna implementacija ovog metoda:

private Node addWaiter(Node mode) {
 // kreira klasu Node, postavlja thread na trenutnu nit i označava je kao ekskluzivnu bravu
 Node node = new Node(Thread.currentThread(), mode);
 // preuzima krajnji čvor reda u AQS-u
 Node pred = tail;
 // ako je tail == null, red je prazan;
 // ako nije null, u redu postoje podaci
 if (pred != null) {
  // prev tekućeg čvora se usmerava na prethodni krajnji čvor, a tekući čvor treba postati novi krajnji čvor
  node.prev = pred;
  // CAS postavlja tail čvor na tekući čvor
  if (compareAndSetTail(pred, node)) {
   // next prethodnog krajnjeg čvora se usmerava na tekući čvor
   pred.next = node;
   // vraća tekući čvor
   return node;
  }
 }
 enq(node);
 return node;
}

// spin-CAS umeće u red čekanja
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // must initialize — moramo inicijalizovati
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

Ova dva metoda su prilično lako razumljiva: na kraj reda se umeće novi Node čvor; treba obratiti pažnju na to da u AQS-u više niti istovremeno može započeti borbu za resurse, pa će sigurno doći do istovremenog umetanja čvorova od strane više niti — ovde se atomičnost i bezbednost niti postižu kroz spin-CAS.

Dobro, sada se vratimo na metod acquire sa početka. Pozivom addWaiter čvor je već smešten na kraj reda čekanja. Čvorovi koji se nalaze u redu čekanja acquire-uju resurse polazeći od head-a, jednog po jednog. Konkretnu realizaciju pogledajmo u metodi acquireQueued:

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        // interrupted beleži da li je nit prekidana
        boolean interrupted = false;
        for (;;) { // operacija spinovanja
            // preuzima prethodni čvor tekućeg čvora
            final Node p = node.predecessor();
            // ako je prethodni čvor head i pokušaj acquire sinhronizacionog stanja uspe
            if (p == head && tryAcquire(arg)) {
                // postavlja tekući čvor kao head
                setHead(node);
                // next referencu prethodnog čvora postavlja na null, čime se pomaže sakupljaču smeća
                p.next = null; 
                // acquire sinhronizacionog stanja je uspeo, failed se postavlja na false
                failed = false;
                // vraća da li je nit prekidana
                return interrupted;
            }
            // ako trenutna nit treba da bude blokirana i zaista je prekinuta tokom blokade, interrupted se postavlja na true
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // ako acquire sinhronizacionog stanja ne uspe, otkazuje pokušaj acquire-ovanja
        if (failed)
            cancelAcquire(node);
    }
}

Ovde metod parkAndCheckInterrupt interno koristi LockSupport.park(this), pa ukratko predstavimo i metod park.

Klasa LockSupport je uvedena u Javi 6 i pruža osnovne primitive sinhronizacije niti. LockSupport zapravo poziva metode klase Unsafe, što se na kraju svodi na dva metoda:

• park(boolean isAbsolute, long time): blokira trenutnu nit
• unpark(Thread jthread): zaustavlja blokadu date niti

Dakle, nakon što čvor uđe u red čekanja, park ga prevodi u blokirano stanje. Samo nit head čvora je u aktivnom stanju.

Naravno, pored acquire, postoje još tri metoda za acquire resursa:

• acquireInterruptibly: acquire resursa koji može biti prekinut (ekskluzivni režim)
• acquireShared: acquire resursa u režimu deljenja
• acquireSharedInterruptibly: acquire resursa koji može biti prekinut (režim deljenja)

„Može biti prekinut" znači da pri prekidanju niti može biti bačen InterruptedException.

Jedan dijagram toka koji sve sumira:

Tok acquire

Oslobađanje resursa

U poređenju sa acquire-ovanjem, oslobađanje resursa je mnogo jednostavnije. U AQS-u postoji samo jedan kratki deo implementacije. Izvorni kod:

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

private void unparkSuccessor(Node node) {
    // ako je status negativan, pokušavamo da ga postavimo na 0
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    // preuzima naslednika head čvora, head.next
    Node s = node.next;
    // ako je taj naslednik null ili mu je status veći od 0
    // prema gornjoj definiciji, veće od 0 može značiti samo da je čvor otkazan (samo Node.CANCELLED(=1) je veće od 0)
    if (s == null || s.waitStatus > 0) {
        s = null;
        // polazeći od kraja, unazad traži prvi ne-otkazani čvor (pravi naslednik)
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    // ako naslednik nije null
    if (s != null)
        LockSupport.unpark(s.thread);
}

U implementaciji java.util.concurrent.locks.ReentrantLock, tryRelease(arg) smanjuje broj bravu koje se drže; kada broj postane 0, oslobađa bravu i vraća true.

Ako tryRelease(arg) uspešno oslobodi bravu, zatim se proverava head čvor reda. Ako head postoji i waitStatus mu nije 0 (to znači da nit čeka), poziva se unparkSuccessor(Node h) koji budi niti koje čekaju.

Kratak pregled

AQS je radni okvir za izgradnju brava i sinhronizatora; uz AQS možemo jednostavno i efikasno izgraditi široko primenjive sinhronizatore kao što su ReentrantLock, Semaphore, ReentrantReadWriteLock, SynchronousQueue, FutureTask itd., koji se svi temelje na AQS-u.

Naravno, možemo iskoristiti AQS i da lako napravimo sopstveni sinhronizator — dovoljno je implementirati njegovih nekoliko protected metoda.

Hajde da napravimo jednu mutex bravu (bravu koja u istom trenutku dozvoljava samo jednoj niti da je drži).

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class Mutex {

    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
    }

    private final Sync sync = new Sync();

    public void lock() {
        sync.acquire(1);
    }

    public void unlock() {
        sync.release(1);
    }

    public boolean isLocked() {
        return sync.isHeldExclusively();
    }
}

Gornja klasa Mutex je jedna mutex brava. Ona interno koristi klasu Sync koja nasleđuje AQS.

• tryAcquire: pokušava da acquire-uje resurs. Ako je trenutno stanje 0 (nije zaključano), postavlja ga na 1 (zaključano) i označava trenutnu nit kao onu koja isključivo drži resurs.
• tryRelease: pokušava da oslobodi resurs. Postavlja stanje na 0 i briše nit koja drži resurs.
• isHeldExclusively: proverava da li je resurs isključivo zauzet.

Pretpostavimo da imamo resurs koji nije bezbedan za niti, a želimo da osiguramo da u svakom trenutku samo jedna nit može pristupiti njemu — možemo iskoristiti ovu Mutex bravu da osiguramo bezbednost niti.

public class Resource {
    private Mutex mutex = new Mutex();

    public void use() {
        mutex.lock();
        try {
            // operacije nad resursom
        } finally {
            mutex.unlock();
        }
    }
}

U ovom scenariju, nebezbednom resursu smo dodali mutex bravu koja osigurava da u istom trenutku samo jedna nit može koristiti taj resurs, čime se postiže bezbednost niti.

Autor: Chenmo Wang Er. Sadržaj pre uređivanja potiče iz ovog otvorenog repozitorijuma prijatelja: Duboko i pristupačno o Java višenitnosti, toplo preporučujemo. Vredi pogledati i: Jun Ge o tehnologiji: 20 hiljada reči + 40 slika koje vas vode kroz Java AQS, A Q o kodu: 20 slika koje vas vode kroz zaključavanje i otključavanje.