Message Passing Interface (MPI) (z ang. Interfejs Transmisji Wiadomości) – protokół komunikacyjny będący standardem przesyłania komunikatów pomiędzy procesami programów równoległych działających na jednym lub więcej komputerach. Interfejs ten wraz z protokołem oraz semantyką specyfikuje, jak jego elementy winny się zachowywać w dowolnej implementacji. Celami MPI są wysoka jakość, skalowalność oraz przenośność. MPI jest dominującym modelem wykorzystywanym obecnie w klastrach komputerów oraz superkomputerach. Pierwsza wersja standardu ukazała się w maju 1994 r. Standard MPI implementowany jest najczęściej w postaci bibliotek, z których można korzystać w programach tworzonych w różnych językach programowania, np. C, C++, Ada, Fortran.

Opis

MPI jest specyfikacją biblioteki funkcji opartych na modelu wymiany komunikatów dla potrzeb programowania równoległego. Transfer danych pomiędzy poszczególnymi procesami programu wykonywanymi na procesorach maszyn będących węzłami klastra odbywa się za pośrednictwem sieci.

Główny model MPI-1 nie wspiera koncepcji współdzielonej pamięci, MPI-2 wspiera (w sposób ograniczony) rozproszony system pamięci dzielonej. Mimo tego programy MPI są bardzo często uruchamiane na komputerach o współdzielonej pamięci.

Program w MPI składa się z niezależnych procesów operujących na różnych danych (MIMD). Każdy proces wykonuje się we własnej przestrzeni adresowej, aczkolwiek wykorzystanie pamięci współdzielonej też jest możliwe.

Zaletami MPI nad starszymi bibliotekami przekazywania wiadomości są przenośność oraz prędkość. Przenośność, ponieważ MPI został zaimplementowany dla każdej architektury opartej na rozproszonej pamięci. Prędkość, ponieważ każda implementacja jest zoptymalizowana pod sprzęt, na którym działa. Standard udostępnia zbiór precyzyjnie zdefiniowanych metod, które mogą być efektywnie zaimplementowane. Stał się on punktem wyjściowym do stworzenia praktycznego, przenośnego, elastycznego i efektywnego narzędzia do przesyłania komunikatów (ang. message passing). Standard MPI pozwala na jego zastosowanie zarówno w komputerach równoległych, jak i heterogenicznych sieciach stacji roboczych.

Standard nie zabrania, aby poszczególne procesy były wielowątkowe. Nie są też udostępnione mechanizmy związane z rozłożeniem obciążenia pomiędzy poszczególne procesy, z architekturą rozkładu procesorów, z dynamicznym tworzeniem i usuwaniem procesów. Procesy są identyfikowane poprzez ich numer w grupie w zakresie 0 .. groupsize – 1.

Główne własności MPI

• umożliwia efektywną komunikację bez obciążania procesora operacjami kopiowania pamięci,

• udostępnia funkcje dla języków C/C++, Fortran oraz Ada,

• specyfikacja udostępnia hermetyczny interfejs programistyczny, co pozwala na skupienie się na samej komunikacji, bez wnikania w szczegóły implementacji biblioteki i obsługi błędów,

• definiowany interfejs zbliżony do standardów takich jak: PVM, NX czy Express,

• udostępnia mechanizmy komunikacji punkt – punkt oraz grupowej,

• może być używany na wielu platformach, tak równoległych, jak i skalarnych, bez większych zmian w sposobie działania.

Zalety MPI

• dobra efektywność w systemach wieloprocesorowych,

• dobra dokumentacja,

• bogata biblioteka funkcji,

• posiada status „public domain”,

• przyjął się jako standard.

Wady MPI

• statyczna konfiguracja jednostek przetwarzających,

• statyczna struktura procesów w trakcie realizacji programu (dotyczy to implementacji opartych na MPI-1). Wersja MPI-2 (wspierana np przez LAM 7.0.4) umożliwia dynamiczne zarządzanie strukturą procesów biorących udział w obliczeniach – MPI_Spawn(),

• brak wielowątkowości.

Interfejs MPI ma na celu dostarczenie wirtualnej topologii, synchronizacji oraz funkcjonalności komunikacji pomiędzy zestawem procesów (które zostały zmapowane do węzłów/serwerów/instancji komputerów) w niezależny od języka sposób, przy podobnej do niego składni. Programy MPI zawsze współpracują z procesami, aczkolwiek programiści powszechnie odnoszą się do procesów jako procesorów. Zazwyczaj dla uzyskania maksymalnej wydajności, każdy CPU (lub rdzeń w wielordzeniowym procesorze) ma przypisany pojedynczy proces. Przypisanie to ma miejsce w czasie rzeczywistym przez agenta, który uruchamia program MPI (zazwyczaj nazywany mpirun lub mpiexec).

Koncepcje

Komunikator

Komunikatory są obiektami łączącymi, tj. grupującymi procesy podczas sesji MPI. W obrębie każdego komunikatora każdy zawarty proces ma niezależny identyfikator, zaś same procesy organizowane są w uporządkowaną topologię.

Punkt – punkt

Szereg ważnych funkcji w API MPI dotyczy komunikacji pomiędzy dwoma określonymi procesami. Operacje punkt – punkt są szczególnie użyteczne w nieregularnej komunikacji, np. w architekturze opartej na równoległości danych, w której każdy procesor stale wymienia regiony danych z innymi określonymi procesorami pomiędzy kolejnymi krokami obliczeń. Innym przykładem jest architektura master-slave, w której master wysyła nowe dane zadania do slave’a niezależnie od tego, czy poprzednie zostało ukończone.

Funkcje zbiorowe

Funkcje zbiorowe w API MPI dotyczą komunikacji pomiędzy wszystkimi procesami w grupie procesów. Wywołania te są często używane na początku lub końcu dużych rozproszonych obliczeń, gdzie każdy z procesorów operuje na części danych, po czym łączy je w wynik.

Typy danych

Wiele funkcji MPI wymaga, ażeby określić typ danych, jaki jest przesyłany pomiędzy procesorami. Dzieje się tak, ponieważ argumentami funkcji są zmienne, nie zaś typy definiowane. Jeśli typ danych jest standardowy, jak np. int, char, double itd., można użyć predefiniowanych w MPI typów takich jak: MPI_INT, MPI_CHAR, MPI_DOUBLE. Dane mogą być również w postać klas lub struktur danych. Można wykorzystywać pochodne typy danych z typów predefiniowanych.

Komunikacja jednostronna (MPI-2)

MPI-2 definuje trzy operacje jednostronnej komunikacji: Put, Get i Accumulate, które to umożliwiają zapis do zdalnej pamięci, odczyt z niej oraz zmniejszają liczbę operacji pamięci dla wielu zadań. Definiowane są również trzy różne metody dla synchronizacji komunikacji – globalna, parami oraz przy wykorzystaniu zdalnych blokad – specyfikacja nie gwarantuje jednak, iż operacje te będą wykonywane aż do momentu punktu synchronizacji.

Dynamiczne zarządzanie procesami (MPI-2)

Kluczowym aspektem dynamicznego zarządzania procesami w MPI-2 jest zdolność procesów do tworzenia nowych procesów bądź ustanowienia komunikacji z procesami, które rozpoczęły się oddzielnie.

MPI I/O (MPI-2)

Równoległe I/O wprowadzone w MPI-2 jest często nazywane w skrócie MPI-IO oraz odnosi się do zbioru funkcji przeznaczonych do ułatwienia zarządzania I/O w systemach rozproszonych w sposób abstrakcyjny oraz umożliwienia łatwego dostępu do plików korzystając z funkcjonalności istniejących typów pochodnych.

1. Ogólna koncepcja systemu

a) Podstawy

MPI realizuje model przetwarzania współbieżnego zwany MIMD ( Multiple Instruction Multiple Data), a dokładniej SPMD (Single Program Multiple Data). Zakłada on, że ten sam kod źródłowy wykonuje się jednocześnie na kilku maszynach i procesy mogą przetwarzać równocześnie różne fragmenty danych, wymieniając informacje przy użyciu komunikatów

Takie podejście ma wiele zalet, z których najbardziej spektakularną jest chyba możliwość współbieżnych obliczeń wykonywanych na maszynach o zupełnie różnych architekturach (np. Linux-x86 oraz Solaris-Sparc).

Realizując ten model, MPI umożliwia:

• Wymianę komunikatów między procesami
  (Główny nacisk jest położony na wymianę danych, ale możliwe jest również wysyłanie komunikatów kontrolnych, czy synchronizacja procesów)

• Uzyskiwanie informacji o środowisku
  (Typowy przykład to ilość aktywnych proces[-ów/-orów], czy numer aktualnego procesu)

• Kontrolę nad systemem
  (Inicjalizacja/kończenie programu, kontrola poprawności przesyłanych komunikatów itp.)

Wszystkie te rzeczy są realizowane przy minimalnym stopniu skomplikowania kodu źródłowego.

b) Komunikaty

Przy przesyłaniu komunikatów między procesami MPI stara się zachować niezależność od platformy (np. kolejności bajtów). Dla standardowych typów jest to proste, natomiast dla typów niestandardowych MPI dostarcza funkcje pozwalające na zdefiniowanie typów użytkownika dla potrzeb przesyłania komunikatów.

Możliwe jest adresowanie komunikatów zarówno do konkretnych procesów, jak i do określonych grup odbiorców. Dostępne są funkcje do definiowania grup procesów i późniejszego rozsyłania komunikatów do tych grup. Komunikaty opatrzone są tagami pozwalającymi na późniejsze selektywne odbieranie ich z kolejki w zależności od rodzaju.

Możliwa jest wymiana komunikatów w trybie non-blocking pozwalającym na jeszcze większe zrównoleglenie obliczeń.

c) Zaawansowana komunikacja

Główną zaletą MPI przy bardziej złożonych schematach wymiany danych jest ukrywanie przed programistą szczegółów implementacyjnych oraz możliwość optymalizacji ścieżki przepływu danych. Wyobraźmy sobie przykładowo, że mamy 8 procesów i pierwszy z nich ma przekazać pewną porcję danych wszystkim pozostałym. Najprostsza możliwość:

jest oczywiście niezbyt optymalna - widać, że w ogólnym przypadku złożoność czasowa procesu przesyłania jest zależna liniowo od liczby procesów biorących udział w tej operacji.

Kolejny przykład jest już krokiem w dobrym kierunku - przesyłanie danych trwa dwukrotnie krócej niż poprzednio

Oczywiście i tak każdy wie, że optymalny schemat broadcastu danych w obrębie grupy procesów wygląda następująco :)

jednak zaletą MPI jest fakt, że takie decyzje (dotyczące wyboru dróg przesyłania danych) są ukryte przed programistą - może on po prostu założyć, że dane "kiedyś" i "jakoś" dotrą na miejsce przeznaczenia, a "kiedy" i "jak" - decyduje system i stara się to zrobić w sposób optymalny.

Wszystkie te rozważania można również przeprowadzić przeprowadzić w odwrotną stronę - dla agregacji danych z kilku procesorów. MPI definiuje zbiór funkcji służących do zbierania danych z kilku procesów. Typowy przykład to obliczanie sumy elementów wektora na podstawie zbioru sum częściowych (lub np. wyszukiwanie maksymalnego elementu).

2. Środowisko pracy

Z punktu widzenia programisty MPI składa się z dwóch części:

• Biblioteki (do języka C lub Fortranu) zawierającej niezbędne funkcje i pliki nagłówkowe

• Środowiska uruchamiania (runtime)
  (Można tu zauważyć pewną użytkową analogię np. z Javą - program mpirun jest tu odpowiednikiem polecenia java)

Najprostszy sposób kompilacji programu to użycie któregoś ze standardowych skryptów dostarczanych z MPI. Są to:

• mpicc - dla programów w C

• mpiCC - dla programów w C++

• mpif77 - dla programów w Fortranie 77

• mpif90 - dla programów w Fortranie 90

Uruchamianie programów w MPI jest równie proste i sprowadza się do uruchomienia skryptu mpirun. Jego jedynym interesującym nas w tej chwili parametrem jest -np <n>. Określa on ilość równoległych procesów, jakie ma uruchomić system w celu wykonania obliczeń.

Przed uruchomieniem mpirun musimy upewnić się, że w naszym katalogu domowym na wszystkich maszynach potencjalnie mogących wchodzić w grę przy wykonywaniu naszego programu w odpowiednim katalogu znajduje się kopia pliku wykonywalnego programu. Najczęściej w takich przypadkach stosuje się montowanie katalogów przez NFS.

Szczegóły techniczne ukryte za mechanizmem rozsyłania procesów na różne maszyny w zasadzie nas tu nie interesują, warto może tylko wspomnieć o ADI (Abstract Device Interface). Jest to wewnętrzny standard mpich określający sposób komunikacji między procesami na niskim poziomie.

Zaletą używania takiego modelu jest oddzielenie części systemu zależnej od konkretnej architektury, od części identycznej dla wszystkich. Wynika z tego na przykład, że ten sam program uruchomiony na maszynie wieloprocesorowej może używać do komunikacji pamięci wspólnej (shared memory), a wykonywany na klastrze kilku silnych pecetów używa TCP/IP bez rekompilacji kodu!

3. Ważniejsze funkcje MPI

a) Podstawy

Podstawowe funkcje służące do inicjalizacji/zamykania programu oraz do wysyłania najprostszych komunikatów (typu Point-to-point) to:

• int MPI_Init(int *argc, char ***argv);

  Funkcja inicjalizuje środowisko wykonywania programu, m.in. tworzy domyślny komunikator MPI_COMM_WORLD. Dopiero od momentu wywołania MPI_Init można używać pozostałych funkcji MPI.

• int MPI_Finalize();

  Funkcja zwalnia zasoby używane przez MPI i przygotowuje system do zamknięcia.

• int MPI_Comm_rank(MPI_Comm comm, int *rank);

  Funkcja pobiera numer aktualnego procesu (w obrębie komunikatora comm) i umieszcza go w zmiennej rank.

• int MPI_Comm_size(MPI_Comm comm, int *size);

  Funkcja pobiera ilość procesów (w obrębie komunikatora comm i umieszcza ją w zmiennej size.

• int MPI_Send(void *msg, int count, MPI_Datatype datatype,
               int dest, int tag, MPI_Comm comm);

  Funkcja wysyła komunikat typu datatype do procesu numer dest oznaczony znacznikiem tag w obrębie komunikatora comm.

  Typ komunikatu jest zawarty w zmiennej datatype i może to być któryś z predefiniowanych typów takich jak MPI_INT, MPI_FLOAT, MPI_DOUBLE, MPI_CHAR, lub inne (zob. instrukcja), jak i typy zdefiniowane przez użytkownika (o tym jeszcze będzie).

  Tag jest liczbą w zakresie [0..MPI_TAG_UB] i określa dodatkowy typ komunikatu wykorzystywany przy selektywnym odbiorze funkcją MPI_Recv.

• int MPI_Recv(void *msg, int count, MPI_Datatype datatype,
               int source, int tag, MPI_Comm comm,
               MPI_Status *status);

  Funkcja odczytuje z kolejki komunikatora comm (z ewentualnym blokowaniem do czasu nadejścia) pierwszy komunikat od procesu source oznaczony znacznikiem tag typu datatype. Wynik umieszczany jest w buforze msg a status operacji w zmiennej status.

  Jeżeli proces ustawi source==MPI_ANY_SOURCE to odczytany będzie pierwszy komunikat od dowolnego procesu. Podobnie, dla tag==MPI_ANY_TAG nie będzie sprawdzany znacznik typu wiadomości.

  Bufor stanu status musi zostać uprzednio stworzony przez programistę. Dla C jego pojedynczy element składa się z trzech liczb całkowitych: MPI_SOURCE, MPI_TAG oraz MPI_STATUS. W ogólnym przypadku bowiem (przy odbieraniu komunikatów w których count>1) tablica status określa nam źródło i typ każdego komunikatu z osobna. Do pobierania ilości odebranych komunikatów na podstawie zmiennej stanu służy kolejna funkcja:

• int MPI_Get_count(MPI_Status *status, MPI_Datatype datatype, int *count);

  która umieszcza szukaną ilość w zmiennej count.

b) Rozsyłanie/zbieranie danych

• int MPI_Bcast(void *msg, int count,
                MPI_Datatype datatype, int root, MPI_Comm comm);

  Funkcja rozsyła komunikat do wszystkich procesów w obrębie komunikatora comm poczynając od procesu root. Pozostałe argumenty - podobnie jak w MPI_Send().

• int MPI_Reduce(void *operand, void *result,
                 int count, MPI_Datatype datatype, MPI_Op op,
                 int root, MPI_Comm comm);

  Bardzo ważna funkcja - pozwala wykonać na przykład sumowanie wszystkich częściowych wyników otrzymanych w procesach i umieszczenie wyniku w zmiennej. Argument root wskazuje dla którego procesu wynik ma być umieszczony w zmiennej result. Oto przykład użycia tej funkcji:

    MPI_Reduce(&suma_cz,&suma,1,MPI_FLOAT,MPI_SUM,0,MPI_COMM_WORLD);

  Przykładowe operatory to MPI_MAX, MPI_MIN, MPI_SUM.

• int MPI_Allreduce(void *operand, void *result,
                    int count, MPI_Datatype datatype, MPI_Op op,
                    MPI_Comm comm);

  Funkcja identyczna z poprzednią, różniąca się jedynie tym, że po jej wykonaniu wynik agregacji z użyciem operatora op znajduje się w zmiennej result we wszystkich procesach.

• int MPI_Scatter(void *send_buf, int send_count,
                  MPI_Datatype send_type, void *recv_buf,
                  int recv_count, MPI_Datatype recv_type,
                  int root, MPI_Comm comm);

  Funkcja rozproszenia ("scatter") danych między procesami. Działa w ten sposób, że proces root rozsyła zawartość send_buff między wszystkie procesy. Jest ona dzielona na p segmentów, każdy składający się z send_count elementów. Pierwszy segment trafia do procesu 0, drugi do procesu 1 itp. Oczywiście argumenty, których nazwy zaczynają się na Send mają znaczenie tylko dla procesu który jest nadawcą.

• int MPI_Gather(void *send_buf, int send_count,
                 MPI_Datatype send_type, void *recv_buf,
                 int recv_count, MPI_Datatype recv_type,
                 int root, MPI_Comm comm);

  Każdy proces w grupie comm wysyła zawartość send_buff do procesu root. Ten proces układa przysłane dane w recv_buff w kolejności numerów procesów.

• int MPI_Allgather(void *send_buf, int send_count,
                    MPI_Datatype send_type, void *recv_buf,
                    int recv_count, MPI_Datatype recv_type,
                    MPI_Comm comm);

  Funkcja ta jest analogiczna do MPI_Gather z tą różnicą, że wynik jest umieszczany w recv_buff każdego procesu. Można tą funkcję traktować jako ciąg kolejnych wywołań MPI_Gather każdorazowo z innym numerem procesu root

Zależność między trzema ostatnimi funkcjami ilustruje następujący rysunek:

c) Struktury danych

W zastosowaniach rzeczywistych najczęściej wysyłamy większą ilość danych, bądź to w strukturach, albo w postaci wektorów. Przesyłanie ich w postaci oddzielnych komunikatów powodowałoby powstanie dużego narzutu czasowego związanego z organizacją przesyłania danych. Potrzebna jest więc możliwość "pakowania" danych i wysyłania większej ich ilości za jednym razem.

W MPI jest to dość spory problem, gdyż najczęściej typ przesyłanych danych jest różny od standardowych typów zdefiniowanych w bibliotece. Zdecydowano się na rozwiązanie w którym programista ma możliwość tworzenia nowych typów danych w czasie wykonywania programu.

• int MPI_Address(void *data, MPI_Aint *address);

  Funkcja zapisuje adres zmiennej data do zmiennej address.

• int MPI_Type_struct(int count,
                      int *array_of_block_lengths,
                      MPI_Aint *array_of_displacements,
                      MPI_Datatype *array_of_types,
                      MPI_Datatype *newtype);

  Funkcja tworzy nowy typ - strukturę o count składowych. Ilość elementów każdej składowej jest zapisana w tablicy array_of_block_lengths. Długości składowych (a dokładniej - offsety względem początku rekordu) należy umieścić w tablicy array_of_displacements. Typy pojedynczych danych składowych umieszczamy w array_of_types. Wynikowy typ MPI jest umieszczany w strukturze wskazywanej przez newtype.

• int MPI_Type_vector(int count, int block_length,
                      int stride, MPI_Datatype element_type,
                      MPI_Datatype *newtype);

  Funkcja tworzy nowy typ - wektor długości count. Każdy element wektora zawiera block_length elementów typu element_type, elementy wektora są zaś rozdzielone dodatkowo stride elementami tego typu.

• int MPI_Type_contiguous(int count, MPI_Datatype oldtype,
                          MPI_Datatype *newtype);

  Tworzenie prostszego (jednowymiarowego) wektora elementów typu oldtype długości count.

• int MPI_Type_indexed(int count,
                       int *array_of_block_lengths,
                       int *array_of_displacements,
                       MPI_Datatype element_type,
                       MPI_Datatype *newtype);

  Jest to prostsza wersja funkcji MPI_Type_struct. Różni się tym, że wszystkie elementy struktury mają ten sam typ bazowy (element_type) i mogą być przesunięte względem początku o zadaną (tablica array_of_displacements) ilość elementów.

• int MPI_Type_commit(MPI_Datatype *newtype);

  Tą funkcję należy wywołać po zdefiniowaniu nowego typu przy użyciu którejś z poprzednich funkcji - dopiero wtedy zacznie on być widziany przez MPI.

d) Grupowanie danych

Istnieje również prostsza metoda wysyłania większej ilości danych za jednym zamachem - polega ona na tym, że przed każdą operacją typu MPI_Send() ręcznie pakujemy wszystkie dane do jednego obszaru pamięci, a druga strona po odebraniu komunikatu może je rozpakować w analogiczny sposób. Odpowiednie funkcje mają następującą postać:

• int MPI_Pack(void *pack_data, int in_count,
               MPI_Datatype datatype, void *buffer,
               int size, int *position_ptr, MPI_Comm comm);

  Parametr pack_data powinien zawierać in_count elementów typu datatype. Parametr position_ptr zawiera adres początkowy w buforze wyjściowym buffer i po wykonaniu funkcji jego wartość jest uaktualniana (dodawana jest długość zapisanych danych). Parametr size określa rozmiar obszaru pamięci wskazywanej przez buffer.

• int MPI_Unpack(void *buffer, int size,
                 int *position_ptr, void *unpack_data, int count,
                 MPI_Datatype datatype, MPI_Comm comm);

  Funkcja odwrotna do poprzedniej - rozpakowuje count elementów typu datatype ze zmiennej buffer począwszy od pozycji position do obszaru pamięci wskazywanego przez unpack_data i odpowiednio uaktualnia zmienną position.

e) Synchronizacja

Podstawową (i najczęściej stosowaną) funkcją MPI służącą do synchronizacji procesów jest

• int MPI_Barrier(MPI_Comm comm);

Wywołanie jej w pewnym miejscu programu powoduje, będzie on czekał, aż wszystkie pozostałe jego instancje dojdą do tego miejsca i dopiero potem ruszy dalej.

Komunikacja non-blocking

MPI umożliwia, oprócz klasycznej komunikacji typu MPI_Send/ MPI_Recv również przesyłanie wiadomości bez blokowania programu. W skrócie wygląda to tak, że po wywołaniu funkcji MPI_Isend program może wykonywać inne czynności (oczywiście nie zmieniające bufora wysyłanego komunikatu), natomiast w chwili, gdy chce się upewnić, że komunikat został już wysłany w całości, wywołuje MPI_Wait, lub MPI_Test.

W podobny sposób realizowany jest odbiór komunikatów bez blokowania - analogiczna funkcja nosi nazwę MPI_Irecv.