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.
