...
1) Each process computes the value of Pi locally, and 2) sends it to master task using send() function. Then, 3) the master task can recieve the messages using sync() function. Finally, we can calculate the average value of sum of PI values from each peers as below:
| Code Block | ||
|---|---|---|
| ||
@Override | ||
| No Format | ||
public void bsp(BSPPeerProtocol bspPeer) throws IOException,
BSPPeer<NullWritable, NullWritable, Text, DoubleWritable> peer)
KeeperExceptionthrows IOException, SyncException, InterruptedException {
int in = 0, out = 0;
for (int i = 0; i < iterations; i++) {
double x = 2.0 * Math.random() - 1.0, y = 2.0 * Math.random() - 1.0;
if ((Math.sqrt(x * x + y * y) < 1.0)) {
in++;
} else {
out++;
}
}
byte[] tagName = Bytes.toBytes(bspPeer.getPeerName());
byte[] myDatadouble data = Bytes.toBytes(4.0 * (double) in / (double) iterations);
BSPMessageDoubleMessage estimate = new BSPMessage(tagNameDoubleMessage(peer.getPeerName(), myDatadata);
bspPeerpeer.send(masterTask, estimate);
bspPeerpeer.sync();
}
@Override
public void setup(
BSPPeer<NullWritable, NullWritable, Text, DoubleWritable> peer)
throws IOException {
// Choose one as a master
this.masterTask = peer.getPeerName(peer.getNumPeers() / 2);
}
public void cleanup(
BSPPeer<NullWritable, NullWritable, Text, DoubleWritable> peer)
throws IOException {
if (peer.getPeerName().equals(masterTask)) {
double pi = 0.0;
int numPeers = bspPeerpeer.getNumCurrentMessages();
BSPMessageDoubleMessage received;
while ((received = (DoubleMessage) bspPeerpeer.getCurrentMessage()) != null) {
pi += Bytes.toDouble(received.getData());
}
pi if (bspPeer.getPeerName().equals(masterTask)) {
= pi / numPeers;
peer
pi = pi / numPeers;
writeResult .write(new Text("Estimated value of PI is"), new DoubleWritable(pi));
}
}
}
|