## Pi Estimator

The value of PI can be calculated in a number of ways. Consider the following method of estimating PI

- Inscribe a circle in a square
- Randomly generate points in the square
- Determine the number of points in the square that are also in the circle
- Let r be the number of points in the circle divided by the number of points in the square
- PI ~ 4 r

Serial pseudo code for this procedure as below:

iterations = 10000 circle_count = 0 do j = 1,iterations generate 2 random numbers between 0 and 1 xcoordinate = random1 ycoordinate = random2 if (xcoordinate, ycoordinate) inside circle then circle_count = circle_count + 1 end do PI = 4.0*circle_count/iterations

## The BSP implementation for Pi

A distributed strategy in HAMA with BSP programming model, is break the loop into portions that can be executed by the tasks.

- Each task executes its portion of the loop a number of times.
- Each task can do its work without requiring any information from the other tasks (there are no data dependencies).
- One task acts as master and collects the results.

public class PiEstimator { private static String MASTER_TASK = "master.task."; public static class MyEstimator extends BSP { public static final Log LOG = LogFactory.getLog(MyEstimator.class); private Configuration conf; private String masterTask; private static final int iterations = 10000; @Override public void bsp(BSPPeer bspPeer) throws IOException, KeeperException, 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(getName().toString()); byte[] myData = Bytes.toBytes(4.0 * (double) in / (double) iterations); BSPMessage estimate = new BSPMessage(tagName, myData); bspPeer.send(bspPeer.getAddress(masterTask), estimate); bspPeer.sync(); double pi = 0.0; BSPMessage received; while ((received = bspPeer.getCurrentMessage()) != null) { LOG.info("Receives messages:" + Bytes.toDouble(received.getData())); if(pi == 0.0) { pi = Bytes.toDouble(received.getData()); } else { pi = (pi + Bytes.toDouble(received.getData())) / 2; } } if (pi != 0.0) { LOG.info("\nEstimated value of PI is " + pi); } } @Override public Configuration getConf() { return conf; } @Override public void setConf(Configuration conf) { this.conf = conf; this.masterTask = conf.get(MASTER_TASK); } } public static void main(String[] args) throws InterruptedException, IOException { // BSP job configuration HamaConfiguration conf = new HamaConfiguration(); // Execute locally // conf.set("bsp.master.address", "local"); BSPJob bsp = new BSPJob(conf, PiEstimator.class); // Set the job name bsp.setJobName("pi estimation example"); bsp.setBspClass(MyEstimator.class); BSPJobClient jobClient = new BSPJobClient(conf); ClusterStatus cluster = jobClient.getClusterStatus(true); // Choose one as a master for (String name : cluster.getActiveGroomNames()) { conf.set(MASTER_TASK, name); break; } BSPJobClient.runJob(bsp); } }