Day 101

Day 101: a note on PMI communications

There are a few ways for the Controller to communicate with the Workers. In most cases I'm using pmi.invoke. Invoke call takes function name to be executed on workers (as a string value), a list of arguments to be passed to workers and returns a list of results returned from workers (if any). For example, let's say we have a function "get_data" of "SomeClass" on workers that takes "data" as an argument and returns "modified_data". First we need to initialise SomeClass on all workers and get reference to it. pmi.create does the job. Let's say the reference is stored in "pmiobj" object. Now, calling

pmi.invoke(pmiobj,"get_data",data)

Would return (assume we have 4 nodes, and all run the same code) a list:

[modified_data,modified_data,modified_data,modified_data]

How that works: pmi.invoke simply passes data to the worker on node 0, which saves it (by reference!) and broadcasts to all other nodes, which involves pickling it on node 0 and unpickling on receiving by other nodes. If this data object does not implement Single Segment Buffer Interface (like numpy arrays, for instance) it sent, generally, 10 times slower than if it does. Ok, nothing new, I guess

However, the communication can be organised in different, slightly lower-level, fashion. Let's say, now instead of passing data and processing it on the nodes, we want workers to store the reference to "some_object" and then use data stored in that object later in some computations. This could be done by issuing pmi.invoke again. However, what if data in "some_object" gets modified on controller? All workers apart from the one on node 0 would have out-dated (which in many situations would mean useless) copy of "some_object", and would need to somehow get a new one, e.g. from the worker on node 0. In order to avoid redundant pickling/unpickling and network overheads we can use pmi localcall (or was it local_call?), that just passes data to the Worker on node 0. Node 0 can now store the reference and when needed by all nodes, broadcast data from the object using MPI bcast. This would ensure that all nodes will work with up to date copy of the data. Simple trick, but very useful, especially since sometimes pickling objects can have unpredictable results,(like for example, in my case, bloody dest..... spent hours debugging this!).

Day 101

Day 101

A side note on PMI terminology I'm using: "Serial" can be also referred to as "the Controller" as well a "Parallel" or "Nodes" can be referred to as "Workers". Tonight while debugging my code I had to, for the first time, have a look at PMI source code. That actually helped... Ok, sleep now. I have to take care of other uni work, so probably I'll resume the Project not sooner than Friday. I really wish I will finish it in one go then...

Day 100

Day 100: PMI. Interesting observation #2

In this example I will demonstrate how changing numpy printoptions threshold affects the communication speed between Serial and Parallel. Honestly, I have no idea what printoptions threshold has to do with MPI or PMI communication speed and I am really tempted to investigate this issue further, but sadly don't have time for this at the moment. Anyway, output first (using mpirun with 2 nodes), then source code:

numpy_printoptions threshold set to 0
sending 200x200 numpy matrix with random elements
Sending x 10 times took 1.40654301643
numpy_printoptions threshold set to 201*201 (greater than the size of communicated matrix)
sending 200x200 numpy matrix with random elements
Sending x 10 times took 12.252920866

The numbers speak for themselves. The source code:

if __name__ != 'pmi':
    ##################################################
    ## Serial code
    ##################################################
    import pmi
    from time import time
    from numpy.random import beta
    from numpy import array, set_printoptions
    
    
    pmi.setup()
    
    pmi.execfile_(__file__)
    
    
    x = beta(a=1,b=1000,size=(200,200))

    # create a frontend class
    class Hello(object):
        def __init__(self, name):
            self.pmiobj = pmi.create('HelloLocal', name)
        def __call__(self):
            return pmi.invoke(self.pmiobj, '__call__')
        def time_send(self):
            t0 = time()
            for i in range(0,10):
                pmi.invoke(self.pmiobj,'recv',x)
            print "Sending x 10 times took",(time() - t0)
        def time_recv(self):
            t0 = time()
            for i in range(0,10):
                pmi.invoke(self.pmiobj,'send')
            print "Receiving x 10 times took",(time() - t0)

    # use the class
    hello = Hello('Konstantin')
    
    print "numpy_printoptions threshold set to 0"
    set_printoptions(threshold=0)
    print "sending 200x200 numpy matrix with random elements"
    hello.time_send()
    set_printoptions(threshold=201*201)
    print "numpy_printoptions threshold set to 201*201 (greater than the size of communicated matrix)"
    print "sending 200x200 numpy matrix with random elements"
    hello.time_send()
    
else:
    ##################################################
    ## Parallel code
    ##################################################
    from mpi4py import MPI
    from numpy.random import beta
    
    y = beta(a=1,b=1000,size=(100,100))

    class HelloLocal(object):
        def __init__(self, name):
            self.name = name
        def __call__(self):
            return 'Hello %s, this is MPI task %d!' % (self.name, MPI.COMM_WORLD.rank)
        def recv(self,x):
            "do nothing"
            
        def send(self):
            return y

Day 100

Day 100: PMI. Interesting observation #1

In this example I will be testing if PMI passes objects from Serial into Parallel by reference or by value. First I will give the output of my little test script, and then the source code.

############ TEST 1: PASSING FROM SERIAL INTO PARALLEL ############
Testing PMI pass by reference: an array is created in Serial and then passed into Parallel
array before mpi.invoke: [1 2 3]
If the array was passed by _value_ then after modifying it in Parallel it should remain
unchanged in Serial
array after mpi.invoke: [1000    2    3]
As you can see, the array has been changed in serial as well, which proves that objects
are passed by reference in pmi.

############ TEST 2: PASSING FROM PARALLEL INTO SERIAL ############
Now we'll check if modifying the same array in Serial will make it change in Parallel
The array: [1000    2    3]
Changed the first element to be 999: [999   2   3]
Now calling pmi.invoke that will make Parallel print out the array (it has been stored
as a class atribute during the previous experiment)
Node 0 array: [999   2   3]
Node 1 array: [1000    2    3]
Oops! It seems like objects are passed by reference only between Serial and node 0 of parallel.
Let's verify that: the next pmi.invoke will call a method in Parallel that will attempt to set
the second array element to be 1000 on node 0 (and only on node 0) and the third element to be
 1000 on node 1 (and node 1 only)

############ TEST 3: CHANGING DIFFERENT ELEMENTS ON DIFFERENT NODES ############
Once again, array before pmi.invoke: [999   2   3]
Array after pmi.invoke: [ 999 1000    3]

Ok, that proves the case that objects are indeed passed by reference in pmi between Serial
and Parallel,but only for the first node. For all other nodes they are passed by reference.
 This is a bit tricky, but might be useful.

and the source code:

if __name__ != 'pmi':
    ##################################################
    ## Serial code
    ##################################################
    import pmi
    from time import time
    from numpy import array
    
    pmi.setup()
    
    pmi.execfile_(__file__)
    

    # create a frontend class
    class Hello(object):
        def __init__(self, name):
            self.pmiobj = pmi.create('HelloLocal', name)
        def __call__(self):
            return pmi.invoke(self.pmiobj, '__call__')
        def check_pass_by_ref1(self,arr):
            pmi.invoke(self.pmiobj,"modify_array",arr)
        def check_pass_by_ref2(self):
            pmi.invoke(self.pmiobj,"print_array")
        def check_pass_by_ref3(self):
            pmi.invoke(self.pmiobj,"modify_array2")
        

    # use the class
    hello = Hello('Konstantin')
    
    a = array([1,2,3])
    
    print "############ TEST 1: PASSING FROM SERIAL INTO PARALLEL ############"
    print "Testing PMI pass by reference: an array is created in Serial and then passed into Parallel"
    print "array before mpi.invoke:",a
    
    hello.check_pass_by_ref1(a)
    
    print "If the array was passed by _value_ then after modifying it in Parallel it should remain"
    print "unchanged in Serial"
    print "array after mpi.invoke:",a
    print "As you can see, the array has been changed in serial as well, which proves that objects"
    print "are passed by reference in pmi."
    
    # test 2
    
    print ""
    print "############ TEST 2: PASSING FROM PARALLEL INTO SERIAL ############"
    print "Now we'll check if modifying the same array in Serial will make it change in Parallel"
    print "The array:", a
    
    a[0] = 999
    
    print "Changed the first element to be 999:", a
    print "Now calling pmi.invoke that will make Parallel print out the array (it has been stored"
    print "as a class atribute during the previous experiment)"
    
    hello.check_pass_by_ref2()
    
    print "Oops! It seems like objects are passed by reference only between Serial and node 0 of parallel."
    print "Let's verify that: the next pmi.invoke will call a method in Parallel that will attempt to set"
    print "the second array element to be 1000 on node 0 (and only on node 0) and the third element to be"
    print " 1000 on node 1 (and node 1 only)"
    
    print ""
    print "############ TEST 3: CHANGING DIFFERENT ELEMENTS ON DIFFERENT NODES ############"
    print "Once again, array before pmi.invoke:",a
    hello.check_pass_by_ref3()
    print "Array after pmi.invoke:",a
    print ""
    print "Ok, that proves the case that objects are indeed passed by reference in pmi between Serial"
    print "and Parallel,but only for the first node. For all other nodes they are passed by reference."
    print " This is a bit tricky, but might be useful."
else:
    ##################################################
    ## Parallel code
    ##################################################
    from mpi4py import MPI
    from numpy.random import beta
    
    y = beta(a=1,b=1000,size=(210,210))

    class HelloLocal(object):
        def __init__(self, name):
            self.name = name
            self.rank = MPI.COMM_WORLD.Get_rank()
        def __call__(self):
            return 'Hello %s, this is MPI task %d!' % (self.name, MPI.COMM_WORLD.rank)
        def modify_array(self,arr):
            self.arr = arr
            self.arr[0] = 1000
        def modify_array2(self):
            if self.rank== 0:
                self.arr[1] = 1000
            if self.rank== 1:
                self.arr[2] = 1000
        def print_array(self):
            print "Node",self.rank,"array:",self.arr

I think, the reason for this is very simple: PMI seems to be broadcasting the data to all nodes, when pmi.invoke is issued, (perhaps, using MPI bcast()) from node 0 and keep the broadcast data on node 0 as it is, i.e. reference to the object passed into pmi.invoke, and not the copy of this object.

Day 88

Day 88: The moment of truth

Benchmarking. The time has come. This evening I will test the performance of my implementation and update this post as I get results with each test. The machine I'm going to use is a shared-memory multicore with 8 Intel(R) Xeon(R) X5450 CPU's, 3GHz, 6MB cache, 4 cores each. This little monster is called Jupiter III and together with Jupiters I and II forms a 3-piece cluster.

So, at the moment I am testing the test-bench and soon will be ready to launch. Don't switch the channel=)

[Update]

Phew! Preparation for testing took me way more time than I expected... However, look at this beautiful script! (./topographica line's were created by Chris, everything else is mine)

#!/bin/bash
# Argument =  -c cortex density  -n number of cpu's  -i number of iterations

usage()
{
cat << EOF
usage: $0 options

This script tests the performance of parallel Topographica implementation 
    
OPTIONS:
    -c        cortex density
    -n        number of cpu's
    -i        number of iterations
EOF
}
        
C_DENSITY=
N_CPU=
ITERATIONS=
while getopts “c:n:i:” OPTION
do
    case $OPTION in
        c)
        C_DENSITY=$OPTARG
        ;;
        n)
        N_CPU=$OPTARG
        ;;
        i)
        ITERATIONS=$OPTARG
        ;;
        ?)
        usage
        exit
        ;;
    esac
done

if [[ -z $C_DENSITY ]] || [[ -z $N_CPU ]] || [[ -z $ITERATIONS ]]
then
    usage
    exit 1
fi

# Mock run to make sure that all C code has compiled before proceeding to actual testing
./topographica -p cortex_density=1 examples/tiny.ty -c "topo.sim.run(1)"

# Serial run
./topographica -p cortex_density=$C_DENSITY examples/tiny.ty -c "import timeit; print 'Serial:', timeit.Timer('topo.sim.run($ITERATIONS)','import topo').timeit(number=1)" -c 'import pickle; pickle.dump(topo.sim["V1"].activity,open("results.pickle","wb"))'


# Parallel run
mpirun -n $N_CPU ./topographica -p mpi=True -p cortex_density=$C_DENSITY examples/tiny.ty -c "import timeit; print 'MPI:', timeit.Timer('topo.sim.run($ITERATIONS)','import topo').timeit(number=1)" -c 'import pickle; from numpy.testing import assert_array_equal; previous_result=pickle.load(open("results.pickle","r")); assert_array_equal(topo.sim["V1"].activity,previous_result)'

Now to do testing you just need to run this script with 3 parameters: -c for cortex density, i for the number of iterations and -n for the number of processes to use. I already tried this out on my machine and so far the results are not that inspiring:

./test_script -c 25 -n 2 -i 100
Serial: 5.36015510559
MPI: 6.15542793274

On the bright side, the results computed in MPI are still correct. Now I'm going to try this out on Jupiter.

[update 2] Ok, found a bug in the script: it ran tiny.ty instead of tiny_mpi.ty. How the hell did it get correct results? Anyway, new output (my laptop):
./test_script -c 25 -n 2 -i 100
Not using MPI
Not using MPI
Serial: 1.02229189873
Using MPI
NODE 0 initialized
NODE 1 initialized
MPI: 3.37540221214

[update 3] Ok, I need to make a break now. For some reason Jupiter won't let me run the tests, so I'll have to deal with that somehow. But later.