MPI (part 2)¶

In [ ]:
%%writefile mpi_master_worker.py
from mpi4py import MPI
import numpy as np

Comm = MPI.COMM_WORLD
size = Comm.Get_size()
rank = Comm.Get_rank()
pname = MPI.Get_processor_name()

def log(*s):
    print(f'LOG {rank:02d}/{pname}: ', *s)
    
log('started')

if rank == 0: 
    #MASTER
    log('I am master')
else:
    #WORKER
    log('I am worker')

log('stopped')
In [1]:
!mpiexec python mpi_master_worker.py

LOG 02/pf-cai-p03-pc0:  started
LOG 02/pf-cai-p03-pc0:  I am worker
LOG 02/pf-cai-p03-pc0:  stopped
LOG 05/pf-cai-p03-pc0:  started
LOG 05/pf-cai-p03-pc0:  I am worker
LOG 05/pf-cai-p03-pc0:  stopped
LOG 04/pf-cai-p03-pc0:  started
LOG 04/pf-cai-p03-pc0:  I am worker
LOG 04/pf-cai-p03-pc0:  stopped
LOG 03/pf-cai-p03-pc0:  started
LOG 03/pf-cai-p03-pc0:  I am worker
LOG 03/pf-cai-p03-pc0:  stopped
LOG 00/pf-cai-p03-pc0:  started
LOG 00/pf-cai-p03-pc0:  I am master
LOG 00/pf-cai-p03-pc0:  stopped
LOG 01/pf-cai-p03-pc0:  started
LOG 01/pf-cai-p03-pc0:  I am worker
LOG 01/pf-cai-p03-pc0:  stopped

We will use TAGS for the communication between master and workers

In [ ]:
%%writefile mpi_master_worker.py
from mpi4py import MPI
import numpy as np
from enum import IntEnum, auto

class PDSTag(IntEnum):
    ASK = auto()
    WORK = 3
    FINISH = auto()

Comm = MPI.COMM_WORLD
size = Comm.Get_size()
rank = Comm.Get_rank()
pname = MPI.Get_processor_name()

def log(*s):
    print(f'LOG {rank:02d}/{pname}: ', *s)
    
log('started')

if rank == 0: 
    #MASTER
    log('I am master')
    data = np.zeros(1, dtype='i8')
    n = 20
    #distribute n tasks
    for i in range(n):
        st = MPI.Status()
        #wait for ASK
        Comm.Recv(data, status=st, tag=PDSTag.ASK)
        #send task
        data[0] = i
        log(f'assign work {i} to worker {st.Get_source()}')
        Comm.Send(data, dest=st.Get_source(), tag=PDSTag.WORK)
    #send FINISH to ALL
    ....
else:
    #WORKER
    log('I am worker')
    data = np.zeros(1, dtype='i8')
    while True:
        #send ASK
        Comm.Send(data, dest=0, tag=PDSTag.ASK)
        #receive WORK/FINISH
        st = MPI.Status()
        Comm.Recv(data, source=0, status=st)
        if st.Get_tag() == PDSTag.FINISH:
            log('received FINISH')
            break
        log(f'received WORK {data[0]}')
        

log('stopped')

Send to all:

In [ ]:
#send to all
for r in range(1, size):
    Comm.Send(data, dest=r, tag=PDSTag.FINISH)

First wait for ASK....

In [ ]:
#send to all
for r in range(1, size):
    Comm.Recv(data, source=r, tag=PDSTag.ASK)
    Comm.Send(data, dest=r, tag=PDSTag.FINISH)

This implies sequential FINISH ...

In [ ]:
#send to all
for r in range(size-1):
    Comm.Recv(data, status=st, tag=PDSTag.ASK)
    Comm.Send(data, dest=st.Get_source(), tag=PDSTag.FINISH)
In [5]:
%%writefile mpi_master_worker.py
from mpi4py import MPI
import numpy as np
from enum import IntEnum, auto

class PDSTag(IntEnum):
    ASK = auto()
    WORK = 3
    FINISH = auto()

Comm = MPI.COMM_WORLD
size = Comm.Get_size()
rank = Comm.Get_rank()
pname = MPI.Get_processor_name()

def log(*s):
    print(f'LOG {rank:02d}/{pname}: ', *s)
    
log('started')

if rank == 0: 
    #MASTER
    log('I am master')
    data = np.zeros(1, dtype='i8')
    n = 20
    #distribute n tasks
    for i in range(n):
        st = MPI.Status()
        #wait for ASK
        Comm.Recv(data, status=st, tag=PDSTag.ASK)
        #send task
        data[0] = i
        log(f'assign work {i} to worker {st.Get_source()}')
        Comm.Send(data, dest=st.Get_source(), tag=PDSTag.WORK)
    #send FINISH to ALL
    for r in range(size-1):
        Comm.Recv(data, status=st, tag=PDSTag.ASK)
        Comm.Send(data, dest=st.Get_source(), tag=PDSTag.FINISH)
else:
    #WORKER
    log('I am worker')
    data = np.zeros(1, dtype='i8')
    while True:
        #send ASK
        Comm.Send(data, dest=0, tag=PDSTag.ASK)
        #receive WORK/FINISH
        st = MPI.Status()
        Comm.Recv(data, source=0, status=st)
        if st.Get_tag() == PDSTag.FINISH:
            log('received FINISH')
            break
        log(f'received WORK {data[0]}')
        

log('stopped')

Overwriting mpi_master_worker.py
In [6]:
!mpiexec python mpi_master_worker.py

LOG 01/pf-cai-p03-pc0:  started
LOG 01/pf-cai-p03-pc0:  I am worker
LOG 01/pf-cai-p03-pc0:  received WORK 2
LOG 01/pf-cai-p03-pc0:  received WORK 7
LOG 01/pf-cai-p03-pc0:  received WORK 11
LOG 01/pf-cai-p03-pc0:  received WORK 14
LOG 01/pf-cai-p03-pc0:  received WORK 16
LOG 01/pf-cai-p03-pc0:  received FINISH
LOG 01/pf-cai-p03-pc0:  stopped
LOG 03/pf-cai-p03-pc0:  started
LOG 03/pf-cai-p03-pc0:  I am worker
LOG 03/pf-cai-p03-pc0:  received WORK 5
LOG 03/pf-cai-p03-pc0:  received WORK 18
LOG 03/pf-cai-p03-pc0:  received FINISH
LOG 03/pf-cai-p03-pc0:  stopped
LOG 05/pf-cai-p03-pc0:  started
LOG 05/pf-cai-p03-pc0:  I am worker
LOG 05/pf-cai-p03-pc0:  received WORK 0
LOG 05/pf-cai-p03-pc0:  received WORK 4
LOG 05/pf-cai-p03-pc0:  received WORK 10
LOG 05/pf-cai-p03-pc0:  received WORK 17
LOG 05/pf-cai-p03-pc0:  received FINISH
LOG 05/pf-cai-p03-pc0:  stopped
LOG 00/pf-cai-p03-pc0:  started
LOG 00/pf-cai-p03-pc0:  I am master
LOG 00/pf-cai-p03-pc0:  assign work 0 to worker 5
LOG 00/pf-cai-p03-pc0:  assign work 1 to worker 2
LOG 00/pf-cai-p03-pc0:  assign work 2 to worker 1
LOG 00/pf-cai-p03-pc0:  assign work 3 to worker 4
LOG 00/pf-cai-p03-pc0:  assign work 4 to worker 5
LOG 00/pf-cai-p03-pc0:  assign work 5 to worker 3
LOG 00/pf-cai-p03-pc0:  assign work 6 to worker 2
LOG 00/pf-cai-p03-pc0:  assign work 7 to worker 1
LOG 00/pf-cai-p03-pc0:  assign work 8 to worker 4
LOG 00/pf-cai-p03-pc0:  assign work 9 to worker 2
LOG 00/pf-cai-p03-pc0:  assign work 10 to worker 5
LOG 00/pf-cai-p03-pc0:  assign work 11 to worker 1
LOG 00/pf-cai-p03-pc0:  assign work 12 to worker 4
LOG 00/pf-cai-p03-pc0:  assign work 13 to worker 2
LOG 00/pf-cai-p03-pc0:  assign work 14 to worker 1
LOG 00/pf-cai-p03-pc0:  assign work 15 to worker 2
LOG 00/pf-cai-p03-pc0:  assign work 16 to worker 1
LOG 00/pf-cai-p03-pc0:  assign work 17 to worker 5
LOG 00/pf-cai-p03-pc0:  assign work 18 to worker 3
LOG 00/pf-cai-p03-pc0:  assign work 19 to worker 4
LOG 00/pf-cai-p03-pc0:  stopped
LOG 04/pf-cai-p03-pc0:  started
LOG 04/pf-cai-p03-pc0:  I am worker
LOG 04/pf-cai-p03-pc0:  received WORK 3
LOG 04/pf-cai-p03-pc0:  received WORK 8
LOG 04/pf-cai-p03-pc0:  received WORK 12
LOG 04/pf-cai-p03-pc0:  received WORK 19
LOG 04/pf-cai-p03-pc0:  received FINISH
LOG 04/pf-cai-p03-pc0:  stopped
LOG 02/pf-cai-p03-pc0:  started
LOG 02/pf-cai-p03-pc0:  I am worker
LOG 02/pf-cai-p03-pc0:  received WORK 1
LOG 02/pf-cai-p03-pc0:  received WORK 6
LOG 02/pf-cai-p03-pc0:  received WORK 9
LOG 02/pf-cai-p03-pc0:  received WORK 13
LOG 02/pf-cai-p03-pc0:  received WORK 15
LOG 02/pf-cai-p03-pc0:  received FINISH
LOG 02/pf-cai-p03-pc0:  stopped

Generating¶

In [9]:
n, k = 4, 3

arr = [0]*n

def generate(level): #we have level numbers, generating arr[level]
    if level == n:
        print(arr)
        return
    for elem in range(k):
        arr[level] = elem
        generate(level+1)
        
generate(0)        

[0, 0, 0, 0]
[0, 0, 0, 1]
[0, 0, 0, 2]
[0, 0, 1, 0]
[0, 0, 1, 1]
[0, 0, 1, 2]
[0, 0, 2, 0]
[0, 0, 2, 1]
[0, 0, 2, 2]
[0, 1, 0, 0]
[0, 1, 0, 1]
[0, 1, 0, 2]
[0, 1, 1, 0]
[0, 1, 1, 1]
[0, 1, 1, 2]
[0, 1, 2, 0]
[0, 1, 2, 1]
[0, 1, 2, 2]
[0, 2, 0, 0]
[0, 2, 0, 1]
[0, 2, 0, 2]
[0, 2, 1, 0]
[0, 2, 1, 1]
[0, 2, 1, 2]
[0, 2, 2, 0]
[0, 2, 2, 1]
[0, 2, 2, 2]
[1, 0, 0, 0]
[1, 0, 0, 1]
[1, 0, 0, 2]
[1, 0, 1, 0]
[1, 0, 1, 1]
[1, 0, 1, 2]
[1, 0, 2, 0]
[1, 0, 2, 1]
[1, 0, 2, 2]
[1, 1, 0, 0]
[1, 1, 0, 1]
[1, 1, 0, 2]
[1, 1, 1, 0]
[1, 1, 1, 1]
[1, 1, 1, 2]
[1, 1, 2, 0]
[1, 1, 2, 1]
[1, 1, 2, 2]
[1, 2, 0, 0]
[1, 2, 0, 1]
[1, 2, 0, 2]
[1, 2, 1, 0]
[1, 2, 1, 1]
[1, 2, 1, 2]
[1, 2, 2, 0]
[1, 2, 2, 1]
[1, 2, 2, 2]
[2, 0, 0, 0]
[2, 0, 0, 1]
[2, 0, 0, 2]
[2, 0, 1, 0]
[2, 0, 1, 1]
[2, 0, 1, 2]
[2, 0, 2, 0]
[2, 0, 2, 1]
[2, 0, 2, 2]
[2, 1, 0, 0]
[2, 1, 0, 1]
[2, 1, 0, 2]
[2, 1, 1, 0]
[2, 1, 1, 1]
[2, 1, 1, 2]
[2, 1, 2, 0]
[2, 1, 2, 1]
[2, 1, 2, 2]
[2, 2, 0, 0]
[2, 2, 0, 1]
[2, 2, 0, 2]
[2, 2, 1, 0]
[2, 2, 1, 1]
[2, 2, 1, 2]
[2, 2, 2, 0]
[2, 2, 2, 1]
[2, 2, 2, 2]

Task KNAPSACK¶

  • master chooses number of elements N (15..25), capacity C (10..1000), random weights W (1..500) and prices P (1..10000) for them

  • [BROADCAST] send 4 messages (N)+(C)+(W)+(P) or 2 messages (N, C)+(W, P), resp. 1 message (N, C, W, P) if workers create array big enough

  • master generates prefixes (e.g. first 10 bits) and send them to workers

In [11]:
from random import randrange

#20 random integers (from 10 to 99)
print(*[randrange(10, 100) for _ in range(20)])

32 37 99 11 90 86 94 27 99 63 44 59 17 93 91 56 95 28 42 60
In [ ]: