Summing Python Objects with MPI Allreduce
I am using sparse tensor array arrays that I have built using dictionaries and counters in Python. I would like to use parallel array manipulation. The bottom line is that I have counters on each node that I would like to add together using MPI.Allreduce (or some other nice solution). For example, using counters you can do this
A = Counter({a:1, b:2, c:3})
B = Counter({b:1, c:2, d:3})
such that
C = A+B = Counter({a:1, b:3, c:5, d:3}).
I would like to do the same operation, but with all the corresponding nodes,
MPI.Allreduce(send_counter, recv_counter, MPI.SUM)
however MPI doesn't seem to recognize this operation on dictionaries / counters, causing an error expecting a buffer or a list/tuple
. Is my best option "User-Defined Operation", or is there a way to force Allreduce to add counters? Thank,
EDIT (7/14/15): I tried to create a custom operation for dictionaries, but there were some inconsistencies. I wrote the following
def dict_sum(dict1, dict2, datatype):
for key in dict2:
try:
dict1[key] += dict2[key]
except KeyError:
dict1[key] = dict2[key]
and when I told MPI about the function, I did this:
dictSumOp = MPI.Op.Create(dict_sum, commute=True)
and in the code i used it like
the_result = comm.allreduce(mydict, dictSumOp)
However, he gave up unsupported operand '+' for type dict
. so i wrote
the_result = comm.allreduce(mydict, op=dictSumOp)
and now he is throwing dict1[key] += dict2[key]
TypeError: 'NoneType' object has no attribute '__getitem__'
, apparently he wants to know what are dictionaries? How can I tell they have a type dictionary?
source to share
Neither MPI nor MPI4py knows anything about counters, so you need to create your own pruning operation for this to work; this would be the same for any other type of python object:
#!/usr/bin/env python
from mpi4py import MPI
import collections
def addCounter(counter1, counter2, datatype):
for item in counter2:
counter1[item] += counter2[item]
return counter1
if __name__=="__main__":
comm = MPI.COMM_WORLD
if comm.rank == 0:
myCounter = collections.Counter({'a':1, 'b':2, 'c':3})
else:
myCounter = collections.Counter({'b':1, 'c':2, 'd':3})
counterSumOp = MPI.Op.Create(addCounter, commute=True)
totcounter = comm.allreduce(myCounter, op=counterSumOp)
print comm.rank, totcounter
Here we took a function that adds two counter objects and created an MPI operator from them with MPI.Op.Create; mpi4py will flatten the objects, run this function to pair these elements together, then sort the partial result and send it to the next task.
Note that we are using (lowercase) allreduce, which works on arbitrary python objects, not uppercase Allreduce, which works on numpy arrays or their moral equivalents (buffers that map to Fortran / C arrays for which the MPI API is designed ).
Running gives:
$ mpirun -np 2 python ./counter_reduce.py
0 Counter({'c': 5, 'b': 3, 'd': 3, 'a': 1})
1 Counter({'c': 5, 'b': 3, 'd': 3, 'a': 1})
$ mpirun -np 4 python ./counter_reduce.py
0 Counter({'c': 9, 'd': 9, 'b': 5, 'a': 1})
2 Counter({'c': 9, 'd': 9, 'b': 5, 'a': 1})
1 Counter({'c': 9, 'd': 9, 'b': 5, 'a': 1})
3 Counter({'c': 9, 'd': 9, 'b': 5, 'a': 1})
And only with minor changes it works with a common dictionary:
#!/usr/bin/env python
from mpi4py import MPI
def addCounter(counter1, counter2, datatype):
for item in counter2:
if item in counter1:
counter1[item] += counter2[item]
else:
counter1[item] = counter2[item]
return counter1
if __name__=="__main__":
comm = MPI.COMM_WORLD
if comm.rank == 0:
myDict = {'a':1, 'c':"Hello "}
else:
myDict = {'c':"World!", 'd':3}
counterSumOp = MPI.Op.Create(addCounter, commute=True)
totDict = comm.allreduce(myDict, op=counterSumOp)
print comm.rank, totDict
Completing the task
$ mpirun -np 2 python dict_reduce.py
0 {'a': 1, 'c': 'Hello World!', 'd': 3}
1 {'a': 1, 'c': 'Hello World!', 'd': 3}
source to share