...
- Each task gets its own hostname (hostname:port pair) and a sorted list containing the hostnames of all the other peers.
- Each task prints the LOG string "Hello BSP" only when its turn comes at intervals of 5 seconds.
Sequence Diagram
BSP implementation of Serialize Printing of "Hello BSP"
- See the BSP programming model of HAMA if you didn't read yet.
| No Format |
|---|
public class SerializePrinting { public static class HelloBSPClassSerializePrinting extends BSP { publicBSP<NullWritable, staticNullWritable, finalIntWritable, Log LOG = LogFactory.getLog(HelloBSP.class); private Configuration conf;Text> { public privatestatic final static int PRINTNUM_INTERVALSUPERSTEPS = 500015; @Override public void bsp(BSPPeer bspPeer) throws IOException, KeeperException,BSPPeer<NullWritable, NullWritable, IntWritable, Text> bspPeer) throws IOException, SyncException, InterruptedException { for (int numi = Integer.parseInt(conf.get("bsp.peers.num")); int i = 0;0; i < NUM_SUPERSTEPS; i++) { for (String otherPeer : bspPeer.getAllPeerNames()) { if (bspPeer.getPeerName().equalssend(otherPeer)), { LOG.info("Hello BSP from " + i + " of " + num + ": " + new IntegerMessage(bspPeer.getPeerName(), i)); } Thread.sleep(PRINT_INTERVAL); bspPeer.sync(); IntegerMessage msg = i++null; } while ((msg } @Override public Configuration getConf() { return conf; } @Override public void setConf(Configuration conf) { this.conf = conf; } } 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, SerializePrinting.class= (IntegerMessage) bspPeer.getCurrentMessage()) != null) { bspPeer.write(new IntWritable(msg.getData()), new Text(msg.getTag())); // Set the job name} bsp.setJobName("serialize printing"); bsp.setBspClass(HelloBSP.class); BSPJobClient.runJob(bsp);} } } |