Title: Przetwarzanie równoległe i strumieniowe

PRS

Przetwarzanie równoległe i strumieniowe

Wykład 4

Anomalie związane z przetwarzaniem równoległym, zagłodzenie, race condition, zakleszczenie

Teoria vs praktyka

https://devrant.com/rants/88296/knock-knock-race-condition-whos-there

Zaliczenie

Priorytety wątków

Modul Thread w języku Python jest luźno oparty na modelu wątków Javy. Klasa Thread w Pythonie obsługuje podzbiór zachowania klasy Thread w Javie.

W Javie blokady są powiązane z obiektami, natomiast w Python są one osobnymi obiektami (Lock,RLock).

Obecnie w Python nie ma priorytetów ani grup wątków, a wątków nie można niszczyć, zatrzymywać, zawieszać, wznawiać ani przerywać.

Wątki

Atomowość operacji i zmiennych

Operacja atomowa to jedna lub sekwencja instrukcji kodu, które są wykonywane bez przerw.

Wątek nie może zostać przełączony kontekstu w środku operacji atomowej.

Określenie zmiennej jako atomowej oznacza to, że dostęp do zmiennej jest bezpieczny dla wątków i że wątek nie może zostać przełączony kontekstowo w trakcie uzyskiwania dostępu do wartości zmiennej.

Atomowość w Python

Wiele operacji w Pythonie ma charakter atomowy.

Atomowe są:

- przypisania wartości do zmiennych i parametrów
- dodawanie, usuwanie, odzwyskanie, pobieranie, ściągane, sortowanie elementów na listach jest atomowe.
- przypisanie wartości do klucza w słowniku jest atomowe.

Atomowość w Python

a = 1;
b.value = 33

d[x] = 22
d.update(a) # add dict to a dict
keys = d.keys()

l.append(3)
l.extend(b)
value = l[0]
value = l.pop()
l[1:4] = a
l.sort()

Atomowość w Python

Większość operacji w Pythonie nie jest atomowa. Oznacza to, że te operacje nie są bezpieczne dla wątków.

Dodawanie lub odejmowanie wartości od zmiennej, takiej jak zmienna całkowita, nie jest atomowe.

Powodem tego jest to, że w grę wchodzą co najmniej trzy operacje, są to:

- Odczytaj wartość zmiennej.
- Oblicz nową wartość zmiennej
- Przypisz obliczoną wartość zmiennej.

Łączenie dostępu i przypisania wartości na liście lub słowniku nie jest atomowe.

Atomowość w Python

# adding and subtracting is not atomic
a = a + 1

>>> import dis
>>> dis.dis("self.value += 1")
  1           0 LOAD_NAME                0 (self)
              2 DUP_TOP
              4 LOAD_ATTR                1 (value)
              6 LOAD_CONST               0 (1)
              8 INPLACE_ADD
             10 ROT_TWO
             12 STORE_ATTR               1 (value)
             14 LOAD_CONST               1 (None)
             16 RETURN_VALUE

# access and assign is not atomic
a[0] = b[0]

Atomowość w Python

Chociaż niektóre operacje w Pythonie są atomowe, nigdy nie powinniśmy polegać na tym, że operacja jest atomowa.

Błąd inkrementacji dla 2 wątków:

Thread-1 reads the value as 23
Thread-1 adds 1 to 23 and get 24
Thread-2 reads the value as 23
Thread-1 stores 24 in value
Thread-2 adds 1 to 23
Thread-2 stores 24 in value

Atomowość w Python

Istnieje wiele powodów by nie zakładać domyślnie atomowości:

- Operacje mogą nie być atomowe, jeśli kod jest wykonywany przez inny interpreter języka Python
- Zachowanie referencyjnego interpretera języka Python może ulec zmianie w przyszłości
- Inni programiści, którzy muszą czytać twój kod, mogą nie być tak dobrze zaznajomieni z operacjami atomowymi w Pythonie.
- Możesz wprowadzić bardziej rozbudowane race condition, np. operacje są atomowe, ale już wiele takich operacji po sobie nie jest wpólnie chronionych.

Wymuszenie "atomowości"

# protect a critical section

with lock:

# ....

Użycie blokady do ochrony bloku kodu nie zapobiega przełączaniu kontekstu wątku w środku instrukcji lub pomiędzy instrukcjami w bloku.

Zamiast tego zapobiega wykonywaniu tego samego bloku przez inne wątki, podczas gdy wątek utrzymuje blokadę.

Volatile

Słowo kluczowe Java volatile służy do oznaczania zmiennej Java jako „przechowywanej w pamięci głównej”.

Oznacza to, że każdy odczyt zmiennej volatile będzie odczytywany z pamięci głównej komputera, a nie z pamięci podręcznej procesora, oraz że każdy zapis zmiennej będzie zapisywany do pamięci głównej, a nie tylko do pamięci podręcznej procesora .

Słowo kluczowe volatile gwarantuje widoczność zmian w zmiennych w różnych wątkach.

Volatile używany we współczesnych skompilowanych językach programowania, takich jak Java i C#.

Volatile

Dla języków kompilowanych:

Odczyty i zapisy do innych zmiennych nie mogą zostać zmienione tak, aby następowały po zapisie do zmiennej volatile, jeśli odczyty/zapisy pierwotnie miały miejsce przed zapisem do zmiennej volatile.

Odczyty i zapisy do innych zmiennych nie mogą zostać zmienione tak, aby miały miejsce przed odczytem zmiennej volatile, jeśli odczyty/zapisy pierwotnie miały miejsce po odczycie zmiennej volatile.

Volatile nie wystarczy!

Wątek 1 odczytuje współdzieloną zmienną licznika o wartości 0 do pamięci podręcznej procesora, zwiększa ją do 1 i nie zapisuje zmienionej wartości z powrotem do pamięci głównej.

Wątek 2 mógłby następnie odczytać tę samą zmienną licznika z pamięci głównej, gdzie wartość zmiennej nadal wynosi 0, do własnej pamięci podręcznej procesora.

https://jenkov.com/tutorials/java-concurrency/volatile.html

Volatile w Python

Python nie ma zmiennych volatile.

Interpreter Python zawsze odwołuje się do zmiennych z pamięci głównej.

Dostęp do wartości zmiennej w Pythonie jest zawsze atomowy.

Obecność globalnej blokady interpretera (GIL) w interpreterze Pythona (CPython) oznacza również, że tylko jeden wątek może jednocześnie wykonywać kod python. Dlatego podczas uzyskiwania dostępu do zmiennej współdzielonej między wątkami wiemy, że nie może ona być aktualizowana jednocześnie przez inny wątek.

Problemy przetwarzania równoległego

https://www.reddit.com/r/ProgrammerHumor/comments/bzv8q4/parallelism_be_like/

Race condition

Race condition występuje, gdy co najmniej dwa wątki mogą uzyskać dostęp do tych samych danych i jednocześnie próbują je zmienić.

Ponieważ algorytm planowania wątków może przełączać się między wątkami w dowolnym momencie, nie znasz kolejności, w jakiej wątki będą próbowały uzyskać dostęp do udostępnionych danych. Dlatego wynik zmiany danych jest zależny od algorytmu planowania wątków, tj. oba wątki „ścigają się” w celu uzyskania dostępu/zmiany danych.

http://www.quickmeme.com/meme/36fvwp

Race condition - przykład

Sprzedajmy bilety na wydarzenie i zostaje nam jeden bilet do sprzedaży. Dwóch klientów w jednym momencie próbuje dokonywać zakupu. Jeżeli oba wątki przetwarzania odczytają, że jeszcze jeden bilet jest dostępny i następnie przejdą do kolejnych kroków może się okazać że oba dokonają sprzedaży biletu i na końcu sprzedamy o jeden więcej bilet niż powinniśmy.

https://dev.to/victorhazbun/best-practices-avoid-race-conditions-i31

Race condition - przykład

if (x == 5):
    y = x * 2; # co jak ktoś zmieni wartość x w międzyczasie?


# Uzyskaj blokadę dla x
if (x == 5):
    y = x * 2; # Teraz nic nie może zmienić x,
               # dopóki blokada nie zostanie zwolniona.
               # Dlatego y = 10
# zwolnij blokadę dla x

Przykładem jest np. wielokrotne wzbudzenie alarmu na podstawie tego samego odczytu z sensora.

Race condition - przeciwdziałanie

Programiści mogą zapobiegać race condition w systemach operacyjnych i innym oprogramowaniu na dwa sposoby:

- Przez unikanie dzielenia zasobów. Oznacza to przeglądanie kodu w celu upewnienia się, że gdy współużytkowane zasoby są częścią systemu lub procesu, interakcja z nimi jest atomowa. Można również używać obiektów atomowych, lub niezmienialnych (immutable).
- Przez synchronizacje wątków - dana część programu może wykonywać tylko jeden wątek na raz:

Zakleszczenie (deadlock)

Zakleszczenie występuje, gdy co najmniej dwa wątki czekają w nieskończoność na blokadę lub zasób utrzymywany przez inny z wątków.

W związku z tym aplikacja może się zatrzymać lub zakończyć niepowodzeniem, ponieważ przetwarzanie nie może być kontynuowane prez zakleszczenie wątków.

https://devrant.com/rants/1756955/this-is-the-real-deadlock

Zakleszczenie (deadlock)

https://www.geeksforgeeks.org/deadlock-in-java-multithreading/

Zakleszczenie (deadlock) - przeciwdziałanie

Zakleszczenie może się pojawiać gdy wątek do działania potrzebuje uzyskać dostęp do więcj niż jednej blokady (locka).

Jak przeciwdziałać zakleszczeniu:

- używając interfejsu lock.acquire(False) na obiekcie Lock, który próbuje pobrać blokadę a w przypadku niepowodzenia po jakimś czasie przechodzi dalej
- implementując pobieranie blokad w taki sposób, by każdy z wątków pobierał je w tej samej kolejności

Zagłodzenie (starvation)

O zagłodzeniu mówimy, gdy określony wątek nie uzyskuje dostępu do obiektu lub zasobu, co prowadzi do wydłużenia czasu oczekiwania i wykonania.

Dzieje się tak, gdy współdzielone zasoby są niedostępne przez długi czas przez „chciwe” wątki. Załóżmy na przykład, że obiekt udostępnia zsynchronizowaną metodę, której zwrócenie często zajmuje dużo czasu.

Jeśli jeden wątek często wywołuje tę metodę, inne wątki, które również wymagają częstego zsynchronizowanego dostępu do tego samego obiektu, będą mogą być przez niego blokowane.

https://imgflip.com/i/5gasaq

Zagłodzenie (starvation)

Przyczyn zagłodzenia wątków może być wiele, niektóre z nich opisano poniżej:

- Działający wątek o wysokim priorytecie: Może zaistnieć przypadek, w którym wątek o wysokim priorytecie, zajmuje CPU i wymaga intensywnego przetwarzania, które wymaga dużo czasu na ukończenie, więc aby ta praca została całkowicie wykonana, inne wątki o niskim priorytecie muszą czekać przez długi czas, co prowadzi do zagłodzenia.

Zagłodzenie (starvation)

- Może zaistnieć przypadek, w którym kolejność, w której wątki mogą wejść do zsynchronizowanego bloku, ma ustaloną kolejność, co powoduje oczekiwanie na zasoby i obiekty przez inny wątek prowadzący do zagłodzenia.
- Wątki oczekujące na obiekt pozostaje w nieskończoność: metoda notify() nie śledzi w wątkach tego, który konkretny wątek jest wybudzany, jeśli istnieje wiele wątków, może istnieć ryzyko, że którykolwiek z wątków jest przetwarzany, a inne wątki nigdy nie są wywoływane do wykonania.

Zagłodzenie (starvation) - przeciwdziałanie

Pomocą w załodzeniu może być jawne (za pomocą Lock) pozyskiwanie i zwalnianie zasobu zamiast stosowania menedżera kontekstu (with lock) gdyż dodatkowa optymalizacja kodu w maszynie wirtualnej Pythona może skutkować mniej sprawiedliwym dostępem do sekcji krytycznej.

Zamiast wait() i notify() możemy korzystać z semaforów które kolejkują wątki czekające na dostęp do semafora.

Livelock

Blokada na żywo to przypadek niepowodzenia współbieżności, w którym wątek nie jest zablokowany, ale nie może poczynić postępów z powodu działań innego wątku.

Livelock zwykle obejmuje dwa lub więcej wątków, które współużytkują blokady wzajemnego wykluczania.

Oba wątki mogą jednocześnie próbować uzyskać blokadę lub serię blokad, wykryć, że konkurują o ten sam zasób, a następnie wycofać się. Ten proces jest powtarzany i żaden wątek nie jest w stanie poczynić postępów i zostaje „zablokowany”.

http://15418.courses.cs.cmu.edu/spring2013/article/27

Livelock przykład

Jest to porównywalne z dwojgiem ludzi, którzy próbują się minąć na korytarzu:

Osoba A przesuwa się w lewo, aby przepuścić osobę B, podczas gdy B przesuwa się w prawo, aby przepuścić A. Widząc, że nadal się blokują, A przesuwa się w prawo, a B w lewo. Wciąż się blokują i tak w nieskończoność.

Livelock przeciwdziałania

Livelocks można uniknąć, wykorzystując RLock jako sposób na określenie, który wątek czeka dłużej, aby można było przypisać mu blokadę.

W ramach najlepszej praktyki jest nie gromadzenie blokad; jeśli wątek nie może uzyskać blokady, powinien zwolnić wcześniej nabyte blokady, aby spróbować ponownie później. Blokady powinny być też pobierane w tej samej kolejności przez różne wątki!

Przykłady ograniczeń przetwarzania równoległego

* Ograniczenie: zadania przychodzą w określonej kolejności, podczas przetwarzania kolejność nie ma znaczenia - ale po zakończeniu przetwarzania kolejność zadań musi zostać przywrócona.

* Ograniczenia kolejnościowe, zadania danego typu powinny być zawsze wykonane w tej samej kolejności, w której przychodzą.

* Ograniczenia dostępu do zasobów (sekcja krytyczna), tylko jeden wątek może na raz korzystać np. z jakiegoś urządzenia itp.

* Ograniczenia na liczbę wątków których uruchomienie w systemie przyspiesza przetwarzanie danych. Dlatego nie możemy np. tworzyć nowego wątku dla każdego nowego zapytania / nowo przychodzących danych.

* Priorytet - niektóre zadania powinny być wykonywane z wyższym priorytetem niż inne

Moduł multiprocessing

multiprocessing to pakiet obsługujący tworzenie procesów przy użyciu interfejsu API podobnie do modułu thread.

Pakiet multiprocessing oferuje zarówno lokalną, jak i zdalną współbieżność, skutecznie omijając globalną blokadę interpretera, używając podprocesów zamiast wątków.

Dzięki temu moduł wieloprocesorowy pozwala programiście na pełne wykorzystanie wielu procesorów na danej maszynie. Działa zarówno na systemie Unix, jak i Windows.

Moduł multiprocessing

W przetwarzaniu multiprocessing procesy są tworzone przez utworzenie obiektu Process, a następnie wywołanie jego metody start(). Proces podąża za interfejsem API Threading.Thread.

Przykładem programu wieloprocesowego jest

from multiprocessing import Process

def f(name):
    print('hello', name)

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

Moduł multiprocessing

W zależności od platformy przetwarzanie wieloprocesowe obsługuje różne sposoby uruchamiania procesu:

spawn -  Proces nadrzędny uruchamia nowy proces interpretera Pythona. Proces potomny odziedziczy tylko te zasoby, które są niezbędne do uruchomienia metody run() obiektu procesu. Rozpoczęcie procesu przy użyciu tej metody jest wolniejsze w porównaniu z użyciem fork. Dostępne w systemach Unix i Windows. Domyślnie w systemach Windows i macOS.

fork -  Proces nadrzędny używa os.fork() do wielowątkowego uruchomienia interpretera Pythona. Proces potomny, gdy się zaczyna, jest faktycznie identyczny z procesem nadrzędnym. Wszystkie zasoby rodzica są dziedziczone przez proces potomny. Dostępne tylko w systemie Unix. Domyślnie w systemie Unix.

Moduł multiprocessing - współdzielone dane

Dane mogą być przechowywane na mapie pamięci współdzielonej przy użyciu Value lub Array. Na przykład:

from multiprocessing import Process, Value, Array

def f(n, a):
    n.value = 3.1415927
    for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))
    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()
    print(num.value)
    print(arr[:])

Moduł multiprocessing

Moduł multiprocessing wprowadza również interfejsy API, które nie mają odpowiedników w module wątków.

Doskonałym tego narzędziem jest obiekt Pool, który oferuje wygodny sposób zrównoleglenia korzystania z funkcji dla wielu wartości wejściowych, dystrybuując dane wyjściowe między procesami (równoległość danych).

Moduł multiprocessing

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

Wypisze:

[1, 4, 9]