Ref to zaawansowany typ referencyjny przygotowany do obsługi transakcji. Możemy dzięki niemu wyrażać częste i jednoczesne zmiany stanów wielu współdzielonych tożsamości w synchroniczny sposób. Wykorzystywany jest tam, gdzie kilka zmieniających się w czasie wartości (wskazywanych przez referencje) zależy od siebie, a modyfikacja ich wszystkich powinna być jednoczesna (np. przekazywanie środków między rachunkami bankowymi). Do obsługi Refów wykorzystywana jest programowa pamięć transakcyjna (STM).

Współbieżność

Wykonywanie współbieżne (ang. concurrent) to cecha systemów i programów komputerowych, w których te same obliczenia dokonywane są jednocześnie (w tym samym czasie) przez więcej niż jeden komponent, przy opcjonalnej komunikacji między komponentami.

Więcej o współbieżności w Clojure można przeczytać w odcinku 15.

Referencje transakcyjne

Referencja transakcyjna (ang. transactional reference) to mechanizm, dzięki któremu można wyrażać częste i koordynowane i zmiany stanów wielu współdzielonych tożsamości w tym samym czasie. Brzmi to nieco enigmatycznie, lecz w praktyce odpowiedzialny jest za to typ danych o nazwie Ref, którego instancje podobnie do Atomów mogą wskazywać na pewne wartości bieżące i umożliwiają podmianę tych wskazań, ale nie tylko w sposób atomowy (cała operacja udaje się bądź jest wycofywana), lecz także synchroniczny.

Aktualizacja odniesień do wartości w obiektach typu Ref przez pewien czas blokuje realizowanie wątku programu, w którym została zapoczątkowana. Dzięki odpowiednim optymalizacjom czas ten jest krótki, a większość kalkulacji dokonywana jest poza okresem trwania blokady, jeżeli jest taka możliwość.

Referencje transakcyjne wykorzystywane są do obsługi sytuacji, w których należy zmienić stan więcej niż jednego obiektu referencyjnego w tym samym czasie, a zmiana ta powinna być atomowa: albo wszystkie odniesienia do wartości zostaną uaktualnione, albo żadne. Przykładem może być tu przelew między rachunkami bankowymi, gdzie z jednego musi zostać odjęta pewna kwota, a na drugim musi się ona pojawić. Niedopuszczalna byłaby sytuacja, w której operacja ta zostałaby zakończona połowicznie (po odjęciu kwoty z rachunku źródłowego) lub powiodłaby się tylko jej druga część (zasilenie rachunku docelowego).

Łatwo skojarzyć, że z podobnym mechanizmem zapewniania spójności danych, na których prowadzone są obliczenia, mamy do czynienia w bazach danych. Można więc wyobrażać sobie, że dochodzi tu do interakcji z pamięcią operacyjną w sposób przypominający wykonywanie bazodanowych transakcji.

Refy (ang. Refs) obsługiwane są z wykorzystaniem programowej pamięci transakcyjnej (ang. Software Transactional Memory, skr. STM) systemu gospodarza. Dzięki użyciu STM korzystanie z Refów nie powinno prowadzić do powstawania tzw. zakleszczeń (ang. deadlocks) bądź sytuacji wyścigów (ang. race conditions).

Korzystanie z Refów polega na utworzeniu odpowiednich obiektów, a następnie aktualizowaniu ich powiązań z wartościami bieżącymi z możliwością grupowania takich operacji w jeden blok, który będzie wyrażał zmiany wartości tych tożsamości, które logicznie od siebie zależą.

Wartością współdzieloną (lub wartością bieżącą) Refa nazwiemy taką wartość, na którą wskazuje dany obiekt typu Ref, a odwołując się do niego w programie, możemy ją uzyskiwać. Ten ostatni proces nazywamy dereferencją.

Zazwyczaj obiekt typu Ref będzie w programie identyfikowany symboliczną nazwą odnoszącej się do niego zmiennej globalnej lub powiązania leksykalnego.

Transakcje

W przeciwieństwie do Varów, Agentów czy Atomów, modyfikacje powiązań obiektów typu Ref muszą być dokonywane w obrębie tak zwanych transakcji (ang. transactions). Odczyty zwolnione są z tego warunku, ale ryzykujemy niespójność wartości bieżących lub pobieranie ich zdezaktualizowanych wersji.

Transakcja jest zbiorem operacji, które powinny być traktowane jako niepodzielna całość: albo uda się wykonać je wszystkie i zaktualizować wiele powiązań referencyjnych, albo żadne zmiany nie zostaną wprowadzone (transakcja zostanie wycofana). Dzięki temu gwarantowane jest, że operacje na Refach będą:

• atomowe – zmiany każdego Refa są niepodzielne;

• spójne – każdą aktualizowaną wartość referencji można poddać testowi poprawności z użyciem tzw. walidatorów lub dodatkowych funkcji umieszczonych w transakcji;

• izolowane – powiązania obiektów typu Ref z wartościami nie zostaną nigdy w obrębie transakcji zmienione przez inną transakcję – są wersjami z momentu jej rozpoczęcia, a nie wartościami bieżącymi (z pewnymi dozwolonymi wyjątkami).

Wywołania funkcji operujących na powiązaniach obiektów typu Ref z wartościami w ramach konkretnej transakcji powinny być podane w S-wyrażeniach przekazywanych jako argumenty makra dosync:

1(let [x (ref 0)      ; referencja x (wartość początkowa 0)
2      y (ref 0)]     ; referencja y (wartość początkowa 0)
3  (dosync            ; transakcja:
4    (ensure x)       ;  · zabezpieczamy x przed równoległymi zmianami
5    (alter x inc)    ;  · zwiększamy x
6    (alter y + @x))  ;  · zwiększamy y o wartość x
7  (println @x @y))   ; wyświetlamy wartości x oraz y
8
9; >> 1 1

Każda transakcja ma trzy główne etapy:

1. rozpoczęcie,
2. wykonanie,
3. zakończenie (zamknięcie).

Transakcja rozpoczyna się wraz z ewaluacją ciała makra dosync, a jej realizowanie polega na wyliczaniu nowych wartości bieżących obecnego w niej zestawu Refów.

Zamknięcie transakcji oznacza albo jej udane zatwierdzenie (ang. commit), albo wycofanie (ang. rollback), które w przypadku STM wiąże się z jej unicestwieniem (ang. kill).

Zatwierdzanie to nazwa procesu, w którym dochodzi do zastąpienia wartości bieżących Refów nowymi wartościami, obliczonymi w trakcie jej trwania.

Z kolei unicestwienie oznacza, że do aktualizacji nie dochodzi i zmiany są porzucane. Nie wyklucza to jednak ponownego zrealizowania transakcji, nawet automatycznego, co jest w STM częstą reakcją na konflikty.

Wewnętrznie, w implementacji STM z Clojure działającej na maszynie wirtualnej Javy, możemy wyróżnić następujące stany transakcji:

• działająca (RUNNING),
• zatwierdzana (COMMITTING),
• ponawiana (RETRY),
• unicestwiona (KILLED),
• zatwierdzona (COMMITTED).

Wielowersyjność i kolejkowanie

Zastosowana w Clojure implementacja programowej pamięci transakcyjnej bazuje na mechanizmie wielowersyjnej kontroli współbieżności (ang. multiversion concurrency control, skr. MCC lub MVCC) z izolowaniem migawek (ang. snapshot isolation) realizowanym przez adaptacyjne kolejkowanie historii zmian.

Używając analogii, możemy powiedzieć, że STM przypomina trochę system kontroli wersji Git, gdzie każdy wątek programu to osobny programista korzystający z repozytorium.

W takim wyimaginowanym scenariuszu mamy wiele gałęzi. Każdy wątek, zaczynając transakcję, tworzy wirtualną gałąź, aby pracować na osobnej wersji (odpowiednik git checkout), czyli odizolowanej migawce. W jej ramach jest w stanie swobodnie modyfikować wartości bieżące Refów, a inne wątki nie widzą tych zmian, a więc nie muszą na nic oczekiwać (brak blokowania na tym etapie).

Na końcu transakcji system próbuje włączyć zmienione w wątku obsługującym migawkę wartości Refów do jedynej słusznej wersji (odpowiednik operacji git commit), która zaraz potem musi być zsynchronizowana z oficjalną wersją międzywątkową (odpowiednik zdalnego repozytorium).

Przydaje się tu wspomniane kolejkowanie historii zmian. Przypomina ono zautomatyzowaną operację git rebase, gdzie system stara się złączyć zgromadzone w czasie zmiany w głównej gałęzi we właściwej kolejności. Jeżeli inny wątek równolegle dokonał zmian, których nie da się złączyć, wtedy istnieje możliwość, że nie zostaną one uwzględnione z powodu konfliktu i będzie musiał spróbować ponownie (odpowiednik odrzucenia automatycznego git merge, gdy główna gałąź w zdalnym repozytorium uległa zmianie).

Migawki

Migawka (ang. snapshot) jest zapamiętanym stanem wszystkich obiektów STM z danego punktu w czasie. Mechanizmy obsługi pamięci transakcyjnej korzystają z migawek, aby wykrywać zmiany i wytwarzać kopie Refów na użytek realizowanych transakcji. Dzięki temu w obrębie tych ostatnich będziemy mieli do czynienia z niezmieniającymi się, spójnymi zestawami danych.

Migawki są również wykorzystywane do utrzymywania tzw. historii zmian, która pozwala na dostęp do przeszłych wartości wybranych referencji transakcyjnych.

Wartości wewnątrztransakcyjne

Aktualizowanie powiązań Refów z wartościami w obrębie transakcji nie będzie polegało na bezpośrednim operowaniu na ich aktualnych wartościach bieżących, lecz na wersjach zapamiętywanych w chwilach rozpoczynania transakcji. Te momenty nazywamy punktami odczytu (ang. read points).

Modyfikacje powiązań Refów z wartościami w obrębie transakcji są widoczne tylko w niej, do momentu zatwierdzenia zmian. Wartości tych powiązań nazywamy wartościami w transakcji lub wartościami wewnątrztransakcyjnymi (ang. in-transaction-values).

Pierwsza wartość wewnątrztransakcyjna powstaje, gdy dokonywana jest aktualizacja odpowiadającego jej Refa. Jeżeli modyfikacji nie dokonano, wartość taka nie jest tworzona, a ewentualne odczyty Refa polegają na bezpośrednim sięgnięciu do wartości z migawki (z punktu odczytu lub z historii zmian).

Propagacja wartości zaktualizowanych Refów wewnątrztransakcyjnych do odpowiadających im współdzielonych wartości bieżących nastąpi wtedy, gdy transakcja pomyślnie się zakończy i zostanie zatwierdzona (stan COMMITTED). Wyjątkiem będzie użycie funkcji commute, która zmieniając stan Refa, operuje bezpośrednio na jego najbardziej aktualnej wartości współdzielonej i nie korzysta z wartości wewnątrztransakcyjnych, chociaż stwarza je i aktualizuje (w celu kompatybilności z funkcjami, które operują na tych wartościach). Modyfikacje zlecone z użyciem tej funkcji również oczekują na zatwierdzenie w odpowiedniej fazie transakcji.

Refy, które są wyłącznie odczytywane w ramach transakcji (nie dochodzi do aktualizowania ich powiązań), również korzystają z kopii (migawek) całej pamięci transakcyjnej. Opcjonalnie można więc użyć dosync nie tylko do aktualizowania powiązań Refów z nowymi wartościami, lecz także po to, aby zachować spójność relacji między logicznie zależnymi Refami, których wartości bieżących używamy w obliczeniach.

Obsługa błędów

Jeżeli podczas realizowania transakcji wystąpi jakiś krytyczny błąd, transakcja zostanie unicestwiona (stan KILLED), a żaden współdzielony obiekt typu Ref nie zostanie przez nią zaktualizowany.

Jeżeli podczas realizowania transakcji pojawi się wyjątek spowodowany brakiem akceptacji nowej wartości Refa przez przypisany do niego walidator lub funkcja walidująca zwróci wartość false, transakcja zostanie przerwana, a więc również unicestwiona (stan KILLED).

Obsługa konfliktów

Jeżeli w trakcie wykonywania transakcji nie wystąpił błąd, lecz doszło do równoległej zmiany (w ramach innej, już zatwierdzonej transakcji) powiązania któregoś obiektów typu Ref, możemy mieć do czynienia z tzw. usterką (ang. fault).

Do usterki dochodzi wtedy, gdy:

• w bieżącej transakcji wystąpi próba odczytu wartości obiektu typu Ref, ale nie można pobrać odpowiadającej jej wartości wewnątrztransakcyjnej (np. jeszcze trwa jej obliczanie), zaś wartość zapamiętana w punkcie odczytu lub w którymś z punktów historii zmian nie jest aktualną wartością współdzieloną, ponieważ inna transakcja już zdążyła zatwierdzić nową, zanim bieżąca się rozpoczęła;

• w bieżącej transakcji wystąpi próba odczytu wartości obiektu typu Ref, lecz inna transakcja wygra z nią tzw. spór;

• w bieżącej transakcji wystąpi próba aktualizacji wartości obiektu typu Ref (z użyciem ref-set lub alter), lecz wystąpi jedna z trzech sytuacji:

  • nie można zablokować możliwości zapisu do odniesienia (bo blokada zapisu lub odczytu należy do innego wątku),

  • w międzyczasie już zmieniono wartość współdzieloną,

  • inna transakcja wytworzyła wartość wewnątrztransakcyjną danego Refa i bieżąca rozpoczęła z nią tzw. spór, lecz przegrała go;

• bieżąca transakcja jest w fazie zatwierdzania zmian, ale inna uruchomiona transakcja dokonała zmiany wartości wewnątrztransakcyjnej aktualizowanego Refa, a tzw. spór z nią został przegrany;

• w bieżącej transakcji dokonano próby zmiany wartości (z użyciem ref-set, alter lub commute) lub użyto ensure w stosunku do danego Refa, lecz inna transakcja rozpoczęła i wygrała z nią tzw. spór.

Warto pamiętać, że odczytywanie wartości bieżących Refów to nie tylko ich wyrażone wprost przez programistę dereferencje (z użyciem funkcji deref czy makra @), ale również aktualizacje, z wykorzystaniem alter, które wymagają wcześniejszego poznania aktualnych wartości współdzielonych lub wartości wewnątrztransakcyjnych.

Reakcją na usterkę jest ponowienie transakcji (bieżącej lub pozostającej z nią w konflikcie). Wyrażenia przekazane do makra dosync będą w takim wypadku przeliczone ponownie z przyjęciem nowego punktu odczytu. W ten sposób realizowana jest synchroniczność aktualizowania referencji transakcyjnych.

Realizowanie transakcji będzie ponawiane do czasu, aż uda się zatwierdzić ją bez konfliktów, chyba że przekroczona zostanie maksymalna liczba prób określona stałą wartością równą 10 000. Jeżeli tak się stanie, zgłoszony zostanie wyjątek RetryEx.

Spory

Gdy wykryty jest potencjalny konflikt (dwie transakcje operują na tych samych obiektach typu Ref i przy próbie zatwierdzania pojawiłby się problem), jedna z transakcji może rozpocząć tzw. spór (ang. barge), aby udowodnić, że druga powinna zostać przerwana i ponowiona. Aby tak się stało, spełnione muszą być następujące warunki:

• bieżąca transakcja rozpoczęła się wcześniej, niż druga;

• bieżąca transakcja musi być realizowana co najmniej przez czas określony stałą BARGE_WAIT_NANOS (domyślnie 1/100 sekundy);

• druga transakcja jest realizowana (stan RUNNING) i jeszcze nie zdążyła rozpocząć procesu zatwierdzania zmian.

Wygranie sporu przez bieżącą transakcję oznacza, że jest ona kontynuowana, a konfliktująca z nią zostaje przerwana (stan KILLED). Do ponowienia drugiej transakcji (stan RETRY) dojdzie, gdy wejdzie ona w fazę zatwierdzania zmian – pierwszym warunkiem jest wtedy sprawdzenie, czy nie przegrała sporu z inną transakcją.

Przegranie sporu przez bieżącą transakcję sprawia, że jej wykonywanie jest wstrzymywane na przynajmniej 1/100 sekundy (wspomniana stała BARGE_WAIT_NANOS) w oczekiwaniu na zakończenie drugiej transakcji.

Aby wykrywać, która transakcja zdarzyła się wcześniej, STM korzysta z globalnego licznika zwiększanego przy każdym rozpoczęciu transakcji. Wartość tego licznika jest następnie używana podczas obsługi konkretnej transakcji, aby pamiętać jej czas rozpoczęcia i czas zatwierdzenia.

Dzięki rozpoczynaniu procesu sporu transakcja, która aktualizuje wartości bieżące obiektów referencyjnych, może upewnić się, że to ona kontroluje ich globalne, współdzielone stany i przygotowuje się do wejścia w fazę COMMITTING.

W praktyce powyższy sposób reakcji na konflikty zostanie zastosowany z uwzględnieniem wszystkich Refów, w odniesieniu do których w transakcji wywołano funkcję ref-set, alter lub ensure. Ta ostatnia nie aktualizuje co prawda powiązania z wartością, ale zapewnia, że obiekt typu Ref będzie chroniony przed modyfikacjami, które mogą być efektem równoległego realizowania innych transakcji.

Z uwagi na to, że transakcje mogą być automatycznie ponawiane, należy w nich unikać funkcji, które mają efekty uboczne (np. korzystają z podsystemu wejścia/wyjścia).

Komutacja wartości bieżącej

Ciekawy mechanizm aktualizowania referencji wprowadza wspomniana już funkcja commute. Z jej pomocą można zaktualizować powiązanie obiektu typu Ref, jednak w przypadku wykrycia równoległej aktualizacji nie będzie konieczne ponawianie transakcji. Dzieje się tak dlatego, że podczas zatwierdzania Refów zmienianych z użyciem commute nie korzysta się z wartości wewnątrztransakcyjnej, lecz wywołuje funkcję modyfikującą na ostatniej wartości współdzielonej. Jeżeli więc inna transakcja zmieni powiązanie obiektu typu Ref w trakcie realizowania bieżącej transakcji, i tak dojdzie do aktualizacji odniesienia.

Dlaczego tak się dzieje? Operacje komutatywne to takie, dla których kolejność prowadzenia na nich obliczeń nie ma wpływu na wynik. Znamy na przykład operator dodawania liczb całkowitych, który jest funkcją sumującą argumenty. Niezależnie od uporządkowania tych argumentów, rezultat dodawania będzie zawsze taki sam. Jeżeli więc w naszym programie mamy takie Refy, których aktualizacja wartości bieżących nie zależy od kolejności zmian w czasie (np. jakiś rodzaj sumatora bądź licznika), możemy zyskać na wydajności, korzystając z komutatywnego trybu wprowadzania zmian.

Podczas zatwierdzania zmiany spowodowanej użyciem commute wywołana zostanie funkcja przeliczająca, która została do niej przekazana. Jako argument tej ostatniej przekazana będzie dopiero co zaktualizowana wartość bieżąca Refa (ustawiona przez wcześniej zakończoną transakcję). Przypomina to trochę działanie funkcji swap! w odniesieniu do Atomów.

Zmiany będą wciąż atomowe i synchroniczne, chociaż bez zachowania kolejności operacji. Może na przykład zdarzyć się, że jedna z kilku równolegle realizowanych transakcji zostanie wykonana wcześniej, niż przewidywano, więc odniesienia, których zmiana jest efektem użycia commute, zostaną zaktualizowane w różnej kolejności. Należy mieć to na uwadze, korzystając z tej funkcji.

Historia zmian

Obsługa konfliktów i utrzymywanie spójności danych na poziomie STM możliwe jest dzięki użyciu migawek, a dokładniej porównywaniu stanu współdzielonych referencji zapisanych podczas rozpoczęcia transakcji ze stanami w momencie jej zatwierdzania. Przez spójność rozumiemy tu zestaw zależnych od siebie logicznie wartości bieżących, który jest niezmienny w pewnym okresie czasu, więc można bezpiecznie przeprowadzać na nim operacje.

Powyższe podejście zadziała w prostych przypadkach, jednak możemy mieć kłopoty, jeżeli zmiany będą bardzo częste. Wyobraźmy sobie dwie referencje transakcyjne nazwane x oraz y, których wartościami początkowymi są odpowiednio 1 i 2. Następnie załóżmy, że wykonujemy transakcję, wewnątrz której obliczamy sumę wartości wskazywanych przez Refy. Jeżeli będzie to jedna transakcja, rezultatem będzie 3:

transakcja              x   y
------------------------------------------
1.  sumowanie [start]   1   2
1.  sumowanie [stop]    1   2 (x + y = 3)

Komplikacje zaczną się, gdy równolegle rozpoczęta zostanie kolejna transakcja, zmieniająca wartość bieżącą powiązaną z x:

transakcja              x   y
------------------------------------------
1.  sumowanie [start]   1   2
2.  zmiana x  [start]   1   2
2.  zmiana x  [stop]    2   2
1.  sumowanie [stop]    1   2 (konflikt x)
1'. sumowanie [start]   2   2
1'. sumowanie [stop]    2   2

Ponieważ współdzielona wartość bieżąca x zmieniła się w trakcie wykonywania transakcji nr 1 (sumowanie), więc transakcja ta została wycofana i ponowiona (nr 1’), aby skorzystać z wartości zaktualizowanej w transakcji nr 2.

 1(def x (ref 1))
 2(def y (ref 2))
 3
 4(future
 5  (dosync
 6    (Thread/sleep 350)
 7    (println "2. zapis  [start] x =" @x "+ 1")
 8    (alter x inc)
 9    (println "2. zapis   [stop] x =" @x)))
10
11(dosync
12  (println "1. odczyt [start]")
13  (Thread/sleep 500)
14  (println "1. odczyt  [stop] x =" @x))

Zobaczmy, co się stanie, gdy zmiany x będą częstsze:

transakcja              x   y
------------------------------------------
1.  sumowanie [start]   1   2
2.  zmiana x  [start]   1   2
2.  zmiana x  [stop]    2   2
1.  sumowanie [stop]    1   2 (konflikt x)
1'. sumowanie [start]   2   2
3.  zmiana x  [start]   2   2
3.  zmiana x  [stop]    3   2
1'. sumowanie [stop]    2   2 (konflikt x)

W transakcji nr 1 chcieliśmy sumować x i y, ale w trakcie realizowania tej operacji zdążyła rozpocząć się i zakończyć transakcja nr 2, która zmieniła powiązanie referencji x z wartością. Porównanie migawek doprowadziło do wykrycia tego konfliktu i transakcja nr 1 została wycofana, a następnie ponowiona (jako transakcja nr 1’). Niestety, podczas ponawiania znowu doszło do zmiany współdzielonego stanu referencji x i jej poprzednia wartość bieżąca przestała być aktualna. Porównanie migawek znów doprowadziło do wykrycia konfliktu i operacja sumowania kolejny raz została wycofana…

Widzimy więc, że bardzo częste zmiany x (w transakcjach nr 2 i 3) prowadzą do sytuacji teoretycznie nieskończonego ponawiania transakcji nr 1. Jak można sobie z tym poradzić?

Do przeprowadzenia sensownych obliczeń potrzebujemy spójnych wartości bieżących referencji transakcyjnych, jednak w tym przypadku pojawia się problem z uzyskaniem aktualnej wersji jednej z nich, ponieważ w praktyce przy tak częstych operacjach nigdy nie uda się trafić z odczytem dokładnie w moment następujący po ostatniej aktualizacji – ulega ona zbyt szybkim zmianom.

Problemem nie jest jednak to, że aktualnych danych nie ma, lecz to, że przyjmujemy optymistyczne założenie o odczycie i użyciu wartości pochodzących z bieżącego punktu w czasie. Warto sobie w tym miejscu przypomnieć, że priorytetem STM nie jest wcale dostarczanie wartości teraźniejszych, lecz przede wszystkim spójnych zestawów wartości, na których kopiach można przeprowadzać sensowne obliczenia.

Rozwiązaniem problemu jest użycie historii zmian. Dzięki niej można będzie skorzystać z wartości minionej, pochodzącej z momentu, w którym również była ona spójna z innymi wartościami. Aby to osiągnąć, implementacja STM w Clojure tworzy dodatkowe migawki stanów dla tych obiektów typu Ref, które są aktualizowane. Są one obsługiwane z użyciem specjalnych kolejek, do których trafiają poprzednie wartości Refów podczas ich aktualizowania. W efekcie porównywanie migawek wykonywane jest nie tylko dla początku i końca transakcji, ale też dla momentów, gdy z powodu następujących po sobie konfliktów nie można użyć najnowszych wartości bieżących.

transakcja              x   y
-------------------------------------------------------
1.  sumowanie [start]   1   2
2.  zmiana x  [start]   1   2
2.  zmiana x  [stop]    2   2 (historia: x0 = 1)
3.  zmiana x  [start]   2   2
3.  zmiana x  [stop]    3   2 (historia: x1 = 2)
1'. sumowanie [stop]    1   2 (x0 + y = 3)

Odczyt wartości bieżącej Refa polega więc na sprawdzeniu czasu aktualizacji jego bieżącej, współdzielonej wartości, a jeżeli jest on późniejszy niż czas rozpoczęcia bieżącej transakcji, sprawdzany jest łańcuch historii zmian. Dla każdej umieszczonej w nim wartości historycznej (od najnowszej do najstarszej) sprawdzony będzie w taki sam sposób zapisany tam czas aktualizacji. Gdy wartość spełniająca warunek zostanie odnaleziona, będzie wykorzystana, a blokada odczytu zwolniona.

Może zdarzyć się tak, że łańcuch historii nie będzie zawierał odpowiedniego wpisu. W takim przypadku transakcja jest ponawiana, a w obiekcie typu Ref zwiększany jest wewnętrzny licznik usterek. Licznik ten używany jest do dynamicznego zwiększania długości łańcucha historii zmian w fazie zatwierdzania transakcji (po rozszerzeniu długości łańcucha licznik będzie zerowany).

 1(def x (ref 1))
 2(ref-min-history x 5)
 3
 4(future
 5  (Thread/sleep 350)
 6  (dotimes [n 5]
 7    (dosync
 8      (println (str (+ 2 n) ".") "zapis  [start] x =" @x "+ 1")
 9      (alter x inc)
10      (println (str (+ 2 n) ".") "zapis   [stop] x =" @x))))
11
12(dosync
13  (println "1. odczyt [start]")
14  (Thread/sleep 500)
15  (println "1. odczyt  [stop] x =" @x))

W naszym przykładzie pierwsza transakcja nie zostanie przerwana z powodu zmiany współdzielonej wartości referencji x w trakcie jej trwania, ponieważ wykorzystana zostanie wartość pochodząca z historii zmian.

Warto zauważyć, że użyliśmy funkcji ref-min-history. Bez niej transakcja odczytująca x zostałaby po prostu ponowiona, bo historia zaczęłaby się tworzyć już po wykryciu konfliktu. Dopiero każda kolejna próba odczytania nieaktualnej wartości bieżącej zwiększałaby rozmiar łańcucha historii o jeden. W ten sposób STM dba o to, aby zasoby systemowe nie były trwonione na zapamiętywanie i przeszukiwanie dodatkowych struktur, gdy nie ma takiej potrzeby. W przykładzie zależało nam jednak na tym, aby historia była używana od razu i stąd ustawienie jej minimalnej długości.

Realizowanie transakcji

Transakcja rozpoczyna się wywołaniem makra dosync, które przyjmuje zestaw wyrażeń stanowiących jej ciało. Zostaną one przekształcone do obiektu typu Callable i skojarzone z obiektem reprezentującym transakcję. W obiekcie transakcji znajdziemy też odwołania do dwóch tworzonych list:

• pierwsza zawierała będzie Refy, na które nałożono blokadę;

• druga zawierała będzie obiekty typu Notify (kapsułkujące obiekty typu Ref, oraz powiązane z nimi poprzednie i bieżące wartości), które będą wykorzystane podczas wywoływania funkcji obserwujących.

Kolejnym krokiem jest uruchomienie pętli ponowień, która wykonuje 10 000 powtórzeń, chyba że transakcja zakończyła się (z powodu błędu lub zatwierdzenia). Gdy wszystkie powtórzenia są zrealizowane, a transakcja nadal nie została zamknięta, zgłaszany jest wyjątek RetryEx.

Na początku pętli aktualizowana jest wartość tzw. punktu początkowego. W przypadku, gdy pętla wykonuje pierwszy przebieg, punkt ten będzie tożsamy z punktem odczytu. Dodatkowo stan transakcji (wyrażony instancją klasy Info) jest ustawiany na wartość RUNNING.

Kolejnym krokiem jest uruchomienie podprogramu obiektu typu Callable, czyli rozpoczęcie przeliczania wyrażeń przekazanych do makra dosync. Po udanym wykonaniu wyrażeń sprawdzany jest status transakcji, aby wykluczyć ewentualny spór rozpoczęty przez inną transakcję. Jeżeli takowy wystąpił, a bieżąca transakcja go przegrała, jej stan będzie ustawiony na KILLED. Jeżeli nie wykryto zmiany stanu, realizowanie jest kontynuowane, a stan zmieniany jest na COMMITTING i rozpoczyna się faza zatwierdzania.

Zatwierdzanie transakcji

Jeżeli operacje aktualizowania Refów przebiegną pomyślnie, podejmowana jest próba zatwierdzenia transakcji (stan COMMITTING).

Funkcje używane do wprowadzania zmian, które zostaną obsłużone podczas tego etapu to ref-set, alter i commute. Dwie pierwsze sprawiają, że dochodzi do aktualizacji wartości wewnątrztransakcyjnych, a w momencie zatwierdzenia zmian do atomowej podmiany powiązań w odpowiednich, współdzielonych obiektach typu Ref. Odniesieniami nowych powiązań są wtedy pochodzące z transakcji, obliczone wcześniej wartości wewnątrztransakcyjne.

Inaczej działa ostatnia z wymienionych wyżej funkcji (commute). Dla każdego Refa, na którym ją zastosowano, jego wartość wewnątrztransakcyjna również będzie na bieżąco aktualizowana, ale podczas zatwierdzania transakcji nie zostanie ona użyta do podmiany powiązania współdzielonego Refa. Zamiast tego dojdzie do ponownego wywołania funkcji przeliczającej przekazanej do commute. Przekazywanym jej argumentem będzie aktualna wartość współdzielona, do której odnosi się obiekt typu Ref, a nie izolowana w transakcji wartość pochodząca z punktu odczytu. W tym czasie żaden inny wątek nie będzie mógł dokonywać aktualizacji współdzielonej wartości Refa. W praktyce oznacza to, że zachowana jest synchroniczność operacji, ale może dojść do zamiany kolejności ich realizowania w czasie.

Jeżeli przed wywołaniem commute stworzono wartość wewnątrztransakcyjną Refa z użyciem alter lub ref-set, powyższy krok zostanie pominięty, a aktualizacja współdzielonych obiektów będzie polegała na powiązaniu ich z wartościami wewnątrztransakcyjnymi (które zostały zaktualizowane również z użyciem commute i przekazanej jej funkcji).

Dalsza obsługa referencji aktualizowanych z użyciem commute polega na zarządzaniu blokadami dostępu. Jeżeli w transakcji użyto ensure w odniesieniu do jakichś referencji, zwalniane są ich blokady odczytu. Zaraz po tym na obiekty zakładane są blokady zapisu, a każdy z nich dodawany jest do listy zablokowanych Refów utrzymywanej w obiekcie reprezentującym transakcję. Wykonywane jest to w takiej kolejności, w jakiej tworzone były obiekty typu Ref, aby uniknąć sytuacji wyścigów.

Poza tym, jeżeli w stosunku do którejś referencji użyto funkcji ensure, a w trakcie realizowania transakcji inna transakcja jednak zmieniła jej wartość współdzieloną (co nie powinno się wydarzyć), bieżąca transakcja jest ponawiana. Dzieje się tak również wtedy, gdy równolegle realizowana transakcja zaktualizowała referencję zaraz przed tym, gdy w bieżącej transakcji doszło do wywołania ensure.

Jeżeli inna transakcja dokonała w międzyczasie zmiany wewnątrztransakcyjnej wartości Refa, który był modyfikowany z użyciem commute, bieżąca transakcja rozpoczyna z nią spór. Jeżeli spór zostanie przegrany, bieżąca transakcja jest ponawiana, a jeżeli wygrany, odczytana zostanie najbardziej aktualna, współdzielona wartość powiązana z referencją i będzie ona użyta jako wartość wewnątrztransakcyjna.

Po tych operacjach wywoływane są funkcje przekazane wcześniej do commute użytych względem Refów i przekazywane są im (poza innymi argumentami) wartości wewnątrztransakcyjne obiektów. Wartości wewnątrztransakcyjne są aktualizowane wartościami zwracanymi przez wywołania funkcji przeliczających.

Kolejnym etapem zatwierdzania transakcji jest założenie blokady zapisu na każdego Refa, które ma być zaktualizowany. Dotyczy to zarówno obiektów, których wartości wewnątrztransakcyjne zostały zmienione z użyciem alter i ref-set, jak również tych, dla których wcześniej wyliczono odpowiadające im wartości wewnątrztransakcyjne opracowywane przy okazji obsługi wywołań commute. Jeżeli dany obiekt typu Ref już został zablokowany przez inną transakcję, bieżąca transakcja jest ponawiana (następuje ustawienie stanu RETRY, a potem powrót do początku pętli).

Jeżeli uda się zablokować możliwość zapisu wszystkich aktualizowanych Refów, dochodzi do wywołania przypisanych do nich walidatorów, gdy takowe ustawiono. Jeżeli któraś funkcja walidująca zwróci false lub zgłosi wyjątek, transakcja jest przerywana.

Jeżeli podczas walidacji, ale przed zablokowaniem możliwości zapisu do aktualizowanych Refów, wykryta zostanie zmiana jednego z nich (dokonana przez inną transakcję), bieżąca transakcja jest ponawiana.

Ostatni etap zatwierdzania polega na umieszczeniu zaktualizowanych wartości Refów w łańcuchach historii zmian. Łańcuch historii jest rozszerzany o nowy element, jeżeli jego długość jest zerowa lub mniejsza niż wartość określająca minimalną długość (można ją ustawić podczas tworzenia obiektu typu Ref lub korzystając z funkcji ref-min-history). Dodanie nowego elementu nastąpi również wtedy, gdy dany Ref uległ wcześniej usterce (konfliktowi z inną transakcją) i przypisany mu licznik usterek jest niezerowy, a długość łańcucha historii jest mniejsza, niż wartość określająca maksymalną długość (można ją ustawić, korzystając z funkcji ref-max-history lub podczas tworzenia obiektu typu Ref).

Jeżeli żaden z powyższych warunków nie zostanie spełniony, łańcuch historii nie będzie rozszerzony o nowy element, lecz ostatni element stanie się początkowym, a jego wartość zastąpiona bieżącą wartością referencyjną danego Refa. Dojdzie więc do rotacji historii zmian (przesunięcia wszystkich elementów o jeden w dół z nadpisaniem pierwszego).

Po aktualizacji historii zmian przygotowywana jest lista obiektów typu Notify. Zawierają one zmienione Refy wraz z poprzednimi i bieżącymi wartościami, a użyte zostaną później, aby wywołać [funkcje obserwujące], które zostały z nimi skojarzone.

Kolejny etap to przejście do ostatniej fazy zatwierdzania. Zwalniane są założone wcześniej i nie zdjęte jeszcze blokady odczytu (np. z powodu użycia ensure). Jeżeli transakcja zakończyła się pomyślnie (jest w stanie COMMITTING), zostaje oznaczona jako zatwierdzona (stan COMMITTED), a w przeciwnym przypadku jako ponawiana (stan RETRY). Poza tym uruchamiane są funkcje obserwujące zgromadzone w obiektach typu Notify utworzonej wcześniej listy.

Gdy zatwierdzanie się powiedzie, wartości bieżące współdzielonych Refów, których wersje wewnątrztransakcyjne zmieniły się, zostaną zastąpione tymi ostatnimi. Moment, w którym dochodzi do podmiany współdzielonych powiązań podczas zatwierdzania transakcji nazywamy punktem zapisu (ang. write point).

Blokowanie równoległych operacji

Operacje transakcyjne można rozpatrywać pod kątem tego, czy powodują blokowanie realizacji równolegle wykonywanych operacji na obiektach referencyjnych. Do blokowania dochodzi wtedy, gdy mechanizmy programowej pamięci transakcyjnej muszą zapewnić synchroniczność i jednolitość wprowadzania zmian.

Zasady są następujące:

• Funkcje używane do odczytywania wartości (dereferencji) obiektów typu Ref nie blokują równoległego wykonywania funkcji używanych do: odczytywania, aktualizowania i aktualizowania przemiennego, chyba że dereferencja następuje w transakcji i dotyczy wartości współdzielonej (bieżącej lub historycznej) a nie wewnątrztransakcyjnej – w takim przypadku na czas odczytu blokowana jest możliwość aktualizowania danego Refa.

• Funkcje używane do aktualizowania przemiennego obiektów typu Ref nie blokują równoległego wykonywania funkcji używanych do: odczytywania, aktualizowania i aktualizowania przemiennego.

• Funkcje używane do aktualizowania obiektów typu Ref nie blokują równoległego wykonywania funkcji używanych do: odczytywania, i aktualizowania przemiennego.

Kiedy dojdzie do blokowania, czyli oczekiwania jednej operacji na drugą? Gdy w stosunku do obiektu typu Ref pojawi się operacja aktualizowania jego współdzielonego powiązania. Z blokowaniem będziemy też mieli do czynienia, gdy w jednej transakcji dochodzi do aktualizowania wartości bieżącej, a w równolegle realizowanej wywołano na tym samym Refie funkcję ensure.

Struktury danych

Obiekty typu Ref powinny zawierać odniesienia do niemutowalnych typów danych i struktur trwałych. Powiązanie referencji transakcyjnej z pamięciowym obiektem, który jest mutowalny, sprawi, że jej obsługa z użyciem mechanizmów języka Clojure przestanie być bezpieczna pod względem przetwarzania współbieżnego.

Zaleca się ponadto, aby tam, gdzie to możliwe, jako wartości wskazywanych referencjami transakcyjnymi używać kolekcji, ponieważ wyposażone są one w wydajne i oszczędne pod względem wykorzystania pamięci mechanizmy tworzenia wersji. STM może więc uczynić z nich użytek, tworząc wartości wewnątrztransakcyjne, czy aktualizując współdzielone stany bieżące obiektów typu Ref.

Tworzenie

Tworzenie obiektów typu Ref, ref

Do tworzenia obiektów typu Ref służy funkcja ref. Przyjmuje ona jeden obowiązkowy argument – wartość początkową (z którą powiązana zostanie referencja) – a także argumenty nieobowiązkowe opcji wyrażane w sposób nazwany, czyli w postaci par klucz–wartość.

Funkcja zwraca obiekt typu Ref.

Użycie:

• (ref wartość-początkowa & opcje).

Możliwe opcje to:

• :meta mapa-metadanowa,
• :validator funkcja,
• :min-history liczba (domyślnie 0),
• :max-history liczba (domyślnie 10).

Mapa metadanowa po słowie kluczowym :meta powinna być mapą wyrażoną z użyciem powiązanego z nią symbolu lub osadzoną w kodzie jako tzw. mapowe S-wyrażenie. Podane asocjacje staną się metadanymi tworzonego obiektu referencyjnego.

Wartość podawana po słowie kluczowym :validator powinna być po ewaluacji jednoargumentowym obiektem funkcyjnym, który nie generuje efektów ubocznych, albo wartością nil. Podana funkcja będzie wywoływana za każdym razem, gdy zażądano zmiany stanu obiektu typu Ref i przekazywana jej będzie proponowana, nowa wartość. Powinna ona zwracać false lub zgłaszać wyjątek, gdy wartość ta jest nieakceptowalna. Operację taką nazywamy walidacją, a wyrażającą ją funkcję walidatorem.

Parametry :min-history i :max-history pozwalają ustawiać minimalną i maksymalną liczbę elementów łańcucha historii zmian, które będą używane do zapamiętywania poprzednich wartości współdzielonych (z powodu aktualizacji lub w efekcie zastosowania funkcji ensure).

1(ref  0                     )  ; => #<Ref@e019b8e: 0>
2(ref  0         :meta {:a 1})  ; => #<Ref@7f8ca927: 0>
3(ref                 [1 2 3])  ; => #<Ref@31470ee1: [1 2 3]>
4(ref 10 :validator #(> %1 1))  ; => #<Ref@68a43065: 10>
5(ref  0 :validator #(> %1 1))  ; >> java.lang.IllegalStateException

• Wyrażenie z linii nr 1 tworzy obiekt referencyjny typu Ref, który odnosi się do wartości 0.

• Wyrażenie drugie (z linii nr 2) działa tak samo, lecz dodatkowo ustawia metadaną.

• Kolejne wyrażenie przykładu (z linii nr 3) ustawia wartość początkową na wektor stworzony z użyciem wyrażenia wektorowego.

• Dwa ostatnie wyrażenia dotyczą sytuacji, w których przekazywana jest funkcja walidująca. Ostatnie wywołanie powoduje zgłoszenie wyjątku, ponieważ podana wartość początkowa nie jest poprawna.

Identyfikowanie Refów

Refy są obiektami, z którymi, podobnie jak z innymi pamięciowymi wartościami, można powiązać symboliczne identyfikatory i w ten sposób określać ich tożsamości. Możemy więc stwarzać odnoszące się do obiektów typu Ref globalne zmienne, jak również korzystać z innych rodzajów powiązań (np. leksykalnych).

1(def referencja (ref 0))     ; zmienna globalna
2(let [refik (ref 0)] refik)  ; powiązanie leksykalne

Zarządzanie transakcją

Tworzenie transakcji, dosync

Makro dosync służy do tworzenia transakcji. Przyjmuje ono zero lub więcej wyrażeń, które staną się ciałem transakcji, tzn. będą przeliczane podczas jej realizowania.

Transakcja rozpoczyna się, jeżeli inna transakcja jeszcze nie działa w bieżącym wątku. Jeżeli działa, ciało drugiej transakcji (syntaktycznie zagnieżdżonej) zostanie dołączone do ciała pierwszej.

Jeżeli podczas obliczania wartości podanych wyrażeń zgłoszony zostanie nieprzechwycony wyjątek, transakcja zostanie przerwana i nastąpi wyjście z dosync.

S-wyrażenia podane w dosync mogą być uruchomione więcej niż raz, ponieważ transakcja może być ponawiana. Należy więc unikać obsługi wejścia/wyjścia i stosowania innych funkcji mających efekty uboczne.

Wartością zwracaną przez makro dosync jest wartość ostatnio przeliczanego wyrażenia z jej ciała.

Użycie:

• (dosync & wyrażenie…).

 1(let [x (ref 0)
 2      y (ref 1)]
 3  (dosync
 4    (ref-set x 10)
 5    (alter   x inc)
 6    (alter   y + (ensure x))))
 7
 8; => 12
 9
10;; transakcje złączone – jeden wątek
11(dosync
12  (println "[start] pierwsza")
13  (dosync
14    (println "[start] druga")
15    (Thread/sleep 500)
16    (println "[stop]  druga"))
17  (println "[stop]  pierwsza"))
18
19; >> [start] pierwsza
20; >> [start] druga
21; >> [stop]  druga
22; >> [stop]  pierwsza
23
24;; transakcje równoległe – dwa wątki
25(dosync
26  (println "[start] pierwsza")
27  (future (dosync
28            (println "[start] druga")
29            (Thread/sleep 500)
30            (println "[stop]  druga")))
31  (println "[stop]  pierwsza"))
32
33; >> [start] pierwsza
34; >> [start] druga
35; >> [stop]  pierwsza
36; >> [stop]  druga

Blokowanie wejścia/wyjścia, io!

W transakcjach powinniśmy unikać stosowania operacji, które są blokujące. Należą do nich m.in. funkcje obsługi wejścia/wyjścia. Dzięki makru io! możemy oznaczyć te fragmenty programów, które są blokujące (związane z obsługą wejścia lub wyjścia).

Jeżeli kod zawierający wywołanie io! zostanie uruchomiony z wnętrza transakcji, dojdzie do zgłoszenia wyjątku IllegalStateException, a jej wykonywanie zostanie przerwane.

Argumentami io! powinny być wyrażenia, które mają być przeliczone. Jeżeli pierwszym przekazywanym argumentem będzie literał łańcucha znakowego, wyrażony nim tekst zostanie użyty jako komunikat towarzyszący wyjątkowi.

Użycie:

• (io! komunikat & wyrażenie…),
• (io! & wyrażenie…).

Pobieranie wartości

Odczytywanie wartości Refów, deref

Żeby odczytać wartość referencji transakcyjnej, możemy użyć funkcji deref. Przyjmuje ona jeden argument, którym powinien być obiekt typu Ref, a zwraca wartość, do której odniesienie jest przechowywane w tym obiekcie.

Dereferencji można użyć w dowolnym miejscu programu, ale jeżeli zależy nam na spójności danych w obrębie pewnego zestawu Refów, powinniśmy rozważyć wywołanie tej funkcji w transakcji. Zostaną wtedy założone odpowiednie blokady, a pobierana wartość będzie wartością spójną z innymi. Jeżeli którejś z używanych wartości nie modyfikujemy, zaleca się w odniesieniu do niej użyć funkcji ensure.

Wywołanie deref w transakcji zwróci najbardziej aktualną wartość wewnątrztransakcyjną, natomiast poza transakcją wartość współdzieloną (bieżącą) obiektu typu Ref.

Użycie:

• (deref ref).

1(def referencja (ref 0))
2(deref referencja)
3; => 0

Zobacz także:

• „Historia zmian”, przykład dereferencji w transakcjach równoległych

Dereferencja Refów, makro @

Makro czytnika @ (znak małpki) umieszczone przed wyrażeniem sprawia, że jeżeli zwracany przez nie obiekt jest typem referencyjnym, wywołana zostanie na nim funkcja odpowiedzialna za odczyt wskazywanej wartości. W przypadku Refów będzie to omówiona wyżej funkcja deref.

Dereferencji można użyć w dowolnym miejscu programu, ale jeżeli zależy nam na spójności danych, powinniśmy rozważyć wywołanie tej funkcji w transakcji. Gdy w obliczeniach korzystamy z wartości, których nie modyfikujemy, zaleca się użycie w odniesieniu do nich funkcji ensure, aby przyjęty zestaw Refów był spójny.

Wywołanie makra @ w transakcji zwróci najbardziej aktualną wartość wewnątrztransakcyjną, natomiast poza transakcją wartość współdzieloną (bieżącą) obiektu typu Ref.

Użycie:

• @ref.

1(def referencja (ref 0))
2@referencja
3; => 0

Zmiany wartości

Aktualizowanie wartości, alter

Zmiana stanu obiektu typu Ref, czyli aktualizacja współdzielonego odniesienia do wartości, możliwa jest z zastosowaniem funkcji alter.

Pierwszy przyjmowany argument powinien być symbolem identyfikującym obiekt typu Ref lub inne wyrażenie, które przeliczone zwróci ten obiekt. Drugim argumentem powinna być funkcja, która jako pierwszy argument przyjmie wartość bieżącą (do której odnosi się Ref) i na jej podstawie obliczy nową wartość. Zostanie ona użyta do zastąpienia poprzedniego, referencyjnego odniesienia w Refie. Opcjonalnie możemy podać dodatkowe argumenty, które zostaną przekazane jako kolejne argumenty wywoływanej funkcji.

Wywołanie funkcji alter musi odbywać się w transakcji. Efektem jej działania jest natychmiastowa zmiana lub utworzenie wartości wewnątrztransakcyjnej, natomiast do ewentualnej aktualizacji wartości współdzielonej dochodzi dopiero wtedy, gdy cała transakcja z powodzeniem się zakończy.

Przekazana funkcja obliczająca nową wartość bieżącą Refa powinna być nieblokująca i bez efektów ubocznych, ponieważ będzie realizowana w wątku należącym do transakcji, która może być ponawiana, zakładać blokady na obiekty referencji transakcyjnych i sprawiać, że inne transakcje będą działały mniej wydajnie.

Wartość bieżąca Refa przekazywana funkcji przeliczającej będzie wartością wewnątrztransakcyjną (jeżeli taką już utworzono poprzednią modyfikacją) lub wartością współdzieloną, uzyskaną w punkcie odczytu. Jeżeli podczas odczytywania wartości okaże się, że wartość współdzielona uległa zmianie (z powodu działania innej transakcji) i wartość z punktu odczytu przestała być aktualna, przeszukany zostanie łańcuch historii zmian. Jeżeli nawet tam nie zostanie znaleziona aktualna wartość współdzielona Refa, transakcja będzie ponowiona.

Gdyby któryś z kroków pośrednich (między odczytem wartości bieżącej a aktualizacją referencji) nie powiódł się, będziemy mieli do czynienia z usterką Refa, czyli sytuacją wyjątkową. Może ona wystąpić z powodu błędu lub niespełnienia wymogów skojarzonej z Refem funkcji walidującej. Innym powodem usterki może być spór z równolegle realizowaną transakcją podczas próby odczytania aktualnej wartości bieżącej. W przypadku nieudanej walidacji, transakcja jest przerywana, a w przypadku przegranego sporu ponawiana zaraz po wejściu w fazę zatwierdzania.

Wartością zwracaną przez funkcję alter jest ustawiona wartość obiektu typu Ref, która jest wartością wewnątrztransakcyjną.

Użycie:

• (alter ref funkcja & argument…).

1(def referencja (ref 0))
2
3@referencja                      ; => 0
4(dosync (alter referencja inc))  ; => 1
5@referencja                      ; => 1

Ustawianie wartości, ref-set

Ustawienie stanu obiektu typu Ref, czyli bezwarunkowa zmiana współdzielonego odniesienia do wartości, możliwa jest z zastosowaniem funkcji ref-set.

Pierwszy przyjmowany argument powinien być symbolem identyfikującym obiekt typu Ref lub inne wyrażenie, które przeliczone zwróci ten obiekt. Drugim argumentem powinna być wartość, z którą chcemy powiązać Refa, zastępując obecną wartość współdzieloną.

Wywołanie funkcji ref-set musi odbywać się w transakcji. Efektem jej działania jest natychmiastowa zmiana lub utworzenie wartości wewnątrztransakcyjnej, natomiast do ewentualnej aktualizacji wartości współdzielonej dochodzi dopiero wtedy, gdy cała transakcja z powodzeniem się zakończy.

Tuż przed zatwierdzeniem zmian dokonanych w transakcji z użyciem ref-set wywołane będą przypisane do Refów walidatory (jeżeli je ustawiono). W wypadku niepowodzenia (walidator zwróci false lub zgłosi wyjątek) cała transakcja zostanie ponowiona.

Istotną różnicą względem alter – poza tym, że nie mamy do czynienia z funkcją przeliczającą – jest brak fazy odczytywania wartości bieżącej referencji. Oznacza to mniej wewnętrznych blokad i znacznie mniejszą szansę na wystąpienie konfliktów.

Wartością zwracaną przez funkcję ref-set jest ustawiona wartość obiektu typu Ref, która jest wartością wewnątrztransakcyjną.

Użycie:

• (ref-set ref wartość).

1(def referencja (ref 0))
2
3@referencja                      ; => 0
4(dosync (ref-set referencja 1))  ; => 1
5@referencja                      ; => 1

Podmiana wartości, commute

Komutacja stanu obiektu typu Ref, czyli podmiana współdzielonego odniesienia do wartości bieżącej, możliwa jest z zastosowaniem funkcji commute. Działa ona podobnie do alter, jednak przekazana funkcja przeliczająca będzie wywołana na samym końcu transakcji, w momencie jej zatwierdzania, a zwracana przez nią wartość zastąpi wartość współdzieloną Refa. Możemy zaobserwować, że analogiczna w działaniu jest funkcja swap! związana z typem Atom, chociaż nie korzysta ona z dobrodziejstw STM.

Podczas użycia pozostałych funkcji aktualizujących najpierw dochodzi do podmiany wartości wewnątrztransakcyjnej, a dopiero ona podczas finalizowania transakcji staje się wartością współdzieloną. W przypadku commute to nie wartość wewnątrztransakcyjna stanie się wartością współdzieloną, lecz w fazie zatwierdzania transakcji po raz kolejny wywołana zostanie podana jako argument commute funkcja przeliczająca, a zwrócony przez nią wynik będzie użyty do zmiany stanu.

W tym samym czasie tylko jedna funkcja przeliczająca będzie zmieniała stan danego obiektu typu Ref, a przekazywaną jako jej argument wartością będzie aktualna wartość bieżąca Refa, a nie wartość wewnątrztransakcyjna. Wyjątek stanowią sytuacje, gdy w transakcji poza commute skorzystano z innych funkcji aktualizujących w odniesieniu do tego samego obiektu.

Używając commute, pomijamy fazę porównywania współdzielonej wartości bieżącej Refa z migawką wykonaną w punkcie odczytu, ponieważ nie korzystamy z wartości wewnątrztransakcyjnej, która w czasie realizowania transakcji mogłaby się zdezaktualizować. Zmiana dokonywana jest w sposób atomowy i polega na zablokowaniu zapisu do danego Refa na czas uruchomienia podanej funkcji.

W efekcie zyskujemy na prędkości realizacji w przypadku wielu wątków, ponieważ dochodzi do mniejszej liczby ponowień transakcji spowodowanej brakiem konfliktów. Tracimy na tym, że może nie być zachowana kolejność operacji. Następcze wywoływanie commute w jednym wątku może zajmować tyle samo lub nawet więcej czasu, co korzystanie alter. Zysk wydajnościowy osiągniemy wówczas, gdy wykorzystamy tę funkcję w wywołaniach współbieżnych.

Pierwszy przyjmowany przez funkcję commute argument powinien być symbolem identyfikującym obiekt typu Ref lub inne wyrażenie, które przeliczone zwróci ten obiekt. Drugim argumentem powinna być funkcja przeliczająca, która jako pierwszy argument przyjmie wartość bieżącą Refa i na jej podstawie obliczy nową wartość. Ta ostatnia zostanie wskazana odniesieniem w Refie i zastąpi tym samym zastaną wartość bieżącą. Opcjonalnie możemy podać dodatkowe argumenty, które zostaną przekazane jako kolejne argumenty funkcji przeliczającej.

Wywołanie funkcji commute musi odbywać się w transakcji. Efektem jej działania jest natychmiastowa zmiana lub utworzenie wartości wewnątrztransakcyjnej, a dodatkowo zmiana wartości współdzielonej.

Przekazana funkcja obliczająca nową wartość bieżącą Refa powinna być nieblokująca i bez efektów ubocznych, ponieważ będzie realizowana w wątku należącym do transakcji, która może być ponawiana, zakładać blokady na obiekty referencji transakcyjnych i sprawiać, że inne transakcje będą działały mniej wydajnie.

Wartość bieżąca Refa przekazywana funkcji przeliczającej będzie ostatnią wartością współdzieloną, a nie wartością wewnątrztransakcyjną, chyba że przed wywołaniem commute utworzono wartość wewnątrztransakcyjną danego Refa z użyciem alter lub ref-set. W takim przypadku będzie to wartość wewnątrztransakcyjna.

Jeżeli w transakcji używamy commute (bez wcześniejszego wywołania na danym obiekcie typu Ref alter lub ref-set), wtedy w przypadku wykrycia równoległej aktualizacji, która miała miejsce w trakcie realizowania bieżącej, nie będzie konieczne jej ponawianie. Jeżeli więc inna transakcja zmieni powiązanie obiektu typu Ref, i tak dojdzie do pomyślnej aktualizacji odniesienia.

Zmiany wprowadzane z użyciem commute są atomowe i synchroniczne, ale wartość bieżąca Refa może być niespójna z innymi wartościami migawki utworzonej na potrzeby realizowania transakcji. Oznacza to, że funkcja przekazana do commute, jak i projektowana w transakcji operacja na danym Refie, powinny być przemienne (komutatywne), czyli zakładać, że kolejność aktualizowania wartości nie jest istotna. Przykładem takiej operacji jest dokonywanie wpłat na rachunek bankowy (ale już nie wykonywanie przelewu!). Nie jest ważne w jakiej kolejności wpłaty się pojawią – po jakimś czasie stan rachunku będzie się zgadzał.

Powyższy proces można obsłużyć, używając alter, jednak zajmie on wtedy więcej czasu, bo w przypadku transakcji równoległych dojdzie do sporów i ponowień. Stanie się tak, ponieważ korzystając z alter przyjmujemy założenie, że różne wartości Refów obsługiwanych w ramach tej samej transakcji muszą tworzyć spójne zestawy, a więc jedne transakcje powinny oczekiwać na zakończenia innych, jeżeli dotyczą obiektów, których wartości były aktualizowane. Funkcja commute jest wolna od tego założenia.

Tuż przed zatwierdzeniem zmian dokonanych w transakcji z użyciem commute wywołane będą przypisane do Refów walidatory (jeżeli je ustawiono). W wypadku niepowodzenia (walidator zwróci false lub zgłosi wyjątek) transakcja zostanie wycofana.

Wartością zwracaną przez funkcję commute jest ustawiona wartość obiektu typu Ref, która jest wartością wewnątrztransakcyjną. Warto jednak pamiętać, że rzeczywiście ustawiona wartość współdzielona może być od niej różna, gdy dojdzie do ostatecznego wywołania przekazanej do commute funkcji. Dzieje się tak, ponieważ do jej wyliczenia użyta może zostać wartość współdzielona Refa ustawiona przez równolegle realizowaną transakcję, która zakończyła się wcześniej.

Użycie:

• (commute ref funkcja & argument…).

 1(def x (ref 0))
 2(time
 3  (dorun
 4    (apply pcalls (repeat 50
 5                          #(dosync
 6                             (Thread/sleep 50)
 7                             (commute x inc))))))
 8
 9; >> "Elapsed time: 511.852803 msecs"
10
11(def x (ref 0))
12(time
13  (dorun
14    (apply pcalls (repeat 50
15                          #(dosync
16                             (Thread/sleep 50)
17                             (alter x inc))))))
18
19; >> "Elapsed time: 2512.555332 msecs"

Zmiany stanów struktur złożonych

Nierzadko będziemy mieli do czynienia z sytuacjami, w których Ref został powiązany z wartością, która jest złożoną strukturą danych, na przykład wektorem czy mapą. Podmiany wartości można wtedy dokonać, przekazując odpowiednią funkcję przeliczającą, na przykład w taki sposób:

1(def x (ref {:imię "Paweł" :wynik 107}))
2
3(dosync (commute x assoc :wynik (inc (:wynik @x))))
4; => {:imię "Paweł" :wynik 108}

W powyższym przykładzie założyliśmy, że klucz :wynik identyfikuje wartość, w przypadku której kolejność prowadzenia obliczeń nie ma znaczenia i do zmiany stanu użyliśmy commute.

Problematyczne jest jednak to, że do funkcji assoc odpowiadającej za aktualizację mapy, przekazujemy przez wartość rezultat uprzedniego wywołania funkcji inc na odczytanym elemencie mapy, którą uzyskaliśmy dzięki dereferencji. Operacja ta nie jest więc atomowa i istnieje ryzyko, że funkcja przeliczająca otrzyma rezultat bazujący na nieaktualnej już strukturze pochodzącej z migawki, a nie na wartości bieżącej. Bezpiecznym wariantem byłby więc kod korzystający z alter i ensure:

1(def x (ref {:imię "Paweł" :wynik 107}))
2
3(dosync
4  (ensure x)
5  (alter x assoc :wynik (inc (:wynik @x))))
6; => {:imię "Paweł" :wynik 108}

W powyższym przykładzie tracimy jednak na ewentualnej prędkości operacji wykonywanych współbieżnie, ponieważ dokonujemy wyłącznej blokady zapisu do Refa x na czas trwania transakcji.

Można zaradzić opisywanemu tu problemowi w nieco inny sposób: korzystając z jednej struktury, która jest przekazywana jako argument do funkcji przeliczającej. Chcemy pozbyć się przekazywania wartości pochodzącej z migawki, więc powinniśmy utworzyć funkcję aktualizującą samodzielnie (np. z użyciem fn lub lambda-wyrażenia):

1(def x (ref {:imię "Paweł" :wynik 107}))
2
3(dosync
4  (commute x (fn [mapa]
5               (assoc mapa :wynik (inc (:wynik mapa))))))
6; => {:imię "Paweł" :wynik 108}

Nasz przykład można jednak zoptymalizować, ponieważ język Clojure wyposażono w odpowiednie funkcje służące do funkcyjnego aktualizowania wbudowanych struktur złożonych:

1(def x (ref {:imię "Paweł" :wynik 107}))
2
3(dosync (commute x update :wynik inc))
4; => {:imię "Paweł" :wynik 108}

Prawda, że bardziej elegancko? Skorzystaliśmy z funkcji commute, której przekazaliśmy funkcję update w celu zaktualizowania elementu mapy wskazywanej przez x o podanym kluczu (:wynik). Sama operacja aktualizacji również jest obiektem funkcyjnym przekazywanym jako argument wywołania update – w tym przypadku używamy wbudowanej funkcji inc, zwiększającej wartość o 1.

Powyższa czynność zostanie wykonana w momencie zatwierdzania transakcji, a w trakcie realizowania obliczeń w obrębie transakcji nie będziemy dokonywali dereferencji, ani tworzyli własnych funkcji.

W odniesieniu do wbudowanych, złożonych struktur możemy używać następujących operacji:

• mapy:
  • update – aktualizacja wartości elementu o podanym kluczu,
  • update-in – jak update, ale dla struktur zagnieżdżonych;
• wektory:
  • update – aktualizacja wartości elementu o podanym numerze indeksu,
  • update-in – jak update, ale dla struktur zagnieżdżonych.

Oczywiście w przypadkach, gdzie element złożonej struktury musi zostać zmieniony na bazie wartości stanowiącej element innej struktury ten sposób nie wystarczy. Warto wtedy korzystać z funkcji ensure, a zmian dokonywać z użyciem alter, aby mieć pewność, że w obrębie transakcji będziemy mieli do czynienia ze spójnym zestawem wartości, które nie zostaną zmodyfikowane w trakcie jej trwania.

Sterowanie odczytem

Ochrona przed zmianami, ensure

Funkcja ensure sprawia, że podany obiekt typu Ref będzie chroniony przed modyfikacjami, które mogłyby być efektem zatwierdzenia zmian przez równolegle realizowane transakcje. Uzyskana wartość będzie zawsze aktualną wartością współdzieloną. Jeżeli nie uda się takiej uzyskać od razu, transakcja będzie ponowiona, chociaż najczęściej dojdzie do sytuacji, w których równoległe transakcje (dokonujące aktualizacji danego Refa) będą ponawiane, dopóki transakcja z wywołaniem ensure się nie zakończy.

Funkcja ensure musi być wywołana w transakcji.

Jeżeli w transakcji użyto ensure w odniesieniu do obiektu typu Ref, a w trakcie jej realizowania inna transakcja jednak zmieniła wartość współdzieloną tego Refa (co nie powinno się wydarzyć), bieżąca transakcja zostanie ponowiona. Stanie się tak również wtedy, gdy równolegle realizowana transakcja zaktualizowała referencję zaraz przed tym, gdy w bieżącej transakcji doszło do wywołania ensure.

Wartością zwracaną jest wartość wewnątrztransakcyjna obiektu typu Ref.

Użycie:

• (ensure ref).

 1(def x (ref 1))
 2(def y (ref 2))
 3
 4(future
 5  (dosync
 6    (Thread/sleep 500)
 7    (println "2. zapis  [start] x =" @x "+ 1")
 8    (alter x inc)
 9    (println "2. zapis   [stop] x =" @x)))
10
11(dosync
12  (ensure x)
13  (println "1. odczyt [start]")
14  (Thread/sleep 2000)
15  (println "1. odczyt  [stop] x =" @x))
16
17; >> 1. odczyt [start]
18; >> 2. zapis  [start] x = 1 + 1
19; >> 2. zapis  [start] x = 1 + 1
20; >> 2. zapis  [start] x = 1 + 1
21; >> 1. odczyt  [stop] x = 1
22; >> 2. zapis  [start] x = 1 + 1
23; >> 2. zapis   [stop] x = 2

Sterowanie historią zmian

Max. długość historii, ref-max-history

Funkcja ref-max-history służy do pobierania bądź ustawiania wartości określającej maksymalną długość łańcucha historii zmian podanego obiektu typu Ref.

Pierwszym, obowiązkowym argumentem powinien być obiekt referencji transakcyjnej. Jeżeli nie podano więcej argumentów, funkcja odczyta bieżącą wartość określającą maksymalną długość łańcucha. Jeżeli podano kolejny argument, oczekuje się, że będzie to liczba całkowita stanowiąca nową wartość maksymalnej długości łańcucha.

Mechanizmy STM odpowiedzialne za utrzymywanie łańcucha historii zmian dla Refów aktualizowanych w transakcjach korzystają z ustalonej z użyciem ref-max-history wartości podczas automatycznego rozszerzania historii. Domyślną wartością jest 10, ale możemy ją zwiększyć przez co historia zmian będzie dłuższa, co może być konieczne w pewnych sytuacjach.

Funkcja zwraca maksymalną długość historii zmian dla podanego Refa.

Użycie:

• (ref-max-history ref),
• (ref-max-history ref długość).

 1(def x (ref 1))
 2(ref-max-history x 20)
 3
 4(future
 5  (Thread/sleep 350)
 6  (dotimes [n 5]
 7    (dosync
 8      (Thread/sleep 100)
 9      (alter x inc))))
10
11(dotimes [_ 3]
12  (dosync
13    (Thread/sleep 100)
14    (println "x =" @x)))
15
16; >> x = 1
17; >> x = 1
18; >> x = 2
19
20(def x (ref 1))
21(ref-max-history x 0)
22
23(future
24  (Thread/sleep 350)
25  (dotimes [n 5]
26    (dosync
27      (Thread/sleep 100)
28      (alter x inc))))
29
30(dotimes [_ 3]
31  (dosync
32    (Thread/sleep 100)
33    (println "x =" @x)))
34
35; >> x = 1
36; >> x = 1
37; >> x = 6

Min. długość historii, ref-min-history

Funkcja ref-min-history służy do pobierania bądź ustawiania wartości określającej minimalną długość łańcucha historii zmian podanego obiektu typu Ref.

Pierwszym, obowiązkowym argumentem powinien być obiekt referencji transakcyjnej. Jeżeli nie podano więcej argumentów, funkcja odczyta bieżącą wartość określającą minimalną długość łańcucha. Jeżeli podano kolejny argument, oczekuje się, że będzie to liczba całkowita stanowiąca nową wartość minimalnej długości łańcucha.

Mechanizmy STM odpowiedzialne za utrzymywanie łańcucha historii zmian dla Refów aktualizowanych w transakcjach korzystają z ustalonej z użyciem ref-min-history wartości podczas automatycznego rozszerzania historii. Domyślną wartością jest 0, ale możemy ją zwiększyć przez co historia będzie tworzona już przy pierwszej aktualizacji wartości danego Refa, a nie dopiero w reakcji na wcześniejsze konflikty i niepowodzenia.

Funkcja zwraca minimalną długość historii zmian dla podanego Refa.

Użycie:

• (ref-min-history ref),
• (ref-min-history ref długość).

 1(def x (ref 1))
 2(ref-min-history x 0)
 3
 4(future
 5  (Thread/sleep 350)
 6  (dotimes [n 5]
 7    (dosync
 8      (alter x inc))))
 9
10(dosync
11  (Thread/sleep 500)
12  (println "x =" @x))
13
14; >> x = 6
15
16(def x (ref 1))
17(ref-min-history x 20)
18
19(future
20  (Thread/sleep 350)
21  (dotimes [n 5]
22    (dosync
23      (alter x inc))))
24
25(dosync
26  (Thread/sleep 500)
27  (println "x =" @x))
28
29; >> x = 1

Długość historii, ref-history-count

Funkcja ref-history-count pozwala poznać aktualną długość historii zmian, czyli liczbę zapisanych, poprzednich wartości współdzielonych danego Refa.

Funkcja przyjmuje jeden obowiązkowy argument, którym powinien być obiekt typu Ref, a zwraca liczbę całkowitą wyrażającą liczbę elementów łańcucha historii zmian.

Użycie:

• (ref-history-count ref).

 1(def x (ref 1))
 2(ref-min-history x 20)
 3
 4(future
 5  (Thread/sleep 350)
 6  (dotimes [n 5]
 7    (dosync
 8      (alter x inc))))
 9
10(dosync
11  (Thread/sleep 500)
12  (ref-history-count x))
13
14; => 5

Walidatory

Referencje transakcyjne można opcjonalnie wyposażyć w mechanizmy testujące poprawność ustawianych wartości bieżących. Służą do tego funkcje set-validator! i get-validator.

Walidator (ang. validator) to w tym przypadku czysta (wolna od efektów ubocznych), jednoargumentowa funkcja, która przyjmuje wartość poddawaną sprawdzaniu. Jeżeli jest ona niedopuszczalna, funkcja powinna zwrócić wartość false lub wygenerować wyjątek.

Zarządzanie walidatorami, set-validator!

Funkcja set-validator! umożliwia ustawianie lub usuwanie walidatora dla podanego jako pierwszy argument obiektu typu Ref.

Jako drugi argument należy podać funkcję, która powinna przyjmować jeden argument i nie generować efektów ubocznych. Będzie ona wywoływana za każdym razem, gdy dojdzie do zmiany stanu referencji, a jako argument przekazywana będzie jej wartość, z którą zażądano powiązania. Funkcja ta powinna zwracać wartość false lub zgłaszać wyjątek, gdy przyjmowana wartość jest nieakceptowalna.

Pierwsze sprawdzenie stanu jest dokonywane już w momencie ustawiania walidatora i w wątku, w którym doszło do wywołania funkcji. Bieżąca wartość współdzielona wskazywana referencją musi więc już wtedy być odpowiednia. Kolejne wywołania walidatora odbywają się w transakcjach, podczas ich zatwierdzania (faza COMMITTING).

Jeżeli sprawdzanie poprawności nie powiedzie się, transakcja, w której ustawiono Refa zostanie ponowiona.

Jeżeli zamiast obiektu funkcyjnego podamy wartość nil, walidator zostanie odłączony od podanego obiektu typu Ref.

Funkcja set-validator! zwraca wartość nil, jeżeli udało się ustawić walidator.

 1(def referencja (ref 1))
 2(def druga      (ref 1))
 3
 4(set-validator! referencja pos?)  ;; akceptujemy tylko liczby dodatnie
 5
 6(dosync (ref-set referencja 2))   ; => 2
 7(dosync (ref-set referencja 1))   ; => 1
 8(dosync (ref-set referencja -1))  ; >> java.lang.IllegalStateException:
 9                                  ; >> Invalid reference state
10
11(dosync
12  (println "start")
13  (alter referencja dec)
14  (alter druga inc)
15  (println "stop"))
16
17; >> start
18; >> stop
19; >> java.lang.IllegalStateException: Invalid reference state
20
21@referencja                       ; => 1
22@druga                            ; => 1

Pobieranie walidatora, get-validator

Mając obiekt typu Ref, możemy pobrać funkcyjny obiekt jego walidatora, posługując się funkcją get-validator. Przyjmuje ona jeden argument, który powinien być obiektem typu Ref, a zwraca obiekt funkcji lub nil, jeżeli walidatora nie ustawiono.

Funkcja zwraca wartość nil.

Użycie:

• (get-validator ref).

1(def referencja (ref 0))
2(def nasz-walidator #(< % 5))
3
4(set-validator! referencja nasz-walidator)
5; => nil
6
7(get-validator referencja)
8; => #<user$nasz_walidator user$nasz_walidator@533f2bec>

Przykłady zastosowań

Globalne tożsamości

Korzystając ze zmiennych globalnych, jesteśmy w stanie tworzyć stałe tożsamości dla wyrażających zmienne stany (przez odniesienia do różnych wartości) obiektów typu Ref. W ten sposób możemy w całym programie odwoływać się do konkretnego Refa z użyciem nadanej mu symbolicznej nazwy o globalnym zasięgu.

Spróbujmy zaimplementować przykładowy licznik w postaci funkcji, która za każdym wywołaniem będzie zwiększała jego wartość o jeden, a jeżeli podamy argument, dokonane zostanie przypisanie konkretnej wartości początkowej:

 1(defonce ref-licznika (ref -1))
 2
 3(defn licznik
 4  ([x] (dosync (ref-set ref-licznika (int x))))
 5  ([]  (dosync (commute ref-licznika inc))))
 6
 7(licznik)        ; => 0
 8(licznik)        ; => 1
 9(licznik)        ; => 2
10
11(licznik 100)    ; => 100
12(licznik)        ; => 101

Przelew bankowy

Spójrzmy na przykładową implementację mechanizmu przelewu między rachunkami bankowymi z wykorzystaniem referencji transakcyjnych. Załóżmy, że mamy do dyspozycji dwa obiekty referencyjne reprezentujące rachunki bankowe: rachunek-a i rachunek-b. Wartościami bieżącymi tych rachunków będą mapy, zawierające odpowiednie, opisujące je elementy, włączając w to salda. Saldo wyrażane będzie liczbą zdawkowych jednostek waluty (np. groszy), aby uniknąć konieczności osobnego przeprowadzania operacji na jednostkach podstawowych i mniejszych oraz wykorzystania typów danych, które przechowują części ułamkowe.

Dodatkowo wprowadzimy też współdzieloną strukturę danych zawierającą historię operacji wykonywanych na rachunkach: wektor, do którego dopisywane będą kolejne operacje jako mapy. Każdy rachunek również będzie wyposażony w historię operacji. Jej elementami będą te same obiekty, co historii globalnej, jednak będą one przechowywane w postaci sortowanej mapy, gdzie kluczem sortowania jest znacznik czasowy raportowanej operacji.

  1;; 
  2;; dziennik operacji
  3
  4(def dziennik-operacji (ref []))
  5
  6;; 
  7;; pobieranie aktualnego czasu w milisekundach
  8
  9(def aktualny-czas #(System/currentTimeMillis))
 10
 11;; 
 12;; przeliczanie kwot na grosze i odwrotnie
 13
 14(defn PLN
 15  ([podstawa] (* 100 podstawa))
 16  ([podstawa reszta] (+ (* 100 podstawa) reszta)))
 17
 18(defn PLN-podziel
 19  [reszta]
 20  (let [podstawa (long (quot reszta 100))
 21        reszta   (long (mod  reszta 100))]
 22    [podstawa reszta]))
 23
 24(defn w-PLN
 25  [reszta]
 26  (/ reszta 100))
 27
 28;; 
 29;; odczytywanie numerów rachunków i sald
 30
 31(defn nr-rachunku
 32  [rachunek]
 33  (:numer @rachunek))
 34
 35(defn nr-rachunku-lub-nil
 36  [rachunek]
 37  (when-not (nil? rachunek) (:numer @rachunek)))
 38
 39(defn rachunek-nr?
 40  [rachunek]
 41  (and
 42   (= clojure.lang.Ref (type rachunek))
 43   (integer? (:numer @rachunek))))
 44
 45(defn rachunek-nr-lub-nil?
 46  [rachunek]
 47  (or (nil? rachunek) (rachunek-nr? rachunek)))
 48
 49(defn saldo
 50  [rachunek]
 51  (:stan @rachunek))
 52
 53(defn saldo-lub-nil
 54  [rachunek]
 55  (when-not (nil? rachunek) (:stan @rachunek)))
 56
 57;; 
 58;; obsługa raportowania
 59
 60(defn log-operacji
 61  [{:keys [czas rodzaj kwota tytułem komunikat]}]
 62  {:pre [(keyword? rodzaj)]}
 63  (let [kwota     (or kwota 0)
 64        komunikat (or komunikat tytułem)]
 65    {:rodzaj    rodzaj,
 66     :kwota     kwota,
 67     :komunikat komunikat}))
 68
 69(defn log-kasowa
 70  [args]
 71  (dosync
 72    (let [log (log-operacji args)
 73          s   (saldo-lub-nil (:rachunek args))]
 74      (assoc log :saldo s))))
 75
 76(defn log-przel
 77  [kierunek args]
 78  (dosync
 79    (let [log (log-operacji args)
 80          r-z (:na-rachunek args)
 81          r-n (:z-rachunku  args)
 82          [r2 r op] (if (= :z kierunek)
 83                        [r-z r-n :przelew-wychodzący]
 84                        [r-n r-z :przelew-przychodzący])
 85          s      (saldo-lub-nil r)
 86          nr-2   (nr-rachunku-lub-nil r2)]
 87      (assoc log :rachunek nr-2 :saldo s :rodzaj op))))
 88
 89(defn log-historia-kasowa
 90  [args]
 91  (dosync
 92    (let [nr    (nr-rachunku-lub-nil (:rachunek args))
 93          czas  (or (:czas args) (aktualny-czas))]
 94      [(assoc (log-operacji args)
 95              :rachunek nr
 96              :czas czas)])))
 97
 98(defn log-historia-przelew
 99  [args]
100  (dosync
101    (let [nr-z  (nr-rachunku-lub-nil (:z-rachunku  args))
102          nr-na (nr-rachunku-lub-nil (:na-rachunek args))
103          czas  (or (:czas args) (aktualny-czas))]
104      [(assoc (log-operacji args)
105              :z-rachunku  nr-z
106              :na-rachunek nr-na
107              :czas czas)])))
108
109;; 
110;; obsługa operacji kasowych
111
112(defn op-kasa
113  [op rodzaj rachunek {:keys [kwota] :as fargs}]
114  {:pre [(integer?      kwota)
115         (pos?          kwota)
116         (keyword?      rodzaj)
117         (fn?           op)
118         (rachunek-nr?  rachunek)]}
119  (dosync
120    (let [czas (aktualny-czas)
121          args (assoc fargs :rodzaj   rodzaj
122                            :czas     czas
123                            :rachunek rachunek)]
124      (alter   rachunek update-in [:stan] #(op % kwota))
125      (commute rachunek assoc-in  [:hist czas] (log-kasowa args))
126      (commute dziennik-operacji into (log-historia-kasowa args)))
127    (saldo-lub-nil rachunek)))
128
129(defn wpłata
130  [& {:keys [na-rachunek] :as args}]
131  (op-kasa + :wpłata na-rachunek (assoc args :komunikat "wp. własna")))
132
133(defn wypłata
134  [& {:keys [z-rachunku] :as args}]
135  (op-kasa - :wypłata z-rachunku (assoc args :komunikat "wyp. własna")))
136
137;;
138;; obsługa przelewów
139
140(defn przelew
141  [& {:keys [na-rachunek z-rachunku kwota tytułem] :as fargs}]
142  {:pre [(integer? kwota)
143         (pos?     kwota)]}
144  (dosync
145    (let [czas (aktualny-czas)
146          args (assoc fargs :rodzaj :przelew :czas czas)]
147      (alter   z-rachunku   update-in [:stan]      #(- % kwota))
148      (alter   na-rachunek  update-in [:stan]      #(+ % kwota))
149      (commute z-rachunku   assoc-in  [:hist czas] (log-przel :z  args))
150      (commute na-rachunek  assoc-in  [:hist czas] (log-przel :na args))
151      (commute dziennik-operacji into (log-historia-przelew args)))
152    [(saldo-lub-nil z-rachunku) (saldo-lub-nil na-rachunek)]))
153;; 
154;; dane rachunków
155
156(def rachunek-a (ref {:numer 123456,
157                      :stan 0,
158                      :hist (sorted-map)}))
159
160(def rachunek-b (ref {:numer 543454,
161                      :stan 0,
162                      :hist (sorted-map)}))
163
164;; 
165;; operacje na rachunkach
166
167(w-PLN (wpłata  :kwota (PLN 400) :na-rachunek rachunek-a))  ; => 400
168(w-PLN (wpłata  :kwota (PLN 200) :na-rachunek rachunek-b))  ; => 200
169(w-PLN (wypłata :kwota (PLN 100) :z-rachunku  rachunek-b))  ; => 100 
170
171(map w-PLN (przelew :z-rachunku  rachunek-a
172                    :na-rachunek rachunek-b
173                    :kwota       (PLN 200)
174                    :tytułem     "za jagody"))
175; => (200 300)
176
177;;
178;; stan po operacjach
179
180@rachunek-a
181
182; => {:numer 123456
183; =>  :stan 20000
184; =>  :hist {1440861812143 {:komunikat "wp. własna"
185; =>                            :kwota 40000
186; =>                            :rodzaj :wpłata
187; =>                            :saldo 40000}
188; =>             1440861823220 {:komunikat "za jagody"
189; =>                            :kwota 20000
190; =>                            :rachunek 543454
191; =>                            :rodzaj :przelew-wychodzący
192; =>                            :saldo 20000}}}
193
194@rachunek-b
195
196; =>{:numer 543454
197; => :stan 30000
198; => :hist {1440861812223 {:komunikat "wp. własna"
199; =>                           :kwota 20000
200; =>                           :rodzaj :wpłata
201; =>                           :saldo 20000}
202; =>            1440861812299 {:komunikat "wyp. własna"
203; =>                           :kwota 10000
204; =>                           :rodzaj :wypłata
205; =>                           :saldo 10000}
206; =>            1440861823220 {:komunikat "za jagody"
207; =>                           :kwota 20000
208; =>                           :rachunek 123456
209; =>                           :rodzaj :przelew-przychodzący
210; =>                           :saldo 30000}}
211
212@dziennik-operacji
213
214; => [{:czas 1440861812143
215; =>   :komunikat "wp. własna"
216; =>   :kwota 40000
217; =>   :rachunek 123456
218; =>   :rodzaj :wpłata}
219; =>  {:czas 1440861812223
220; =>   :komunikat "wp. własna"
221; =>   :kwota 20000
222; =>   :rachunek 543454
223; =>   :rodzaj :wpłata}
224; =>  {:czas 1440861812299
225; =>   :komunikat "wyp. własna"
226; =>   :kwota 10000
227; =>   :rachunek 543454
228; =>   :rodzaj :wypłata}
229; =>  {:czas 1440861823220
230; =>   :komunikat "za jagody"
231; =>   :kwota 20000
232; =>   :na-rachunek 543454
233; =>   :rodzaj :przelew
234; =>   :z-rachunku 123456}]
235
236;;
237;; sprawdzanie współbieżnego realizowania operacji
238
239(w-PLN (saldo rachunek-a))  ; => 200
240(w-PLN (saldo rachunek-b))  ; => 300
241
242(future
243  (Thread/sleep 500)
244  (dotimes [_ 500]
245    (Thread/sleep 10)
246    (przelew :z-rachunku  rachunek-a
247             :na-rachunek rachunek-b
248             :kwota       (PLN 100 50)
249             :tytułem     "za maliny")))
250
251(future
252  (Thread/sleep 500)
253  (dotimes [_ 500]
254    (Thread/sleep 20)
255    (przelew :z-rachunku  rachunek-b
256             :na-rachunek rachunek-a
257             :kwota       (PLN 100 50)
258             :tytułem     "za jabłka")))
259
260(future
261  (Thread/sleep 300)
262  (dotimes [_ 100]
263    (Thread/sleep 100)
264    (przelew :z-rachunku  rachunek-a
265             :na-rachunek rachunek-b
266             :kwota       (PLN 10 50)
267             :tytułem     "za ojczyznę")))
268
269(future
270  (Thread/sleep 200)
271  (dotimes [_ 100]
272    (Thread/sleep 40)
273    (przelew :z-rachunku  rachunek-b
274             :na-rachunek rachunek-a
275             :kwota       (PLN 10 50)
276             :tytułem     "za tych z Radomia")))
277
278(future
279  (dotimes [_ 200]
280    (Thread/sleep 100)
281    (wpłata :kwota (PLN 5 10) :na-rachunek rachunek-a)
282    (wpłata :kwota (PLN 15 1) :na-rachunek rachunek-b)))
283
284(future
285  (dotimes [_ 200]
286    (Thread/sleep 110)
287    (wypłata :kwota (PLN 5 10) :z-rachunku rachunek-a)
288    (wypłata :kwota (PLN 15 1) :z-rachunku rachunek-b)))
289
290(Thread/sleep 22000)
291
292(w-PLN (saldo rachunek-a))  ; => 200
293(w-PLN (saldo rachunek-b))  ; => 300
294
295;;
296;; obliczamy prędkość wykonywania przelewów
297
298(time (dotimes [n 6000]
299  (przelew :z-rachunku  rachunek-b
300           :na-rachunek rachunek-a
301           :kwota       (PLN n 1)
302           :tytułem     "za tych z Radomia")
303  (przelew :z-rachunku  rachunek-a
304           :na-rachunek rachunek-b
305           :kwota       (PLN n 1)
306           :tytułem     "za tych z Radomia")))
307
308; >> "Elapsed time: 818.965037 msecs"
309
310;; (ok. 12000 koordynowanych operacji między rachunkami na sekundę
311;;  przy 1 rdzeniu)