NumPy v MPI4Py

NumPy in MPI4Py se uporabljata skupaj in sta učinkovita predvsem zaradi tega ker je ndarray kontunuirani blok podatkov, ki se lahko posreduje med procesi učinkovito v enem kosu in ne tako kot običajno, ko se v MPI4Py pošiljajo objekti, ki se morajo v procesih pakirati/razpakirati ob prenosu. Pomembna razlika pri pošiljanji/sprejemanju je v prvi črki ukaza, ki je z veliko začetnico (Send/Recv) in ne tako kot je običaj (send/recv). Druga pomemna značilnost NumPy prenosa je v sprejemu, saj mora biti polje v katerega sprejemamo že predalocirano.

Broadcast – Emitiranje ali razpošiljanje vsem se izvaja iz procesa 0.

# -*- coding: utf-8 -*-

import numpy as np

from mpi4py import MPI



comm = MPI.COMM_WORLD

if comm.rank == 0:

  print "-"*78

  print " Tečemo na %d jedrih" % comm.size

  print "-"*78

comm.Barrier()



# Pripravimo vektor velikosti N=5 elements za razpošiljanje

N = 5

if comm.rank == 0:

  A = np.arange(N, dtype=np.float64); # rank 0 ima vse podatke

else:

  A = np.empty(N, dtype=np.float64); # vsi ostali pa le prazna polja



# Razpošljimo A iz procesa 0 vsem 

comm.Bcast( [A, MPI.DOUBLE] )



# Vsi bi morali sedaj imeti enako polje...

print "[%02d] %s" % (comm.rank, A)

Scatter – razpršimo problem in ga razpečamo po obdelavi z Allgather

Razdelimo večji problem na manjši tako, da pošljemo podatke razdeljeno po koščkih, potem pa vsak pridobi rezultate še od ostalih in nadaljuje z delom s tem, da ima ponovno vse rezultate.

# -*- coding: utf-8 -*-

import numpy as np

from mpi4py import MPI

 

comm = MPI.COMM_WORLD

 

if comm.rank == 0:

  print "-"*78

  print " Tečemo na %d jedrih" % comm.size

  print "-"*78

comm.Barrier()

 

my_N = 4

N = my_N * comm.size

 

if comm.rank == 0:

  A = np.arange(N, dtype=np.float64)

else:

  A = np.empty(N, dtype=np.float64)

 

my_A = np.empty(my_N, dtype=np.float64)

 

# Razdeli  in pošlji (razpošlji) podatke data v my_A polja

comm.Scatter( [A, MPI.DOUBLE], [my_A, MPI.DOUBLE] )

 

if comm.rank == 0:

    print("Po razpošiljanju:")

for r in xrange(comm.size):

  if comm.rank == r:

    print "[%d] %s" % (comm.rank, my_A)

 

comm.Barrier()

 

# Vsi sedaj množijo z 2

my_A *= 2

 

# Vsi poberejo od vseh (preostalo) nazaj v A

comm.Allgather( [my_A, MPI.DOUBLE], [A, MPI.DOUBLE] )

if comm.rank == 0:

    print("Po sestavljanju:")

for r in xrange(comm.size):

  if comm.rank == r:

    print "[%d] %s" % (comm.rank, A)

 

comm.Barrier()

Razprševanje objektov

Razprševanje in sestavljanje je uporabno zaradi manjše komunikacije, saj velikokrat ni potrebno, da se vsi podatki razpošljejo vsem ampak le nujno potrebni.

# -*- coding: utf-8 -*-

# half mpirun -np 4&nbsp; python ex4.py

from mpi4py import MPI



comm = MPI.COMM_WORLD

size = comm.Get_size()

rank = comm.Get_rank()



if rank == 0:

   data = [[1, 2], [3, 4], [5, 6], [7, 8]]

else:

   data = None



print "Prvotni podatki za proces", rank, "so",  data

razprseno = comm.scatter(data)

print "Proces", rank, "je sprejel", razprseno

pomnozeno = [i*2 for i in razprseno]

print "Proces", rank, "je pomnožil z 2 na", pomnozeno

pobrano = comm.gather(razprseno)

print "Pobrano oz oddano za proces", rank, "je", pobrano

Tag je oznaka sporočila

Pošiljanje in sprejemanje podatkov med dvema procesoma. tag je uporabniško določljiva oznaka sporočila.

# -*- coding: utf-8 -*-

from mpi4py import MPI

comm = MPI.COMM_WORLD

rank = comm.Get_rank()

if rank == 0:

    data = [1, 2, 3]

    comm.send(data, dest=1, tag=12)

    print "Poslani podatki so: ", data

elif rank == 1:

    data = comm.recv(source=0, tag=12)

    print "Sprejeti podatki so: ", data

Redukcija

Z redukcijo izračunamo Pi

# -*- coding: utf-8 -*-

from mpi4py import MPI

import math



def compute_pi(n, start, step):

    h = 1.0 / n

    s = 0.0

    for i in range(start, n, step):

        x = h * (i + 0.5)

        s += 4.0 / (1.0 + x**2)

    return s * h



comm = MPI.COMM_WORLD

nprocs = comm.Get_size()

myrank = comm.Get_rank()



if myrank == 0:

    n = 1000000

else:

    n = None



n = comm.bcast(n, root=0)

mypi = compute_pi(n, myrank, nprocs)

pi = comm.reduce(mypi, op=MPI.SUM, root=0)



if myrank == 0:

    error = abs(pi - math.pi)

    print ("pi je približno %.16f, "

           "napaka je %.16f" % (pi, error))

Mandelbrot

Mandelbrotov fraktal se ob zaključku shrani v datoteko, ki si jo ogledamo z ukazom

gimp mandelbrot.png

# -*- coding: utf-8 -*-

def mandelbrot(x, y, maxit):

    c = x + y*1j

    z = 0 + 0j

    it = 0

    while abs(z) < 2 and it < maxit:

        z = z**2 + c

        it += 1

    return it



x1, x2 = -2.0, 1.0

y1, y2 = -1.0, 1.0

w, h = 6000, 4000

maxit = 127



from mpi4py import MPI

import numpy



comm = MPI.COMM_WORLD

size = comm.Get_size()

rank = comm.Get_rank()



# število vrstic, ki jih računamo v tem procesu

N = h // size + (h % size > rank)



# prva vrstica se začne na

start = comm.scan(N)-N



# polje za shranitev lokalnih rezultatov

Cl = numpy.zeros([N, w], dtype='i')



# preračunaj dodeljene vrstice

dx = (x2 - x1) / w

dy = (y2 - y1) / h

for i in range(N):

    y = y1 + (i + start) * dy

    for j in range(w):

        x = x1 + j * dx

        Cl[i, j] = mandelbrot(x, y, maxit)



# gather results at root (process 0)

counts = comm.gather(N, root=0)

C = None

if rank == 0:

    C = numpy.zeros([h, w], dtype='i')



rowtype = MPI.INT.Create_contiguous(w)

rowtype.Commit()



comm.Gatherv(sendbuf=[Cl, MPI.INT],

             recvbuf=[C, (counts, None), rowtype],

             root=0)



rowtype.Free()



if comm.rank == 0:

    from matplotlib import pyplot

    pyplot.imshow(C, aspect='equal')

    pyplot.spectral()

    pyplot.savefig("mandelbrot.png")

Tiskalniku prijazna oblika

Main menu

High Performance Computing

Navigacija

Nahajate se tukaj