Implementacja synchronizowanej warstwy zasilania

operator wstawia krotki do strumienia. Podobnie tylko jeden operator pobiera krotki ze strumienia. Oznacza to, że kolejka działa w konfiguracji 1 producent – 1 konsument. Kolejka FIFO obsługująca krotki z flagą URGENT działa w konfiguracji wielu producentów - jeden konsument, ponieważ wstawiane są do niej krotki z n wejść.

W [63,85,26,62,94] przestawiono algorytm oraz dowód poprawności szybkiej kolejki FIFO działającej w konfiguracji wielu producentów - wielu konsumentów. Jej zaletą jest brak operacji lock/unlock, korzysta ona z operacji atomowych i operacji typu CAS (Compare-And-Swap). Powyższy algorytm użyto

do implementacji kolejki zawierającej krotki z flagą URGENT. Strumień krotek realizuje funkcję kolejki FIFO o własności: 1 producent – 1 konsument, dlatego wprowadzono dodatkowe uproszczenia algorytmu.

head tail

null next

Rys. 4.11. Kolejka z elementem wartownikiem na początku listy

Na rys. 4.11 przedstawiono przykładową kolejkę. Dodanie nowego elementu

new składa się z podoperacji: ustawienia new.next = null, odczytu ostatniego

elementu t przy użyciu zmiennej tail, ustawienia wartości pola t.next = new, aktualizacji zmiennej tail = new. Odczyt elementu składa się z podoperacji: odczyt elementu t wskazywanego przez zmienną head, odczyt pola t.next i zapisanie go do zmiennej tmp, aktualizacji zmiennej head = tmp, przekazanie na wyjście elementu

tmp. Zauważmy, że przez cały czas pobytu elementu w kolejce zachodzą następujące

własności. Operacja wstawiania zmienia tylko wartość zmiennych tail oraz next, z kolei operacja odczytu zmienia tylko wartość zmiennej head. Efekt ten uzyskujemy dzięki wprowadzeniu dodatkowego elementu na początku kolejki. Jeżeli przedstawiony algorytm działa w modelu 1 producent – 1 konsument, nie ma wtedy konieczności synchronizacji operacji wstawiania oraz odczytu. Implementacja powyższego algorytmu w języku Java wymaga dodatkowo uwzględnienia modelu pamięci i specyfikacji wątków w JVM [84,61]. Zgodnie z tą specyfikacją instrukcja

synchronized nie tylko zakłada oraz zdejmuje blokadę na obiekcie monitora,

ale również zapewnia aktualność odczytanych wartości zmiennych. Przyjmijmy, że wątek A zmienił wartość zmiennej x, następnie wątek B odczytał wartość zmiennej x. Otrzymane wartości mogą być różne, jeżeli dostęp do zmiennej x nie był objęty instrukcją synchronized. Przyczyna takiej sytuacji tkwi w tym, że procesor w celu przyspieszenia operacji IO korzysta z mechanizmu cache. Jeżeli wątek B działa na innym rdzeniu procesora niż wątek A, zmiany zawartości zmiennej wymagają dodatkowej synchronizacji. Dokument JSR-133 tłumaczy zasady widoczności zmian zmiennych volatile oraz final w modelu pamięci JVM. Zmienna final raz zainicjalizowana pozostaje niezmienna, dodatkowo każdy wątek może bezpiecznie odczytać jej zawartość bez potrzeby synchronizacji. Jeżeli zmienna final jest typu obiektowego, JVM gwarantuje, że z chwilą jej udostępnienia w systemie, wskazuje ona na obiekt w pełni zainicjalizowany obiekt. Wyobraźmy sobie, że nowo utworzony

obiekt zapisujemy do zwykłej zmiennej obiektowej. Specyfikacja JVM zezwala na podział tej operacji na kilka faz, w których optymalizator wpierw tworzy surowy obiekt, następnie go przypisuje do zmiennej, na koniec go inicjalizuje. Powyższy scenariusz tłumaczy, dlaczego spójny odczyt zmiennej obiektowej przez kilka wątków nie jest gwarantowany, nawet jeżeli jest ona ustawiona jednokrotnie wewnątrz konstruktora klasy. Specyfikacja modelu pamięci JVM gwarantuje, że zapis oraz odczyt zmiennych typu volatile jest atomowy oraz zmiana wartości zmiennej

volatile jest natychmiast widoczna przez inne wątki. Podsumowując, zmienne typu volatile oraz final zapewniają odczyt spójnych wartości bez konieczności użycia

instrukcji synchronized. Dodatkowo, wprowadzane zmiany wartości zmiennych są natychmiast widoczne przez inne wątki bez konieczności korzystania z instrukcji

synchronized. Specyfikacja modelu pamięci JVM wskazuje także, że zapis/odczyt

zmiennych final jest szybszy aniżeli volatile. Po uwzględnieniu dokumentu JSR-133 algorytm kolejki FIFO pracującej w konfiguracji 1 producent – 1 konsument wymaga następujących modyfikacji. Zmienne tail, head oraz next należy zadeklarować jako zmienne typu volatile. Zmienną item należy zadeklarować jako final. Poniżej zamieszczono implementację algorytmu w języku Java.

Algorytm 5.2. Kolejka FIFO 1 producent – 1 konsument.

class TupleNode { final TupleI item;

volatile TupleNode next;

public TupleNode(TupleI aItem, TupleNode aNext) { item = aItem;

next = aNext; }

}

volatile TupleNode head; volatile TupleNode tail; public PipeTupleList() {

head = tail = new TupleNode(null,null); }

public final void pushNode(TupleI newNode) { TupleNode node = new TupleNode(newNode, null); tail.next = node;

tail = node; }

TupleNode tmp;

if((tmp = head) == tail) return null;

head = tmp.next; tmp.next = null; return head; }

W celu osiągnięcia dodatkowej optymalizacji czasowej, implementując warstwę zasilania wprowadzono wersje dedykowane ze względu na liczbę strumieni zasilających. Wyróżniamy warstwę zasilania z jednym lub wieloma strumieniami wejściowymi. Dla osiągnięcia większej czytelności, w przedstawionej poniżej analizie pomijamy obsługę krotek z ustawioną flagą URGENT. Implementacja warstwy zasilania z wieloma strumieniami korzysta z następującej obserwacji. Przyjmijmy, że zmienna minSize jest równa minimalnej długości wszystkich strumieni wejściowych. Zauważmy, że jeżeli wartość minSize = 0, wtedy zachodzi punkt 2) algorytmu 5.2. Jeżeli minSize > 0, wtedy zachodzi punkt 1) algorytmu 5.2. Sygnał niedostępności nie jest generowany, jeżeli funkcja getNextTuple jest wywołana w chwili minSize > 1. Sygnał niedostępności może zostać wygenerowany tylko, gdy minSize = 1 lub minSize = 0. Sygnał dostępności jest generowany tylko, gdy do strumienia wstawiamy pierwszą krotkę. Warstwa zasilania obsługuje kilka wątków, dlatego należy zaimplementować synchronizację gwarantującą, że sygnał niedostępności będzie poprzedzany sygnałem dostępności. Utworzony mechanizm synchronizacji opisano poniżej, korzysta on z obserwacji wartości zmiennej minSize oraz rozmiaru strumienia wejściowego. Operacja wstawiania krotki do strumienia jest synchronizowana przez monitor m, gdy strumień jest pusty. Operacja odczytu jest synchronizowana przez monitor m, gdy minSize = 0 lub minSize = 1. Powyższy algorytm nie wymaga dostępu do dokładnej wartości minSize tylko do informacji kiedy minSize > 1. Własność ta pozwala sprowadzić minSize do trzech wartości. Jeżeli

minSize = 2 oznacza to, że minimalny rozmiar strumieni jest > 1. Poniżej przestawiono

algorytm synchronizacji funkcji getNextTuple.

Algorytm 5.3. Synchronizacja funkcji getNextTuple.

if(minSize.get() > 1) {

pobranie następnej krotki do przetworzenia punkt 2) algorytm 5.1

if(po wyjęciu krotki strumień ma rozmiar 1) minSize = 1 else { synchronized(m) { switch(minSize.get()) { case 0:

pobranie następnej krotki do przetworzenia punkt 1) algorytm 5.1

aktualizacja minSize

if(brak krotki do przetworzenia w następnym kroku iteracji)

wysłanie sygnału niedostępności break;

case 1:

pobranie następnej krotki do przetworzenia punkt 2) algorytm 5.1

aktualizacja minSize

if(brak krotki do przetworzenia w następnym kroku iteracji)

wysłanie sygnału niedostępności break;

default:

pobranie następnej krotki do przetworzenia punkt 2) algorytm 1

if(po wyjęciu krotki strumień ma rozmiar 1) minSize = 1 break; } } } }

Jeżeli minimalna długość strumieni jest >1, wtedy funkcja getNextTuple nie korzysta z monitora m. Zyskujemy dzięki temu dodatkowe przyspieszenie algorytmu. Aby mechanizm synchronizacji obsługiwał krotki z ustawioną flagą URGENT, należy zmodyfikować minSize = minimalny_rozmiar_strumieni_wejściowych + rozmiar_kolejki_z_krotkami_priorytetowymi. Zmienna minSize jest wtedy

aktualizo-wana zarówno przez wątki wstawiające krotki jak i wątek odczytujący, dlatego wymagana jest synchronizacja odczytu/zapisu, do czego użyto poleceń typu CAS. W przypadku pojedynczego strumienia wejściowego wybór następnej krotki sprowadza się do sprowadzenia zawartości: 1) kolejki z krotkami priorytetowymi, 2) bufora strumienia.