Detaljno objašnjenje mehanizma čekanja/obaveštenja await i signal u Java Condition
O uslovu čekanja/obaveštenja Condition smo već govorili kada smo obrađivali ReentrantLock — sećate li se?
Svaki objekat može pozvati metode wait/notify iz klase Object da ostvari mehanizam čekanja/obaveštenja. Interfejs Condition pruža slične metode.
Interfejs Condition ukupno pruža sledećih 7 metoda:
await(): nit čeka dok ne bude obaveštena ili prekinuta. SličnoObject.wait().awaitUninterruptibly(): nit čeka dok ne bude obaveštena; čak i ako je prekinuta tokom čekanja, ne vraća se. Nema odgovarajući metod u klasi Object.await(long time, TimeUnit unit): nit čeka zadato vreme, dok ne bude obaveštena ili prekinuta. SličnoObject.wait(long timeout), ali sa fleksibilnijom jedinicom vremena.awaitNanos(long nanosTimeout): nit čeka zadato vreme u nanosekundama, dok ne bude obaveštena ili prekinuta. Nema odgovarajući metod u klasi Object.awaitUntil(Date deadline): nit čeka do zadatog roka, dok ne bude obaveštena ili prekinuta. Nema odgovarajući metod u klasi Object.signal(): budi jednu nit koja čeka. SličnoObject.notify().signalAll(): budi sve niti koje čekaju. SličnoObject.notifyAll().
Podsetimo se još jednom glavnih metoda klase Object:
wait(): nit čeka dok ne bude obaveštena ili prekinuta.wait(long timeout): nit čeka zadato vreme, dok ne bude obaveštena ili prekinuta.wait(long timeout, int nanos): nit čeka zadato vreme, dok ne bude obaveštena ili prekinuta.notify(): budi jednu nit koja čeka.notifyAll(): budi sve niti koje čekaju.
Analiza izvornog koda Condition
Da bismo duboko razumeli princip implementacije Condition, moramo zaviriti u njen izvorni kod.
Condition objekat se kreira preko lock.newCondition(); ovaj metod zapravo kreira nov objekat ConditionObject, pri čemu je ConditionObject interni razred AQS-a; uzmimo ReentrantLock kao primer.
public class ReentrantLock implements Lock, java.io.Serializable {
abstract static class Sync extends AbstractQueuedSynchronizer {
final ConditionObject newCondition() {
return new ConditionObject();
}
}
public Condition newCondition() {
return sync.newCondition();
}
}Ranije smo naučili da AQS interno održava FIFO dvosmerni red, sa referencama head i tail koje označavaju početak i kraj reda.
I Condition interno koristi isti pristup — interno održava FIFO jednosmerni red koji nazivamo red čekanja.
Sve niti koje pozovu metod await se dodaju u red čekanja, a stanje niti je stanje čekanja. firstWaiter pokazuje na prvi čvor, a lastWaiter na poslednji; izvorni kod je:
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
}nextWaiter u Node-u pokazuje na sledeći čvor u redu. Čvor koji uđe u red čekanja dobija status CONDITION (vidi se u demo primeru ispod).
Rekli smo da je red čekanja Condition-a jednosmerni red; hajde to da proverimo kroz demo sa debug-om.
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
Thread thread = new Thread(() -> {
lock.lock();
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
});
thread.start();
}
}Ovaj kod nema nikakav konkretan smisao, čak je i prilično loš. Kreiramo 10 niti; svaka nit prvo acquire-uje bravu, a zatim poziva metod condition.await koji oslobađa bravu i dodaje trenutnu nit u red čekanja; pri prelasku u 10. nit debug-om proveravamo firstWaiter, odnosno head čvor reda čekanja:
Sa ove slike jasno se uočavaju dve stvari:
- Nakon poziva metode condition.await, niti su redom dodavane na kraj reda čekanja — redom Thread-0, Thread-1, Thread-2...Thread-8;
- Red čekanja je jednosmeran red. Šematski prikaz:
Takođe treba napomenuti: metod newCondition() možemo pozivati više puta i kreirati više Condition objekata, odnosno jedan lock može držati više redova čekanja.
Kod pristupa preko Object, s druge strane, postoji samo jedan sinhronizacioni red i jedan red čekanja.
Prema tome, ReentrantLock i slični AQS sinhronizatori mogu držati jedan sinhronacioni red i više redova čekanja — dovoljno je kreirati više Condition objekata. Šematski prikaz:
Koja je korist od držanja više redova čekanja? Možemo je pokazati sledećim primerom:
public class BoundedBuffer<T> {
private final LinkedList<T> buffer; // LinkedList kao bafer
private final int capacity; // maksimalni kapacitet bafera
private final ReentrantLock lock; // mutex brava
private final Condition notEmpty; // uslov da bafer nije prazan
private final Condition notFull; // uslov da bafer nije pun
public BoundedBuffer(int capacity) {
this.capacity = capacity;
this.buffer = new LinkedList<>();
this.lock = new ReentrantLock();
this.notEmpty = lock.newCondition();
this.notFull = lock.newCondition();
}
// ubacuje jedan element
public void put(T item) throws InterruptedException {
lock.lock();
try {
// ako je bafer pun, čeka
while (buffer.size() == capacity) {
notFull.await();
}
buffer.add(item);
// obaveštava potencijalne potrošače koji čekaju
notEmpty.signal();
} finally {
lock.unlock();
}
}
// uzima jedan element
public T take() throws InterruptedException {
lock.lock();
try {
// ako je bafer prazan, čeka
while (buffer.isEmpty()) {
notEmpty.await();
}
T item = buffer.removeFirst();
// obaveštava potencijalne proizvođače koji čekaju
notFull.signal();
return item;
} finally {
lock.unlock();
}
}
}Razmotrimo ovaj jednostavan ograničeni bafer BoundedBuffer u koji proizvođači ubacuju elemente, a potrošači ih uzimaju. Koristićemo dva Condition-a: jedan koji označava da bafer nije prazan (za čekanje potrošača), i drugi da bafer nije pun (za čekanje proizvođača).
Proizvođač poziva metod put da ubaci element; ako je bafer pun, čeka na uslov notFull. Potrošač poziva metod take da uzme element; ako je bafer prazan, čeka na uslov notEmpty. Kada se element ubaci ili uzme, odgovarajući uslov signalizira i budi nit koja čeka.
Glavna prednost korišćenja više Condition objekata jeste finija kontrola nad bravama, što omogućava složenije scenarije sinhronizacije, kao što je gore pomenuti ograničeni bafer.
Dobro, nastavimo sa analizom izvornog koda Condition.
Metod await klase Condition
Kada se pozove condition.await(), nit koja je acquire-ovala bravu ulazi u red čekanja; ako ta nit uopšte i može da se vrati iz metoda await(), nužno je acquire-ovala bravu povezanu sa Condition-om.
Već je rečeno da je Condition samo interfejs čija je implementaciona klasa ConditionObject, podklasa AQS-a.
Izvorni kod metoda await klase ConditionObject:
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 1. trenutna nit se umotava u Node i dodaje na kraj reda čekanja
Node node = addConditionWaiter();
// 2. oslobađa bravu koju drži trenutna nit; pri oslobađanju se budi sledeći čvor u sinhronizacionom redu
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// 3. trenutna nit ulazi u stanje čekanja
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 4. acquire sinhronizacionog stanja (odnosno brave) se čeka u spinu
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // cleanup ako je otkazano
unlinkCancelledWaiters();
// 5. obrada slučaja prekida
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}Glavna logika koda je u komentarima. Kada trenutna nit pozove condition.await(), oslobađa bravu, dodaje se u red čekanja i ostaje tamo dok je metodi signal/signalAll ne probude.
Mogu se postaviti sledeća pitanja:
- Kako se trenutna nit dodaje u red čekanja?
- Kako izgleda proces oslobađanja brave?
- Kako se izlazi iz metoda await?
Gornji kod daje odgovore na sva tri pitanja.
Odgovor na pitanje 1
Poziv metoda addConditionWaiter dodaje trenutnu nit u red čekanja; izvorni kod:
private Node addConditionWaiter() {
Node t = lastWaiter;
if (t != null && t.waitStatus != Node.CONDITION) {
// uklanja iz reda čekanja čvorove koji nisu u stanju čekanja
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
// ako je krajnji čvor null
if (t == null)
// prvi čvor se usmerava na node
firstWaiter = node;
else
// nextWaiter krajnjeg čvora se usmerava na node
t.nextWaiter = node;
// krajnji čvor se usmerava na node
lastWaiter = node;
return node;
}Prvo se t usmerava na krajnji čvor; ako krajnji čvor nije null i njegov waitStatus != -2 (-2 je CONDITION i označava da čeka na Condition uslov), uklanjaju se čvorovi koji nisu u stanju čekanja iz reda čekanja, a zatim se t usmerava na novi krajnji čvor.
Zatim se trenutna nit umotava u čvor sa waitStatus = -2 i dodaje na kraj reda čekanja.
Ako je krajnji čvor null, red je prazan, pa se i prvi i krajnji čvor usmeravaju na tekući čvor.
Ako krajnji čvor nije null, u redu postoje i drugi čvorovi, pa se nextWaiter tekućeg krajnjeg čvora usmerava na tekući čvor, a tekući čvor postaje novi krajnji čvor.
Ukratko, uloga ovog koda je da umotani Node trenutne niti umetne u red čekanja dodavanjem na kraj. Takođe se vidi da je red čekanja Condition-a lančani red bez head čvora, dok smo dok smo učili AQS videli da je sinhronizacioni red lančani red sa head čvorom — to je jedna od razlika.
Ulogu head čvora ćemo ovde ukratko objasniti.
Lančana lista bez head čvora je lista čiji je prvi čvor ujedno i prvi stvarni element sa podacima, a ne poseban „head" čvor bez stvarnih podataka.
- Lista bez head čvora:
- Prvi čvor liste jeste prvi stvarni čvor sa podacima.
- Kada je lista prazna, head referenca (često nazvana head) pokazuje na null.
- Lista sa head čvorom:
- Lista ima poseban čvor na početku — taj čvor se zove head čvor.
- Head čvor obično ne skladišti nikakve stvarne podatke, ili se njegovo polje sa podacima ne koristi.
- Bez obzira na to da li je lista prazna, head čvor uvek postoji. Kada je lista prazna, sledeći čvor head čvora pokazuje na null.
- Korišćenje head čvora pojednostavljuje neke operacije nad listom jer ne zahteva posebno tretiranje umetanja i brisanja prvog elementa.
Da bismo bolje objasnili ove dve strukture liste, za svaku ćemo dati jednostavan primer metode umetanja u listu celih brojeva.
- Lista bez head čvora
public class Node {
public int data;
public Node next;
public Node(int data) {
this.data = data;
this.next = null;
}
}
public class LinkedListWithoutHead {
public Node head;
public void insert(int value) {
Node newNode = new Node(value);
if (head == null) {
head = newNode;
} else {
Node temp = head;
while (temp.next != null) {
temp = temp.next;
}
temp.next = newNode;
}
}
}- Lista sa head čvorom
public class NodeWithHead {
public int data;
public NodeWithHead next;
public NodeWithHead(int data) {
this.data = data;
this.next = null;
}
}
public class LinkedListWithHead {
private NodeWithHead head;
public LinkedListWithHead() {
head = new NodeWithHead(-1); // inicijalizacija head čvora
}
public void insert(int value) {
NodeWithHead newNode = new NodeWithHead(value);
NodeWithHead temp = head;
while (temp.next != null) {
temp = temp.next;
}
temp.next = newNode;
}
}Sada je sve jasno? Pošto smo raščistili head čvor, vratimo se na metod await klase Condition.
Odgovor na pitanje 2
Nakon umetanja trenutnog čvora u red čekanja, trenutna nit oslobađa bravu, što radi metod fullyRelease; izvorni kod:
final int fullyRelease(Node node) {
// true ako oslobađanje ne uspe, false ako uspe
boolean failed = true;
try {
// preuzima state trenutne brave
int savedState = getState();
// ako oslobađanje uspe
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
// ako oslobađanje ne uspe, status čvora se postavlja na otkazano
node.waitStatus = Node.CANCELLED;
}
}I ovaj kod je lako razumeti — poziva se metod šablon release iz AQS-a koji oslobađa sinhronizaciono stanje AQS-a i budi nit na koju ukazuje referenca naslednika head čvora sinhronizacionog reda; ako oslobađanje uspe, vraća se normalno, a ako ne uspe, baca se izuzetak.
Odgovor na pitanje 3
Kako se izlazi iz metoda await? Sada se vratimo na metod await; u njemu postoji sledeća logika:
while (!isOnSyncQueue(node)) {
// 3. trenutna nit ulazi u stanje čekanja
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}Metod isOnSyncQueue služi da proveri da li se Node u kom je trenutna nit nalazi u sinhronizacionom redu.
Ako waitStatus tekućeg čvora = -2, to znači da je u redu čekanja, pa se vraća false; ako tekući čvor ima prethodni čvor, to dokazuje da je u AQS redu, ali ako je prethodni čvor prazan, to znači da je to head čvor — a head čvor ne učestvuje u borbi za bravu, pa se takođe vraća false.
Ako tekući čvor nije ni u redu čekanja, a nije ni head čvor AQS-a i pri tome ima čvor next, onda se nalazi u AQS-u i vraća se true.
Sada je nužno prikazati dijagram odnosa sinhronizacionog reda i reda čekanja.
Kada nit prvi put pozove metod condition.await, ući će u ovaj while petlju, a zatim se LockSupport.park(this) prevodi u stanje čekanja. Da bi se izašlo iz await-a, prvi uslov je izlazak iz ove while petlje, a izlaza su samo dva:
- Izlazak iz while petlje preko break;
- Logički uslov u while petlji postane false.
Prvi slučaj nastaje kada se nit koja čeka prekine — tada kod ide na break i izlazi iz while petlje; drugi slučaj je kada se tekući čvor prebaci u sinhronizacioni red (odnosno druga nit je pozvala metod signal ili signalAll klase condition), pa logički uslov u while postaje false i while se završava.
Ukratko, preduslov za izlazak iz metoda await je da je trenutna nit prekinuta ili da je poziv condition.signal ili condition.signalAfter prebacio tekući čvor u sinhronizioni red.
Nakon izlaska iz while petlje poziva se acquireQueued(node, savedState); ovaj metod u spinu stalno pokušava da acquire-uje sinhronizaciono stanje, sve dok ne uspe (nit acquire-uje lock). To takođe pokazuje da za izlazak iz metoda await nit nužno mora biti acquire-ovala lock koji je referenciran (povezan) sa condition-om.
Do sada smo na sva tri pitanja došli čitanjem izvornog koda i time produbili razumevanje metoda await. Dijagram metoda await je ispod:
Kao na slici, nit koja poziva metod condition.await mora biti acquire-ovala lock — odnosno trenutna nit je head čvor sinhronizacionog reda. Pozivom ovog metoda, Node u koji je umotana trenutna nit se dodaje na kraj reda čekanja.
Podrška za mehanizam isteka vremena
Condition dodatno podržava mehanizam isteka vremena; korisnik može pozvati metode awaitNanos i awaitUntil, čija je implementacija gotovo ista kao metod tryAcquire iz AQS-a.
Podrška za ignorisanje prekida
Da biste ignorisali prekid, pozovite metod condition.awaitUninterruptibly(); njegov izvorni kod:
public final void awaitUninterruptibly() {
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
boolean interrupted = false;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if (Thread.interrupted())
interrupted = true;
}
if (acquireQueued(node, savedState) || interrupted)
selfInterrupt();
}Ovaj metod je gotovo identičan gorenjem metodu await, samo je smanjena obrada prekida.
Princip implementacije signal/signalAll
Poziv metoda signal ili signalAll klase condition prebacuje čvor koji najduže čeka u redu čekanja u sinhronizioni red, čime taj čvor dobija šansu da acquire-uje lock. Red čekanja je FIFO, pa head čvor reda čekanja nužno jeste čvor koji najduže čeka; drugim rečima, svaki poziv metoda signal klase condition prebacuje head čvor u sinhronizioni red.
Proverimo kroz izvorni kod da li je ta pretpostavka tačna. Izvorni kod metoda signal:
public final void signal() {
//1. prvo proverava da li je trenutna nit acquire-ovala lock
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//2. preuzima prvi čvor iz reda čekanja; sve naredne operacije se odnose na taj čvor
Node first = firstWaiter;
if (first != null)
doSignal(first);
}Metod signal prvo proverava da li je trenutna nit acquire-ovala lock; ako nije, baca izuzetak; ako jeste, preuzima head čvor reda čekanja, a zatim i metod doSignal radi na tom čvoru. Pogledajmo šta radi metod doSignal; njegov izvorni kod:
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
//1. uklanja head čvor iz reda čekanja
first.nextWaiter = null;
//2. while petlja: transferForSignal vrši stvarnu obradu head čvora
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}Konkretna logika je u komentarima; stvarna obrada head čvora se nalazi u metodu transferForSignal, čiji je izvorni kod:
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
//1. status se ažurira na 0
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
//2. taj čvor se prebacuje u sinhronizioni red
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}Ključna logika je u komentarima; ovaj kod uglavnom radi dve stvari:
- Status head čvora se menja iz CONDITION;
- Poziva se metod enq koji dodaje taj čvor na kraj sinhronizionog reda; o metodu enq pogledajte članak implementacija AQS-a.
Sada možemo izvesti sledeći zaključak:
Preduslov za poziv metoda condition.signal jeste da je trenutna nit acquire-ovala lock; ovaj metod prebacuje head čvor reda čekanja — odnosno čvor koji najduže čeka — u sinhronizioni red, a tek prebacivanjem u sinhronioni red dobija šansu da bude probuđen, odnosno da se vrati iz LockSupport.park(this) metoda u metodi await, kako bi nit koja je pozvala await uspela da izađe.
Dijagram izvršenja metoda signal je ispod:
signalAll
Razlika između signalAll i signal metoda ogleda se u metodi doSignalAll; već znamo da metod doSignal radi samo sa head čvorom reda čekanja; izvorni kod metode doSignalAll:
private void doSignalAll(Node first) {
lastWaiter = firstWaiter = null;
do {
Node next = first.nextWaiter;
first.nextWaiter = null;
transferForSignal(first);
first = next;
} while (first != null);
}Ovaj metod prebacuje svaki čvor iz reda čekanja u sinhronioni red, odnosno „obaveštava" svaku nit koja je pozvala condition.await().
await i signal/signalAll
Mehanizam čekanja/obaveštenja pomenut na početku članka može se ostvariti preko metoda await i signal/signalAll klase condition, a ovaj mehanizam može rešiti najklasičniji problem — „problem proizvođača i potrošača", koji je inače i česta tema na intervjuima; biće detaljno objašnjen kasnije.
Metodi await, signal i signalAll su poput prekidača koji kontroliše nit A (ona koja čeka) i nit B (ona koja obaveštava). Njihov odnos se najbolje može opisati sledećom slikom:
Nit awaitThread prvo acquire-uje bravu preko metoda lock.lock(), a nakon uspeha poziva metod condition.await i ulazi u red čekanja; druga nit signalThread acquire-uje bravu preko metoda lock.lock(), a nakon uspeha poziva metod condition.signal ili signalAll, čime nit awaitThread dobija šansu da bude prebačena u sinhronioni red; kada druge niti oslobode lock, nit awaitThread dobija šansu da acquire-uje lock i tako izađe iz metode await i nastavi sa narednim operacijama. Ako awaitThread ne uspe da acquire-uje lock, direktno ulazi u sinhronioni red.
Primer korišćenja Condition
Prikažimo upotrebu Condition-a kroz jednostavan primer:
public class AwaitSignal {
private static ReentrantLock lock = new ReentrantLock();
private static Condition condition = lock.newCondition();
private static volatile boolean flag = false;
public static void main(String[] args) {
Thread waiter = new Thread(new waiter());
waiter.start();
Thread signaler = new Thread(new signaler());
signaler.start();
}
static class waiter implements Runnable {
@Override
public void run() {
lock.lock();
try {
while (!flag) {
System.out.println(Thread.currentThread().getName() + " trenutno uslov nije ispunjen — čeka");
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + " primio obaveštenje, uslov ispunjen");
} finally {
lock.unlock();
}
}
}
static class signaler implements Runnable {
@Override
public void run() {
lock.lock();
try {
flag = true;
condition.signalAll();
} finally {
lock.unlock();
}
}
}
}Izlaz je:
Thread-0 trenutno uslov nije ispunjen — čeka
Thread-0 primio obaveštenje, uslov ispunjenPokrenute su dve niti — waiter i signaler. Kada nit waiter počne sa izvršenjem, pošto uslov nije ispunjen, poziva metod condition.await koji prevodi tu nit u stanje čekanja i oslobađa bravu. Nit signaler acquire-uje bravu, menja uslov, obaveštava sve niti koje čekaju, a zatim oslobađa bravu. Tada nit waiter acquire-uje bravu; pošto je nit signaler promenila uslov, iz ugla niti waiter uslov je sada ispunjen, pa se nastavlja izvršenje.
Kratak pregled
Interfejs Condition je važna komponenta Java konkurentnog programiranja, namenjena koordinaciji i komunikaciji između niti. Obično se koristi zajedno sa bravom (naročito sa ReentrantLock) i pruža nitima mehanizam za čekanje da neki uslov postane istinit, dok drugim nitima dozvoljava da obaveste niti koje čekaju kada se taj uslov promeni. Time se za koordinaciju između niti dobija fleksibilniji i moćniji alat.
Autor: Chenmo Wang Er. Sadržaj pre uređivanja potiče uglavnom iz GitHub repozitorijuma CL0610 https://github.com/CL0610/Java-concurrency, a deo sadržaja i slika potiče iz članka čitaoca A Q: Konačno temeljno objašnjen principa Condition, toplo preporučujemo.