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.

Day 83

Day 83: ssh me!

Hi, Mr Internet. Yesterday I was finally able to run my first experiment with PMI implementation! This means that now I can start benchmarking and see where I really am. At the moment I am working on setting up my first cluster network - something I started over Christmas Holidays, spent a couple of days working on, hadn't managed to make it and decided to shift my attention to more important things. However, now is the time. I want to see for myself that it is possible to make a scalable network of arbitrary PC's (not specialized hardware) with different architectures to work as a single unit doing computations in parallel and then gathering results on a single node. Currently setting up ssh-agent to allow password-less access from my computer to DICE (university) network, which is absolutely necessary for running parallel jobs between these machines, is giving me the most headache. Anyway, should be manageable.

Also, since it's finally woring now, here's the (part of) actual implementation of MPI_CFProjection:

class MPI_CFProjection(CFProjection):
    
    def __init__(self,initialize_cfs=True, **params):
        pmi.execfile_('topo/base/mpi_cf.py')
        self.pmiobj = pmi.create('MPI_CFProjection_node')
        super(MPI_CFProjection,self).__init__(initialize_cfs=True,**params)
        self.allow_skip_non_responding_units = True # = self.dest.allow_skip_non_responding_units
        
        self.mask = self.dest.mask
            
    
    """>>>>>>>>>>>>>>>>>>>>>>>>> PROPERTIES >>>>>>>>>>>>>>>>>>>>>>>>>"""
    
    def __set_flatcfs(self,flatcfs):
        pmi.invoke(self.pmiobj,'_set_flatcfs_chunk',flatcfs)
    def __get_flatcfs(self):
        flatcfs_list = pmi.invoke(self.pmiobj,'_get_flatcfs_chunk')
        flatcfs = []
        for flatcfs_row in flatcfs_list:
            flatcfs.extend(flatcfs_row)
        return flatcfs
    def __del_flatcfs(self):
        pmi.invoke(self.pmiobj,'_set_flatcfs_chunk',None)
    flatcfs = property(__get_flatcfs,__set_flatcfs,__del_flatcfs)
    
    
    def __set_strength(self,strength):
        pmi.invoke(self.pmiobj,'_set_strength',strength)
    def __get_strength(self):
        strength = pmi.invoke(self.pmiobj,'_get_strength')
        return strength[0]
    def __del_strength(self):
        pmi.invoke(self.pmiobj,'_set_strength',None)
    strength = property(__get_strength,__set_strength,__del_strength)
    

    def __set_activity(self,activity):
        #flattening activity matrix
        self.activity_shape = activity.shape
        #reshaping into one-dimensional matrix (2d matrix that has only one row)
        activity = activity.reshape(1,self.activity_shape[0] * self.activity_shape[1])
        pmi.invoke(self.pmiobj,'_set_activity',list(activity[0]))
    def __get_activity(self):
        activity_list = pmi.invoke(self.pmiobj,'_get_activity')
        activity = []
        for activity_row in activity_list:
            activity.extend(activity_row)
        activity = numpy.array(activity)
        return activity.reshape(self.activity_shape[0],self.activity_shape[1])
    def __del_activity(self):
        pmi.invoke(self.pmiobj,'_set_activity', None)
    activity = property(__get_activity,__set_activity,__del_activity)

This is the serial part of implementation. This means that this code is run as if mpirun command wasn't issued. Members of the super-class (CFProjection) that have to be distributed in order to do computations are implemented as properties. For each property, get and set methods call pmi.invoke which, in turn, calls the MPI method specified as the parameter to mpi.invoke call. If something else is passed as parameter to pmi.invoke, it goes through to parallel method as a parameter, one copy per node. Thus, if you do pmi.invoke(self.pmiobj, "some_method", "abc"), then some_method will be called in the parallel mode and an instance of "abc" string will be passed to it. Obviously, some_method(self, some_string) has to exist somewhere.

Here's the (part of) parallel code that is being run every time pmi.invoke call occurs:

class MPI_CFProjection_node(CFProjection):
    def __init__(self):
        self.comm = MPI.COMM_WORLD
        self.rank = self.comm.Get_rank()
        self.size = self.comm.Get_size()
        print "NODE", MPI.COMM_WORLD.Get_rank(), "initialized"
    
    def _set_flatcfs_chunk(self, flatcfs):
        if flatcfs==None:
            self.flatcfs = None
        else:
            cfs_per_node = int(round(len(flatcfs)/self.size))
            if self.rank+1<self.size:
                self.flatcfs = flatcfs[self.rank * cfs_per_node : (self.rank+1) * cfs_per_node]
            else:
                self.flatcfs = flatcfs[self.rank * cfs_per_node : len(flatcfs)]
    def _get_flatcfs_chunk(self):
        return self.flatcfs

    
    def _set_activity(self, activity):
        if activity==None:
            self.activity = None
        else:
            items_per_node = int(round(len(activity) / self.size))
            if self.rank+1<self.size:
                self.activity = activity[self.rank * items_per_node : (self.rank+1) * items_per_node]
            else:
                self.activity = activity[self.rank * items_per_node : ]
            self.activity = numpy.array([self.activity])
    def _get_activity(self):
        if self.activity==None:
            return None
        else:
            return list(self.activity[0])

That's how it works.

Day 67

Day 67. "Wow, PMI rules!"

Yes, indeed I was wrong. Lets have a look at this file (my edited version of hello_class.py, which comes with the PMI archive):

if __name__ != 'pmi':
    ##################################################
    ## Serial code
    ##################################################
    import pmi

    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 func(self):
            return pmi.invoke(self.pmiobj,'func')

    # use the class
    hello = Hello('Olaf')
    print('\n'.join(hello()))

    a = hello.func()

    print "Wow, PMI rules!"

else:
    ##################################################
    ## Parallel code
    ##################################################
    from mpi4py import MPI

    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 func(self):
            if MPI.COMM_WORLD.rank == 0:
                return raw_input()
            else:
                return "Slave node" 

Things to pay attention to:

1) First, name object is passed as a parameter from the serial code into the parallel class, and than a separate call to parallel code uses this object. This means that all objects persist between separate calls to parallel code.

2) While calling the func function, only the master node (rank 0) is waiting for user input. However, as opposed to using just mpi, all other nodes are waiting for the master node to complete operation before all nodes join again in the serial code.

3) Notice that "Wow, PMI rules!" is printed out only once, no matter how many nodes you create with mpirun. Now comment out everything else, leave just this print statement, and mpirun this file once again, creating say 4 processes. What will you get?

"Wow, PMI rules!"
"Wow, PMI rules!"
"Wow, PMI rules!"
"Wow, PMI rules!"

Indeed, it does.

Day 65

Day 65

I just thought I haven't posted for a while, so this is a quick update on what's currently going on.

First, this Monday my laptop has refused to switch on and now seems to be pretty much dead, but hopefully this is just coma and anyways my warranty hasn't expired yet, so this Wednesday it's going to be picked up and hopefully returned in a week, fully functional. Not a big issue, just a bit annoying.

Second, some time ago Chris suggested I have a look at PMI - "a pure python module that allows libraries to provide functions that are parallelized using MPI but that can nonetheless be called from serial Python scripts." I'll write about it if I find the way to utilize this thing, but at the moment it seems to be quite a handy module. However, here's what I want to get rid of in my implementation by using PMI:

'''
Created on 20 Jan 2010

@author: megatelevizor
'''
from mpi4py import MPI
import time

from MPI_client import client


""" Vocabulary

a   = assign flatcfs
c   = compute activity
i   = Initialize
i_a = Get a copy of input_activity matrix
k   = Kill
p   = Print
r   = Return activity
s   = Set strength


"""

class communicator(object):

   def listen(self):
       self.comm = MPI.COMM_WORLD
       self.rank = self.comm.Get_rank()
      
       self.client_node = client()
      
       while True:
           while not self.comm.Iprobe(source=0, tag=0):
               time.sleep(0.1)
           command_prefix = self.comm.recv(source=0, tag=0)
          
           """ Print """
           if command_prefix == "p":
               i_command_body = self.comm.recv(source=0,tag=1)
               self.client_node.print_command(command = i_command_body)
          
           """ Kill """
           if command_prefix == "k":
               exit()
          
           """ Initialize MPI_CFProjection_slave"""
           if command_prefix == "i":
               self.client_node.init_mpi_cfprojection_slave(node_id=self.rank)
          
           """ Assign flatcfs"""
           if command_prefix == "a":
               new_flatcfs = self.comm.recv(source=0,tag=1)
               print "Node",self.rank,": received", len(new_flatcfs),"flatcfs"
               self.client_node.assign_cfs(new_flatcfs)
          
           """ Set a copy of input activity matrix """
           if command_prefix == "i_a":
               input_activity = self.comm.recv(source=0,tag=1)
               self.client_node.set_input_activity(input_activity)
                      
           """ Compute activity """
           if command_prefix == "c":
               self.client_node.compute_activity()
              
           """ Return activity """
           if command_prefix == "r":
               activity = self.client_node.get_activity()
               print "Node",self.rank,"activity:",activity
               self.comm.send(obj=activity, dest=0, tag=2)
          
           """ Set strength"""
           if command_prefix == "s":
               strength = self.comm.recv(source=0,tag=1)
               self.client_node.set_strength(strength)
              

              
              

""" TAGS:
0 - command prefix
1 - input data
2 - output data
"""

This class allows master node to communicate with slave nodes. Potentially, PMI will eliminate all need to have such class.



[update] No, it will not. What PMI allows you to do is hide the parallel implementation, so that your programme looks like it is a serial (single-processed) code. However, it does not enhance the underlying MPI implementation in any way, so it still needs to be implicit. What does that mean in my case? Well, if I was to introduce PMI into my code as it is now, then there would have been a class - something like CFProjection_proxy, linked to the real (i.e. parallel) implementation of CFProjection_slave using PMI mechanism. CFPRojection_slave's would still have to communicate with the master node CFProjection using MPI_io (the code listed above) because PMI does not influence or change the way different nodes of an MPI application communicate. In simple words, it just makes your code look like it's not using MPI, when it actually does.

Well, so what's so bad about it? Why not use PMI anyway, since from all that it looks like it's actually doing a good job? Actually, there's nothing wrong with it and it might be a good idea to use it. No, it won't add any new functionality, but it will definitelly make things look a lot nicer. However, I prefer dealing with things one at a time, and also finishing what I started, so I think I'll concentrate on looking for the solution to my initial problem.

To summarise what I'm looking for, have a look at this: Java RMI . Without going into too much detail, RMI is a mechanism that allows Java objects calling each other's methods accross several different JVM's (which can be run on different hosts!) and it stands for Remote Method Invocation. Combined with serialization (something similar to Python's pickling), it allows to hide all explicit communication-level code away from the user, which is great! Thus, my assumption is that if something exists in Java, it has a chance to exist in Python. As simple as that.

Searching goes on.