incubator-hama-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tomm...@apache.org
Subject svn commit: r1152788 [7/9] - in /incubator/hama/trunk: ./ bin/ conf/ core/ core/bin/ core/conf/ core/src/ core/src/main/ core/src/main/java/ core/src/main/java/org/ core/src/main/java/org/apache/ core/src/main/java/org/apache/hama/ core/src/main/java/o...
Date Mon, 01 Aug 2011 14:12:56 GMT
Added: incubator/hama/trunk/core/src/main/webapp/bspmaster/index.html
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/webapp/bspmaster/index.html?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/main/webapp/bspmaster/index.html (added)
+++ incubator/hama/trunk/core/src/main/webapp/bspmaster/index.html Mon Aug  1 14:12:46 2011
@@ -0,0 +1,36 @@
+<!--
+   Licensed to the Apache Software Foundation (ASF) under one or more
+   contributor license agreements.  See the NOTICE file distributed with
+   this work for additional information regarding copyright ownership.
+   The ASF licenses this file to You under the Apache License, Version 2.0
+   (the "License"); you may not use this file except in compliance with
+   the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+-->
+<meta HTTP-EQUIV="REFRESH" content="0;url=bspmaster.jsp"/>
+<html>
+
+<head>
+<title>Hama Administration</title>
+</head>
+
+<body>
+
+<h1>Hama Administration</h1>
+
+<ul>
+
+<li><a href="bspmaster.jsp">BSPMaster</a></li>
+
+</ul>
+
+</body>
+
+</html>

Added: incubator/hama/trunk/core/src/main/webapp/bspmaster/machines.jsp
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/main/webapp/bspmaster/machines.jsp?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/main/webapp/bspmaster/machines.jsp (added)
+++ incubator/hama/trunk/core/src/main/webapp/bspmaster/machines.jsp Mon Aug  1 14:12:46 2011
@@ -0,0 +1,61 @@
+<!--
+   Licensed to the Apache Software Foundation (ASF) under one or more
+   contributor license agreements.  See the NOTICE file distributed with
+   this work for additional information regarding copyright ownership.
+   The ASF licenses this file to You under the Apache License, Version 2.0
+   (the "License"); you may not use this file except in compliance with
+   the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+-->
+<%@ page contentType="text/html; charset=UTF-8" import="javax.servlet.*"
+	import="javax.servlet.http.*" import="java.io.*" import="java.util.*"
+	import="java.text.DecimalFormat" import="org.apache.hama.bsp.*"
+	import="org.apache.hama.util.*"%>
+<%!private static final long serialVersionUID = 1L;%>
+<%
+	BSPMaster tracker = (BSPMaster) application
+			.getAttribute("bsp.master");
+	ClusterStatus status = tracker.getClusterStatus(true);
+	String trackerName = tracker.getBSPMasterName();
+	String type = request.getParameter("type");
+%>
+<%!public void generateGroomsTable(JspWriter out, String type,
+			ClusterStatus status, BSPMaster master) throws IOException {
+	out.print("<center>\n");
+    out.print("<table border=\"2\" cellpadding=\"5\" cellspacing=\"2\">\n");
+    out.print("<tr><td align=\"center\" colspan=\"6\"><b>Groom Servers</b></td></tr>\n");
+    out.print("<tr><td><b>Name</b></td>" + "<td><b>Host</b></td>"
+        + "<td><b># running tasks</b></td></tr>\n");
+    for (Map.Entry<String, String> entry : status.getActiveGroomNames()
+        .entrySet()) {
+      out.print("<tr><td><a href=\"http://");
+      out.print(entry.getKey() + ":" + master.getHttpPort() + "/\">");
+      out.print(entry.getValue() + "</a></td><td>");
+      out.print(entry.getValue() + "</td>" + "<td>" + 1 + "</td></tr>\n");
+    }
+    out.print("</table>\n");
+    out.print("</center>\n");
+  }%>
+
+<html>
+
+<title><%=trackerName%> Hama Machine List</title>
+
+<body>
+<h1><a href="bspmaster.jsp"><%=trackerName%></a> Hama Machine List</h1>
+
+<h2>Grooms</h2>
+<%
+  generateGroomsTable(out, type, status, tracker);
+%>
+
+<%
+  out.println(BSPServletUtil.htmlFooter());
+%>
\ No newline at end of file

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaCluster.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaCluster.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaCluster.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaCluster.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,47 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Forming up the miniDfs and miniZooKeeper
+ */
+public abstract class HamaCluster extends HamaClusterTestCase {
+  public static final Log LOG = LogFactory.getLog(HamaCluster.class);
+  protected final static HamaConfiguration conf = new HamaConfiguration();
+
+  public HamaCluster(){
+    super();
+  }
+
+  public HamaCluster(boolean startDfs) {
+    super(startDfs);
+  }
+
+  protected void setUp() throws Exception {
+    super.setUp();
+  }
+
+  protected static HamaConfiguration getConf() {
+    return conf;
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaClusterTestCase.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaClusterTestCase.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaClusterTestCase.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaClusterTestCase.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,107 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama;
+
+import java.io.File;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+
+public abstract class HamaClusterTestCase extends HamaTestCase {
+  public static final Log LOG = LogFactory.getLog(HamaClusterTestCase.class);
+  protected MiniDFSCluster dfsCluster;
+  protected MiniBSPCluster bspCluster;
+  protected MiniZooKeeperCluster zooKeeperCluster;
+  protected boolean startDfs;
+  protected int numOfGroom = 2;
+
+  /** default constructor */
+  public HamaClusterTestCase() {
+    this(false);
+  }
+
+  public HamaClusterTestCase(boolean startDfs) {
+    super();
+    this.startDfs = startDfs;
+  }
+
+  /**
+   * Actually start the MiniBSP instance.
+   */
+  protected void hamaClusterSetup() throws Exception {
+    File testDir = new File(getUnitTestdir(getName()).toString());
+
+    // Note that this is done before we create the MiniHamaCluster because we
+    // need to edit the config to add the ZooKeeper servers.
+    this.zooKeeperCluster = new MiniZooKeeperCluster();
+    int clientPort = this.zooKeeperCluster.startup(testDir);
+    conf.set("hama.zookeeper.property.clientPort", Integer.toString(clientPort));
+    bspCluster = new MiniBSPCluster(this.conf, numOfGroom); 
+    bspCluster.startBSPCluster();
+  }
+
+  @Override
+  protected void setUp() throws Exception {
+    try {
+      if (this.startDfs) {
+        // This spews a bunch of warnings about missing scheme. TODO: fix.
+        this.dfsCluster = new MiniDFSCluster(0, this.conf, 2, true, true, true,
+          null, null, null, null);
+
+        // mangle the conf so that the fs parameter points to the minidfs we
+        // just started up
+        FileSystem filesystem = dfsCluster.getFileSystem();
+        conf.set("fs.defaultFS", filesystem.getUri().toString());
+        Path parentdir = filesystem.getHomeDirectory();
+        
+        filesystem.mkdirs(parentdir);
+      }
+
+      // do the super setup now. if we had done it first, then we would have
+      // gotten our conf all mangled and a local fs started up.
+      super.setUp();
+
+      // start the instance
+      hamaClusterSetup();
+    } catch (Exception e) {
+      if (zooKeeperCluster != null) {
+        zooKeeperCluster.shutdown();
+      }
+      if (dfsCluster != null) {
+        shutdownDfs(dfsCluster);
+      }
+      throw e;
+    }
+  }
+
+  @Override
+  protected void tearDown() throws Exception {
+    super.tearDown();
+    try {
+      if (startDfs) {
+        shutdownDfs(dfsCluster);
+      }
+      bspCluster.shutdown();
+    } catch (Exception e) {
+      LOG.error(e);
+    }
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/HamaTestCase.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,179 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama;
+
+import java.io.File;
+import java.io.IOException;
+
+import junit.framework.AssertionFailedError;
+import junit.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hama.util.Bytes;
+
+public abstract class HamaTestCase extends TestCase {
+  private static Log LOG = LogFactory.getLog(HamaTestCase.class);
+  
+  /** configuration parameter name for test directory */
+  public static final String TEST_DIRECTORY_KEY = "test.build.data";
+
+  private boolean localfs = false;
+  protected Path testDir = null;
+  protected FileSystem fs = null;
+  
+  static {
+    initialize();
+  }
+
+  public volatile HamaConfiguration conf;
+
+  /** constructor */
+  public HamaTestCase() {
+    super();
+    init();
+  }
+  
+  /**
+   * @param name
+   */
+  public HamaTestCase(String name) {
+    super(name);
+    init();
+  }
+  
+  private void init() {
+    conf = new HamaConfiguration();
+    conf.setStrings("bsp.local.dir", "/tmp/hama-test");
+    conf.set("bsp.master.address", "localhost");
+    conf.set("bsp.groom.report.address", "127.0.0.1:0");
+  }
+
+  /**
+   * Note that this method must be called after the mini hdfs cluster has
+   * started or we end up with a local file system.
+   */
+  @Override
+  protected void setUp() throws Exception {
+    super.setUp();
+    localfs =
+      (conf.get("fs.defaultFS", "file:///").compareTo("file:///") == 0);
+
+    if (fs == null) {
+      this.fs = FileSystem.get(conf);
+    }
+    try {
+      if (localfs) {
+        this.testDir = getUnitTestdir(getName());
+        if (fs.exists(testDir)) {
+          fs.delete(testDir, true);
+        }
+      } else {
+        this.testDir =
+          this.fs.makeQualified(new Path("/tmp/hama-test"));
+      }
+    } catch (Exception e) {
+      LOG.fatal("error during setup", e);
+      throw e;
+    }
+  }
+
+  @Override
+  protected void tearDown() throws Exception {
+    try {
+      if (localfs) {
+        if (this.fs.exists(testDir)) {
+          this.fs.delete(testDir, true);
+        }
+      }
+    } catch (Exception e) {
+      LOG.fatal("error during tear down", e);
+    }
+    super.tearDown();
+  }
+
+  protected Path getUnitTestdir(String testName) {
+    return new Path(
+        conf.get(TEST_DIRECTORY_KEY, "/tmp/hama-test/build/data"), testName);
+  }
+
+  /**
+   * Initializes parameters used in the test environment:
+   *
+   * Sets the configuration parameter TEST_DIRECTORY_KEY if not already set.
+   * Sets the boolean debugging if "DEBUGGING" is set in the environment.
+   * If debugging is enabled, reconfigures logging so that the root log level is
+   * set to WARN and the logging level for the package is set to DEBUG.
+   */
+  public static void initialize() {
+    if (System.getProperty(TEST_DIRECTORY_KEY) == null) {
+      System.setProperty(TEST_DIRECTORY_KEY, new File(
+          "build/hama/test").getAbsolutePath());
+    }
+  }
+
+  /**
+   * Common method to close down a MiniDFSCluster and the associated file system
+   *
+   * @param cluster
+   */
+  public static void shutdownDfs(MiniDFSCluster cluster) {
+    if (cluster != null) {
+      LOG.info("Shutting down Mini DFS ");
+      try {
+        cluster.shutdown();
+      } catch (Exception e) {
+        /// Can get a java.lang.reflect.UndeclaredThrowableException thrown
+        // here because of an InterruptedException. Don't let exceptions in
+        // here be cause of test failure.
+      }
+      try {
+        FileSystem fs = cluster.getFileSystem();
+        if (fs != null) {
+          LOG.info("Shutting down FileSystem");
+          fs.close();
+        }
+        FileSystem.closeAll();
+      } catch (IOException e) {
+        LOG.error("error closing file system", e);
+      }
+    }
+  }
+
+  public void assertByteEquals(byte[] expected,
+                               byte[] actual) {
+    if (Bytes.compareTo(expected, actual) != 0) {
+      throw new AssertionFailedError("expected:<" +
+      Bytes.toString(expected) + "> but was:<" +
+      Bytes.toString(actual) + ">");
+    }
+  }
+
+  public static void assertEquals(byte[] expected,
+                               byte[] actual) {
+    if (Bytes.compareTo(expected, actual) != 0) {
+      throw new AssertionFailedError("expected:<" +
+      Bytes.toStringBinary(expected) + "> but was:<" +
+      Bytes.toStringBinary(actual) + ">");
+    }
+  }
+
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/MiniBSPCluster.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/MiniBSPCluster.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/MiniBSPCluster.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/MiniBSPCluster.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,252 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama;
+
+import java.io.IOException;
+
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import static java.util.concurrent.TimeUnit.*;
+
+import static junit.framework.Assert.*;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hama.bsp.BSPMaster;
+import org.apache.hama.bsp.GroomServer;
+import org.apache.hama.HamaConfiguration;
+
+
+public class MiniBSPCluster {
+
+  public static final Log LOG = LogFactory.getLog(MiniBSPCluster.class);
+
+  private ScheduledExecutorService scheduler;
+
+  private HamaConfiguration configuration;
+  private BSPMasterRunner master;
+  private List<GroomServerRunner> groomServerList = 
+    new CopyOnWriteArrayList<GroomServerRunner>();
+  private int grooms;
+
+  public class BSPMasterRunner implements Runnable{
+    BSPMaster bspm;
+    HamaConfiguration conf;
+
+    public BSPMasterRunner(HamaConfiguration conf){
+      this.conf = conf;
+      if(null == this.conf) 
+        throw new NullPointerException("No Configuration for BSPMaster.");
+    }  
+
+    public void run(){
+      try{
+        LOG.info("Starting BSP Master.");
+        this.bspm = BSPMaster.startMaster(this.conf); 
+        this.bspm.offerService();
+      }catch(IOException ioe){
+        LOG.error("Fail to startup BSP Master.", ioe);
+      }catch(InterruptedException ie){
+        LOG.error("BSP Master fails in offerService().", ie);
+        Thread.currentThread().interrupt();
+      }
+    }
+
+    public void shutdown(){
+      if(null != this.bspm) this.bspm.shutdown();
+    }
+
+    public boolean isRunning(){
+      if(null == this.bspm) return false;
+
+      if(this.bspm.currentState().equals(BSPMaster.State.RUNNING)){
+        return true;
+      } 
+      return false;
+    }
+
+    public BSPMaster getMaster(){
+      return this.bspm;
+    }
+  }
+
+  public class GroomServerRunner implements Runnable{
+    GroomServer gs;
+    HamaConfiguration conf;
+
+    public GroomServerRunner(HamaConfiguration conf){
+      this.conf = conf;
+    }
+ 
+    public void run(){
+      try{
+        this.gs = GroomServer.constructGroomServer(GroomServer.class, conf);
+        GroomServer.startGroomServer(this.gs).join();
+      }catch(InterruptedException ie){
+        LOG.error("Fail to start GroomServer. ", ie);
+        Thread.currentThread().interrupt();
+      }
+    }
+
+    public void shutdown(){
+      try{
+        if(null != this.gs) this.gs.shutdown();
+      }catch(IOException ioe){
+        LOG.info("Fail to shutdown GroomServer.", ioe);
+      }
+    }
+    
+    public boolean isRunning(){
+      if(null == this.gs) return false;
+      return this.gs.isRunning(); 
+    }
+
+    public GroomServer getGroomServer(){
+      return this.gs;
+    }
+  }
+
+  public MiniBSPCluster(HamaConfiguration conf, int groomServers) {
+    this.configuration = conf;
+    this.grooms = groomServers;
+    if(1 > this.grooms) {
+      this.grooms = 2;  
+    }
+    LOG.info("Groom server number "+this.grooms);
+    int threadpool = conf.getInt("bsp.test.threadpool", 10);
+    LOG.info("Thread pool value "+threadpool);
+    scheduler = Executors.newScheduledThreadPool(threadpool);
+  }
+
+  public void startBSPCluster(){
+    startMaster();
+    startGroomServers();
+  }
+
+  public void shutdownBSPCluster(){
+    if(null != this.master && this.master.isRunning())
+      this.master.shutdown();
+    if(0 < groomServerList.size()){
+      for(GroomServerRunner groom: groomServerList){
+        if(groom.isRunning()) groom.shutdown();
+      }
+    }
+  }
+
+
+  public void startMaster(){
+    if(null == this.scheduler) 
+      throw new NullPointerException("No ScheduledExecutorService exists.");
+    this.master = new BSPMasterRunner(this.configuration);
+    scheduler.schedule(this.master, 0, SECONDS);
+  }
+
+  public void startGroomServers(){
+    if(null == this.scheduler) 
+      throw new NullPointerException("No ScheduledExecutorService exists.");
+    if(null == this.master) 
+      throw new NullPointerException("No BSPMaster exists.");
+    int cnt=0;
+    while(!this.master.isRunning()){
+      LOG.info("Waiting BSPMaster up.");
+      try{
+        Thread.sleep(1000);
+        cnt++;
+        if(100 < cnt){
+          fail("Fail to launch BSPMaster.");
+        }
+      }catch(InterruptedException ie){
+        LOG.error("Fail to check BSP Master's state.", ie);
+        Thread.currentThread().interrupt();
+      }
+    }
+    for(int i=0; i < this.grooms; i++){
+      HamaConfiguration c = new HamaConfiguration(this.configuration);
+      randomPort(c);
+      GroomServerRunner gsr = new GroomServerRunner(c);
+      groomServerList.add(gsr);
+      scheduler.schedule(gsr, 0, SECONDS);
+      cnt = 0;
+      while(!gsr.isRunning()){
+        LOG.info("Waitin for GroomServer up.");
+        try{
+          Thread.sleep(1000);
+          cnt++;
+          if(10 < cnt){
+            fail("Fail to launch groom server.");
+          }
+        }catch(InterruptedException ie){
+          LOG.error("Fail to check Groom Server's state.", ie);
+          Thread.currentThread().interrupt();
+        }
+      }
+    }
+
+  }
+
+  private void randomPort(HamaConfiguration conf){
+    try{
+      ServerSocket skt = new ServerSocket(0);
+      int p = skt.getLocalPort(); 
+      skt.close();
+      conf.set(Constants.PEER_PORT, new Integer(p).toString());
+      conf.setInt(Constants.GROOM_RPC_PORT, p+100);
+    }catch(IOException ioe){
+      LOG.error("Can not find a free port for BSPPeer.", ioe);
+    }
+  }
+
+  public void shutdown() {
+    shutdownBSPCluster();
+    scheduler.shutdown();
+  }
+
+  public List<Thread> getGroomServerThreads() {
+    List<Thread> list = new ArrayList<Thread>();
+    for(GroomServerRunner gsr: groomServerList){
+      list.add(new Thread(gsr));
+    }
+    return list;
+  }
+
+  public Thread getMaster() {
+    return new Thread(this.master);
+  }
+
+  public List<GroomServer> getGroomServers(){
+    List<GroomServer> list = new ArrayList<GroomServer>();
+    for(GroomServerRunner gsr: groomServerList){
+      list.add(gsr.getGroomServer());
+    }
+    return list;
+  }
+
+  public BSPMaster getBSPMaster(){
+    if(null != this.master)
+      return this.master.getMaster();
+    return null;
+  }
+
+  public ScheduledExecutorService getScheduler(){
+    return this.scheduler;
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMasterGroomServer.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,94 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.bsp;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.Constants;
+import org.apache.hama.HamaCluster;
+import org.apache.hama.HamaConfiguration;
+
+public class TestBSPMasterGroomServer extends HamaCluster {
+
+  private static Log LOG = LogFactory.getLog(TestBSPMasterGroomServer.class);
+  private static String TMP_OUTPUT = "/tmp/test-example/";
+  private HamaConfiguration configuration;
+  private String TEST_JOB = "src/test/java/testjar/testjob.jar";
+
+  public TestBSPMasterGroomServer() {
+    configuration = new HamaConfiguration();
+    configuration.set("bsp.master.address", "localhost");
+    assertEquals("Make sure master addr is set to localhost:", "localhost",
+        configuration.get("bsp.master.address"));
+    configuration.setStrings("bsp.local.dir", "/tmp/hama-test");
+    configuration.set(Constants.ZOOKEEPER_QUORUM, "localhost");
+    configuration.setInt(Constants.ZOOKEEPER_CLIENT_PORT, 21810);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+  }
+
+  public void testSubmitJob() throws Exception {
+    BSPJob bsp = new BSPJob(configuration);
+    bsp.setJobName("Test Serialize Printing");
+    bsp.setBspClass(testjar.ClassSerializePrinting.HelloBSP.class);
+    bsp.setJar(System.getProperty("user.dir")+"/"+TEST_JOB);
+
+    // Set the task size as a number of GroomServer
+    BSPJobClient jobClient = new BSPJobClient(configuration);
+    configuration.setInt(Constants.ZOOKEEPER_SESSION_TIMEOUT, 600);
+    ClusterStatus cluster = jobClient.getClusterStatus(false);
+    assertEquals(this.numOfGroom, cluster.getMaxTasks());
+    
+    // TODO test the multi-tasks 
+    bsp.setNumBspTask(1);
+    
+    FileSystem fileSys = FileSystem.get(conf);
+
+    if (bsp.waitForCompletion(true)) {
+      checkOutput(fileSys, cluster, conf);
+    }
+    LOG.info("Client finishes execution job.");
+  }
+
+  private static void checkOutput(FileSystem fileSys, ClusterStatus cluster,
+      HamaConfiguration conf) throws Exception {
+    for (int i = 0; i < 1; i++) { // TODO test the multi-tasks
+      SequenceFile.Reader reader = new SequenceFile.Reader(fileSys, new Path(
+          TMP_OUTPUT + i), conf);
+      LongWritable timestamp = new LongWritable();
+      Text message = new Text();
+      reader.next(timestamp, message);
+      assertTrue("Check if `Hello BSP' gets printed.", message.toString()
+          .indexOf("Hello BSP from") >= 0);
+      reader.close();
+    }
+  }
+
+  public void tearDown() throws Exception {
+    super.tearDown();
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMessageBundle.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMessageBundle.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMessageBundle.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestBSPMessageBundle.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,81 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.bsp;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import junit.framework.TestCase;
+
+public class TestBSPMessageBundle extends TestCase {
+
+  public void testEmpty() throws IOException {
+    BSPMessageBundle bundle = new BSPMessageBundle();
+    // Serialize it.
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    bundle.write(new DataOutputStream(baos));
+    baos.close();
+    // Deserialize it.
+    BSPMessageBundle readBundle = new BSPMessageBundle();
+    readBundle.readFields(new DataInputStream(new ByteArrayInputStream(baos
+        .toByteArray())));
+    assertEquals(0, readBundle.getMessages().size());
+  }
+
+  public void testSerializationDeserialization() throws IOException {
+    BSPMessageBundle bundle = new BSPMessageBundle();
+    ByteMessage[] testMessages = new ByteMessage[16];
+    for (int i = 0; i < testMessages.length; ++i) {
+      // Create a one byte tag containing the number of the message.
+      byte[] tag = new byte[1];
+      tag[0] = (byte) i;
+      // Create a four bytes data part containing serialized number of the
+      // message.
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      baos.write(i);
+      baos.close();
+      byte[] data = baos.toByteArray();
+      testMessages[i] = new ByteMessage(tag, data);
+      bundle.addMessage(testMessages[i]);
+    }
+    // Serialize it.
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    bundle.write(new DataOutputStream(baos));
+    baos.close();
+    // Deserialize it.
+    BSPMessageBundle readBundle = new BSPMessageBundle();
+    readBundle.readFields(new DataInputStream(new ByteArrayInputStream(baos
+        .toByteArray())));
+    // Check contents.
+    int messageNumber = 0;
+    for (BSPMessage message : readBundle.getMessages()) {
+      ByteMessage byteMessage = (ByteMessage) message;
+      assertTrue(Arrays.equals(testMessages[messageNumber].getTag(),
+          byteMessage.getTag()));
+      assertTrue(Arrays.equals(testMessages[messageNumber].getData(),
+          byteMessage.getData()));
+      ++messageNumber;
+    }
+    assertEquals(testMessages.length, messageNumber);
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestClusterStatus.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestClusterStatus.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestClusterStatus.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestClusterStatus.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.bsp;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hama.bsp.BSPMaster;
+import org.apache.hama.bsp.ClusterStatus;
+
+public class TestClusterStatus extends TestCase {
+  Random rnd = new Random();
+
+  protected void setUp() throws Exception {
+    super.setUp();
+  }
+
+  public final void testWriteAndReadFields() throws IOException {
+    DataOutputBuffer out = new DataOutputBuffer();
+    DataInputBuffer in = new DataInputBuffer();
+
+    ClusterStatus status1;
+    Map<String, String> grooms = new HashMap<String, String>();
+
+    for (int i = 0; i < 10; i++) {
+      int num = rnd.nextInt();
+      String groomName = "groom_" + num;
+      String peerName = "peerhost:" + num;
+      grooms.put(groomName, peerName);
+    }
+
+    int tasks = rnd.nextInt(100);
+    int maxTasks = rnd.nextInt(100);
+    BSPMaster.State state = BSPMaster.State.RUNNING;
+
+    status1 = new ClusterStatus(grooms, tasks, maxTasks, state);
+    status1.write(out);
+
+    in.reset(out.getData(), out.getLength());
+
+    ClusterStatus status2 = new ClusterStatus();
+    status2.readFields(in);
+
+    Map<String, String> grooms_s = new HashMap<String, String>(status1
+        .getActiveGroomNames());
+    Map<String, String> grooms_o = new HashMap<String, String>(status2
+        .getActiveGroomNames());
+
+    assertEquals(status1.getGroomServers(), status2.getGroomServers());
+
+    assertTrue(grooms_s.entrySet().containsAll(grooms_o.entrySet()));
+    assertTrue(grooms_o.entrySet().containsAll(grooms_s.entrySet()));
+
+    assertEquals(status1.getTasks(), status2.getTasks());
+    assertEquals(status1.getMaxTasks(), status2.getMaxTasks());
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestMessages.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestMessages.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestMessages.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestMessages.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.bsp;
+
+import junit.framework.TestCase;
+
+import org.apache.hama.util.Bytes;
+
+public class TestMessages extends TestCase {
+
+  public void testByteMessage() {
+    int dataSize = (int) (Runtime.getRuntime().maxMemory() * 0.60);
+    ByteMessage msg = new ByteMessage(Bytes.toBytes("tag"), new byte[dataSize]);
+    assertEquals(msg.getData().length, dataSize);
+    msg = null;
+    
+    byte[] dummyData = new byte[1024];
+    ByteMessage msg2 = new ByteMessage(Bytes.tail(dummyData, 128), dummyData);
+    assertEquals(
+        Bytes.compareTo(msg2.getTag(), 0, 128, msg2.getData(),
+            msg2.getData().length - 128, 128), 0);
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestIPC.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestIPC.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestIPC.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestIPC.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,232 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.ipc;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Random;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.util.StringUtils;
+
+public class TestIPC extends TestCase {
+  public static final Log LOG = LogFactory.getLog(TestIPC.class);
+
+  final private static Configuration conf = new Configuration();
+  final static private int PING_INTERVAL = 1000;
+
+  static {
+    Client.setPingInterval(conf, PING_INTERVAL);
+  }
+
+  public TestIPC(String name) {
+    super(name);
+  }
+
+  private static final Random RANDOM = new Random();
+
+  private static final String ADDRESS = "0.0.0.0";
+
+  private static class TestServer extends Server {
+    private boolean sleep;
+
+    public TestServer(int handlerCount, boolean sleep) throws IOException {
+      super(ADDRESS, 0, LongWritable.class, handlerCount, conf);
+      this.sleep = sleep;
+    }
+
+    @Override
+    public Writable call(Class<?> protocol, Writable param, long receiveTime)
+        throws IOException {
+      if (sleep) {
+        try {
+          Thread.sleep(RANDOM.nextInt(2 * PING_INTERVAL)); // sleep a bit
+        } catch (InterruptedException e) {
+        }
+      }
+      return param; // echo param as result
+    }
+  }
+
+  private static class SerialCaller extends Thread {
+    private Client client;
+    private InetSocketAddress server;
+    private int count;
+    private boolean failed;
+
+    public SerialCaller(Client client, InetSocketAddress server, int count) {
+      this.client = client;
+      this.server = server;
+      this.count = count;
+    }
+
+    public void run() {
+      for (int i = 0; i < count; i++) {
+        try {
+          LongWritable param = new LongWritable(RANDOM.nextLong());
+          LongWritable value = (LongWritable) client.call(param, server, null,
+              null);
+          if (!param.equals(value)) {
+            LOG.fatal("Call failed!");
+            failed = true;
+            break;
+          }
+        } catch (Exception e) {
+          LOG.fatal("Caught: " + StringUtils.stringifyException(e));
+          failed = true;
+        }
+      }
+    }
+  }
+
+  private static class ParallelCaller extends Thread {
+    private Client client;
+    private int count;
+    private InetSocketAddress[] addresses;
+    private boolean failed;
+
+    public ParallelCaller(Client client, InetSocketAddress[] addresses,
+        int count) {
+      this.client = client;
+      this.addresses = addresses;
+      this.count = count;
+    }
+
+    public void run() {
+      for (int i = 0; i < count; i++) {
+        try {
+          Writable[] params = new Writable[addresses.length];
+          for (int j = 0; j < addresses.length; j++)
+            params[j] = new LongWritable(RANDOM.nextLong());
+          Writable[] values = client.call(params, addresses, null, null);
+          for (int j = 0; j < addresses.length; j++) {
+            if (!params[j].equals(values[j])) {
+              LOG.fatal("Call failed!");
+              failed = true;
+              break;
+            }
+          }
+        } catch (Exception e) {
+          LOG.fatal("Caught: " + StringUtils.stringifyException(e));
+          failed = true;
+        }
+      }
+    }
+  }
+
+  public void testSerial() throws Exception {
+    testSerial(3, false, 2, 5, 100);
+  }
+
+  public void testSerial(int handlerCount, boolean handlerSleep,
+      int clientCount, int callerCount, int callCount) throws Exception {
+    Server server = new TestServer(handlerCount, handlerSleep);
+    InetSocketAddress addr = NetUtils.getConnectAddress(server);
+    server.start();
+
+    Client[] clients = new Client[clientCount];
+    for (int i = 0; i < clientCount; i++) {
+      clients[i] = new Client(LongWritable.class, conf);
+    }
+
+    SerialCaller[] callers = new SerialCaller[callerCount];
+    for (int i = 0; i < callerCount; i++) {
+      callers[i] = new SerialCaller(clients[i % clientCount], addr, callCount);
+      callers[i].start();
+    }
+    for (int i = 0; i < callerCount; i++) {
+      callers[i].join();
+      assertFalse(callers[i].failed);
+    }
+    for (int i = 0; i < clientCount; i++) {
+      clients[i].stop();
+    }
+    server.stop();
+  }
+
+  public void testParallel() throws Exception {
+    testParallel(10, false, 2, 4, 2, 4, 100);
+  }
+
+  public void testParallel(int handlerCount, boolean handlerSleep,
+      int serverCount, int addressCount, int clientCount, int callerCount,
+      int callCount) throws Exception {
+    Server[] servers = new Server[serverCount];
+    for (int i = 0; i < serverCount; i++) {
+      servers[i] = new TestServer(handlerCount, handlerSleep);
+      servers[i].start();
+    }
+
+    InetSocketAddress[] addresses = new InetSocketAddress[addressCount];
+    for (int i = 0; i < addressCount; i++) {
+      addresses[i] = NetUtils.getConnectAddress(servers[i % serverCount]);
+    }
+
+    Client[] clients = new Client[clientCount];
+    for (int i = 0; i < clientCount; i++) {
+      clients[i] = new Client(LongWritable.class, conf);
+    }
+
+    ParallelCaller[] callers = new ParallelCaller[callerCount];
+    for (int i = 0; i < callerCount; i++) {
+      callers[i] = new ParallelCaller(clients[i % clientCount], addresses,
+          callCount);
+      callers[i].start();
+    }
+    for (int i = 0; i < callerCount; i++) {
+      callers[i].join();
+      assertFalse(callers[i].failed);
+    }
+    for (int i = 0; i < clientCount; i++) {
+      clients[i].stop();
+    }
+    for (int i = 0; i < serverCount; i++) {
+      servers[i].stop();
+    }
+  }
+
+  public void testStandAloneClient() throws Exception {
+    testParallel(10, false, 2, 4, 2, 4, 100);
+    Client client = new Client(LongWritable.class, conf);
+    InetSocketAddress address = new InetSocketAddress("127.0.0.1", 10);
+    try {
+      client.call(new LongWritable(RANDOM.nextLong()), address, null, null);
+      fail("Expected an exception to have been thrown");
+    } catch (IOException e) {
+      String message = e.getMessage();
+      String addressText = address.toString();
+      assertTrue("Did not find " + addressText + " in " + message, message
+          .contains(addressText));
+      Throwable cause = e.getCause();
+      assertNotNull("No nested exception in " + e, cause);
+      String causeText = cause.getMessage();
+      assertTrue("Did not find " + causeText + " in " + message, message
+          .contains(causeText));
+    }
+  }
+
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestRPC.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestRPC.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestRPC.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/ipc/TestRPC.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,173 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.ipc;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.ipc.VersionedProtocol;
+
+import junit.framework.TestCase;
+
+public class TestRPC extends TestCase {
+  private static final int PORT = 1234;
+  private static final String ADDRESS = "0.0.0.0";
+
+  public static final Log LOG = LogFactory
+      .getLog("org.apache.hadoop.ipc.TestRPC");
+
+  private static Configuration conf = new Configuration();
+
+  public TestRPC(String name) {
+    super(name);
+  }
+
+  public interface TestProtocol extends VersionedProtocol {
+    public static final long versionID = 1L;
+
+    void ping() throws IOException;
+
+    String echo(String value) throws IOException;
+
+    String[] echo(String[] value) throws IOException;
+
+    Writable echo(Writable value) throws IOException;
+
+    int add(int v1, int v2) throws IOException;
+
+    int add(int[] values) throws IOException;
+
+    int error() throws IOException;
+
+    void testServerGet() throws IOException;
+  }
+
+  public class TestImpl implements TestProtocol {
+
+    public long getProtocolVersion(String protocol, long clientVersion) {
+      return TestProtocol.versionID;
+    }
+
+    public void ping() {
+    }
+
+    public String echo(String value) throws IOException {
+      return value;
+    }
+
+    public String[] echo(String[] values) throws IOException {
+      return values;
+    }
+
+    public Writable echo(Writable writable) {
+      return writable;
+    }
+
+    public int add(int v1, int v2) {
+      return v1 + v2;
+    }
+
+    public int add(int[] values) {
+      int sum = 0;
+      for (int i = 0; i < values.length; i++) {
+        sum += values[i];
+      }
+      return sum;
+    }
+
+    public int error() throws IOException {
+      throw new IOException("bobo");
+    }
+
+    public void testServerGet() throws IOException {
+      if (!(Server.get() instanceof RPC.Server)) {
+        throw new IOException("Server.get() failed");
+      }
+    }
+
+  }
+
+  public void testCalls() throws Exception {
+    Server server = RPC.getServer(new TestImpl(), ADDRESS, PORT, conf);
+    server.start();
+
+    InetSocketAddress addr = new InetSocketAddress(PORT);
+    TestProtocol proxy = (TestProtocol) RPC.getProxy(TestProtocol.class,
+        TestProtocol.versionID, addr, conf);
+
+    proxy.ping();
+
+    String stringResult = proxy.echo("foo");
+    assertEquals(stringResult, "foo");
+
+    stringResult = proxy.echo((String) null);
+    assertEquals(stringResult, null);
+
+    String[] stringResults = proxy.echo(new String[] { "foo", "bar" });
+    assertTrue(Arrays.equals(stringResults, new String[] { "foo", "bar" }));
+
+    stringResults = proxy.echo((String[]) null);
+    assertTrue(Arrays.equals(stringResults, null));
+
+    int intResult = proxy.add(1, 2);
+    assertEquals(intResult, 3);
+
+    intResult = proxy.add(new int[] { 1, 2 });
+    assertEquals(intResult, 3);
+
+    boolean caught = false;
+    try {
+      proxy.error();
+    } catch (IOException e) {
+      LOG.debug("Caught " + e);
+      caught = true;
+    }
+    assertTrue(caught);
+
+    proxy.testServerGet();
+
+    // try some multi-calls
+    Method echo = TestProtocol.class.getMethod("echo",
+        new Class[] { String.class });
+    String[] strings = (String[]) RPC.call(echo, new String[][] { { "a" },
+        { "b" } }, new InetSocketAddress[] { addr, addr }, null, conf);
+    assertTrue(Arrays.equals(strings, new String[] { "a", "b" }));
+
+    Method ping = TestProtocol.class.getMethod("ping", new Class[] {});
+    Object[] voids = (Object[]) RPC.call(ping, new Object[][] { {}, {} },
+        new InetSocketAddress[] { addr, addr }, null, conf);
+    assertEquals(voids, null);
+
+    server.stop();
+  }
+
+  public static void main(String[] args) throws Exception {
+
+    new TestRPC("test").testCalls();
+
+  }
+
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestBytes.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestBytes.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestBytes.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestBytes.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,162 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.util;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import junit.framework.TestCase;
+
+public class TestBytes extends TestCase {
+  public void testNullHashCode() {
+    byte [] b = null;
+    Exception ee = null;
+    try {
+      Bytes.hashCode(b);
+    } catch (Exception e) {
+      ee = e;
+    }
+    assertNotNull(ee);
+  }
+
+  public void testSplit() throws Exception {
+    byte [] lowest = Bytes.toBytes("AAA");
+    byte [] middle = Bytes.toBytes("CCC");
+    byte [] highest = Bytes.toBytes("EEE");
+    byte [][] parts = Bytes.split(lowest, highest, 1);
+    for (int i = 0; i < parts.length; i++) {
+      System.out.println(Bytes.toString(parts[i]));
+    }
+    assertEquals(3, parts.length);
+    assertTrue(Bytes.equals(parts[1], middle));
+    // Now divide into three parts.  Change highest so split is even.
+    highest = Bytes.toBytes("DDD");
+    parts = Bytes.split(lowest, highest, 2);
+    for (int i = 0; i < parts.length; i++) {
+      System.out.println(Bytes.toString(parts[i]));
+    }
+    assertEquals(4, parts.length);
+    // Assert that 3rd part is 'CCC'.
+    assertTrue(Bytes.equals(parts[2], middle));
+  }
+
+  public void testSplit2() throws Exception {
+    // More split tests.
+    byte [] lowest = Bytes.toBytes("http://A");
+    byte [] highest = Bytes.toBytes("http://z");
+    byte [] middle = Bytes.toBytes("http://]");
+    byte [][] parts = Bytes.split(lowest, highest, 1);
+    for (int i = 0; i < parts.length; i++) {
+      System.out.println(Bytes.toString(parts[i]));
+    }
+    assertEquals(3, parts.length);
+    assertTrue(Bytes.equals(parts[1], middle));
+  }
+
+  public void testToLong() throws Exception {
+    long [] longs = {-1l, 123l, 122232323232l};
+    for (int i = 0; i < longs.length; i++) {
+      byte [] b = Bytes.toBytes(longs[i]);
+      assertEquals(longs[i], Bytes.toLong(b));
+    }
+  }
+
+  public void testToFloat() throws Exception {
+    float [] floats = {-1f, 123.123f, Float.MAX_VALUE};
+    for (int i = 0; i < floats.length; i++) {
+      byte [] b = Bytes.toBytes(floats[i]);
+      assertEquals(floats[i], Bytes.toFloat(b));
+    }
+  }
+
+  public void testToDouble() throws Exception {
+    double [] doubles = {Double.MIN_VALUE, Double.MAX_VALUE};
+    for (int i = 0; i < doubles.length; i++) {
+      byte [] b = Bytes.toBytes(doubles[i]);
+      assertEquals(doubles[i], Bytes.toDouble(b));
+    }
+  }
+
+  public void testBinarySearch() throws Exception {
+    byte [][] arr = {
+        {1},
+        {3},
+        {5},
+        {7},
+        {9},
+        {11},
+        {13},
+        {15},
+    };
+    byte [] key1 = {3,1};
+    byte [] key2 = {4,9};
+    byte [] key2_2 = {4};
+    byte [] key3 = {5,11};
+    
+    assertEquals(1, Bytes.binarySearch(arr, key1, 0, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+    assertEquals(0, Bytes.binarySearch(arr, key1, 1, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+    assertEquals(-(2+1), Arrays.binarySearch(arr, key2_2,
+      Bytes.BYTES_COMPARATOR));
+    assertEquals(-(2+1), Bytes.binarySearch(arr, key2, 0, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+    assertEquals(4, Bytes.binarySearch(arr, key2, 1, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+    assertEquals(2, Bytes.binarySearch(arr, key3, 0, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+    assertEquals(5, Bytes.binarySearch(arr, key3, 1, 1,
+      Bytes.BYTES_RAWCOMPARATOR));
+  }
+  
+  public void testIncrementBytes() throws IOException {
+
+    assertTrue(checkTestIncrementBytes(10, 1));
+    assertTrue(checkTestIncrementBytes(12, 123435445));
+    assertTrue(checkTestIncrementBytes(124634654, 1));
+    assertTrue(checkTestIncrementBytes(10005460, 5005645));
+    assertTrue(checkTestIncrementBytes(1, -1));
+    assertTrue(checkTestIncrementBytes(10, -1));
+    assertTrue(checkTestIncrementBytes(10, -5));
+    assertTrue(checkTestIncrementBytes(1005435000, -5));
+    assertTrue(checkTestIncrementBytes(10, -43657655));
+    assertTrue(checkTestIncrementBytes(-1, 1));
+    assertTrue(checkTestIncrementBytes(-26, 5034520));
+    assertTrue(checkTestIncrementBytes(-10657200, 5));
+    assertTrue(checkTestIncrementBytes(-12343250, 45376475));
+    assertTrue(checkTestIncrementBytes(-10, -5));
+    assertTrue(checkTestIncrementBytes(-12343250, -5));
+    assertTrue(checkTestIncrementBytes(-12, -34565445));
+    assertTrue(checkTestIncrementBytes(-1546543452, -34565445));
+  }
+  
+  private static boolean checkTestIncrementBytes(long val, long amount) 
+  throws IOException {
+    byte[] value = Bytes.toBytes(val);
+    byte [] testValue = {-1, -1, -1, -1, -1, -1, -1, -1};
+    if (value[0] > 0) {
+      testValue = new byte[Bytes.SIZEOF_LONG];
+    }
+    System.arraycopy(value, 0, testValue, testValue.length - value.length,
+        value.length);
+
+    long incrementResult = Bytes.toLong(Bytes.incrementBytes(value, amount));
+
+    return (Bytes.toLong(testValue) + amount) == incrementResult;
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestNumeric.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestNumeric.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestNumeric.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestNumeric.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,34 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.util;
+
+import junit.framework.TestCase;
+
+public class TestNumeric extends TestCase {
+  final static int TEST_INT = 3;
+  final static double TEST_DOUBLE = 0.4;
+
+  /**
+   * Double conversion test
+   */
+  public void testDouble() {
+    assertEquals(Bytes.toDouble(Bytes.toBytes(TEST_DOUBLE)), TEST_DOUBLE);
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestRandomVariable.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestRandomVariable.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestRandomVariable.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/util/TestRandomVariable.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,74 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.util;
+
+import org.apache.log4j.Logger;
+
+import junit.framework.TestCase;
+
+/**
+ * Random variable generation test
+ */
+public class TestRandomVariable extends TestCase {
+  static final Logger LOG = Logger.getLogger(TestRandomVariable.class);
+  final static int COUNT = 50;
+
+  /**
+   * Random object test
+   * 
+   * @throws Exception
+   */
+  public void testRand() throws Exception {
+    for (int i = 0; i < COUNT; i++) {
+      double result = RandomVariable.rand();
+      assertTrue(result >= 0.0d && result <= 1.0);
+    }
+  }
+
+  /**
+   * Random integer test
+   * 
+   * @throws Exception
+   */
+  public void testRandInt() throws Exception {
+    final int min = 122;
+    final int max = 561;
+
+    for (int i = 0; i < COUNT; i++) {
+      int result = RandomVariable.randInt(min, max);
+      assertTrue(result >= min && result <= max);
+    }
+  }
+
+  /**
+   * Uniform test
+   * 
+   * @throws Exception
+   */
+  public void testUniform() throws Exception {
+    final double min = 1.0d;
+    final double max = 3.0d;
+
+    for (int i = 0; i < COUNT; i++) {
+      double result = RandomVariable.uniform(min, max);
+      assertTrue(result >= min && result <= max);
+    }
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/org/apache/hama/zookeeper/TestZKTools.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/org/apache/hama/zookeeper/TestZKTools.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/org/apache/hama/zookeeper/TestZKTools.java (added)
+++ incubator/hama/trunk/core/src/test/java/org/apache/hama/zookeeper/TestZKTools.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.zookeeper;
+
+import org.apache.hama.Constants;
+import org.apache.hama.HamaConfiguration;
+
+import junit.framework.TestCase;
+
+public class TestZKTools extends TestCase {
+
+  public void testZKProps() {
+    HamaConfiguration conf = new HamaConfiguration();
+    conf.set(Constants.ZOOKEEPER_QUORUM, "test.com:123");
+    conf.set(Constants.ZOOKEEPER_CLIENT_PORT, "2222");
+
+    assertEquals("test.com:2222", QuorumPeer.getZKQuorumServersString(conf));
+  }
+}

Added: incubator/hama/trunk/core/src/test/java/testjar/ClassSerializePrinting.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/testjar/ClassSerializePrinting.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/core/src/test/java/testjar/ClassSerializePrinting.java (added)
+++ incubator/hama/trunk/core/src/test/java/testjar/ClassSerializePrinting.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package testjar;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.zookeeper.KeeperException;
+
+public class ClassSerializePrinting {
+  private static String TMP_OUTPUT = "/tmp/test-example/";
+
+  public static class HelloBSP extends BSP {
+    public static final Log LOG = LogFactory.getLog(HelloBSP.class);
+    private Configuration conf;
+    private final static int PRINT_INTERVAL = 1000;
+    private FileSystem fileSys;
+    private int num;
+
+    public void bsp(BSPPeer bspPeer) throws IOException,
+        KeeperException, InterruptedException {
+
+      int i = 0;
+      for (String otherPeer : bspPeer.getAllPeerNames()) {
+        String peerName = bspPeer.getPeerName();
+        if (peerName.equals(otherPeer)) {
+          writeLogToFile(peerName, i);
+        }
+
+        Thread.sleep(PRINT_INTERVAL);
+        bspPeer.sync();
+        i++;
+      }
+    }
+
+    private void writeLogToFile(String string, int i) throws IOException {
+      SequenceFile.Writer writer = SequenceFile.createWriter(fileSys, conf,
+          new Path(TMP_OUTPUT + i), LongWritable.class, Text.class,
+          CompressionType.NONE);
+      writer.append(new LongWritable(System.currentTimeMillis()), new Text(
+          "Hello BSP from " + (i + 1) + " of " + num + ": " + string));
+      writer.close();
+    }
+
+    public Configuration getConf() {
+      return conf;
+    }
+
+    public void setConf(Configuration conf) {
+      this.conf = conf;
+      num = Integer.parseInt(conf.get("bsp.peers.num"));
+      try {
+        fileSys = FileSystem.get(conf);
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
+    }
+
+  }
+
+}

Added: incubator/hama/trunk/core/src/test/java/testjar/testjob.jar
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/core/src/test/java/testjar/testjob.jar?rev=1152788&view=auto
==============================================================================
Binary file - no diff available.

Propchange: incubator/hama/trunk/core/src/test/java/testjar/testjob.jar
------------------------------------------------------------------------------
    svn:mime-type = application/octet-stream

Propchange: incubator/hama/trunk/examples/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Mon Aug  1 14:12:46 2011
@@ -0,0 +1,3 @@
+*.iml
+
+target

Added: incubator/hama/trunk/examples/pom.xml
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/examples/pom.xml?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/examples/pom.xml (added)
+++ incubator/hama/trunk/examples/pom.xml Mon Aug  1 14:12:46 2011
@@ -0,0 +1,76 @@
+<!--
+   Licensed to the Apache Software Foundation (ASF) under one or more
+   contributor license agreements.  See the NOTICE file distributed with
+   this work for additional information regarding copyright ownership.
+   The ASF licenses this file to You under the Apache License, Version 2.0
+   (the "License"); you may not use this file except in compliance with
+   the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+  <parent>
+    <groupId>org.apache.hama</groupId>
+    <artifactId>hama-parent</artifactId>
+    <version>0.4.0-incubating-SNAPSHOT</version>
+  </parent>
+
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>org.apache.hama</groupId>
+  <artifactId>hama-examples</artifactId>
+  <name>Apache Hama Examples</name>
+  <version>0.4.0-incubating-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.hama</groupId>
+      <artifactId>hama-core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+  </dependencies>
+  <build>
+    <finalName>hama-examples-${project.version}</finalName>
+    <pluginManagement>
+      <plugins>
+        <plugin>
+          <artifactId>maven-surefire-plugin</artifactId>
+          <configuration>
+            <skipTests>true</skipTests>
+          </configuration>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>1.4</version>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+            <configuration>
+              <transformers>
+                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                  <mainClass>org.apache.hama.examples.ExampleDriver</mainClass>
+                </transformer>
+              </transformers>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>

Added: incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/ExampleDriver.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/ExampleDriver.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/ExampleDriver.java (added)
+++ incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/ExampleDriver.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,42 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples;
+
+import org.apache.hadoop.util.ProgramDriver;
+import org.apache.hama.examples.graph.PageRank;
+import org.apache.hama.examples.graph.ShortestPaths;
+
+public class ExampleDriver {
+
+  public static void main(String[] args) {
+    ProgramDriver pgd = new ProgramDriver();
+    try {
+      pgd.addClass("pi", PiEstimator.class, "Pi Estimator");
+      pgd.addClass("bench", RandBench.class, "Random Communication Benchmark");
+      pgd.addClass("test", SerializePrinting.class, "Serialize Printing Test");
+      pgd.addClass("sssp", ShortestPaths.class, "Single Source Shortest Path");
+      pgd.addClass("pagerank", PageRank.class, "PageRank");
+      
+      pgd.driver(args);
+    } catch (Throwable e) {
+      e.printStackTrace();
+    }
+  }
+}

Added: incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/PiEstimator.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/PiEstimator.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/PiEstimator.java (added)
+++ incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/PiEstimator.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,158 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPJobClient;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.hama.bsp.ClusterStatus;
+import org.apache.hama.bsp.DoubleMessage;
+import org.apache.zookeeper.KeeperException;
+
+public class PiEstimator {
+  private static String MASTER_TASK = "master.task.";
+  private static Path TMP_OUTPUT = new Path("/tmp/pi-example/output");
+
+  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;
+
+    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++;
+        }
+      }
+
+      double data = 4.0 * (double) in / (double) iterations;
+      DoubleMessage estimate = new DoubleMessage(bspPeer.getPeerName(), data);
+
+      bspPeer.send(masterTask, estimate);
+      bspPeer.sync();
+
+      if (bspPeer.getPeerName().equals(masterTask)) {
+        double pi = 0.0;
+        int numPeers = bspPeer.getNumCurrentMessages();
+        DoubleMessage received;
+        while ((received = (DoubleMessage) bspPeer.getCurrentMessage()) != null) {
+          pi += received.getData();
+        }
+
+        pi = pi / numPeers;
+        writeResult(pi);
+      }
+    }
+
+    private void writeResult(double pi) throws IOException {
+      FileSystem fileSys = FileSystem.get(conf);
+
+      SequenceFile.Writer writer = SequenceFile.createWriter(fileSys, conf,
+          TMP_OUTPUT, DoubleWritable.class, DoubleWritable.class,
+          CompressionType.NONE);
+      writer.append(new DoubleWritable(pi), new DoubleWritable(0));
+      writer.close();
+    }
+
+    public Configuration getConf() {
+      return conf;
+    }
+
+    public void setConf(Configuration conf) {
+      this.conf = conf;
+      this.masterTask = conf.get(MASTER_TASK);
+    }
+
+  }
+
+  private static void initTempDir(FileSystem fileSys) throws IOException {
+    if (fileSys.exists(TMP_OUTPUT)) {
+      fileSys.delete(TMP_OUTPUT, true);
+    }
+  }
+
+  private static void printOutput(FileSystem fileSys, HamaConfiguration conf)
+      throws IOException {
+    SequenceFile.Reader reader = new SequenceFile.Reader(fileSys, TMP_OUTPUT,
+        conf);
+    DoubleWritable output = new DoubleWritable();
+    DoubleWritable zero = new DoubleWritable();
+    reader.next(output, zero);
+    reader.close();
+
+    System.out.println("Estimated value of PI is " + output);
+  }
+
+  public static void main(String[] args) throws InterruptedException,
+      IOException, ClassNotFoundException {
+    // BSP job configuration
+    HamaConfiguration conf = new HamaConfiguration();
+
+    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);
+
+    if (args.length > 0) {
+      bsp.setNumBspTask(Integer.parseInt(args[0]));
+    } else {
+      // Set to maximum
+      bsp.setNumBspTask(cluster.getMaxTasks());
+    }
+
+    // Choose one as a master
+    for (String peerName : cluster.getActiveGroomNames().values()) {
+      conf.set(MASTER_TASK, peerName);
+      break;
+    }
+
+    FileSystem fileSys = FileSystem.get(conf);
+    initTempDir(fileSys);
+
+    long startTime = System.currentTimeMillis();
+
+    if (bsp.waitForCompletion(true)) {
+      printOutput(fileSys, conf);
+
+      System.out.println("Job Finished in "
+          + (double) (System.currentTimeMillis() - startTime) / 1000.0
+          + " seconds");
+    }
+  }
+}

Added: incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/RandBench.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/RandBench.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/RandBench.java (added)
+++ incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/RandBench.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples;
+
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPJobClient;
+import org.apache.hama.bsp.BSPMessage;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.hama.bsp.ByteMessage;
+import org.apache.hama.bsp.ClusterStatus;
+import org.apache.hama.util.Bytes;
+import org.apache.zookeeper.KeeperException;
+
+public class RandBench {
+  private static final String SIZEOFMSG = "msg.size";
+  private static final String N_COMMUNICATIONS = "communications.num";
+  private static final String N_SUPERSTEPS = "supersteps.num";
+
+  public static class RandBSP extends BSP {
+    public static final Log LOG = LogFactory.getLog(RandBSP.class);
+    private Configuration conf;
+    private Random r = new Random();
+    private int sizeOfMsg;
+    private int nCommunications;
+    private int nSupersteps;
+
+    @Override
+    public void bsp(BSPPeer bspPeer) throws IOException,
+        KeeperException, InterruptedException {
+      byte[] dummyData = new byte[sizeOfMsg];
+      BSPMessage msg = null;
+      String[] peers = bspPeer.getAllPeerNames();
+      String peerName = bspPeer.getPeerName();
+
+      for (int i = 0; i < nSupersteps; i++) {
+
+        for (int j = 0; j < nCommunications; j++) {
+          String tPeer = peers[r.nextInt(peers.length)];
+          String tag = peerName + " to " + tPeer;
+          msg = new ByteMessage(Bytes.toBytes(tag), dummyData);
+          bspPeer.send(tPeer, msg);
+        }
+
+        bspPeer.sync();
+
+        ByteMessage received;
+        while ((received = (ByteMessage) bspPeer.getCurrentMessage()) != null) {
+          LOG.info(Bytes.toString(received.getTag()) + " : "
+              + received.getData().length);
+        }
+
+      }
+    }
+
+    @Override
+    public void setConf(Configuration conf) {
+      this.conf = conf;
+      this.sizeOfMsg = conf.getInt(SIZEOFMSG, 1);
+      this.nCommunications = conf.getInt(N_COMMUNICATIONS, 1);
+      this.nSupersteps = conf.getInt(N_SUPERSTEPS, 1);
+    }
+
+    @Override
+    public Configuration getConf() {
+      return conf;
+    }
+
+  }
+
+  public static void main(String[] args) throws Exception {
+    if (args.length < 3) {
+      System.out.println("Usage: <sizeOfMsg> <nCommunications> <nSupersteps>");
+      System.exit(-1);
+    }
+
+    // BSP job configuration
+    HamaConfiguration conf = new HamaConfiguration();
+
+    conf.setInt(SIZEOFMSG, Integer.parseInt(args[0]));
+    conf.setInt(N_COMMUNICATIONS, Integer.parseInt(args[1]));
+    conf.setInt(N_SUPERSTEPS, Integer.parseInt(args[2]));
+
+    BSPJob bsp = new BSPJob(conf, RandBench.class);
+    // Set the job name
+    bsp.setJobName("Random Communication Benchmark");
+    bsp.setBspClass(RandBSP.class);
+
+    // Set the task size as a number of GroomServer
+    BSPJobClient jobClient = new BSPJobClient(conf);
+    ClusterStatus cluster = jobClient.getClusterStatus(false);
+    bsp.setNumBspTask(cluster.getMaxTasks());
+
+    long startTime = System.currentTimeMillis();
+    bsp.waitForCompletion(true);
+    System.out.println("Job Finished in "
+        + (double) (System.currentTimeMillis() - startTime) / 1000.0
+        + " seconds");
+  }
+}

Added: incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/SerializePrinting.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/SerializePrinting.java?rev=1152788&view=auto
==============================================================================
--- incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/SerializePrinting.java (added)
+++ incubator/hama/trunk/examples/src/main/java/org/apache/hama/examples/SerializePrinting.java Mon Aug  1 14:12:46 2011
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples;
+
+import java.io.IOException;
+import java.util.Date;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPJobClient;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.hama.bsp.ClusterStatus;
+import org.apache.zookeeper.KeeperException;
+
+public class SerializePrinting {
+  private static String TMP_OUTPUT = "/tmp/test-example/";
+
+  public static class HelloBSP extends BSP {
+    public static final Log LOG = LogFactory.getLog(HelloBSP.class);
+    private Configuration conf;
+    private final static int PRINT_INTERVAL = 1000;
+    private FileSystem fileSys;
+    private int num;
+
+    public void bsp(BSPPeer bspPeer) throws IOException,
+        KeeperException, InterruptedException {
+
+      int i = 0;
+      for (String otherPeer : bspPeer.getAllPeerNames()) {
+        String peerName = bspPeer.getPeerName();
+        if (peerName.equals(otherPeer)) {
+          writeLogToFile(peerName, i);
+        }
+
+        Thread.sleep(PRINT_INTERVAL);
+        bspPeer.sync();
+        i++;
+      }
+    }
+
+    private void writeLogToFile(String string, int i) throws IOException {
+      SequenceFile.Writer writer = SequenceFile.createWriter(fileSys, conf,
+          new Path(TMP_OUTPUT + i), LongWritable.class, Text.class,
+          CompressionType.NONE);
+      writer.append(new LongWritable(System.currentTimeMillis()), new Text(
+          "Hello BSP from " + (i + 1) + " of " + num + ": " + string));
+      writer.close();
+    }
+
+    public Configuration getConf() {
+      return conf;
+    }
+
+    public void setConf(Configuration conf) {
+      this.conf = conf;
+      num = Integer.parseInt(conf.get("bsp.peers.num"));
+      try {
+        fileSys = FileSystem.get(conf);
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
+    }
+
+  }
+
+  private static void printOutput(FileSystem fileSys, ClusterStatus cluster,
+      HamaConfiguration conf) throws IOException {
+    System.out.println("Each task printed the \"Hello World\" as below:");
+    for (int i = 0; i < cluster.getGroomServers(); i++) {
+      SequenceFile.Reader reader = new SequenceFile.Reader(fileSys, new Path(
+          TMP_OUTPUT + i), conf);
+      LongWritable timestamp = new LongWritable();
+      Text message = new Text();
+      reader.next(timestamp, message);
+      System.out.println(new Date(timestamp.get()) + ": " + message);
+      reader.close();
+    }
+  }
+
+  private static void initTempDir(FileSystem fileSys) throws IOException {
+    if (fileSys.exists(new Path(TMP_OUTPUT))) {
+      fileSys.delete(new Path(TMP_OUTPUT), true);
+    }
+  }
+
+  public static void main(String[] args) throws InterruptedException,
+      IOException, ClassNotFoundException {
+    // BSP job configuration
+    HamaConfiguration conf = new HamaConfiguration();
+
+    BSPJob bsp = new BSPJob(conf, SerializePrinting.class);
+    // Set the job name
+    bsp.setJobName("Serialize Printing");
+    bsp.setBspClass(HelloBSP.class);
+
+    // Set the task size as a number of GroomServer
+    BSPJobClient jobClient = new BSPJobClient(conf);
+    ClusterStatus cluster = jobClient.getClusterStatus(false);
+    bsp.setNumBspTask(cluster.getGroomServers());
+
+    FileSystem fileSys = FileSystem.get(conf);
+    initTempDir(fileSys);
+
+    if (bsp.waitForCompletion(true)) {
+      printOutput(fileSys, cluster, conf);
+    }
+  }
+
+}



Mime
View raw message