Как нам распараллелить программу

А.П.Сапожников Статья из "Информационного бюллетеня ЛИТ" N2[43]
4-8160, Дубна, ОИЯИ, 2003, с.37-50.

Если взглянуть на современную редакцию известного списка Top500 самых мощных вычислительных систем в мире, можно сделать любопытное открытие: там нет ни одной привычной нам машины с одним процессором! Более того - даже систем с малым числом процессоров, таких как наш 8-процессорный SPP-2000, совсем немного. Подавляющее преимущество здесь имеют кластеры, состоящие из тысяч процессоров.
Однако обычные наши фортранные программы просто так не в состоянии использовать одновременно более одного процессора. Их необходимо подвергнуть так называемому распараллеливанию, то есть как-то преобразовать. Настоящий опус и призван показать, как можно это делать, используя инструментальный пакет MPI, ныне входящий в состав практически любой вычислительной системы. Почему именно MPI, и что это за штука?
Вообще говоря, существует два подхода к распараллеливанию программ. Первый из них, и самый привлекательный, состоит в том, что проблема возлагается на компилятор, а пользователь только выдает компилятору ценные указания типа того: вот здесь попробуй распараллелить, а вот тут - не моги! Этот подход принят в системе OpenMP, довольно распространенной в настоящее время. Недостаток здесь только один, но существенный: все это может работать только в вычислительных системах с общей памятью! А кластеры, увы, к таковым не относятся.
Второй подход существенно труднее: пользователь сам, вручную, преобразует текст своей (или чужой!) программы, явно программируя как распределение работы между процессорами, так и все необходимые межпроцессорные коммуникации. А чтобы не думать о процессорах (кто его знает - сколько их всего!), вместо процессоров в ход идет понятие процессов. За последние 30 лет мы все уже привыкли к тому, что даже на единственном процессоре могут успешно сосуществовать много процессов. Этот подход работоспособен практически на всех вычислительных системах, программное обеспечение которых имеет в своем составе библиотеку программ для организации межпроцессных коммуникаций. Самый продвинутый на сегодня инструментарий такого рода - это пакет "Message Passing Interface", или просто MPI, работающий практически везде, претендовавший, начиная с 1995, года на роль стандарта при распараллеливании программ, а ныне благополучно таковым стандартом и ставший.
На базе MPI существует несколько десятков технологий параллельного программирования. Наиболее интересной из них, на наш взгляд, является DVM - Distributed Virtual Memory, разработанная в ИПМ РАН. В чем-то она пытается сочетать оба описанных выше подхода. Здесь так же, как в OpenMP, пользователь имеет возможность указать компилятору, где распараллеливать, а где нет. С другой стороны, аналогично явному программированию межпроцессных коммуникаций, пользователь явно программирует распределение памяти между виртуальными процессорами. Причины, по которым мы все же начинаем с MPI, просты:
a) это базовый инструмент, все прочие - это навороты над ним;
b) это стандартный инструмент, он работает везде, а главное - на наших машинах;
c) это простой и по-человечески привычный инструмент.

А теперь уместно задаться основным вопросом: а надо ли заниматься этим самым распараллеливанием, каков будет предел наших грядущих достижений? Предел этот описывается так называемым законом Амдаля.

Пусть 0 <= S <= 1 - доля вычислительных операций нашей программы, которые должны совершаться сугубо последовательно. Тогда при одновременном использовании P процессоров мы можем ускорить свою программу максимум в

K = 1 / (S + ( 1 - S ) / P )     раз.

Устремляя Р к бесконечности, получим K < 1 / S. В частности, если S > 0.1, то K < 10 при любом P. Если же S = 0 (чего в природе, вообще говоря, не наблюдается), то K = P. Напомним, что в 1998 году с появлением SPP-2000 у нас стало P = 8. Не прошло и 4 лет, как в нашем новом LINUX-кластере стало P = 16 и в ближайшее время возможно дальнейшее увеличение. Так что решайте сами - окупятся Ваши труды, или нет! Впрочем, пакет MPI позволяет распараллеливать программу для выполнения на любом количестве даже разнотипных компьютеров, соединенных в локальную сеть.

Тех, кто хочет получить более подробную информацию о распараллеливании вычислений или подробное описание пакета MPI, мы отсылаем к специализированному серверу МГУ. Здесь же Вас ждут только основы идеологии MPI и несколько конкретных примеров.

Основные понятия MPI. Парадигма SPMD

При запуске задачи создается группа из NProc процессов. Группа идентифицируется целочисленным дескриптором (коммуникатором). Внутри группы процессы нумеруются от 0 до NProc-1. В ходе решения задачи исходная группа (ей присвоено имя MPI_COMM_WORLD) может делиться на подгруппы, подгруппы могут объединяться в новую группу, имеющую свой коммуникатор. Таким образом, процесс может одновременно принадлежать нескольким группам процессов. Каждый процесс может узнать свой номер myProc внутри любой группы, к которой он принадлежит.
Поведение всех процессов описывается одной и той же программой. Межпроцессные коммуникации в ней программируются явно с использованием библиотеки MPI, которая и диктует стандарт программирования.
Квазиодновременный запуск исходной группы процессов производится средствами операционной системы. При этом NProc определяется желанием пользователя, а отнюдь не количеством доступных процессоров!
Итак, все NProc процессов асинхронно выполняют одну и ту же программу. Но у каждого из них свой номер myProc. Поэтому в программе естественно будут такие фрагменты:

Таким образом, в программе "под одной крышей" закодировано поведение всех процессов. В этом и заключена парадигма программирования SPMD: Single Program - Multiple Data.
Не надо путать аббревиатуру SPMD с SIMD. Последняя обозначает специфическую компьютерную архитектуру в классификации Флинна, когда несколько процессоров СИНХРОННО исполняют одну и ту же программу, а о процессах и речи не идет!

Словарь пакета MPI состоит из некоторого количества поименованных целочисленных констант. Все эти имена начинаются с префикса MPI_. Описания всех этих констант хранятся в системном файле mpif.h для Фортрана и mpi.h для языка С. Вот важнейшие из них:

Основные операции MPI:

Пожалуй, уже пора изваять нашу первую MPI-программу. Как запустить ее "на счет", мы расскажем попозже, когда дело дойдет до команды qsub. 
         Program Hello
         Include 'mpif.h'
         call MPI_Init(ierr)
         call MPI_Comm_Size(MPI_Comm_World, NProc, ierr)       ! Сколько нас?
         call MPI_Comm_Rank(MPI_Comm_World, myProc, ierr)      ! Кто я?
         write(*,*) ' Hello, I am ',myProc,' process of ',NProc,
        &           ' in a group ',MPI_COMM_WORLD 
         call MPI_Finalize(ierr)
         End
Забегая вперед, скажем, что выходной файл, полученный этой программой, будет содержать NProc строк. Например, при NProc=3 Вы увидите следующее: 
         Hello, I am 0 process of 3 in a group 91
         Hello, I am 2 process of 3 in a group 91
         Hello, I am 1 process of 3 in a group 91
Порядок строк может быть и иным, ведь процессы трудились асинхронно!

"Point-to-Point" операции MPI (самые главные):

Процесс myProc (тот, кто выполняет операцию!) посылает процессу Whom (принимает от процесса From) Cnt штук данных типа Type в массив Buf с тегом Tag. В качестве From и Tag могут фигурировать джокеры MPI_Any_Source, MPI_Any_Tag. Сие означает "принимаю от кого угодно" и "принимаю с любым тегом". Кстати, почему размер сообщения измеряется в штуках, а не в байтах, словах или в прочих привычных нам единицах? Да просто потому, что один и тот же тип данных в общем случае требует разного количества памяти на разных машинах! Более того, мы избавлены и от забот по преобразованию чисел из одного внутримашинного представления в другое.
Далее, comm - целочисленный дескриптор группы (в терминологии MPI - коммуникатор), к которой принадлежит процесс Whom (From). Напомним, что внутри каждой группы нумерация процессов независимая. Обычно в качестве коммуникатора comm используют MPI_Comm_World - группу, содержащую все запущенные процессы. Status - это массив из 3 целочисленных полей ( integer status(3) ), содержащих основные атрибуты последнего принятого процессом сообщения:
Status(MPI_Source) - от кого оно принято
Status(MPI_Tag) - с каким Tag-ом оно передавалось
Status(MPI_Error) - были ли ошибки при передаче

В переменную ierr MPI запишет 0, если операция выполнена нормально, иначе там будет ненулевой код ошибки. Вообще-то этот параметр присутствует практически во всех MPI-операциях, и всегда является последним в списке параметров. Поскольку программистам вообще свойственно игнорировать проверки кодов ошибок, и дабы упростить изложение, мы в дальнейшем про него забудем. Тем более, что любители языка С этого параметра не видят вообще, он спрятан в выходное значение соответствующей функции. Так, аналогом фортранного MPI_Send будет С - функция 
          int MPI_Send(void* buf, int cnt, MPI_Datatype type, 
                      int whom, int tag, MPI_Comm comm)

Коллективные операции MPI:

Здесь процесс с номером root инициирует операцию, в то время как остальные процессы "работают на подхвате".

Op = ( MPI_Min, MPI_Max, MPI_Sum, MPI_Prod, ...)

Семантику основных коллективных операций поясняет рис.1, взятый из [4], а семантику операций семейства MPI_Reduce - рис.2.
Важно отметить, что коллективные операции исполняются ВСЕМИ ПРОЦЕССАМИ ГРУППЫ! Но исполняются по-разному, в зависимости от того, является процесс главным в коллективе, или нет!
Вообще-то очевидно, что эти операции можно построить самому из двух базовых, SEND и RECV, но для нашего удобства в MPI есть более 200 различных операций. В назидательных целях изобразим самодельную программу MPI_Bcast: 
         Subroutine MPI_Bcast(Buf,Cnt,Type,root,comm) ! передает Cnt штук данных
         Dimension Buf(*)                             ! типа Type, лежащих в BUF,
         Integer Cnt,Type,root,comm, newcomm,status(3)! от процесса root всем прочим 
         Include 'mpif.h'
         call MPI_Comm_Dup(comm,newcomm)              ! создаем временную группу              
         call MPI_Comm_Size(newcomm,NProc)            ! и работаем внутри нее !
         call MPI_Comm_Rank(newcomm,myProc)
         if (myProc.eq.root) then
           do k=0,NProc-1                             !   Наиболее тупым способом
             call MPI_Send(Buf,Cnt,Type, k, 0,newcomm)
           enddo                                      !   Сам себе - тоже посылал !
         endif
         call MPI_Recv(Buf,Cnt,Type, root, 0,newcomm, status)
         call MPI_Comm_Free(newcomm)                  ! уничтожим временную группу
         end

Обратите внимание на то, что мы пользуемся временным коммуникатором newcomm вместо исходного коммуникатора comm. Это необходимо для того, чтобы Send, испущенный внутри этой программы, не нарвался на Recv, испущенный вне ее. Такая техника использования временных, рабочих групп процессов характерна для библиотечных программ, желающих скрыть от внешнего мира свои внутренние межпроцессные коммуникации.

Далее мы увидим пример использования операции MPI_AllReduce, а за разъяснением смысла других коллективных операций отсылаем читателя в Интернет.

Вот этот пример: коллективное интегрирование функции одной переменной F(x)

         Program Cooperative_Integration
         Include 'mpif.h'
         external F                   ! интегрируемая функция
         data A/0/,  B/1/             ! пределы интегрирования
         call MPI_Init
         call MPI_Comm_Size(MPI_Comm_World, NProc)       ! Сколько нас?
         call MPI_Comm_Rank(MPI_Comm_World, myProc)      ! Кто я?
         dx=(B-A)/NProc                      !   Делим интервал поровну
         a1=A+myProc*dx                      !   между всеми процессами группы
         b1=a1+dx                    
         s1=Common_Integration( F, a1, b1 )  ! - обычная библиотечная пр-ма
         call MPI_AllReduce(s1,S,1,MPI_Real, MPI_Sum, MPI_Comm_World)
         if(myProc.eq.0) write(*,*) ' Integral=',S
         call MPI_Finalize
         End

Здесь каждый процесс интегрирует, как умеет, F(x) на своем подинтервале, используя обычную, не распараллеленную процедуру Common_Integration. А операция MPI_AllReduce делает главную работу - суммирует эти подинтегральчики, живущие в переменной S1 каждого процесса, помещая результат суммирования в переменную S. Откуда это видно - да потому, что параметр OP в ней равен MPI_Sum.
Печать, и вообще весь вывод, естественно выполняет кто-то один, и очевидно это должен быть процесс 0, ведь наша программа должна правильно работать при любом NProc, в частности при NProc=1, а только 0-й процесс есть всегда, при любом составе группы процессов, вовлеченных в решение задачи!
Что же касается ввода, то тут чуть сложнее. Ясно, что файлы с входной информацией должны принадлежать кому-то одному, стало быть 0-му процессу в основной группе MPI_COMM_WORLD. Прочитав из файла, он должен поделиться прочитанным со своими партнерами: 
         Real*4  array(100)
           .  .  . 
         if(myProc.eq.0) then 
           open(1,file='input.dat',...)
           read(1) array
           close(1) 
         endif
         call MPI_Bcast(array,100,MPI_REAL,0,comm)
Это типичная техника организации ввода начальных данных при распараллеливании программы. Добавляются две вещи: условный переход и MPI_Bcast.

Столь же доступным примером является программа умножения матриц, где все элементы матрицы-произведения можно вычислять параллельно, т.е. S из закона Амдаля, так же как и в предыдущем примере, очевидно близко к 0. Вот "нераспараллеленный" вариант: 
         program mumu         ! matrix multiplication  С = A * B
         parameter (N=400)    ! matrix dimension
         real*8 A(N,N),B(N,N),C(N,N)
         real*8 t
   C...     мы опустили формирование исходных матриц А и В
         do i=1,N
           do j=1,N
             t=0.0
             do k=1,N
               t=t+A(i,k)*B(k,j)
             enddo
             C(i,j)=t
           enddo
         enddo
         end
Теперь попытаемся проделать эту же работу коллективом из нескольких процессов. Процессы, пронумерованные от 0 до P-1, исполняют один и тот же программный код, используя независимо работающие процессоры. Процесс 0 распределяет работу между всеми исполнителями, пересылая им обе исходные матрицы А и В. Каждый исполнитель (в том числе и сам процесс 0) вычисляет "свои" столбцы матрицы С, после чего пересылает результат своей работы обратно процессу 0.

"Распараллеленный" вариант той же программы.

Итак, мы видим, что даже в простейших случаях распараллеливание программы требует изрядных усилий. Более того, в ходе вычислений как правило необходимы межпроцессные коммуникации, которые могут вообще "съесть" весь эффект от распараллеливания. Здесь все зависит от соотношения цены программы и стоимости Вашего труда:

Еще одно замечание. Интуиция подсказывает, что достаточно легко могут быть распараллелены так называемые Монте-Карловские программы, где вычислительной обработке подвергаются независимые события, сгенерированные с помощью датчика случайных чисел. Здесь важно обеспечить, чтобы каждый из параллельно работающих процессов получал свою, независимую от остальных процессов, серию случайных чисел. Для этого каждый из процессов, начиная свою работу, должен как-то по-своему инициализировать датчик. На наш взгляд, идеальным датчиком для использования в распараллеленной программе является датчик, предложенный G.Marsaglia, способный выдавать до 32000 независимых серий равномерно распределенной на [0,1] случайной величины. Лучше всего инициализировать серию номером своего процесса: 
           .  .  .
         call MPI_Comm_rank(MPI_COMM_WORLD,myProcess)       ! кто я?
         call RandomInitiate(myProcess,myProcess)           ! начинаем свою серию
           .  .  .
         R = Random(1)   !  Real*8  uniform-distributed number on [0,1]

Датчик входит в состав нашей библиотеки JINRLIB. Немаловажным обстоятельством является то, что это самый быстрый из известных нам датчиков: всего 5 сложений и ни одного умножения с плавающей запятой! Кстати: использование пакета MPI вовсе не уменьшит мобильность Вашей программы. На машинах, где нет MPI, Вы можете использовать заглушку:

файл mpif.h:
         parameter(MPI_COMM_WORLD=0)
         parameter(MPI_DOUBLE_PRECISION=0)
         parameter(MPI_STATUS_SIZE=10)
файл mpi.for:
         subroutine MPI_Comm_rank(comm,myProcess,ierr)
         myProcess=0   !    наш процесс имеет номер 0
         return
         subroutine MPI_Comm_size(comm,NProcs,ierr)
         NProcs=1      !    и он единственный
         return
          . . .        ! все остальные "MPI-программы" - пустые !!!

Применение этой заглушки позволит запускать Вашу программу в однопроцессном режиме, не меняя ее текста.

Как заставить работать MPI-программу на нашем SPP-2000?

Вы должны получить доступ к пакету MPI. Для этого надо дополнить свои переменные окружения PATH и MANPATH:
setenv PATH /opt/mpi/bin:$PATH
setenv MANPATH /opt/mpi/share/man:$MANPATH

Добавьте это заклинание к своему .login -файлу.

Вместо трансляторов f77 и cc вызывайте mpif77 и mpicc:
mpif77 example.f -o primer
mpicc example.c -o primer
При запуске программы указывайте, на сколько процессов Вы хотите ее распараллелить:
primer -np 3
- в данном случае на троих.
Если вызовете без параметров - будет работать в одиночку. Кстати - ничто не мешает Вам указывать число процессов большее, чем количество имеющихся в наличии процессоров! Просто Ваши процессы будут простаивать в очереди к процессорам, напрасно расходуя ресурсы системы.
Наши эксперименты с программой mumu при N = 400 и разным числом процессов P показали, что при всех 1 < P < 6 работа выполняется в P раз быстрее, чем в однопроцессном варианте. При P > 5 уменьшения времени уже нет, наоборот - процессы начинают "толкаться" в памяти компьютера. Конечно, эти цифры характерны только для этой программы и только для нашего 8-процессорного SPP-2000.
Действительно, в этой программе не требуется межпроцессных коммуникаций во время счета, обмен информацией требуется только в начале и в конце работы. Поэтому, пока системе хватает ресурсов памяти, ускорение и должно линейно зависеть от числа процессов (три землекопа выкопают ту же самую яму втрое быстрее, чем один. Если же все землекопы сразу в одну яму не поместятся - они будут только мешать друг другу!).

Как пользоваться MPI на LINUX-кластере?

О переменных PATH и MANPATH можно не заботиться: та же сладкая парочка компиляторов mpif77 и mpicc доступна на головных машинах кластера lxpub01-lxpub04 без дополнительных указаний. Запуск задачи на счет осуществляется через систему пакетной обработки Portable Batch System (сокращенно - PBS) с помощью команды qsub:
qsub s
где s - имя заранее заготовленного Вами файла, содержащего "паспорт задачи". Вот образчик нашего собственного паспорта задачи, в котором из превеликого множества возможных PBS-директив фигурируют только самые необходимые:
#!/bin/sh
#PBS -q para
#PBS -N test
 *NAME test
#PBS -l walltime=00:01:00,nodes=3:para
 *TIME:00.01
#PBS -j oe
 *FULL LIST
mpiexec $PBS_O_WORKDIR/a.out
 *EXECUTE

Справа мы нарисовали аналогичные директивы паспорта задачи, принятые в былинные времена на ЭВМ БЭСМ-6 в мониторной системе "Дубна". Полное описание команды qsub и всех PBS-директив Вы получите по приказу
man qsub
Здесь исполняемый файл a.out, изготовленный компилятором mpif77 или mpicc, отправляется во входную очередь (para - это ее имя) задач, предназначенных для счета на myrc - подмножестве нашего кластера. Задаче присваивается имя test. Она требует 1 минуту времени и 3 процессора (точнее - она распараллелена на 3 процесса). Задача получит уникальный номер во входной очереди. Пусть это будет 1234. Тогда листинг задачи будет оформлен как файл test.o1234 в той же Вашей директории, где находился ее исполняемый файл a.out.

Литература

  1. http://parallel.ru
  2. http://www.openmp.org
  3. Воеводин В.В., Воеводин Вл.В. Параллельные вычисления. БХВ-Петербург, 2002.
  4. MPI: The complete Reference. MIT Press, Cambridge, Massachusetts, 1997.

10.02.2004
А.П.Сапожников