Title: Przetwarzanie równoległe i strumieniowe

PRS

Przetwarzanie równoległe i strumieniowe

Wykład 3

Implementacja wątków w Python, synchronizacja, semafory, lock.

Wątki Python

Klasa Thread reprezentuje działanie uruchamiane w osobnym wątku kontroli.

Działanie można określić na dwa sposoby: przekazując funkcję do konstruktora lub przesłaniając metodę run() w podklasie.

Żadne inne metody (z wyjątkiem konstruktora) nie powinny być nadpisywane w podklasie. Innymi słowy, nadpisuj tylko metody __init__() i run() tej klasy.

Wątki Python

Przekazywanie funkcji do konstruktora:

t = threading.Thread(target=thread_function, args=(1,))
t.start()

Kiedy tworzysz wątek, przekazujesz mu funkcję i listę zawierającą parametry tej funkcji.

Wątki Python

Podklasa klasy Thread :

import threading
import time

class MyThread(threading.Thread):

    def run(self):
        print("Working")
        time.sleep(5)
        return

Nadpisujemy metodę run().

Wątki - metody klasy Thread

start()
Uruchamia wątek. Musi być wywołany co najwyżej raz na obiekt wątku. Wywołuje metodę run() obiektu w osobnym wątku. Ta metoda zgłosi RuntimeError, jeśli zostanie wywołana więcej niż raz w tym samym obiekcie wątku.

run()
Uruchamia wątek. Możesz zastąpić tę metodę w podklasie. Standardowa metoda run() wywołuje funkcje przekazywaną do konstruktora obiektu jako argument docelowy, jeśli istnieje, z argumentami pozycyjnymi i słowami kluczowymi wziętymi odpowiednio z argumentów args i kwargs.

Start tworzy nowy wątek, podczas gdy run nie tworzy żadnego wątku i po prostu wykonuje operacji w bieżącym wątku.

Wątki - metody klasy Thread

name
String używany wyłącznie do celów identyfikacyjnych. Wiele wątków może mieć tę samą nazwę. Początkowa nazwa jest ustawiana przez konstruktor.

demon

Wartość logiczna wskazująca, czy ten wątek jest wątkiem demona (prawda), czy nie (fałsz). Cały program w Pythonie kończy działanie, gdy nie ma już żywych wątków innych niż demony.

getName()
setName()
Przestarzały interfejs API pobierający/ustawiający nazwę; zamiast tego użyj go bezpośrednio jako właściwości.

isDaemon()
setDaemon()
Przestarzałe API pobierające/ustawiające dla demona; zamiast tego użyj go bezpośrednio jako właściwości.

Przestarzałe od wersji 3.10.

Wątki Python - join()

Aby powiedzieć jednemu wątkowi, aby czekał na zakończenie innego wątku, wywołujesz .join().

# t.join()


W tym wypadku wątek, który wywołał tę metodę poczeka w tym miejscu na zakończenie wątku t.

Funkjca ta jest przydatna do synchronizowania zakończenia głównego programu dopiero w momencie gdy wszystkie stworzone przez nas wątki się zakończą.

Wątki metody modułu threading

Threading.enumerate()
Zwróć listę wszystkich aktualnie aktywnych obiektów Thread. Lista zawiera demoniczne wątki i obiekty wątków utworzone przez current_thread(). Wyklucza zakończone wątki i wątki, które nie zostały jeszcze rozpoczęte. Jednak główny wątek jest zawsze częścią wyniku, nawet po zakończeniu.

Threading.active_count()
Zwraca liczbę aktualnie aktywnych obiektów Thread. Zwrócona liczba jest równa długości listy zwróconej przez enumerate(). Funkcja activeCount jest przestarzałym aliasem tej funkcji.

Threading.main_thread()
Zwróć główny obiekt wątku. W normalnych warunkach głównym wątkiem jest wątek, z którego został uruchomiony interpreter Pythona.

Wątki local data

Thread-local data to dane, których wartości są związane z wątkiem. Aby zarządzać danymi lokalnymi wątku, po prostu utwórz instancję local (lub podklasę) i przechowuj na niej atrybuty:

mydata = threading.local()
mydata.x = 1

 Inne wątki nie mają dostępu do tych danych.

Wątki sleep()

Python ma moduł o nazwie time, który udostępnia kilka przydatnych metod do obsługi zadań związanych z czasem.

Jedną z najpopularniejszych metod wśród nich jest sleep().

time.sleep(2.4)

Metoda sleep() wstrzymuje wykonywanie bieżącego wątku na określoną liczbę sekund.

Wątki Python

Zadania, które spędzają większość czasu na oczekiwaniu na zdarzenia zewnętrzne, są dobrymi kandydatami do tworzenia wątków.

Jeśli używasz standardowej implementacji Pythona, piszesz tylko w Pythonie to interpreter może wymuszać działanie tylko jednego wątku na raz, w celu równoległości powinieneś więc wykorzystywać moduł multiprocessing.

multiprocessing będzie tematem kolejnych zajęć.

Synchronizacja

Synchronizacja to wspólne działanie dwóch lub więcej wątków, które zapewnia, że ​​każdy wątek osiągnie znany punkt działania w stosunku do innych wątków przed kontynuowaniem działania.

Innymi słowy to możliwość kontrolowania dostępu wielu wątków do dowolnego współdzielonego zasobu.

Próba współdzielenia zasobów bez poprawnego użycia synchronizacji jest najczęstszą przyczyną błędnego działania aplikacji.

Wątki - synchronizacja

W celu kontroli nad wywołaniem programów wielowątkowych stosujemy tzw. metody synchronizacji.

Głównie implementacja korzysta z obiektów, które chronią dostęp do kodu lub zasobów umożliwiając ograniczonej liczbie (najczęściej jednemu) wątkowi w tym samym momencie na dostęp.

Przykładem obiektu służącego do synchronizacji jest Lock. W Pythonie jest to metoda synchronizacji najniższego poziomu, zaimplementowana bezpośrednio przez moduł rozszerzenia _thread.

Wątki - synchronizacja Lock

Zamek (lock) znajduje się w jednym z dwóch stanów: „zablokowany” lub „odblokowany”. Jest tworzony w stanie odblokowanym. Ma dwie podstawowe metody, acquire() i release().

Wątki - synchronizacja Lock

Gdy stan jest odblokowany, acquire() zmienia stan na zablokowany i natychmiast zwraca sterowanie.

Kiedy stan jest zablokowany, acquire() blokuje się do momentu, gdy wywołanie release() w innym wątku zmieni go na odblokowany, wtedy wywołanie acquire() zresetuje go do zablokowanego.

Metoda release() powinna być wywoływana tylko w stanie zablokowanym; zmienia stan na odblokowany i natychmiast wraca. Jeśli zostanie podjęta próba zwolnienia odblokowanej blokady, zostanie zgłoszony RuntimeError.

Wątki - synchronizacja RLock

Lock jest obiektem synchronizacji, który nie jest własnością określonego wątku po zablokowaniu. Reentrant lock jest mechanizmem synchronizacji, który może być uzyskiwany wiele razy przez ten sam wątek.

Wewnętrznie wykorzystuje koncepcje „posiadania wątku” i „rekurencji” oprócz stanu zablokowania/odblokowania używanego przez blokady lock.

W stanie zablokowanym jakiś wątek posiada blokadę; w stanie odblokowanym żaden wątek nie jest jego właścicielem.

Rlock posiada te same metody co Lock.

Producent - konsument

Producent

Konsument

Producent - konsument

Klasycznym wzorcem projektowym programowania współbieżnego jest wzorzec producent-konsument,

Producenci są odpowiedzialni za dodanie danych do jakiejś udostępnionej globalnie struktury danych, a konsumenci są odpowiedzialni za odczytywanie danych z tej struktury.

Tylko jedna strona, albo pojedynczy producent, albo pojedynczy konsument, może mieć dostęp do struktury w danym momencie.

Zaliczenie

Producent - konsument

https://jenkov.com/tutorials/java-concurrency/producer-consumer.html

Producent - konsument

Niezbędnym elementem wzorca producent konsumer jest struktura danych (bufor), która łączy producentów i konsumentów. Powinna być ona bezpieczna dla wątków. Zazwyczaj będzie to kolejka komunikatów.

https://commons.wikimedia.org/wiki/File:Data_queue.svg

Producent - konsument

Jeśli w buforze udostępnionym nie ma elementów, zadanie konsumenta może czekać lub blokować, dopóki elementy nie staną się dostępne.

Wspólny bufor może być ograniczony lub nie, co oznacza, że ​​może mieć ograniczony rozmiar. Jeśli tak, zadania producenta mogą wymagać oczekiwania lub zablokowania, aż zwolni się miejsce na dodanie nowych elementów.

Producent - konsument

Zadania producenta mogą sygnalizować zadaniom konsumenta, że ​​żadne dalsze elementy nie będą generowane, w takim przypadku system może zostać zamknięty.​​​​​​

Producent - konsument

Zadania producenta i konsumenta można zaimplementować za pomocą wątków, a wspólny bufor można zaimplementować za pomocą kolejki.

# producer task
def producer(kolejka):
	while True:
		# stwórz wiadomość
		item = ...
		# prześlij wiadomość
		buffer.push(item)

# consumer task
def consumer(kolejka):
	while True:
		# odbierz wiadomość
		item = buffer.pop()
		# procesuj wiadomość
		# ...

...
# współdzielona kolejka
kolejka = queue.Queue()

Producent - konsument przykład

Jednym z przykładów typowej relacji producent-konsument jest buforowanie wydruku.

Chociaż drukarka może nie być dostępna, gdy chcesz drukować z aplikacji (tj. producenta), nadal możesz „ukończyć” zadanie drukowania, ponieważ dane są tymczasowo umieszczane na dysku, dopóki drukarka nie będzie dostępna.

Podobnie, gdy drukarka (tj. konsument) jest dostępna, nie musi czekać, aż aktualny użytkownik zechce wydrukować. Zbuforowane zadania drukowania można wydrukować, gdy tylko drukarka stanie się dostępna.

Wątki - Condition objects

W przypadku współbieżności condition (nazywany również monitorem) umożliwia powiadamianie wielu procesów (lub wątków) o pewnym wyniku.

Łączy w sobie zarówno blokadę wzajemnego wykluczania (lock), jak i zmienną warunkową.

Condition może służyć do ochrony sekcji krytycznej, ale nie może być używany do ostrzegania innych procesów, że warunek został zmieniony lub spełniony.

Wątki - Condition objects

Condition może zostać uzyskany przez proces, po czym może czekać na powiadomienie przez inny proces, że coś się zmieniło. Podczas oczekiwania proces jest blokowany i zwalnia blokadę dla innych procesów.

Następnie inny proces może uzyskać condition, wprowadzić zmianę i powiadomić jeden, wszystkie lub podzbiór procesów oczekujących na warunek, że coś się zmieniło.

Oczekujący proces może następnie się obudzić, ponownie uzyskać condition i wykonać wymagane działania.

Wątki - Condition objects

Wątek jest wstrzymywany przez operację wait().

Wiele wątków może w tym samym momencie czekać na wait() na wzbudzenie.

Wzbudzenie wykonuje się poprzez wykonanie notify() przez inny wątek. Tylko jeden z oczekiwanych wątków zostanie wybudzony!

Aby wybudzić wszystkie wątki należy wykonać notify_all(). Jeżeli wątki będą w contekscie blokady to będa one wtedy wspólnie "walczyć" między sobą o blokadę.

Wątki - Semafory

W Lock i RLock w danym momencie może działać tylko jeden Wątek, ale czasami naszym wymaganiem jest wykonanie określonej liczby Wątków na raz.

Załóżmy, że musimy zezwolić 10 członkom na dostęp do bazy danych, a możemy utworzyć tylko 4 sesje połączenia z bazą jednocześnie.

Do obsługi tego typu wymagań nie możemy użyć koncepcji Lock i RLock i tutaj powinniśmy przejść do Semaphore. Semafor może służyć do ograniczania dostępu do współdzielonych zasobów o ograniczonej pojemności.

Wątki - Semafory

Semafor posiada stan początkowy.

Wykonanie acquire() zmniejsza stan o 1.

Wykonanie release() zwiększa stan o 1.

Po dojściu do stanu zero kolejne wątki próbujące wykonać acquire() będą blokowane aż nie wpuści ich ktoś inny wykonują relase().