Producent / konsument
Leszek Ciesielski
Mam wiele wątków producenta, jeden wątek konsumenta. Konsument
powinien obsłużyć wszystkich producentów, a po ich zakończeniu -
zakończyć też swój wątek. Napisałem następujący kod i nie jestem
pewien, czy jest poprawny:
import java.util.LinkedList;
import java.util.Queue;
public class MultithreadedQueue {
private Queue<String> queue = new LinkedList<String>();
private boolean shutdown = false;
public synchronized void put(String o) {
queue.add(o);
notify();
}
public synchronized String get() throws InterruptedException {
while (true) {
String result = queue.poll();
if (result != null)
return result;
else {
if (shutdown)
throw new InterruptedException();
else {
wait();
}
}
}
}
public synchronized void prepareForShutdown() {
shutdown = true;
notify();
}
}
O ile się nie mylę, dla 'normalnej sytuacji' ten kod powinien działać.
Dodatkowy haczyk polega na tym, że put() jest wołane przez classloader
(a dokładniej ClassFileTransformer), więc jest możliwe, że metoda do
wykonania będzie potrzebować załadowania dodatkowej klasy, a ta z
kolei zostanie zablokowana... I z tym już nie wiem jak sobie
poradzić. (tak, ten kod czasem się zakleszcza, a to jest najbardziej
sensowne wytłumaczenie, jakie mi przychodzi do głowy).
Czy ktoś może mi wskazać błędy w tej implementacji bufora?
--
MS-DOS user since 5.0
Windows user since 3.11
Linux user since kernel 2.4
Novell Netware user since 2.2
WARCRAFT user since 1.0
Marcin Kosmalski
List list = Collections.synchronizedList(new LinkedList(...));
W Twoim li cie datowanym 14 listopada 2007 (22:08:15) moæna przeczytaę:
> import java.util.LinkedList;
> import java.util.Queue;
--
Pozdrowienia,
Marcin Kosmalski
Dawid Weiss
> Mam wiele wątków producenta, jeden wątek konsumenta. Konsument
> powinien obsłużyć wszystkich producentów, a po ich zakończeniu -
> zakończyć też swój wątek.
Co to znaczy "po ich zakonczeniu"? Tzn. kiedy wiadomo ze sie skonczyli
producenci? Gdy w kolejce nic nie ma? Musialbys dokladnie powiedziec o co
chodzi, bo opis scenariusza jest jakis dziwny.
> import java.util.LinkedList;
Nie uzywaj LinkedList, ArrayList prawie na pewno bedzie szybszy.
> public synchronized String get() throws InterruptedException {
> while (true) {
> String result = queue.poll();
> if (result != null)
> return result;
> else {
> if (shutdown)
> throw new InterruptedException();
> else {
> wait();
> }
> }
> }
> }
> O ile się nie mylę, dla 'normalnej sytuacji' ten kod powinien działać.
> Dodatkowy haczyk polega na tym, że put() jest wołane przez classloader
> (a dokładniej ClassFileTransformer), więc jest możliwe, że metoda do
> wykonania będzie potrzebować załadowania dodatkowej klasy, a ta z
> kolei zostanie zablokowana... I z tym już nie wiem jak sobie
> poradzić. (tak, ten kod czasem się zakleszcza, a to jest najbardziej
> sensowne wytłumaczenie, jakie mi przychodzi do głowy).
1) Ja bym tego nie implementowal sam -- java.util.concurrent.BlockingQueue robi
dokladnie to, co chcesz.
2) Jak masz deadlock, do ctrl-break albo wyslanie sygnalu (chyba sighup, nie
pamietam dokladnie) zrzuca stan watkow i monitory na ktorych oczekuja. Mozna
dokladnie powiedziec co gdzie stoi.
3) W tym kodzie raczej zakleszczenia nie ma. Sytuacja, ktora opisujesz nie jest
mozliwa chyba, ze ten kod, ktory podales to nie jest ten sam kod, o ktorym mowisz.
> Czy ktoś może mi wskazać błędy w tej implementacji bufora?
W tej raczej bledu nie ma. Natomiast tak sie tego nie implementuje -- sa klasy w
java.util.concurrent, ktore robia to lepiej.
D.
Leszek Ciesielski
>
> Witam!
>
> List list = Collections.synchronizedList(new LinkedList(...));
Byłoby fajnie, ale get() ma dwa warianty zakończenia : coś pojawi się
w kolejce, albo dostanie sygnał "zakończ wątek". Z drugiej strony,
"zakończ wątek" mogę zakomunikować poprzez wstawienie specjalnego
znacznika do kolejki...
>
> W Twoim li cie datowanym 14 listopada 2007 (22:08:15) można przeczytać:
>
> > Problem:
> > Mam wiele wŕtków producenta, jeden wŕtek konsumenta. Konsument
> > powinien obsůuýyă wszystkich producentów, a po ich zakończeniu -
> > zakończyă teý swój wŕtek. Napisaůem nastćpujŕcy kod i nie jestem
> > pewien, czy jest poprawny:
>
> > import java.util.LinkedList;
> > import java.util.Queue;
>
> > public class MultithreadedQueue {
> > private Queue<String> queue = new LinkedList<String>();
>
> > private boolean shutdown = false;
>
> > public synchronized void put(String o) {
> > queue.add(o);
> > notify();
> > }
>
> > public synchronized String get() throws InterruptedException {
> > while (true) {
> > String result = queue.poll();
> > if (result != null)
> > return result;
> > else {
> > if (shutdown)
> > throw new InterruptedException();
> > else {
> > wait();
> > }
> > }
> > }
> > }
>
> > public synchronized void prepareForShutdown() {
> > shutdown = true;
> > notify();
> > }
> > }
>
> > O ile sić nie mylć, dla 'normalnej sytuacji' ten kod powinien dziaůaă.
> > Dodatkowy haczyk polega na tym, ýe put() jest woůane przez classloader
> > (a dokůadniej ClassFileTransformer), wićc jest moýliwe, ýe metoda do
> > wykonania bćdzie potrzebowaă zaůadowania dodatkowej klasy, a ta z
> > kolei zostanie zablokowana... I z tym juý nie wiem jak sobie
> > poradziă. (tak, ten kod czasem sić zakleszcza, a to jest najbardziej
> > sensowne wytůumaczenie, jakie mi przychodzi do gůowy).
>
> > Czy ktoú moýe mi wskazaă bůćdy w tej implementacji bufora?
>
>
>
>
> --
> Pozdrowienia,
> Marcin Kosmalski
>
>
> >
>
--
Tomasz Nowak
Co zmieni synchronizedList, jeśli odwołujemy się do obiektu jedynie z
już zsynchronizowanych metod?
Dnia 14-11-2007, śro o godzinie 22:29 +0100, Marcin Kosmalski pisze:
> Witam!
>
> List list = Collections.synchronizedList(new LinkedList(...));
>
> W Twoim li cie datowanym 14 listopada 2007 (22:08:15) można przeczytać:
>
> > Problem:
> > Mam wiele wŕtków producenta, jeden wŕtek konsumenta. Konsument
> > powinien obsůuýyă wszystkich producentów, a po ich zakończeniu -
> > zakończyă teý swój wŕtek. Napisaůem nastćpujŕcy kod i nie jestem
> > pewien, czy jest poprawny:
>
> > import java.util.LinkedList;
> > import java.util.Queue;
>
> > public class MultithreadedQueue {
> > private Queue<String> queue = new LinkedList<String>();
>
> > private boolean shutdown = false;
>
> > public synchronized void put(String o) {
> > queue.add(o);
> > notify();
> > }
>
> > public synchronized String get() throws InterruptedException {
> > while (true) {
> > String result = queue.poll();
> > if (result != null)
> > return result;
> > else {
> > if (shutdown)
> > throw new InterruptedException();
> > else {
> > wait();
> > }
> > }
> > }
> > }
>
> > public synchronized void prepareForShutdown() {
> > shutdown = true;
> > notify();
> > }
> > }
>
Leszek Ciesielski
>
> Witam,
> Co zmieni synchronizedList, jeśli odwołujemy się do obiektu jedynie z
> już zsynchronizowanych metod?
Marcinowi chyba chodziło o zastąpienie całego mojego kodu jedną linijką ;-)
--
Dawid Weiss
> Byłoby fajnie, ale get() ma dwa warianty zakończenia : coś pojawi się
> w kolejce, albo dostanie sygnał "zakończ wątek". Z drugiej strony,
> "zakończ wątek" mogę zakomunikować poprzez wstawienie specjalnego
> znacznika do kolejki...
Ten scenariusz uzycia wydaje mi sie byc naruszeniem obiektowosci -- kontrakt na
kolejke jest prosty i elegancki, to co probujesz zrobic to troche wymuszenie...
Zwroc rowniez uwage, ze rzucanie wyjatku (nawet InterruptedException) aby
zasygnalizowac w gruncie rzeczy _normalny_ i przewidywany stan aplikacji to code
smell. Sugeruje zrobic cos takiego na poziomie klienta (queue nie musi byc sama
w sobie synchronizowana, wystarczy, ze napiszesz w dokumentacji, ze musi byc
kontrolowana zewnetrznie. Wtedy z watku "klienta":
while (true) {
final Consumable c;
synchronized (superqueue) {
if (c.isEmpty() || superqueue.shouldClientsTerminate()) {
break;
}
c = superqueue.remove();
}
// outside of the monitor;
consume(c);
}
w producencie:
Consumable c = produce();
synchronized (superqueue) {
superqueue.put(c);
}
a w implementacji kolejki, jesli juz koniecznie chcesz, wtedy mozesz zrobic
odpowiednie sprawdzenie stanu i rzucanie wyjatku (bo pobranie elementu po
momencie, gdy jest flaga sygnalizuje naruszenie kontraktu):
SuperQueue<T> extends Queue<T> {
@Override
public T remove() {
if (shouldClientsTerminate()) {
throw new IllegalStateException("remove illegal when clients should terminate.");
}
return super.remove();
}
}
To powiedziawszy bez scenariusza ogolnego uzycia tego w aplikacji trudno
powiedziec czy to najlepsze wyjscie.
D.