Аналог глобальных переменных в MPI?

Question

[okazymyrov]

Как можно организовать что-то на подобе глобальной переменной в MPI? Интересует любой вариант (файловая система, межпроцессорное взоимодействие...). Должно быть расширяемым. Тестирование на 4-х процессорах, будет запускаться на 4096-ти.

Заготовка программы:

#include <cstdio>
#include <vector>
#include <mpi.h>

#include "tools.h"

using namespace std;
unsigned long long d = 0;

unsigned long long get_new_d(int ccpu)
{
	d++;

	return d;
}

int main(int argc, char *argv[])
{
    int ccpu, nprocs;

    MPI_Init(&argc, &argv);
    MPI_Comm_size(MPI_COMM_WORLD, &nprocs);
    MPI_Comm_rank(MPI_COMM_WORLD, &ccpu);

    printf("%d: %lld\n",ccpu,get_new_d(ccpu));

    MPI_Finalize();
    return 0;
}

Запуск:

#!/bin/bash

N_PROCESSORS=4

mpiexec -n ${N_PROCESSORS} bent_mpi

Что должно быть? Функция get_new_d должна возвращать каждый раз значение d на 1 больше, чем максимальное d среди всех процессов. Что-то типа:

3: 1
0: 2
1: 3
2: 4

Кто подскажет (а лучше покажет) что-то типа семафора (мьютекса) в MPI тоже буду рад.

Answer 1

[Максим Пелевин]

Первое, что пришло в голову, — выделить процесс (например 0), который меняет и возвращает значение переменной, остальные процессы будут запрашивать значение и изменение у 0-го процесса. Поскольку процессы независимые, то и семафор не нужен. Сам процесс, понятное дело, сам у себя тоже может запрашивать значение переменной.

Comments

[okazymyrov]

1. Как организовать для 4096 процессов? На сколько я понял, то MPI работает P2P. Как узнать в 0-м процессе от кого пришёл запрос (т.е. кому отправлять ответ).
2. Что будет если 2 процесса в одно и тоже время запросят значение? В примере выше это и происходит.

[Максим Пелевин]

Когда-то я писал простой пример на C++ для вычисления числа Pi:

#include "mpi.h"
#include <cstdio>
#include <cstdlib>
#include <cmath>

int main(int argc, char *argv[]) {
	
	long double acc = 0.0;
	long double pi = 0.0;
	long double real_pi = 3.1415926535897932384626433832795028841971693993751058209749;
	
	MPI_Init(&argc, &argv);

	int rank, size;
	const int MASTER = 0, TAG = 0;

  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  MPI_Comm_size(MPI_COMM_WORLD, &size);
	
	__int64 nIter = 10000000000 / size;
	long double start = MPI_Wtime();
	for (__int64 k = rank * nIter; k < nIter * (1 + rank); k++) {
		acc += 4 * (k % 2 == 0 ? 1.0 : -1.0) / (2.0 * k + 1);
	}
	printf("Calculated part = %.16f in %d process\n", acc, rank);

  if (rank == 0) {
		for (int i = 1; i < size; i++) {
			MPI_Status status;
			long double tmp = 0.0;
			MPI_Recv(&tmp, 1, MPI_LONG_DOUBLE, i, TAG, MPI_COMM_WORLD, &status);
			acc += tmp;
		}
		pi = acc;
		long double end = MPI_Wtime();
		printf("Calculated pi = %0.16f with %i processes in %f\n", pi, size, end - start);
	} else {
		MPI_Send(&acc, 1, MPI_LONG_DOUBLE, MASTER, TAG, MPI_COMM_WORLD);
	}

  MPI_Finalize();
	return EXIT_SUCCESS;
}

Функция MPI_Recv получает переменную структуры status, в которой есть поле MPI_Source — источник сообщения, ему и нужно отвечать. Сообщения не могут прийти одновременно — кто-то прийдёт раньше, другой процесс будет ждать, пока главный процесс (0-ой) снова вернёться к функции MPI_Recv. Вместо указания на процесс i, указывается тег MPI_ANY_SOURCE

[okazymyrov]

Спасибо за совет помогло. Только ещё вопрос, можно ли как-то узнать что просесса больше не существует?

#include <cstdio>
#include <vector>
#include <mpi.h>

#include "tools.h"

using namespace std;
unsigned long long d = 1;

unsigned long long get_new_d(int ccpu)
{
	MPI_Status status;

    //printf("%d: Start sender\n",ccpu);
    if ( MPI_SUCCESS != MPI_Sendrecv(NULL,0,MPI_UNSIGNED_LONG_LONG,0,0,&d,1,MPI_UNSIGNED_LONG_LONG,0,MPI_ANY_TAG,MPI_COMM_WORLD,&status))
    	return 0;

	return d;
}

int server()
{
	MPI_Status status;

    for(; d < 10; d++) {
    	MPI_Recv(NULL, 0, MPI_UNSIGNED_LONG_LONG, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
        MPI_Send(&d,1,MPI_UNSIGNED_LONG_LONG,status.MPI_SOURCE,0,MPI_COMM_WORLD);
    }

	return 0;
}

int main(int argc, char *argv[])
{
    int ccpu, nprocs;

    MPI_Init(&argc, &argv);
    MPI_Comm_size(MPI_COMM_WORLD, &nprocs);
    MPI_Comm_rank(MPI_COMM_WORLD, &ccpu);

    if(ccpu == 0)
    {
    	server();
    	printf("end server\n");
    }
    else
    {
    	for(;;)
    	{
    		d=get_new_d(ccpu);
    		if(d == 0)
    			break;
    		printf("%d: %lld\n",ccpu,d);
    	}
    }

    MPI_Finalize();
    return 0;
}

Сейчас у меня deadlock. Сервер (0-й процесс) заканчивает заботать, а все остальные отправляют сообщения и ждут ответа…

[Владислав]

okazymyrov, тебе нужно определиться, когда сервер должен завершать свою работу и на уровне алгоритма сделать так, что-бы пока остальные процессы не разошлют информацию о своем завершении сервер продолжит работать. Например, перед завершением каждый процесс посылает серверу специальное значение и когда число полученных уведомлений станет равным числу остальных процессов, то сервер завершиться.

[okazymyrov]

Сейчас работает (вроде) так:

int server()
{
	unsigned long long start = , end = ; // 1..59049
	MPI_Status status;
	int i=0;
	ofstream output;
	char file_name[FILENAME_MAX] = {};

	sprintf(file_name,"./d.txt");

    for(d=start; d < end; d++) {
		init_output(file_name,&output);
		output << d << endl;
		output.flush();
		close_output(&output);
    	MPI_Recv(NULL, 0, MPI_UNSIGNED_LONG_LONG, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
        MPI_Send(&d,1,MPI_UNSIGNED_LONG_LONG,status.MPI_SOURCE,0,MPI_COMM_WORLD);
    }

    d=0;
    for(i=0;i<4095;i++) {
    	MPI_Recv(NULL, 0, MPI_UNSIGNED_LONG_LONG, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
        MPI_Send(&d,1,MPI_UNSIGNED_LONG_LONG,status.MPI_SOURCE,0,MPI_COMM_WORLD);
    }

	return 0;
}

Т.е. server ждёт от всех остальных последний send и отслылает сигнал на выход (d=0).

Answer 2

[Владислав]

okazymyrov, у тебя довольно странная задача.
Скажи пожалуйста, для чего и на сколько часто ты планируешь использовать этот глобальный счетчик?
По сути ты пытаешься в параллельную систему добавить строго последовательный участок, что может привести к значительной деградации производительности и чем больше ядер — тем больше деградация. 4096 ядер уже достаточно большая величина, чтобы начать следить за потерями производительности из-за синхронизаций.

Напиши задачу, которую ты решаешь, возможно к ней можно придумать более эффективное решение, чем глобальный счетчик.

Comments

[okazymyrov]

Если сильно упростить, то есть цикл, который считает «что-то» от d=1 до d=3^10-1. Длительность цикла на одном процессоре варьируется от 1 мин до 3 ч. Нужно распаралелить цикл средствами MPI с минимальными затратами CPU/ч. Т.е. я могу запросить 4096 процессоров на t часов (до 20) и нужно чтобы они по максимому использовались. Процедуру с «запросом» можно повторить.

[Владислав]

Итерации цикла для разных d независимы?
Если да, то все решается гораздо проще. Если есть N процессов, то процессы перебирают d с шагом N.
Например (N=3):
Пр.1: 1,4,7,10,…
Пр.2: 2,5,8,11,…
Пр.3: 3,6,9,12,…

Если итерации зависимы, те итерация d=k зависит от результатов итерации d=k-p, то распараллеливать по d бессмысленно. нужно распараллеливать итерацию, а d менять во всех процессах синхронно (+ барьерная синхронизация в начале каждой итерации по d).

[okazymyrov]

Спасибо за совет, подсказка действительно дельная. Однако не подходит для моего случая, так как:
а) для одного d длительность подсчёта варьируется от 1 мин до 3 ч, т.е. после определённого времени некоторые процессоры будут отдыхать, в то время, как остальные работать;
б) можно попасть на шаг, где все d перебируться за пару минут, т.е. несколько процессоров будут работать пару минут, остальные, например, 12 часов;
в) на кластере процессоры не однородные, есть очень старые, а есть совсем новые.

В принципе общя проблема описана в а). И ещё есть проблема со временем, т.е. я могу запросить 3, 6, 12 часов. Но не факт, что за это время все посчитается, поэтому мне нужно сохранять промежутучные значения, а после их вводить в программу. Я хочу автоматизировать данный процесс.

[Владислав]

В таком случае согласен, что можно выделить некоторый task-сервер (нулевой процесс, например), который будет раздавать задачи (номера d) процессам.

Кстати тогда лучше сделать не так, как выше я предлагал логику завершения, а наоборот. Не вычислительные процессы уведомляют процесс диспетчера о своем завершении, а диспетчер отдает процессам команды завершения, когда значения d кончатся, после чего сам завершается.
А если задача будет принудительна снята со счета, то первый d можно будет определять по ранее посчитанным и сохраненным данным (необходимо учесть, что задача может быть снята когда значения для больших d посчитаны, а для меньших еще нет, по этому необходимо учитывать возможные дыры в нумерации при перезапуске).

[okazymyrov]

Сейчас так приблизительно и делается. d=0 означает завершение процесса. Т.е. слиенты, получив это значения завершаются. Посчитаенные d записываются в файл, а после парсером выписываю только те значения, которые ещё не были посчитаны. И повторяю процедуру для оставшихся d. Таким образом загружаю по максимому все процессоры.

Answer 3

[icc]

В MPI для таких целей лучше использовать MPI_Bcast. Если же нужны именно глобальные переменные в чистом виде, примените связку OpenMP + MPI.

Comments

[okazymyrov]

MPI_Bcast не подходит без семафора. Если ли пример со связкой OpenMP + MPI?

[icc]

Естественно, MPI_Bcast — блокирующая функция. Если нужен неблокирующий вариант, тогда вызывайте в цикле пачку MPI_Irecv.

Простейший пример OpenMP + MPI: www.slac.stanford.edu/comp/unix/farm/mpi_and_openmp.html
Но, тут многое зависит от архитектуры кластера.

[okazymyrov]

Проблема в том, что если использовать MPI_Bcast, то в каждом процессе нужно выполнять d+=1 перед MPI_Bcast. Так вот, если 2 или более процесса пришли в точку d+=1 одновременно, то не важно MPI_Bcast блокирующая или нет, в обоих процессах d будет инкрементировано на единицу, а только после разосланно. Т.е. в каждом процессе будет предыдущее d плюс 1, а нужно чтобы было предыдущее d от максимального d из всех процессов плюс 1.

Как использовать OpenMP + MPI я знаю, проблема в том, что каждый процесс независимый, только в рамках одного процесса можно организовать глобальную переменную, использую OpenMP. Или я чего-то не понимаю?