Savladajte Java ConcurrentLinkedQueue za intervju
ConcurrentLinkedQueue je implementacija reda bezbednog za niti u okviru paketa java.util.concurrent (JUC). Zasnovana je na neblokirajućem algoritmu (varijanta Michael-Scott neblokirajućeg algoritma), što znači da ConcurrentLinkedQueue više ne koristi tradicionalni mehanizam brava za zaštitu podataka, već se oslanja na atomic operacije nižeg nivoa (kao što je CAS).
Michael-Scott su predložili Maged M. Michael i Michael L. Scott 1996. godine; u tom algoritmu neuspeh ili suspendovanje jedne niti ne dovodi do neuspeha ili suspendovanja drugih niti.
Dobro, hajde da zajedno pogledamo izvornu realizaciju ConcurrentLinkedQueue.
Čvor Node
Počnimo od njegovog čvora Node, kako bismo razumeli strukturu podataka ispod ConcurrentLinkedQueue. Izvorni kod klase Node:
private static class Node<E> {
volatile E item;
volatile Node<E> next;
.......
}Čvor Node sadrži dva polja:
- Jedno je polje sa podacima item
- Drugo je pokazivač next koji ukazuje na sledeći čvor i tako čini lančani red.
Oba polja su obeležena sa volatile kako bi se osigurala vidljivost u memoriji.
Pored toga, ConcurrentLinkedQueue ima još dve sledeće promenljive člana:
private transient volatile Node<E> head;
private transient volatile Node<E> tail;
To znači da ConcurrentLinkedQueue upravlja redom držeći dve reference — na početak i na kraj. Kada pozovemo konstruktor bez argumenata, izvorni kod je:
public ConcurrentLinkedQueue() {
head = tail = new Node<E>(null);
}head i tail tada pokazuju na isti čvor, a stanje ConcurrentLinkedQueue je kao na slici ispod:
head i tail pokazuju na isti čvor Node0; polje item tog čvora je null, a polje next je takođe null.
Pri dequeue/enqueue operacijama neizbežno se manipuliše čvorovima, a u višenitnom okruženju lako dolazi do problema bezbednosti niti. ConcurrentLinkedQueue bira CAS da osigura bezbednost niti:
//menja polje podataka item u Node-u
boolean casItem(E cmp, E val) {
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
//menja polje pokazivača next u Node-u
void lazySetNext(Node<E> val) {
UNSAFE.putOrderedObject(this, nextOffset, val);
}
//menja polje pokazivača next u Node-u
boolean casNext(Node<E> cmp, Node<E> val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}Vidimo da ovi metodi zapravo pozivaju metode klase UNSAFE:
sun.misc.Unsafe je interni razred u Javi koji pruža skup metoda za direktan pristup resursima nižeg nivoa i manipulaciju memorijom. Ova klasa je veoma moćna jer programu dozvoljava da zaobiđe Java kontrolu pristupa i bezbednosne provere i direktno izvršava operacije nižeg nivoa.
Unsafe dozvoljava alokaciju, oslobađanje i pristup lokalnoj memoriji, baš kao malloc i free u jeziku C. To smo detaljno obrađivali kada smo pričali o CAS-u, pa se nadamo da vam je ostalo u sećanju.
offer metod
ConcurrentLinkedQueue je FIFO (First-In-First-Out) red; offer metod se koristi za umetanje elementa na kraj reda. Ako dodavanje uspe, vraća true. Uopštena definicija ovog metoda:
public boolean offer(E e)Pogledajmo sledeći kod:
ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
queue.offer(1);
queue.offer(2);Kreiramo objekat queue tipa ConcurrentLinkedQueue, prvo pozivamo offer 1, a zatim offer 2. Izvorni kod offer:
public boolean offer(E e) {
checkNotNull(e);
final Node<E> newNode = new Node<E>(e);
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
if (q == null) {
// p is last node
if (p.casNext(null, newNode)) {
// Successful CAS is the linearization point
// for e to become an element of this queue,
// and for newNode to become "live".
if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
else if (p == q)
// We have fallen off list. If tail is unchanged, it
// will also be off-list, in which case we need to
// jump to head, from which all live nodes are always
// reachable. Else the new tail is a better bet.
p = (t != (t = tail)) ? t : head;
else
// Check for tail updates after two hops.
p = (p != t && t != (t = tail)) ? t : q;
}
}Provera parametra:
checkNotNull(e)osigurava da prosleđeni element nije null.Kreiranje novog čvora:
final Node<E> newNode = new Node<E>(e)kreira novi čvor koji će sadržati element za dodavanje.Petlja traženja krajnjeg čvora: ova petlja služi da pronađe krajnji čvor reda i bezbedno poveže novi čvor na kraj.
- a. Čitanje sledećeg čvora:
Node<E> q = p.nextčita sledeći čvor tekućeg čvora. - b. Provera krajnjeg čvora: ako je q null, to znači da je tekući čvor p krajnji čvor.
- c. CAS operacija dodavanja novog čvora:
p.casNext(null, newNode)koristi CAS da poveže novi čvor na tekući kraj. Ako uspe, ažurira referencu tail i vraća true. - d. Ažuriranje tail-a dvostrukim skokom:
casTail(t, newNode)ponekad pokušava da ažurira referencu tail tako da pokazuje na novi kraj. To pomaže drugim nitima da brže pronađu kraj. - e. Provera ispadanja sa liste: ako je p == q, to znači da je tekuća nit „ispala" sa liste. U tom slučaju kod pokušava skok na head ili novi tail.
- f. Dodatna provera: inače, kod vrši dodatne provere i ažurira vrednost p — možda na tekući tail ili na sledeći čvor.
Komentare u kodu ćemo ukloniti i dodati brojeve redova.
public boolean offer(E e) {
1. checkNotNull(e);
2. final Node<E> newNode = new Node<E>(e);
3. for (Node<E> t = tail, p = t;;) {
4. Node<E> q = p.next;
5. if (q == null) {
6. // p is last node
7. if (p.casNext(null, newNode)) {
8. if (p != t)
9. casTail(t, newNode);
10. return true;
}
}
11. else if (p == q)
12. p = (t != (t = tail)) ? t : head;
else
13. p = (p != t && t != (t = tail)) ? t : q;
}
}Analiza iz ugla jednonitnog izvršenja
Iz ugla jedne niti analizirajmo proces offer 1.
Red 1 proverava da li je element e null; ako jeste, odmah baca izuzetak NullPointerException.
Red 2 umotava e u Node objekat.
Red 3 je for petlja sa samo inicijalnim uslovom, bez uslova završetka — to odgovara „triku" CAS-a: u telu petlje, ako CAS uspe, direktno se return; ako CAS ne uspe, u for petlji se stalno pokušava dok ne uspe. Ovde se instanca t inicijalizuje na tail, a p na t, odnosno tail.
p se smatra pravim krajnjim čvorom reda; tail nije nužno pravi krajnji čvor jer se u ConcurrentLinkedQueue tail ažurira sa zakašnjenjem.
Kada dođemo do reda 3, t i p pokazuju na item (null) kreiran pri inicijalizaciji, čiji je next takođe null, odnosno Node0.
Red 4: promenljiva q se postavlja na null.
Red 5: if uslov je true.
Red 7: pomoću casNext, umetnuti Node se postavlja kao next čvor tekućeg krajnjeg čvora p; ako CAS ne uspe, petlja se završava, a u sledećoj iteraciji pokušava ponovo.
Ako CAS uspe, dolazimo do reda 8; tada je p==t, uslov if je false, pa se direktno return true. Ako je 1 uspešno umetnuto, stanje ConcurrentLinkedQueue je kao na slici ispod:
Krajnji čvor reda sada treba da bude Node1, ali tail još uvek pokazuje na Node0, što pokazuje da se tail ažurira sa zakašnjenjem.
Nastavimo sa offer 2; očigledno, u redu 4 q sada nije null, već pokazuje na Node1, red 5 if je false, red 11 if je false, pa kod ide na red 13.
Dobro, prilikom umetanja čvora postavimo sebi sledeće pitanje: s obzirom da tail nije pravi krajnji čvor, ne bismo li trebalo prvo pronaći tekući krajnji čvor da bismo umetnuli čvor?
Red 13 služi baš tome — da pronađe pravi krajnji čvor reda.
p = (p != t && t != (t = tail)) ? t : q;U jednonitnom okruženju, pošto je p==t, p će se dodeliti q, a q je Node<E> q = p.next, odnosno Node1.
U prvoj iteraciji p pokazuje na pravi kraj reda Node1; u sledećoj iteraciji reda 4, q je null, pa je red 5 if true; red 7 opet pomoću casNext postavlja next čvora p na novi Node; zatim se dolazi do reda 8; tada je p!=t, pa je red 8 if true, pa se preko casTail(t, newNode) tekući Node postavlja kao krajnji čvor reda; stanje reda je kao na slici ispod:
Čvor na koji pokazuje tail se promenio iz Node0 u Node2. Ovde casTail ne mora da radi ponovne pokušaje; razlog je taj što offer uglavnom pomoću next čvora q čvora p (Node<E> q = p.next) odlučuje o toku logike. Dijagram stanja kada casTail ne uspe je ispod:
Ako casTail ne uspe da ažurira tail, odnosno tail još uvek pokazuje na Node0, jedino što se desi je da će biti par iteracija više, a preko reda 13 će se locirati krajnji čvor.
Analizom iz ugla jedne niti dolazimo do logike offer-a:
Ako je sledeći čvor (polje next) čvora tail null, to znači da je tail pravi krajnji čvor reda, pa se preko casNext može umetnuti tekući čvor, ali tada tail nije promenjen.
Ako sledeći čvor čvora tail nije null, to znači da tail nije pravi krajnji čvor. Pomoću q (
Node<E> q = p.next) se kreće napred dok se ne pronađe krajnji čvor, zatim se preko casNext umetne tekući čvor i preko casTail ažurira tail.
U jednonitnom okruženju, red p = (p != t && t != (t = tail)) ? t : q; nikada neće p postaviti na t; hajde da pokušamo analizu u višenitnom okruženju.
Analiza iz ugla višenitnog izvršenja
U višenitnom okruženju, red p = (p != t && t != (t = tail)) ? t : q; postaje zanimljiv.
Pošto operacija t != (t = tail) nije atomična, postoji sledeća situacija:
Pretpostavimo da nit A pročita promenljivu t, a nit B u tom trenutku uradi offer Node-a, čime menja tail; kada nit A ponovo izvrši t=tail, t će pokazivati na drugi čvor — očigledno, nit A u dva uzastopna čitanja vidi različite čvorove na koje pokazuje t, odnosno t != (t = tail) je true; a kako se t promenio, p != t je takođe true, pa je rezultat tog reda: p i t sada pokazuju na isti čvor, a t je i pravi kraj reda. Drugim rečima, sada je lociran pravi kraj reda i offer operacija se može izvršiti.
Do sada nije analiziran red 11; verovatno pogađate o kom slučaju je reč: deo niti radi offer, deo niti radi poll (o čemu će biti reči niže; poll dohvata i uklanja head element reda i suprotan je od offer).
Kada je if (p == q) true, next čvora p pokazuje na samog sebe — takav čvor se zove sentinel čvor; takvi čvorovi nemaju veliku vrednost u redu i obično označavaju čvor za brisanje ili prazan čvor. Da bismo bolje razumeli tu situaciju, prvo pogledajmo proces poll metode, a zatim se vratimo — u svakom slučaju, vrlo je zanimljivo.
poll metod
Izvorni kod poll metoda:
public E poll() {
restartFromHead:
for (;;) {
for (Node<E> h = head, p = h, q;;) {
E item = p.item;
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
else if (p == q)
continue restartFromHead;
else
p = q;
}
}
}Beskonačna petlja: spolja beskonačna petlja osigurava da se u visokokonkurentnom okruženju elementi ispravno uklanjaju sa početka reda.
Inicijalizacija referenci: inicijalizacija tekućeg head čvora h i čvora p (na početku isti kao head).
Čitanje tekućeg čvora:
E item = p.itemčita element tekućeg čvora.Provera da li tekući element nije null:
- Ako jeste i CAS uspešno postavi element na null (
p.casItem(item, null)), to znači da je element uspešno uklonjen. - Ako tekući čvor nije head (p != h), ažurira se head referenca tako da „preskoči" dva čvora.
- Vraća se uklonjeni element.
- Provera da li je dostignut kraj reda:
- Ako je q = p.next null, stigli smo do kraja reda. Ažurira se head referenca i vraća null što znači da je red prazan.
- Ako je p == q, to ukazuje na anomaliju usled konkurentne izmene;
continue restartFromHeadskače na početak spoljašnje petlje i pokušava ponovo.
- Prelazak na sledeći čvor: p se postavlja na q, odnosno sledeći čvor, i petlja se nastavlja.
Analiza iz ugla jednonitnog izvršenja
Radi lakše analize, uklonićemo komentare i dodati brojeve redova.
public E poll() {
restartFromHead:
1. for (;;) {
2. for (Node<E> h = head, p = h, q;;) {
3. E item = p.item;
4. if (item != null && p.casItem(item, null)) {
5. if (p != h) // hop two nodes at a time
6. updateHead(h, ((q = p.next) != null) ? q : p);
7. return item;
}
8. else if ((q = p.next) == null) {
9. updateHead(h, p);
10. return null;
}
11. else if (p == q)
12. continue restartFromHead;
else
13. p = q;
}
}
}Pretpostavimo da je početno stanje ConcurrentLinkedQueue kao na slici ispod:
Kao i kod offer, promenljivu p tretiramo kao head čvor koji treba ukloniti, dok h (head) nije nužno head čvor reda.
Najpre pogledajmo poll-ovanje Node1; pošto je p=h=head, očigledno podatak u Node1 na koji pokazuje p nije null, pa je u redu 4 uslov item!=null true, a zatim se preko casItem podatak Node1 postavlja na null.
Ako CAS ne uspe, petlja se završava i čeka sledeću iteraciju za ponovni pokušaj.
Ako red 4 uspe i uđemo u red 5, tada p i h oba pokazuju na Node1, uslov if u redu 5 je false, pa se u redu 7 vraća podatak 1 iz Node1 i metod se završava; stanje reda je kao na slici ispod.
Nastavimo sa poll-ovanjem; očigledno podatak u Node1 na koji pokazuju trenutni h i p jeste null, pa je prva stvar lociranje head čvora za brisanje (pronalaženje čvora čiji podatak nije null).
Nastavimo; podatak item u redu 3 je null, uslov if u redu 4 je false, pa se dolazi do reda 8 (q = p.next); taj if je takođe false; pošto q pokazuje na Node2, if u redu 11 je takođe false, pa kod ide na red 13; tada p i q oba pokazuju na Node2 — i time je pronađen pravi head čvor za brisanje.
Proces lociranja čvora za brisanje je: ako je podatak tekućeg čvora null, taj čvor očigledno nije čvor za brisanje, pa se pokušava sa sledećim čvorom tekućeg čvora. Nakon prve iteracije, dijagram stanja je ispod:
U sledećoj iteraciji, operacija u redu 4 je kao što je opisano; pretpostavimo da casItem u redu 4 uspe; pošto p sada pokazuje na Node2, dok h i dalje pokazuje na Node1, if u redu 5 je true, pa se izvršava updateHead(h, ((q = p.next) != null) ? q : p); tada q pokazuje na Node3; izvorni kod metode updateHead:
final void updateHead(Node<E> h, Node<E> p) {
if (h != p && casHead(h, p))
h.lazySetNext(h);
}Ovaj metod uglavnom preko casHead usmerava head reda na Node3, a preko h.lazySetNext usmerava next čvora Node1 na samog sebe. Na kraju, red 7 vraća vrednost Node2. Stanje reda je kao na slici ispod:
next čvora Node1 pokazuje na samog sebe, a head pokazuje na Node3.
Ako je red prazan, izvršavaće se red 8 (q = p.next) == null, uslov if je true, pa se u redu 10 direktno vraća null.
Sumirajmo:
Ako podatak Item čvora na koji pokazuju trenutni head, h i p nije null, taj čvor je pravi head čvor (čvor za brisanje); dovoljno je pomoću casItem postaviti item na null i vratiti stari item.
Ako podatak item čvora na koji pokazuju trenutni head, h i p jeste null, taj čvor nije pravi čvor za brisanje, pa treba nastaviti traženje čvora sa item != null. Pomoću q koji pokazuje na sledeći čvor p (q = p.next) se isprobava; kada se pronađe, preko metode updateHead se ažurira head čvor i kreira sentinel čvor (preko
h.lazySetNext(h)u metodi updateHead).
Analiza višenitnog izvršenja
Sada se vratimo na izvorni kod poll metoda; u njemu postoji deo:
else if (p == q)
continue restartFromHead;Taj deo služi za obradu više niti koje rade poll; q = p.next, odnosno q uvek pokazuje na sledeći čvor od p; u kom slučaju p i q pokazuju na isti čvor?
Na osnovu prethodne analize, samo kada čvor na koji pokazuje p u toku poll-a postane sentinel čvor (preko h.lazySetNext u metodi updateHead).
Kada nit A proverava p==q, nit B je već završila poll, čime je čvor p pretvorila u sentinel čvor i head čvor se već promenio, pa je potrebno krenuti od restartFromHead da bi se koristio najnoviji head.
Zamislimo još jednu situaciju. Ako je tekući red prazan, nit A radi poll, istovremeno nit B radi offer, a zatim nit A ponovo radi poll — da li će tada nit A vratiti null ili onaj čvor koji je nit B upravo umetnula? Hajde da napišemo demo:
public static void main(String[] args) {
Thread thread1 = new Thread(() -> {
Integer value = queue.poll();
System.out.println(Thread.currentThread().getName() + " poll vrednost: " + value);
System.out.println("da li je queue trenutno prazan: " + queue.isEmpty());
});
thread1.start();
Thread thread2 = new Thread(() -> {
queue.offer(1);
});
thread2.start();
}Izlaz je:
Thread-0 poll vrednost: null
da li je queue trenutno prazan: falsethread1 prvo dolazi do reda 8 if ((q = p.next) == null); pošto je red prazan, if je true i ulazi u if blok; tada pauzirajmo thread1, a zatim nek thread2 uradi offer i umetne čvor sa vrednošću 1; thread2 završava. Kada thread1 nastavi, thread1 ne radi ponovni pokušaj, već nastavlja dalje i vraća null, iako red u međuvremenu, zahvaljujući thread2, ima novi čvor sa vrednošću 1.
Rezultat je da thread0 poll vraća null, a red nije prazan.
Prema tome, pri proveri da li je red prazan ne treba koristiti null povratnu vrednost poll-a, već metodu isEmpty.
Deo niti radi offer, deo niti radi poll
Pri analizi offer metoda ostavili smo pitanje — tumačenje reda 11 u offer metodu.
offer->poll->offer
Rekli smo da red 11 offer metoda if (p == q) može biti true samo ako je čvor p sentinel čvor; a kada nastaje sentinel čvor?
U analizi poll metode pronašli smo odgovor: kada je item polje head čvora null, traži se pravi head čvor; nakon umetanja čvora, ažurira se head, a stari head čvor se postavlja za sentinel čvor. Pretpostavimo da je početno stanje reda kao na slici ispod:
Prema tome, dok nit A radi offer, a nit B radi poll, može nastupiti sledeća situacija:
Nit A ima čvor tail čiji je next čvor Node1, pa će kroz q tražiti pravi kraj reda; kada dođe do if (p == q), nit B radi poll; za nit B head i p pokazuju na Node0; pošto je item polje čvora Node0 null, i ona ide napred da pronađe pravi head Node1; nakon što nit B završi poll, Node0 postaje sentinel čvor, što znači da se head reda promenio; stanje reda je kao na slici ispod.
Nit A izvršava proveru if (p == q) true, pa nastavlja sa p = (t != (t = tail)) ? t : head;; pošto se tail nije promenio, p se dodeljuje head i ponovo polazi od head-a da dovrši umetanje.
Strategija odloženog ažuriranja
Analizom offer i poll metoda vidimo da se tail i head ažuriraju sa zakašnjenjem; trenuci okidača za ažuriranje su:
Okidač ažuriranja tail: kada sledeći čvor čvora tail nije null, izvršava se operacija lociranja pravog kraja reda; nakon pronalaska kraja i završetka umetanja, ažurira se tail preko casTail; kada je sledeći čvor čvora tail null, samo se umetne čvor bez ažuriranja tail.
Okidač ažuriranja head: kada je item polje čvora head null, izvršava se operacija lociranja pravog head čvora; nakon pronalaska head-a i završetka brisanja, ažurira se head preko updateHead; kada item polje čvora head nije null, samo se obriše čvor bez ažuriranja head.
Inače, u izvornom kodu postoji komentar: hop two nodes at a time.
Ova strategija odloženog ažuriranja se zove HOPS (otprilike iz tog razloga — nagađamo); iz gornjih dijagrama stanja se vidi da se head i tail ažuriraju „na preskok", odnosno uvek je između njih tačno jedan čvor. Koja je namera ovakvog dizajna?
Ako tail uvek bude pravi kraj reda, kôd bi bio kraći i logika razumljivija.
Međutim, takav pristup ima manu — ako postoji mnogo enqueue operacija, svaka bi zahtevala CAS ažuriranje tail-a, što u zbiru predstavlja veliki gubitak performansi. Smanjenjem broja CAS ažuriranja se može značajno poboljšati efikasnost enqueue-a, pa majstor Doug Lea tail ažurira CAS-om tek na svaki drugi korak (kada razmak između tail i kraja reda iznosi 1).
Isto važi i za ažuriranje head; iako ovakav dizajn uvodi dodatnu operaciju lociranja krajnjeg čvora u petlji, performanse čitanja su daleko bolje od performansi pisanja, pa je dodatni gubitak na lociranje krajnjeg čvora relativno mali.
Primer upotrebe
public class ConcurrentLinkedQueueTest {
public static void main(String[] args) {
ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
queue.offer(1);
queue.offer(2);
queue.offer(3);
queue.offer(4);
queue.offer(5);
System.out.println("da li je queue trenutno prazan: " + queue.isEmpty());
System.out.poll();
System.out.println("da li je queue trenutno prazan: " + queue.isEmpty());
System.out.println("trenutna veličina queue: " + queue.size());
}
}Izlaz:
da li je queue trenutno prazan: false
da li je queue trenutno prazan: false
trenutna veličina queue: 4Kratak pregled
ConcurrentLinkedQueue je FIFO (First-In-First-Out) red; u pitanju je neograničeni red bezbedan za niti, zasnovan na povezanim čvorovima. Elementi tog reda slede principa FIFO — prvi ubačeni je head, poslednji ubačeni je tail. Red ne dozvoljava null elemente.
ConcurrentLinkedQueue koristi HOPS dizajn — head i tail se ažuriraju sa zakašnjenjem; glavni cilj ovog dizajna je smanjenje konkurentnih nadmetanja u višenitnom okruženju i poboljšanje performansi.
Metod offer klase ConcurrentLinkedQueue služi za umetanje elementa na kraj reda. Ako dodavanje uspe, vraća true.
Metod poll klase ConcurrentLinkedQueue dohvata i uklanja head element reda. Ako je red prazan, vraća null.
Metod isEmpty klase ConcurrentLinkedQueue proverava da li je red prazan.
Metod size klase ConcurrentLinkedQueue vraća veličinu reda.
Autor: Chenmo Wang Er. Deo sadržaja potiče iz GitHub repozitorijuma CL0610 https://github.com/CL0610/Java-concurrency.