Разработка параллельных программ в системах с распределенной памятью

распределенные вычисления” Лекция 22

разработку сложных схем обмена между процессами
Для этого разработаны специальные интерфейсы

кода C и Fortran
Существует множество реализаций
mpich
LAM
SCALI
..
По исполняемому коду, протоколу передачи

и др. совместимость не гарантируется
MPI – надстройка над сокетами, общей памятью и др.

обмениваться между собой сообщениями
Для обмена сообщениями процессы объединяются в группы

– коммуникаторы или коммуникационные области (домены)
MPI предоставляют функции обмена сообщениями внутри любых коммуникационных доменов
С помощью библиотеки MPI каждый процесс может определить свой номер внутри домена и выполнять необходимую часть расчетов (SIMD, MISD, MIMD)
Все процессы выполняют один и тот же исполняемый файл
Разные процессы выполняют разные исполняемые файлы
Каждое сообщение имеет свой номер – идентификатор, по которому сообщения можно отличать друг от друга

$LAMRSH
export LAMRSH="rsh“
Виртуальная машина запускается командой
lamboot
Первой в

списке должна быть локальная машина
node1
node2

же исполняемый файл
Первой MPI функцией должна быть вызвана
MPI_Init( &argc,

&argv )
Последней одна из функций
MPI_Finalize() - нормальное завершение
MPI_Abort( описатель области связи, код ошибки MPI ); - завершение с сообщением об ошибке
Между этими вызовами можно вызывать любые функции MPI

основному коммуникационному домену с именем MPI_COMM_WORLD
Каждому процессу после запуска присваивается

номер
Для любого коммуникационного домена каждый процесс может узнать свой номер и общее количество процессов в домене
int size, rank;
MPI_Comm_size( MPI_COMM_WORLD, &size ); - количество процессов в домене
MPI_Comm_rank( MPI_COMM_WORLD, &rank ); - номер процесса в домене
Эти вызовы - только после MPI_Init

int my_number; //Мой номер процесса
int proc_num;

//общее количество процессов

MPI_Init(&argc, &argv);
MPI_Comm_size(MPI_COMM_WORLD, &proc_num);
MPI_Comm_rank(MPI_COMM_WORLD, &my_number);

printf("I an process %d, sending data\n", my_number);

MPI_Finalize();
return 0;
}

с обменом в основном коммуникационном домене
Чтобы теги сообщений повторялись
Чтобы выполнять

коллективные операции обмена
Алгоритм Фокса
Это делается путем создания нового коммуникационного домена
Копированием существующего
Разделением существующего на поддомены
Отображение групп на коммуникационные домены

с обменом в старом
Все процессы старого коммуникатора должны вызвать

эту функцию
MPI_Comm tempComm; MPI_Comm_dup( MPI_COMM_WORLD, &tempComm ); /* ... передаем данные через tempComm ... */ MPI_Comm_free( &tempComm );

процессы старого коммуникатора должны вызвать эту функцию
Все процессы с одинаковым

«цветом» (задается) попадают в один коммуникатор
Номер процесса в новом коммуникаторе задается
MPI_Comm_split( existingComm, /* расщепляемый описатель ( MPI_COMM_WORLD) */ indexOfNewSubComm, /* номер подгруппы, куда попадает вызывающий процесс */ rankInNewSubComm, /* желательный номер в новой подгруппе */ &newSubComm ); /* описатель области связи новой подгруппы */

rank, subrank, subsize;
MPI_Init(&argc,&argv);
MPI_Comm_size(MPI_COMM_WORLD, &size);
MPI_Comm_rank( MPI_COMM_WORLD, &rank );

//разбиваем набор процессов на

3 группы
//нумерация процессов в каждой группе 0-2
MPI_Comm_split(MPI_COMM_WORLD, rank/3, rank%3, &subComm);

MPI_Comm_rank( subComm, &subrank );
MPI_Comm_size( subComm, &subsize );
printf("old size = %d new size %d ",size, subsize);
printf("old rank = %d new rank %d\n",rank, subrank);
MPI_Comm_free(&subComm);
MPI_Finalize();
return 0;
}

-np 10 -wd /net/s1/$PWD ./comm_split
old size = 10 new size

1 old rank = 9 new rank 0
old size = 10 new size 3 old rank = 8 new rank 2
old size = 10 new size 3 old rank = 6 new rank 0
old size = 10 new size 3 old rank = 7 new rank 1
old size = 10 new size 3 old rank = 0 new rank 0
old size = 10 new size 3 old rank = 1 new rank 1
old size = 10 new size 3 old rank = 2 new rank 2
old size = 10 new size 3 old rank = 4 new rank 1
old size = 10 new size 3 old rank = 5 new rank 2
old size = 10 new size 3 old rank = 3 new rank 0

коммуникатора
В/из группы добавляются/удаляются номера процессов
На базе группы создается новый коммуникатор
Все

процессы с общим коммуникатором должны вызвать соответствующие функции

каждый процесс параллельной программы
В каждом процессе можно создавать свои группы
MPI_Group

group;
MPI_Comm_group(MPI_COMM_WORLD, &group);

групп
MPI_Group group, subgroup;
int ranks[5], i;
for(i = 0; i

MPI_Comm subComm;
int size, rank,

subrank, subsize,i;
int ranks[5];
MPI_Group group, subgroup;
MPI_Init(&argc,&argv);
MPI_Comm_size(MPI_COMM_WORLD, &size);
MPI_Comm_rank( MPI_COMM_WORLD, &rank );
MPI_Comm_group(MPI_COMM_WORLD, &group);
for(i = 0; i<5; i++) ranks[i]=i;
MPI_Group_excl(group,5,ranks,&subgroup);
MPI_Comm_create(MPI_COMM_WORLD, subgroup,&subComm);
if( subComm == MPI_COMM_NULL) {
printf ("%d not in communicator\n",rank);
} else {
MPI_Comm_rank( subComm, &subrank );
MPI_Comm_size( subComm, &subsize );
printf("old size = %d new size %d ",size, subsize);
printf("old rank = %d new rank %d\n",rank, subrank);
}
if( subComm != MPI_COMM_NULL) MPI_Comm_free(&subComm);
MPI_Group_free(&subgroup);
MPI_Group_free(&group);
MPI_Finalize();
return 0;
}

-np 10 -wd /net/s1/$PWD ./comm_group
2 not in communicator
0 not in

communicator
1 not in communicator
4 not in communicator
3 not in communicator
old size = 10 new size 5 old rank = 8 new rank 3
old size = 10 new size 5 old rank = 9 new rank 4
old size = 10 new size 5 old rank = 5 new rank 0
old size = 10 new size 5 old rank = 6 new rank 1
old size = 10 new size 5 old rank = 7 new rank 2

MPI_Get_processor_name(nm, &nm_size);

в виде сообщений
Сообщение имеет номер (тег)
В одном сообщении могут быть

данные любой сложности
Операции передачи данных
Точка-точка
Коллективные
Broadcst
Reduce
Синхронные – блокируется пока не завершится
Асинхронные – не блокируется

MPI_COMM_WORLD );
Прием
int buf[10];
MPI_Status status;
MPI_Recv( buf, 10, MPI_INT,

0, 0, MPI_COMM_WORLD, &status );
Аргументы
Буфер данных
Количество элементов данных
Тип каждого элемента данных буфера
Номер процесса кому/от кого передавать/принимать
Может быть MPI_ANY_SOURCE
Тег сообщения
Может быть MPI_ANY_TAG
Коммуникатор в каком передавать
Результат операции (только для приема)

неизвестном количестве
При приеме может возникнуть ошибка
Определение количества данных
MPI_Status status; int

count; MPI_Recv( ... , MPI_INT, ... , &status ); MPI_Get_count( &status, MPI_INT, &count ); /* ... теперь count содержит количество принятых ячеек */

subsize;
int vector[16],i;
int resultVector[16];
MPI_Init(&argc,&argv);
MPI_Comm_size(MPI_COMM_WORLD, &size);
MPI_Comm_rank( MPI_COMM_WORLD, &rank );

for( i=0; i

) vector[i] = rank*100 + i;

MPI_Reduce(
vector, /* каждая задача в коммуникаторе предоставляет вектор */
resultVector, /* задача номер 'root' собирает данные сюда */
16, /* количество ячеек в исходном и результирующем массивах */
MPI_INT, /* тип элемента данных */
MPI_SUM, /* описатель операции: поэлементное сложение векторов */
0, /* номер задачи, собирающей результаты в 'resultVector' */
MPI_COMM_WORLD /* описатель области связи */
);
if( rank==0 ) for(i=0; i<16; i++)
printf("resultVector[%d]=%d ",i,resultVector[i]);
MPI_Finalize();
return 0;
}

resultVector[2]=4520 resultVector[3]=4530 resultVector[4]=4540 resultVector[5]=4550 resultVector[6]=4560 resultVector[7]=4570 resultVector[8]=4580 resultVector[9]=4590 resultVector[10]=4600 resultVector[11]=4610

resultVector[12]=4620 resultVector[13]=4630 resultVector[14]=4640 resultVector[15]=4650

разному
Необходимо при передаче преобразовывать представления данных (XDR)
Для каждого типа своя

функция преобразования
Можно создавать свои типы данных, которые тоже будут передаваться корректно

MPI_FLOAT MPI_DOUBLE MPI_LONG_DOUBLE MPI_BYTE MPI_PACKED
Fortran
MPI_CHARACTER MPI_INTEGER MPI_REAL MPI_DOUBLE_PRECISION

MPI_COMPLEX MPI_DOUBLE_COMPLEX MPI_LOGICAL MPI_BYTE MPI_PACKED

тип
Это упрощает сериализацию
Новый тип создается на основе уже существующих

int *len, /* массив с длинами полей */ /* (на

тот случай, если это массивы) */ MPI_Aint *pos, /* массив со смещениями полей */ /* от начала структуры, в байтах */ MPI_Datatype *types, /* массив с описателями типов полей */ MPI_Datatype *newtype ); /* ссылка на создаваемый тип */

struct {
int i;
double d[3];
long l[8];
char c;

} AnyStruct;

AnyStruct st;
MPI_Datatype anyStructType;

int len[5] = { 1, 3, 8, 1, 1 };
MPI_Aint pos[5] = { offsetof(AnyStruct,i), offsetof(AnyStruct,d), offsetof(AnyStruct,l),
offsetof(AnyStruct,c), sizeof(AnyStruct)
};

MPI_Datatype typ[5] = { MPI_INT,MPI_DOUBLE,MPI_LONG,MPI_CHAR,MPI_UB };
MPI_Type_struct( 5, len, pos, typ, &anyStructType );
MPI_Type_commit( &anyStructType ); /* подготовка закончена */

MPI_Send( st, 1, anyStructType, ... );

массив
int MPI_Pack(void *buf, int count, MPI_Datatype dtype, void *packbuf, int

packsize, int *packpos, MPI_Comm comm)
INPUT PARAMETERS
buf - input buffer start (choice)
count - number of input data items (integer)
dtype - datatype of each input data item (handle)
packsize - output buffer size, in bytes (integer)
comm - communicator for packed message (handle)

INPUT/OUTPUT PARAMETER
packpos - current position in buffer, in bytes (integer)

OUTPUT PARAMETER
packbuf - output buffer start (choice)

*buf, int count, MPI_Datatype dtype, MPI_Comm comm)

INPUT PARAMETERS

packbuf - input buffer start (choice)
packsize - size of input buffer, in bytes (integer)
count - number of items to be unpacked (integer)
dtype - datatype of each output data item (handle)
comm - communicator for packed message (handle)

INPUT/OUTPUT PARAMETERS
packpos - current position in bytes (integer)

OUTPUT PARAMETER
buf - output buffer start (choice)

char c[8];
} s;

Передача
int bufPos = 0;
char tempBuf[

sizeof(s) ];
MPI_Pack(&s.i, 1, MPI_INT, tempBuf, sizeof(tempBuf), &bufPos, MPI_COMM_WORLD );
MPI_Pack( s.f, 4, MPI_FLOAT, tempBuf, sizeof(tempBuf), &bufPos, MPI_COMM_WORLD );
MPI_Pack( s.c, 8, MPI_CHAR, tempBuf, sizeof(tempBuf), &bufPos, MPI_COMM_WORLD );
MPI_Send( tempBuf, bufPos, MPI_BYTE, targetRank, msgTag, MPI_COMM_WORLD );

Прием
int bufPos = 0;
char tempBuf[ sizeof(s) ];
MPI_Recv( tempBuf, sizeof(tempBuf), MPI_BYTE, sourceRank, msgTag, MPI_COMM_WORLD, &status );
MPI_Unpack( tempBuf, sizeof(tempBuf), &bufPos,&s.i, 1, MPI_INT, MPI_COMM_WORLD);
MPI_Unpack( tempBuf, sizeof(tempBuf), &bufPos, s.f, 4, MPI_FLOAT,MPI_COMM_WORLD);
MPI_Unpack( tempBuf, sizeof(tempBuf), &bufPos, s.c, 8, MPI_CHAR, MPI_COMM_WORLD);

данных
К передаваемым данным нельзя обращаться в процессе передачи-приема
Необходимо специально проверять

завершение операции

один
Статус выполнения запроса request

завершения
MPI_Wait(MPI_Request *preq, MPI_Status *stat)

Ожидание завершения нескольких запросов
int MPI_Waitall(int count, MPI_Request

*reqs, MPI_Status *stats)

tag1, MPI_COMM_WORLD, &reqs[0]);
MPI_Irecv(&buf[1], 1, MPI_INT, next, tag2, MPI_COMM_WORLD, &reqs[1]);
MPI_Isend(&rank,

1, MPI_INT, prev, tag2, MPI_COMM_WORLD, &reqs[2]);
MPI_Isend(&rank, 1, MPI_INT, next, tag1, MPI_COMM_WORLD, &reqs[3]);

{ do some work }

MPI_Waitall(4, reqs, stats);

на удаленной машине
Синхронизация
Создание и остановка процессов
Присоединение к работающей программе

передача сообщений
Управление буфером сообщений
Функции управления системой PVM

управления
Все системные переменные указанные в системной переменной PVM_EXPORT экспортируются на

все процессы задачи
export DISPLAY=myworkstation:0.0
export MYSTERYVAR=13
export PVM_EXPORT=“DISPLAY:MYSTERYVAR”
Машины можно добавлять через командную строку - pvm
По умолчанию все программы должны находится в каталогах
$PVM_ROOT
~pvm3/bin/$PVM_ARCH/

локальной машине
Запуск программ на других машинах выполняется через команду $PVM_RSH
export

PVM_RSH="/usr/bin/rsh“
На всех машинах должны быть установлены системные переменные
PVM_ARCH PVM_ROOT PVM_RSH XPVM_ROOT
export PVM_ARCH="LINUXI386“
export PVM_ROOT="/usr/share/pvm3“
export PVM_RSH="/usr/bin/rsh“
Это можно сделать в файле
~/.bashrc

имя программы
char **argv, // набор аргументов программы
int flag, //

особенность интерпретации // аргументов программы
char *where, // масив имен машин
int ntask, // количество заданий
int *tids ) // возвращаются
// идентификаторы заданий

– завершение своего процесса

int nhost, int *infos )
Удаление физических машин
int info = pvm_delhosts(

char **hosts, int nhost, int *infos )
Остановка виртуальной машины
pvm_halt()

char* hosts[]={"node1", "node2", "node3"};
int tids[16];

int status[3];
int ntasks, nhosts;
char* task="/net/node1/home/saa/src/parallel_and_distributed/pvm/create";
char* arg[]={"child",NULL};

if(argc <2){
nhosts = pvm_addhosts(hosts,3,status);
printf("added %d hosts\n",nhosts);
ntasks = pvm_spawn(task,arg,0,NULL,16,tids);
printf("spawned %d tasks\n",ntasks);
pvm_halt();
}

return 0;
}

-o create
/usr/share/pvm3/lib/LINUXI386/libpvm3.a(lpvmgen.o)(.text+0x292): In function `pvmlogperror':
: warning: `sys_errlist' is deprecated; use

`strerror' or `strerror_r' instead
/usr/share/pvm3/lib/LINUXI386/libpvm3.a(lpvmgen.o)(.text+0x289): In function `pvmlogperror':
: warning: `sys_nerr' is deprecated; use `strerror' or `strerror_r' instead
[saa@cluster pvm]$ pvm
pvm> quit

Console: exit handler called
pvmd still running.
[saa@cluster pvm]$ ./create
added 2 hosts
spawned 16 tasks
Terminated

несколько буферов
Один – существует по умолчанию
Буфер инициализируется с помощью функции
int

bufid = pvm_initsend( int encoding )
encoing – PvmDataDefault или 0 соответсвует XDR формату
Новый буфер можно создать и удалить
pvm_mkbuf(int encoding )
pvm_freebuf(int buf)
Для установки буфера по умолчанию
oldbuf = pvm_setrbuf( int bufid )
oldbuf = pvm_setsbuf( int bufid )

разных типов данных
Данные пакуются в буфер, используемый по умолчанию
Упаковка
pvm_pkbyte( char

*xp, int nitem, int stride )
pvm_pkdouble( double *dp, int nitem, int stride )
pvm_pkint( int *ip, int nitem, int stride )
pvm_pkstr( char *sp )
Распаковка
pvm_upkbyte( char *xp, int nitem, int stride)
pvm_upkdouble( double *dp, int nitem, int stride)
pvm_upkint( int *ip, int nitem, int stride)
pvm_upkstr( char *sp )
Аргументы
xp – указатель на буфер памяти откуда/куда будет выполняться операция
nitem – количество элементов массива одного типа для упаковки/распаковки
stride – через сколько элементов “перепрыгивать”

in tag)
Прием
int pvm_recv(int tid, in tag)
Проверка на наличие сообщений в

буфере
int pvm_nrecv(int tid, in tag)

char* hosts[]={"node1", "node2", "node3"};
int

tids[16];
int status[3];
char message[4096];
int ntasks, nhosts,i;
char* task="/net/node1/home/saa/src/parallel_and_distributed/pvm/msg";
char* arg[]={"child",NULL};

if(argc <2){ //parent
nhosts = pvm_addhosts(hosts,3,status);
printf("added %d hosts\n",nhosts);
ntasks = pvm_spawn(task,arg,0,"",16,tids);
printf("spawned %d tasks\n",ntasks);
for (i = 0; i<16;i++){
pvm_recv(tids[i],1);
pvm_upkstr(message);
printf("child %d work at %s\n",i,message);
}
pvm_halt();
} else {
pvm_initsend(PvmDataDefault);
gethostname(message,4096);
pvm_pkstr(message);
pvm_send(pvm_parent(),1);
pvm_exit();
}

return 0;
}

In function `pvmlogperror':
: warning: `sys_errlist' is deprecated; use `strerror' or

`strerror_r' instead
/usr/share/pvm3/lib/LINUXI386/libpvm3.a(lpvmgen.o)(.text+0x289): In function `pvmlogperror':
: warning: `sys_nerr' is deprecated; use `strerror' or `strerror_r' instead
[saa@cluster pvm]$ pvm
pvm> quit

Console: exit handler called
pvmd still running.
[saa@cluster pvm]$ ./msg
added 2 hosts
spawned 16 tasks
child 0 work at ss20-2.univ.kiev.ua
child 1 work at ss20-2.univ.kiev.ua
child 2 work at ss20-2.univ.kiev.ua
child 3 work at ss20-2.univ.kiev.ua
child 4 work at ss20-2.univ.kiev.ua
child 5 work at ss20-2.univ.kiev.ua
child 6 work at ss20-3.univ.kiev.ua
child 7 work at ss20-3.univ.kiev.ua
child 8 work at ss20-3.univ.kiev.ua
child 9 work at ss20-3.univ.kiev.ua
child 10 work at ss20-3.univ.kiev.ua
child 11 work at cluster.univ.kiev.ua
child 12 work at cluster.univ.kiev.ua
child 13 work at cluster.univ.kiev.ua
child 14 work at cluster.univ.kiev.ua
child 15 work at cluster.univ.kiev.ua
Terminated