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!).
Note that you do not actually have to use pmi.localcall(), it is just there for convenience and a consistent interface. You can always access the local (Node 0) copy of the parallel object directly, as the reference that you get when using pmi.create() is nothing else than the object instance itself. So, you can always call "pmiobj.whatever_function()" if you only want to call the local copy.
ReplyDeletePMI provides a much simpler way to do what you want to do via localcall: reduce(), call() and invoke() can also take arguments that are only passed to the controller when you use the prefix __pmictr_ for the argument.
ReplyDeleteIn the following example, I create a function "parallel_function" that gets a huge list of values on the controller and scatters the data as needed. Note that in this case, the data does not get broadcast! The calls to parallel_function on the workers will only see an empty list.
# this is the parallel function
def parallel_function(xs=[]):
# scatter the data obtained on the controller
comm = MPI.COMM_WORLD
allchunks = []
if pmi.isController:
chunksize = len(xs)/comm.size
for i in range(0, comm.size):
allchunks.append(
xs[i*chunksize:(i+1)*chunksize])
mychunk = comm.scatter(allchunks)
# Now do whatever you want with this chunk
.
.
# here is how you can call the function
xs = range(0, 10000)
pmi.invoke("parallel_function", __pmictr_xs=xs)