Day 113
Oh God, I did it! I figured out how to run topographica on a cluster, and it wasn't that easy. A very good guide that helped me was: Setting up a Beowulf Cluster Using Open MPI on Linux. Running MPI jobs points 1-4 are also useful. Just to summarise, in order to run an MPI job on a cluster, you need:
- Hostfile with hostnames of machines in the cluster
- OpenMPI installed on each machine in the cluster, in the same directory
- system PATH needs to know where OpenMPI's bin/ directory is
- system LD_LIBRARY_PATH needs to know where OpenMPI's lib/ directory is
- All common libraries that are going to be used during the runtime need to be installed in the same directory (i.e. path to them should be the same on all nodes). In my case, all topographica libraries are local and therefore I had to have it installed on each node in the same directory, e.g. ~/CODE/topographica
One note: if using an mpirun command with an absolute path to it (e.g. /usr/bin/openmpi/mpirun), then OpenMPI bin/ and lib/ directories will be added automatically to PATH and LD_LIBRARY_PATH respectively on all nodes.
Showing posts with label programming. Show all posts
Showing posts with label programming. Show all posts
Sunday, 21 March 2010
Tuesday, 9 March 2010
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!).
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...
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...
Monday, 8 March 2010
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:
The numbers speak for themselves. The source code:
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.
and the source code:
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.
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.
Wednesday, 24 February 2010
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)
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.
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.
Friday, 19 February 2010
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:
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:
That's how it works.
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.
Subscribe to:
Posts (Atom)