Programación paso de mensajes.

MPI

Es un sistema estandarizado y portable de paso de mensajes, diseñado para funcionar en una amplia variedad de computadoras paralelas. El estándar define la sintaxis y semántica de un conjunto de funciones de librería que permiten a los usuarios escribir programas portables en los principales lenguajes utilizados por la comunidad científica (Fortran, C, C++).

Desde su aparición, la especificación MPI se ha transformado en el estándar dominante en librerías de paso de mensajes para computadoras paralelas. En la actualidad se dispone de diversas implementaciones, algunas provistas por los fabricantes de computadoras de alta performance, hasta otras de reconocidos proyectos open source, tales como MPICH y LAM/MPI, muy utilizadas en los clusters de PC’s tipo Beowulf.

Comunicación básica

El resultado de la combinación de dos modelos y cuatro modos de comunicación nos da 8 diferentes funciones de envío. Funciones de recepción sólo hay dos, una por modelo. Presentamos a continuación los prototipos de las funciones más habituales. Empezamos con MPI_Send() y MPI_Recv que son, respectivamente, las funciones de envío y recepción básicas bloqueantes.

int MPI_Send(void* buf, int count, MPI_Datatype datatype,

int dest, int tag, MPI_Comm comm);

int MPI_Recv(void* buf, int count, MPI_Datatype datatype,

int source, int tag, MPI_Comm comm, MPI_Status *status);

El significado de los pará metros es como sigue.

Buf, count y datatype forman el mensaje a enviar o recibir: count copias de un dato del tipo datatype que se encuentran (o se van a dejar) en memoria a partir de la dirección indicada por buf. Destes, en las funciones de envío, el identificador del proceso destinatario del mensaje. Source es, en las funciones de recepción, el identificador del emisor del cual esperamos un mensaje. Si nonos importa el origen del mensaje, podemos poner MPI_ANY_SOURCE. Tag es una etiqueta que se puede poner al mensaje. El significado de la etiqueta lo asigna el programador. Suele emplearse para distinguir entre diferentes clases de mensajes. El emisor pone siempre una etiqueta a los mensajes, y el receptor puede elegir entre recibir sólo los mensajes que tengan una etiqueta dada, o aceptar cualquier etiqueta, poniendo MPI_ANY_TAG como valor de este parámetro. Comm es un comunicador; en muchas ocasiones se emplea el comunicador universal MPI_COMM_WORLD. Status es un resultado que se obtiene cada vez que se completa una recepción, y nos informa de aspectos tales como el tamaño del mensaje recibido, la etiqueta del mensaje y el emisor del mismo.

Barreras y broadcast

Dos de las operaciones colectivas más comunes son las barreras de sincronización (MPI_Barrier()) y el envío de información en modo difusión

(MPI_Broadcast()). La primera de estas operaciones no exige ninguna clase de intercambio de información. Es una operación puramente de sincronización, que bloquea a los procesos de un comunicador hasta que todos ellos han pasado por la barrera. Suele emplearse para dar por finalizada una etapa del programa, asegurándose de que todos han terminado antes de dar comienzo a la siguiente.

int MPI_Barrier(MPI_Comm comm);

MPI_Broadcast() sirve para que un proceso, el raíz, envíe un mensaje a todos los miembros del comunicador. Esta función ya muestra una característica peculiar de las operaciones colectivas de MPI: todos los participantes invocan la misma función, designando al mismo proceso como raíz de la misma. Una implementación alternativa sería tener una función de envío especial, en modo broadcast, y usar las funciones de recepción normales para leer los mensajes.

int MPI_Bcast(void* buffer, int count, MPI_Datatype

datatype, int root, MPI_Comm comm);

Este esquema representa el significado de un broadcast. En las filas representamos los procesos de un comunicador, y en las columnas los datos que posee cada proceso. Antes del broadcast, el procesador raíz (suponemos que es el 0) tiene el dato A0. Tras realizar el broadcast, todos los procesos (incluyendo la raíz) tienen una copia de A0.

Recolección (gather)

MPI_Gather realiza una recolección de datos en el proceso raíz. Este proceso recopila un vector de datos, al que contribuyen todos los procesos del comunicador con la misma cantidad de datos. El proceso raíz almacena las contribuciones de forma consecutiva.

int MPI_Gather(void* sendbuf, int sendcount, MPI_Datatype

sendtype, void* recvbuf, int recvcount, MPI_Datatype

recv

type, int root, MPI_Comm comm);

Sólo el proceso raíz tiene que preocuparse de los parámetros

recbuf, recvcount y recvtype. Sin embargo, todos los procesos (raíz incluido) tienen que facilitar valores válidos para sendbuf, sendcount y sendtype.

Existe una versión de la función de recolección, llamada MPI_Gatherv(), que permite almacenar los datos recogidos en forma no consecutiva, y que cada proceso contribuya con bloques de datos de diferente tamaño. La tabla recvcounts establece el tamaño del bloque de datos aportado por cada proceso, y displs indica cuál es el desplazamiento entre bloque y bloque

int MPI_Gatherv(void* sendbuf, int sendcount,

MPI_Datatype sendtype, void* recvbuf, int *recvcounts,

int *displs, MPI_Datatype recvtype, int root, MPI_Comm

comm);

Distribución (scatter)

MPI_Scatter() realiza la operación simétrica a MPI_Gather(). El proceso raíz posee un vector de elementos, uno por cada proceso del comunicador. Tras realizar la distribución, cada proceso tiene una copia del vector inicial. Recordemos que MPI permite enviar bloques de datos, no sólo datos individuales.

int MPI_Scatter(void* sendbuf, int sendcount,

MPI_Datatype sendtype, void* recvbuf, int recvcount,

MPI_Datatype recvtype, int root, MPI_Comm comm);

Si los datos a enviar no están almacenados de forma consecutiva en memoria, o los bloques a enviar a cada proceso no son todos del mismo tamaño, tenemos la función MPI_Scatterv(), con un interfaz más complejo pero más flexible.

int MPI_Scatterv(void* sendbuf, int *sendcounts, int

*displs, MPI_Datatype sendtype, void* recvbuf, int

recvcount, MPI_Datatype recvtype, int root, MPI_Comm

comm);

Reducción

Una reducción es una operación realizada de forma cooperativa entre todos los procesos de un comunicador, de tal forma que se obtiene un resultado final que se almacena en el proceso raíz.

int MPI_Reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm

comm);

MPI define una colección de operaciones del tipo MPI_Op que se pueden utilizar en una reducción: MPI_MAX (máximo), MPI_MIN (mínimo), MPI_SUM (suma), MPI_PROD (producto), MPI_LAND (and lógico), MPI_BAND (and bit a bit), MPI_LOR (or lógico), MPI_BOR (or bit a bit), MPI_LXOR (xor lógico), MPI_BXOR (xor bit a bit), etc.

En ocasiones el programador puede necesitar otra operación de reducción distinta, no predefinida. Para ello MPI ofrece la función MPI_Op_create(), que toma como argumento de entrada una función de usuario y devuelve un objeto de tipo MPI_Op que se puede emplear con MPI_Reduce(). Ese objeto se puede destruir más adelante con MPI_Op_free().

int MPI_Op_create(MPI_User_function *function, int commute, MPI_Op *op );

int MPI_Op_free(MPI_Op *op);

(a)

Startup

(LAM/MPI)

$ lamboot nodes.dat

$ mpirun -np 3 ppython

>>> import mpi

(b)

Startup

(MPICH)

$ mpirun -machinefile nodes.dat -np 3 ppython

>>> import mpi

>>> if mpi.rank == 0:

... sendbuf = { ’op1’: True, \

... ’op2’: 2.52, \

... ’op3’: ’yes’ }

(c) MPI

BCAST

... else:

... sendbuf = None

...

>>> print ’[%d]’ % mpi.rank, sendbuf

[0] {’op1’: True, ’op3’: ’yes’, ’op2’: 2.52}

[2] None

[1] None

>>> 

>>> recvbuf = mpi.WORLD[0].Bcast(sendbuf)

>>> print ’[%d]’ % mpi.rank, recvbuf

[0] {’op1’: True, ’op3’: ’yes’, ’op2’: 2.52}

[2] {’op1’: True, ’op3’: ’yes’, ’op2’: 2.52}

[1] {’op1’: True, ’op3’: ’yes’, ’op2’: 2.52}

(d) MPI

SCATTER

>>> root = mpi.size/2

>>> 

>>> sendbuf = None

>>> if mpi.rank = root:

... sendbuf = [ (i,i**2,i**3) \

... for i in [2,3,4] ]

>>> print ’[%d] %s’ % (mpi.rank, sendbuf)

[0] None

[1] [(2, 4, 8), (3, 9, 27), (4, 16, 64)]

[2] None

>>> 

>>> recvbuf = mpi.WORLD[root].Scatter(sendbuf)

>>> print ’[%d] %s’ % (mpi.rank, recvbuf)

[0] (2, 4, 8)

[1] (3, 9, 27)

[2] (4, 16, 64)

(e) MPI

GATHER

>>> sendbuf = [mpi.rank**2 , mpi.rank%2!=0]

>>> print ’[%d] %s’ % (mpi.rank, sendbuf)

[0] [0, False]

[1] [1, True]

[2] [4, False]

>>> 

>>> root = mpi.size/2

>>> recvbuf = mpi.WORLD[root].Gather(sendbuf)

>>> print ’[%d] %s’ % (mpi.rank, recvbuf)

[0] None

[1] [[0, False], [1, True], [4, False]]

[2] None

(f) MPI

ALLTOALL

>>> sendbuf = []

>>> for i in xrange(mpi.size):

... sendbuf += [(mpi.size+mpi.rank)*100]

...

>>> print "[%d] %s" % (mpi.rank, sendbuf)

[0] [300, 300, 300]

[1] [400, 400, 400]

[2] [500, 500, 500]

>>> 

>>> recvbuf = mpi.WORLD.Alltoall(sendbuf)

>>> print "[%d] %s" % (mpi.rank, recvbuf)

[0] [300, 400, 500]

[1] [300, 400, 500]

[2] [300, 400, 500]