0. 2022_II_ST_NPR

Repozytorium przeznaczone na przedmiot Narzędzia Przetwarzania Rozproszonego (NPR) na drugim stopniu studiów.

1. Implementacja rozproszonego monitora w ØMQ

1.0 Wymagania

Python
PyZmq (https://github.com/zeromq/pyzmq)

$ pipenv install pyzmq

1.1 Krótki opis implementacji

W skład implementacji (znajdującej się w pliku Distr_Monit.py) wchodzą dwa obiekty:

ExchangeMsg() --> Komunikat wymiany, który służy do tworzenia wiadomości,
przesyłanych pomiędzy procesami.
DistributedMonitor() --> Rozproszony monitor, oferujący mechanizm synchronizacji
w systemie rozproszonym bazujący głównie na funkcji acquire (zdobycia dostępu do zasobu)
oraz funkcji release (zwolnienie dostępu do zasobu).
Do komunikacji międzyprocesowej wykorzystuje mechanizm ZeroMQ (0MQ) w schemacie PUB-SUB.
Zaimplementowany algorytm wzajemnego wykluczania: suzuki-kasami.

1.2 Opis klasy ExchangeMsg()

# Przesyłana wiadomość
class ExchangeMsg():
    def __init__(self, rcvfrom, Q=None, LN=None, 
        SD=None, sendto=None, rni=None):
        # Nadawca wiadomości (od kogo odebraliśmy wiadomość)
        self.rcvfrom = rcvfrom
        # Zaktualizowana wartość tablicy RNi[rcvfrom]
        self.rni = rni
        # Kolejka procesów żądających tokenu
        self.Q = Q
        # Tablica LN zawierająca lp. ostatnio wykonywanych żądań
        self.LN = LN
        # Shared_Data - zmienna/obiekt współdzielony
        self.SD = SD
        # Odbiorca wiadomości (do kogo wiadomość wysyłamy)
        self.sendto = sendto 
# Przesyłane są komunikaty 4 typów (rozróżniane na podstawie warunków ich zawartości):
#   1.0 Komunikat budzenia - Warunki: rcvfrom = None
#       1.1 sendto = None --> wszystkie procesy oczekujące [metodą going_sleep() lub wait()]
#       są budzone, aby mogły wznowić przetwarzanie.
#       1.2 sendto jest sprecyzowane --> konkretny proces wznawia przetwarzanie.
#   2.0 Komunikat aktualizacji tablicy RN - Warunki: rcvfrom i rni jest sprecyzowane.
#   3.0 Komunikat przesłania tokenu - Warunki: rcvfrom i sendto jest sprecyzowane, a rni = None.
#   4.0 Komunikat zasypiania - Warunki: rcvfrom jest sprecyzowane, a rni i sendto jest równe None.

1.3 Opis klasy DistributedMonitor()

1.3.1 Wstęp

# Monitor rozproszony wykorzystujący:
# Do komunikacji - mechanizm ZMQ PUB-SUB
# Do wzajemnego wykluczania - algorytm suzuki-kasami 
class DistributedMonitor():
    # Init monitora rozproszonego (fr. implementacji)
    def __init__(self, id_ip_port, is_token_acquired, coworkers, init_shared_data):
        # Init zmiennych wykorzystywanych w algorytmie suzuki-kasami oraz przy komunikacji zmq
        # (...)
    # Init rozproszonego monitora (wykorzystanie)
    myDistributedMonitor = DistributedMonitor(
        id_ip_port, # łańcuch znaków: "AdresIP:PORT" wykorzystywany przez dany proces do komunikacji 
        is_token_acquired, # wartość boolowska True or False wykorzystywana do poinformowania, 
        # czy dany proces zaczyna pracę nad współdzielonymi danymi w systemie posiadając token. 
        coworkers, # lista wszystkich identyfikatorów współpracowników (z aktualnym procesem włącznie)
        init_shared_data # początkowa wartość/obiekt zmiennej współdzielonej  
        )

Po utworzeniu obiektu monitora rozproszonego "myDistributedMonitor" rozpoczyna się próba podłączenia do systemu rozproszonego. Inicjowane jest gniazdo publikujące oraz subskrybujące, a także oczekiwanie na podłączenie się wszystkich współpracowników do systemu i połączenie do nich gniazda subskrybenta. W wątku demonicznym działającym w tle odbierane i przetwarzane są komunikaty (opisane w pkt. 1.2.).

1.3.2 Funkcje użytkowe

    # Zdobycie dostępu do zasobu (zmiennej współdzielonej)
    # Wynik działania funkcji: zwrócenie zmiennej współdzielonej
    # NOTE: Funkcja będzie w stanie się zablokować,
    # jeżeli proces nie posiada tokenu, aż wątek odbiornika
    # odbierze token (zawierający zmienną współdzieloną)
    def acquire(self): 
        #(...)
    # Zwolnienie dostępu do zasobu (zmiennej współdzielonej)
    # Wynik działania funkcji: funkcja nic nie zwraca,
    # wychodzi z sekcji krytycznej i odsyła token, jeżeli ktoś o niego aktualnie prosi
    def release(self, shared_data_obj):
        #(...)
    # Jeżeli z jakiś powodów proces nie może teraz przetwarzać zmiennej współdzielonej
    # (np. odebraną zmienną jest bufor, który jest pusty i nie można z niego nic pobrać)
    # to proces zasypia i informuje o tym resztę współpracowników (procesów)
    def going_sleep(self, shared_data_obj): 
        #(...)
    # Jeżeli z jakiś powodów proces chce kogoś obudzić (np. dodał coś do współdzielonego bufora)
    # to proces wysyła komunikat budzenia jednego procesu, żeby wybudzić pierwszy znany mu śpiacy proces
    def wake_up(self): 
        #(...)
    # Podobnie jak metoda wake_up() z tą różnicą, że komunikat jest wysyłany do wszystkich,
    # aby każdy proces, który aktualnie "śpi" został wybudzony
    def wake_up_all(self): 
        #(...)
    # Opakowanie (z ang. wrapper) na funkcje: going_sleep(), wake_up() oraz wake_up_all(), 
    # aby ich nazwy były bardziej intuicyjne (dla innych programistów):
    # NOTE: metoda going_sleep() umożliwia edycję zmiennej w przeciwieństwie do wait()
    # w metodzie wait() oddawana zmienna współdzielona będzie niezmieniona
    # Odpowiednik going_sleep()
    def wait(self): 
        #(...)
    # Odpowiednik wake_up()
    def notify(self): 
        #(...)
    # Odpowiednik wake_up_all()
    def notifyAll(self): 
        #(...)

Mechanizm synchronizacji oparty tylko na metodach acquire() i release() nie umożliwia prostego sposobu na usprawnienie systemu rozproszonego w scenariuszu, gdy proces otrzymuje dostęp do zmiennej współdzielonej, która nie może zostać przez niego przetworzona (np. bufor jest pusty). W takim przypadku proces po oddaniu dostępu do zmiennej współdzielonej (metodą release()) może ponownie próbować otrzymać dostęp (metodą acquire()) do zmiennej, której wciąż nie będzie w stanie przetworzyć. Do tego typu przypadków przydatne są metody going_sleep() oraz wait(), które umożliwiają "poczekanie" na komunikat o zmianie stanu zmiennej współdzielonej wysyłany przez inny proces za pomocą metod notify() lub notifyAll() [lub analogicznie wake_up()/wake_up_all()].

1.4 Przykłady problemu producenta konsumenta

Do przetestowania poprawności implementacji przygotowane zostały przeze mnie przykłady problemu producenta konsumenta.

1.4.1 Podstawowy (wykorzystanie wyłącznie metod acquire() oraz release())

W pliku kons.py znajduje się przykładowy konsument produktu Produkt(), który jest obiektem bufora zawierającym tablicę oraz licznik elementów w tablicy. Konsument próbuje konsumować współdzielony między procesami produkt, a po 3 nieudanych próbach konsumpcji z rzędu kończy swoją pracę w systemie. Czas konsumpcji produktu jest wydłużony o 1 sekunde, w celu lepszego zaobserwowania przetrzymywania zasobu.
W pliku prod.py znajduje się przykładowy producent produktu Produkt(), który wstawia do produktu liczby od 1 do 5 i kończy swoje działanie. Czas produkcji produktu jest wydłużony o 1 sekunde, w celu lepszego zaobserwowania przetrzymywania zasobu.
Parametry wywołania:
[Posiadanie przez proces tokenu: 1/0] [ID procesu --> "IP:PORT"] [lista ID innych procesów --> "IP:PORT"]
Zasada uruchomienia:

# Bardzo prosty przykład producenta konsumenta
pipenv run python prod.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772  
pipenv run python prod.py 0 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777  
pipenv run python prod.py 0 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776  
pipenv run python kons.py 0 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775   
pipenv run python kons.py 0 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774  
pipenv run python kons.py 0 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773

1.4.2 Wydajniejszy (wykorzystanie dodatkowo metod wait() oraz notifyAll())

W pliku kons_sleep.py znajduje się przykładowy konsument produktu, którym jest tablica wartości (bufor). Konsument próbuje konsumować jedną wartość ze współdzielonego między procesami produktu, a po 3 nieudanych próbach konsumpcji z rzędu kończy swoją pracę w systemie. Czas konsumpcji produktu jest wydłużony o 1 sekunde, w celu lepszego zaobserwowania przetrzymywania zasobu. Dodatkowo w przypadku nieudanej konsumpcji wywoływana jest metoda wait(), aby poczekać na zmianę stanu współdzielonej zmiennej.
W pliku prod_wakeupall.py znajduje się przykładowy producent produktu Produkt(), który wstawia do produktu liczby od 1 do 5 i kończy swoje działanie. Dodatkowo po wstawieniu do produktu liczby wywoływana jest metoda notifyAll(), która budzi wszystkich śpiących konsumentów.
Parametry wywołania:
[Posiadanie przez proces tokenu: 1/0] [ID procesu --> "IP:PORT"] [lista ID innych procesów --> "IP:PORT"]
Zasada uruchomienia:

# Przykład producenta konsumenta z możliwością czekania na pojawienie się produktu i budzeniem wszystkich oczekujących konsumentów
pipenv run python prod_wakeupall.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772  
pipenv run python prod_wakeupall.py 0 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777  
pipenv run python prod_wakeupall.py 0 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776  
pipenv run python kons_sleep.py 0 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775   
pipenv run python kons_sleep.py 0 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774  
pipenv run python kons_sleep.py 0 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773

1.4.3 Wykorzystanie metody notify() zamiast notifyAll()

W pliku kons_sleep.py znajduje się przykładowy konsument produktu, którym jest tablica wartości (bufor). Konsument próbuje konsumować jedną wartość ze współdzielonego między procesami produktu, a po 3 nieudanych próbach konsumpcji z rzędu kończy swoją pracę w systemie. Czas konsumpcji produktu jest wydłużony o 1 sekunde, w celu lepszego zaobserwowania przetrzymywania zasobu. Dodatkowo w przypadku nieudanej konsumpcji wywoływana jest metoda wait(), aby poczekać na zmianę stanu współdzielonej zmiennej.
W pliku prod_wakeup.py znajduje się przykładowy producent produktu Produkt(), który wstawia do produktu liczby od 1 do 5 i kończy swoje działanie. Dodatkowo po wstawieniu do produktu liczby wywoływana jest metoda notify(), która budzi jeden ze śpiących konsumentów (który jest znany danemu producentowi).
Parametry wywołania:
[Posiadanie przez proces tokenu: 1/0] [ID procesu --> "IP:PORT"] [lista ID innych procesów --> "IP:PORT"]
Zasada uruchomienia:

# Przykład producenta konsumenta z możliwością czekania na pojawienie się produktu oraz budzenia, 
# ale budzony jest tylko 1 proces na raz (przez inny proces) 
pipenv run python prod_wakeup.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772  
pipenv run python prod_wakeup.py 0 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777  
pipenv run python prod_wakeup.py 0 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776  
pipenv run python kons_sleep.py 0 127.0.0.1:7774 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775   
pipenv run python kons_sleep.py 0 127.0.0.1:7773 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774  
pipenv run python kons_sleep.py 0 127.0.0.1:7772 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775 127.0.0.1:7774 127.0.0.1:7773

1.4.4 Mniejsza ilość procesów

Zasada uruchomienia poprzednich przykładów z 2 konsumentami i 1 producentem: (do testowania)

# Mniejsza ilość procesów
# Podstawowa
pipenv run python prod.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775
pipenv run python kons.py 0 127.0.0.1:7776 127.0.0.1:7777 127.0.0.1:7775
pipenv run python kons.py 0 127.0.0.1:7775 127.0.0.1:7777 127.0.0.1:7776
# notifyAll()
pipenv run python prod_wakeupall.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775
pipenv run python kons_sleep.py 0 127.0.0.1:7776 127.0.0.1:7777 127.0.0.1:7775
pipenv run python kons_sleep.py 0 127.0.0.1:7775 127.0.0.1:7777 127.0.0.1:7776
# notify()
pipenv run python prod_wakeup.py 1 127.0.0.1:7777 127.0.0.1:7776 127.0.0.1:7775
pipenv run python kons_sleep.py 0 127.0.0.1:7776 127.0.0.1:7777 127.0.0.1:7775
pipenv run python kons_sleep.py 0 127.0.0.1:7775 127.0.0.1:7777 127.0.0.1:7776

1.4.5 Prezentacja na 4 wirtualnych maszynach

pipenv run python prod_wakeup.py 1 192.168.0.12:7777 192.168.0.13:7776 192.168.0.14:7775 192.168.0.15:7774
pipenv run python prod_wakeup.py 0 192.168.0.13:7776 192.168.0.14:7775 192.168.0.15:7774 192.168.0.12:7777
pipenv run python kons_sleep.py 0 192.168.0.14:7775 192.168.0.15:7774 192.168.0.12:7777 192.168.0.13:7776 
pipenv run python kons_sleep.py 0 192.168.0.15:7774 192.168.0.12:7777 192.168.0.13:7776 192.168.0.14:7775

1.5 Przydatne linki

Spis linków pomocnych w zrozumieniu implementacji Distr_Monit.py
Zmniejszając ilość komentarzy w pliku Distr_Monit.py, które mogłyby zmniejszyć czytelność kodu.
Przydatne aktywne (do czasu) linki:
[0] https://www.geeksforgeeks.org/suzuki-kasami-algorithm-for-mutual-exclusion-in-distributed-system/
[1] https://docs.python.org/3/library/queue.html?highlight=queue#module-queue
[2] https://docs.python.org/3/library/threading.html
[3] https://github.com/zeromq/pyzmq
[4] https://docs.python.org/3/library/contextlib.html
[5] https://docs.python.org/3/library/socket.html?highlight=socket#module-socket
[6] https://docs.python.org/3/library/pickle.html
[7] https://zguide.zeromq.org/docs/chapter5/
https://dev.to/dansyuqri/pub-sub-with-pyzmq-part-1-2f63
https://dev.to/dansyuqri/pub-sub-with-pyzmq-part-2-2f63
[8] https://docs.python.org/3/library/socket.html
[9] https://www.pythontutorial.net/advanced-python/python-threading/
[10] https://www.geeksforgeeks.org/python-daemon-threads/

Name		Name	Last commit message	Last commit date
Latest commit History 33 Commits
Distr_Monit.py		Distr_Monit.py
GrupoweZalozenia.txt		GrupoweZalozenia.txt
Links.txt		Links.txt
Logger.py		Logger.py
Pipfile		Pipfile
Pipfile.lock		Pipfile.lock
README.md		README.md
kons.py		kons.py
kons_sleep.py		kons_sleep.py
prod.py		prod.py
prod_wakeup.py		prod_wakeup.py
prod_wakeupall.py		prod_wakeupall.py

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Repository files navigation

0. 2022_II_ST_NPR

1. Implementacja rozproszonego monitora w ØMQ

1.0 Wymagania

1.1 Krótki opis implementacji

1.2 Opis klasy ExchangeMsg()

1.3 Opis klasy DistributedMonitor()

1.3.1 Wstęp

1.3.2 Funkcje użytkowe

1.4 Przykłady problemu producenta konsumenta

1.4.1 Podstawowy (wykorzystanie wyłącznie metod acquire() oraz release())

1.4.2 Wydajniejszy (wykorzystanie dodatkowo metod wait() oraz notifyAll())

1.4.3 Wykorzystanie metody notify() zamiast notifyAll()

1.4.4 Mniejsza ilość procesów

1.4.5 Prezentacja na 4 wirtualnych maszynach

1.5 Przydatne linki

About

Releases

Packages

Languages

Nailu776/Studies-2022_II_ST_NPR

Folders and files

Latest commit

History

Repository files navigation

0. 2022_II_ST_NPR

1. Implementacja rozproszonego monitora w ØMQ

1.0 Wymagania

1.1 Krótki opis implementacji

1.2 Opis klasy ExchangeMsg()

1.3 Opis klasy DistributedMonitor()

1.3.1 Wstęp

1.3.2 Funkcje użytkowe

1.4 Przykłady problemu producenta konsumenta

1.4.1 Podstawowy (wykorzystanie wyłącznie metod acquire() oraz release())

1.4.2 Wydajniejszy (wykorzystanie dodatkowo metod wait() oraz notifyAll())

1.4.3 Wykorzystanie metody notify() zamiast notifyAll()

1.4.4 Mniejsza ilość procesów

1.4.5 Prezentacja na 4 wirtualnych maszynach

1.5 Przydatne linki

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages 0

Languages

Packages