MPI (ang. Message Passing Interface) jest specyfikacją API bibliotecznego w założeniu umożliwiającego budowanie równoległych programów, w których procesy komunikują się poprzez jawnie przekazywane komunikaty. Najwięcej zastosowań MPI znajduje w tworzeniu równoległych programów na komputery klastrowe i superkomputery bez rozproszonej pamięci dzielonej.
Modelem współbieżności realizowanym przez MPI jest MIMD (ang. Multiple Instruction Multiple Data). Jednakże bardzo często używa się MPI w modelu SPMD (ang. Single Program Multiple Data). W modelu SPMD pojedynczy program jest uruchamiany na wszystkich maszynach i ewentualne różnice w sterowaniu podejmowane są w konkretnej instancji programu w zależności od numeru maszyny, na której ta instancja działa.
Analizując prosty program "hello world!" (hello-world-p2p.c) przeanalizujemy główne idee MPI. Dokładniej, zobaczymy jak MPI:
Program hello-world-p2p.c jest równoległą wersją trywialnego programu "Hello world!" używanego zwykle do nauki języków programowania. W naszej wersji "Hello world!" dla MPI mamy N procesów, numerowanych od 0..N-1. Procesy o numerach i ∈ {1..N-1} wysyłają komunikaty zawierające napis "Hello world from MPI process <i>!" do procesu o numerze 0, gdzie "<i>" jest numerem procesu. Natomiast proces o numerze 0 wypisuje otrzymane komunikaty na standardowe wyjście w kolejności numeru procesu nadawcy. Wynikiem uruchomienia naszego programu na 4 procesach (o uruchamianiu nieco później) będzie następujący ciąg napisów na standardowym wyjściu:
A parallel MPI-based "Hello world!" application. Hello world from MPI process 1! Hello world from MPI process 2! Hello world from MPI process 3!
Pierwszą interesującą nas instrukcją w pliku hello-world-p2p.c jest:
MPI_Init(&argc, &argv);
Instrukcja ta jest wykonywana przez wszystkie procesy. Ma ona za zadanie zainicjalizowanie środowiska MPI dla procesu, który ją wywołuje (man MPI_Init). Inicjalizacja informuje środowisko uruchomieniowe MPI o nowym procesie, jak również wykonuje inne czynności administracyjne. Wszystkie te czynności są jednak przeźroczyste dla użytkownika — zamknięte w wywołaniu pojedycznej funkcji — co znacznie ułatwia programowanie.
Funkcja MPI_Init musi dostać niezmodyfikowane parametry linii poleceń. Ewentualne przetwarzanie linii poleceń przez aplikację może być przeprowadzone dopiero po powrocie z MPI_Init. Podobnie, zabronione jest wywoływanie innych funkcji MPI przed powrotem z funkcji MPI_Init.
Proszę zauważyć, że w przeciwieństwie do wywołań funkcji systemowych, nasz program "Hello world!" nie testuje wyników wywołań funkcji MPI. Powodem tego jest fakt, że standardowa obsługa błędów przez MPI ma semantykę "all errors are fatal". Oznacza to, że jakikolwiek błąd środowiska uruchomieniowego MPI w jakimkolwiek procesie automatycznie zabija wszystkie procesy naszego programu równoległego. Jest to dość częsta praktyka w równoległych programach obliczeniowych na komputery klastrowe. Jej powodem jest fakt, że programy te są dość trudne nawet bez obsługi błędów. Zakłada się więc, że system uruchomieniowy wybranego środowiska do obliczeń równoległych powinien automatycznie zapewniać obsługę błędów: w sytuacji idealnej — pewien stopień odporności na błędy, minimalistycznie — automatyczne ubijanie błędnego wykonania obliczeń.
Kolejnymi ważnymi instrukcjami są:
MPI_Comm_size(MPI_COMM_WORLD, &numProcesses); MPI_Comm_rank(MPI_COMM_WORLD, &myProcessNo);
Pierwsza z w/w instrukcji (man MPI_Comm_size) pobiera ze środowiska liczbę procesów użytych do uruchomienia naszego programu równoległego. Parametr MPI_COMM_WORLD jest stałą oznaczającą uchwyt do środowiska zawierającego wszystkie procesy, które uruchamiają nasz program. Innym przykładem uchwytu jest MPI_COMM_SELF, odpowiadający środowisku zawierającemu jedynie aktualny proces. Druga z wyżej wymienionych instrukcji (man MPI_Comm_rank) pobiera ze środowiska unikalny numer aktualnego procesu w ramach wszystkich procesów uruchamiających nasz program rozproszony. Procesy są numerowane od 0 do N - 1, gdzie N jest liczbą wszystkich procesów (== numProcesses).
Następnie sterowanie programu rozgałęzia się w zależności od tego czy aktualny proces ma numer 0 czy nie:
if (myProcessNo != 0)
{
...
}
else
{
...
}
Jeśli nie jesteśmy procesem 0, przygotowujemy komunikat zawierający napis "Hello world..." z naszym numerem, a następnie wysyłamy go do procesu 0:
MPI_Send(
message,
messageLen,
MPI_CHAR,
dstProcessNo,
MPI_MESSAGE_HELLO_WORLD_TAG,
MPI_COMM_WORLD
);
Powyższa funkcja (man MPI_Send) przesyła komunikaty punkt-do-punktu, tj. od aktualnego procesu do dokładnie jednego procesu docelowego. Identyfikatorem/adresem procesu docelowego jest jego numer (w naszym programie dstProcessNo równy 0) w środowisku (zwykle zakłada się środowisko zawierające wszystkie procesy wchodzące w skład aplikacji równoległej — MPI_COMM_WORLD — patrz wyżej). Jedną z ogromnych zalet MPI jest fakt, iż oprócz komunikacji punkt-do-punktu, MPI oferuje wiele innych wariantów, które omówimy dalej.
Wysyłany komunikat jest przekazywany do funkcji MPI_Send przez nietypowany wskaźnik (message). Jednakże, mimo użycia nietypowanych wskaźników do wysyłania (i odbierania), każdy komunikat w MPI ma określony format. Dokładniej, pojedynczy komunikat MPI jest tablicą elementów pewnego ustalonego przez użytkownika typu. Typ elementów tablicy jest opisywany przez trzeci parametr funkcji MPI_Send (w naszym przypadku jest to typ znakowy — MPI_CHAR), zaś liczba elementów w tablicy — przez drugi parametr (w naszym przypadku messageLen). W powyższym przykładzie wysyłamy więc tablicę messageLen elementów typu MPI_CHAR (znaków). Innymi słowy, tablicę znaków tworzącą nasz napis "Hello world..." wraz z kończącym znakiem zera (konwencja języka C). Generalnie, MPI definiuje pewną liczbę typów prostych (np. MPI_LONG, MPI_FLOAT, MPI_DOUBLE) oraz umożliwia aplikacjom tworzenie własnych typów złożonych (patrz literatura).
Jako że komunikaty mają ustalone formaty, nadawca musi mieć możliwość przekazania odbiorcy, jakiego formatu komunikat do niego wysyła, tj. jakiego typu są elementy tablicy stanowiącej komunikat. Odbiorca zaś musi mieć możliwość selektywnego odbierania komunikatów, tj. komunikatów o określonym formacie lub też komunikatów, na które obecnie oczekuje. W tym celu komunikaty opatrywane są tzw. znacznikami (piąty parametr funkcji MPI_Send). Znaczniki to po prostu liczby całkowite unikalnie identyfikujące rodzaj komunikatu w aplikacji równoległej. W naszym przykładzie, wysyłany komunikat "Hello world..." jest opatrzony znacznikiem MPI_MESSAGE_HELLO_WORLD_TAG.
Kolejnym zagadnieniem, na którym zatrzymamy się przy okazji funkcji MPI_Send jest semantyka tej funkcji. Dokładniej, jakie gwarancje daje MPI_Send w momencie zakończenia jeśli chodzi dostarczenie wysyłanego komunikatu oraz o obszar pamięci zawierającej wysyłany komunikat. Otóż standard MPI tego nie precyzuje. Generalnie, MPI_Send gwarantuje, że komunikat zostanie kiedyś dostarczony oraz że w momencie zakończenia MPI_Send, obszar pamięci zajmowany przez komunikat może być bezpiecznie użyty. Jednakże nie ma żadnych gwarancji co do tego, że po opuszczeniu funkcji MPI_Send, wysyłany komunikat został odebrany (i przetworzony przez obiorcę). Nie ma nawet gwarancji, że odbiorca zaczął odbierać komunikat. MPI mówi tylko, że funkcja MPI_Send może się zablokować do momentu, gdy kontynuowanie działania będzie bezpieczne z punktu widzenia wysyłanego komunikatu. Jednocześnie, aby umożliwiać bardziej precyzyjną kontrolę nad tym, co dzieje się z wysyłanym komunikatem, MPI udostępnia całą gamę wariantów funkcji MPI_Send z różną semantyką.
Przejdźmy teraz do procesu 0 i odbierania komunikatów, tj. do drugiej gałęzi instrukcji warunkowej w naszym programie. Jako że w programie chcemy odebrać napis "Hello world..." kolejno od wszystkich procesów od 1 do N-1, proces 0 iteruje po tych procesach. Samo odbieranie komunikatu od konkretnego nadawcy odbywa się za pomocą funkcji:
MPI_Recv(
message,
sizeof(message) / sizeof(char),
MPI_CHAR,
srcProcessNo,
MPI_MESSAGE_HELLO_WORLD_TAG,
MPI_COMM_WORLD,
&status
);
W funkcji odbierającej (man MPI_Recv) musimy określić znacznik komunikatu, który chcemy odebrać, czyli w powyższym przykładzie zdefiniowany przez nas wcześniej MPI_MESSAGE_HELLO_WORLD_TAG. Musimy także zdecydować, od którego procesu chcemy odebrać komunikat (proces o numerze srcProcessNo interpretowanym w środowisku zawierającym wszystkie procesy — MPI_COMM_WORLD).
Po drugie, przekazujemy bufor na tablicę elementów stanowiącą komunikat (message), maksymalną liczbe elementów tablicy mieszczących się w buforze (sizeof(message) / sizeof(char)) oraz typ pojedynczego elementu (MPI_CHAR). Jeśli liczba elementów w tablicy odpowiadającej wysłanemu komunikatowi (przekazana do MPI_Send) była mniejsza lub równa długości tablicy bufora odbiorczego (przekazanej do MPI_Recv) komunikat zostanie odebrany poprawnie. Jeśli natomiast wysłany komunikat był dłuższy niż rozmiar bufora do odbioru, nastąpi błąd (i zgodnie z semantyką "all errors are fatal" nasza aplikacja zostanie przerwana).
Wywołanie funkcji MPI_Recv jest blokowane do momentu, gdy komunikat, na który oczekujemy nadejdzie.
Jako ostatni parametr, funkcja MPI_Recv przyjmuje wskaźnik na strukturę status. W momencie odebrania komunikatu (powrotu z funkcji MPI_Recv) struktura ta będzie zawierać dodatkowe dane o komunikacie. Przykładowo, jeśli potrzebowalibyśmy znać dokładną liczbę elementów tablicy reprezentującej otrzymany komunikat (w naszym przykładzie — liczbę znaków tworzących napis "Hello world...") możemy uzyskać tę informację wywołując funkcję:
MPI_Get_count(&status, MPI_CHAR, &messageLen);
Pole status ma także inne zastosowania. Jeśli do MPI_Recv zamiast srcProcessNo przekazalibyśmy stałą MPI_ANY_SOURCE, oznaczałoby to, że chcemy odebrać komunikat od dowolnego procesu. Po odebraniu takiego komunikatu, możemy dowiedzieć się od jakiego procesu go właściwie odebraliśmy używając pola MPI_SOURCE struktury status. Podobnie, jeśli do MPI_Recv zamiast MPI_MESSAGE_HELLO_WORLD_TAG przekazalibyśmy MPI_ANY_TAG, oznaczałoby to, że chcemy odebrać komunikat dowolnego typu. Po odebraniu komunikatu, pole MPI_TAG struktury status zawierało będzie znacznik komunikatu. Uwaga! W tym ostatnim przypadku, ważne jest, aby poprawnie zinterpretować typ elementów tablicy stanowiącej komunikat.
Ostatnią interesującą instrukcją jest wywołanie funkcji:
MPI_Finalize();
Zadaniem tej funkcji (man MPI_Finalize) jest poinformowanie systemu uruchomieniowego MPI, że aktualny proces kończy pracę, i zwolnienie zasobów zaalokowanych przez implementację MPI. Funkcja MPI_Finalize może się zablokować, na przykład, dopóki komunikaty, które wysłaliśmy przy użyciu MPI_Send nie dotrą do odbiorcy. Po powrocie z funkcji MPI_Finalize, proces wołający nie możne wołać innych funkcji MPI.
Jak wspomniano powyżej, MPI_Finalize zwalnia zasoby zaalokowane przez implementację MPI dla kończonego procesu. Z tego powodu ważne jest, aby MPI_Finalize było wołane w każdej ścieżce zakończenia procesu. W szczególności, jeśli w procesie wystąpił błąd nie związany z MPI (np. malloc zwrócił NULL-owy wskaźnik), przez zakończeniem proces powinien wywołać MPI_Finalize.
Przed wywołaniem funkcji MPI_Finalize dobra praktyka nakazuje zadbać, aby każdy komunikat wysłany przez inny proces do aktualnego (kończonego procesu) został odebrany. Innymi słowy, aplikacja rozproszona używająca MPI nie powinna zostawiać tak zwanych "osieroconych" komunikatów w sieci.
Na laboratorium będziemy posługiwać się implementacją MPI zwaną MPICH. Na maszynie students (i w labach) zainstalowana jest wersja 1.2.7. Najnowsza dostępna wersja ma numer 2.x.
Przykładowe programy budujemy wpisując polecenie
make
Do zbudowania programów używających MPI, Makefile wewnętrznie używana specjalnych kompilatorów dostarczonych z MPICH:
W rzeczywistościi mpicc i mpiCC są to odpowiednie opakowane gcc i g++.
Do uruchamiania rozproszonych aplikacji służy polecenie mpirun (przeczytaj man mpirun). Typowym użyciem mpirun jest:
mpirun -np <N> <nazwa_programu>
Parametr <nazwa_programu> to plik wykonywalny z naszą aplikacją rozproszoną. Opcja -np <N> specyfikuje liczbę procesów — N — które zostaną użyte do uruchomienia naszej aplikacji (i które będą zwracane procesom w ramach funkcji MPI_Comm_size). Jeśli opcja -np zostanie pominięta, przyjmowane jest N = 1.
Możemy dodatkowo wyspecyfikować maszyny, na których zostaną uruchomione nasze procesy. Robi się to za pomocą opcji -machinefile:
mpirun -machinefile <machine_file> -np <N> <nazwa_programu>
gdzie plik <machine_file> zawiera listę maszyn do wyboru. Pominięcie tej opcji powoduje użycie standardowej listy maszyn, która w prawidłowo skonfigurowanych środowiskach MPICH zwykle znajduje się w katalogu /usr/local/mpi/share/machines.$ARCH (niestety, na maszynie students standardowa konfiguracja nie jest najlepsza). Przykładowa treść pliku machine file:
pink05 pink06 pink07 pink08
Do odpalania procesów na wyspecyfikowanych maszynach, MPICH domyślnie używa polecenia rsh. Ważne jest, aby wszystkie maszyny "widziały" system plików, z którego uruchamiamy naszą aplikację.
Jeśli nie mamy dostępu do wielu maszyn (lub nasza konfiguracja nie jest najlepsza), możemy użyć:
mpirun -all-local -np <N> <nazwa_programu>
co spowoduje odpalenie wszystkich N procesów na maszynie lokalnej. Takie polecenie jest zalecane do testowania przykładowych programów na maszynie students.
Dla zainteresowanych: Instalacja MPICH jest w miarę prosta, szczególnie w systemie Windows i systemach Ubuntu. Zabawa z własnym środowiskiem MPICH jest więc jak najbardziej zalecana.
Czas wykonania poszczególnych fragmentów kodu naszej aplikacji rozproszonej i tym samym przyspieszenie wynikające z przetwarzania równoległego możemy mierzyć używając standardowych funkcji i bibliotek benchmarkujących dostępnych w systemie (np. funkcji gettimeofday). Standard MPI dodatkowo definiuje własne funkcje:
Przykładowe użycie tych funkcji:
double startTime; double endTime; double executionTime; startTime = MPI_Wtime(); // long-lasting parallelized computation of XXX endTime = MPI_Wtime(); executionTime = endTime - startTime;
Zakładając że czas wykonania obliczeń XXX w programie sekwencyjnym to executionTime_s, zaś w i-tym spośród N procesów równoległych (włączając narzut na komunikację wewnątrz obliczeń) to executionTime_i_N, możemy obliczyć speed-up naszej aplikacji:
speedupWithNProcesses = executionTime_s / max_over_i(executionTime_i_N)
Speed-up podaje nam, jak bardzo efektywny jest użyty przez nas algorytm zrównoleglania. Idealnie speedupWithNProcesses == N. W praktyce speed-up jest zwykle mniejszy z powodu narzutu na komunikację. To jak bardzo funkcja speed-up'u, f(N), odstaje od funkcji id(N) = N jest miarą tego, jak dobry jest nasz algorytm równoległy.
Podstawowa komunikacja punkt-do-punktu pozwala budować dowolne wysokopoziomowe schematy komunikacji. Przykładowo, aby zaimplementować rozgłaszanie (czyli komunikację jeden-do-wszystich), rozgłaszający — proces i — może bezpośredio wysłać komunikat punkt-do-punktu do każdego odbierającego procesu j (j różne od i). Takie podejście jednak jest nieefektywne. Jeśli założymy, że koszty komunikacji pomiędzy dowolną parą procesów są stałe i takie same, najefektywniejsze rozgłaszanie działa w rundach wg następującego algorytmu:
W ten sposób rozgłaszanie będzie skończone w log2(N) rundach a nie w N - 1, jak w naiwnym algorytmie.
Aby uniknąć konieczności pisania podobnych algorytmów przy każdej aplikacji równoległej, MPI definiuje pewne wysokopoziomowe, popularne schematy komunikacji. Konkretna biblioteka implementująca API MPI (np. MPICH) ma za zadanie zapewnić, że implementacje tych schematów są efektywne. Przykładami takich schematów są (patrz odpowienie strony man):
data data
P0: D1 D2 D3 D4 scatter P0: D1 -- -- --
P1: -- -- -- -- ————--> P1: D2 -- -- --
P2: -- -- -- -- <————-- P2: D3 -- -- --
P3: -- -- -- -- gather P3: D4 -- -- --
Dodatkowo, dość częstym mechanizmem synchronizacyjnym w aplikacjach równoległych jest tzw. bariera. Bariera to miejsce w kodzie aplikacji równoległej, do którego muszą dojść wszystkie procesy wykonujące tę aplikację, aby którykolwiek z procesów mógł przejść dalej. MPI udostępnia własną barierę, której używa się przy pomocy funkcji:
MPI_Barrier
Więcej informacji o wysokopoziomowych schematach komunikacyjnych MPI można znaleźć w podanej wyżej literaturze.
Jak wspomniano wyżej, semantyka operacji MPI_Send nie jest ściśle sprecyzowana. Nie ma żadnych gwarancji co do tego, że po opuszczeniu funkcji MPI_Send, wysyłany komunikat został odebrany (i przetworzony przez obiorcę). Nie ma nawet gwarancji, że odbiorca zaczął odbierać komunikat. MPI gwarantuje jedynie, że po powrocie z funkcji, obszar pamięci zajmowany przez komunikat może zacząć zostać modyfikowany oraz wspomina, że operacja MPI_Send może się zablokować. Ta ostatnia cecha sprawia w szczególności, że poniższy kod jest niepoprawny.
// Proces A: ... MPI_Send(..., processB, msgTagA, ...); MPI_Recv(..., processB, msgTagB, ...); ... // Proces B: ... MPI_Send(..., processA, msgTagB, ...); MPI_Recv(..., processA, msgTagA, ...); ...
Powodem jest fakt, że każdy z procesów A i B może się zablokować na swojej funkcji MPI_Send i w rezultacie żaden z nich nie będzie mógł zacząć odbierać komunikatu od drugiego procesu, co doprowadzi do zakleszczenia.
MPI udostępnia jednak operacje o bardziej sprecyzowanej semantyce (patrz odpowiednie strony man). Operacje te dzielą się na blokujące i nieblokujące. Przykłady operacji blokujących to:
Przykłady nieblokujących operacji natomiast to:
Przykład nieblokującej komunikacji znajduje się w pliku ring-nonblocking.c.
Więcej informacji o różnych semantykach komunikatów w MPI można znaleźć w podanej wyżej literaturze. Osoby zainteresowane komunikacją, odpornością na błędy i synchronizacją w środowiskach rozproszonych zapraszam na wykład systemy rozproszone.
Ostatnia modyfikacja: 21/01/2011