hama-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From edwardy...@apache.org
Subject svn commit: r1061087 - in /incubator/hama/trunk: ./ conf/ src/examples/org/apache/hama/examples/ src/java/org/apache/hama/bsp/ src/test/org/apache/hama/bsp/
Date Thu, 20 Jan 2011 01:10:18 GMT
Author: edwardyoon
Date: Thu Jan 20 01:10:17 2011
New Revision: 1061087

URL: http://svn.apache.org/viewvc?rev=1061087&view=rev
Log:
Add implementation of umbilical interface

Added:
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerProtocol.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskRunner.java
Modified:
    incubator/hama/trunk/CHANGES.txt
    incubator/hama/trunk/conf/hama-default.xml
    incubator/hama/trunk/src/examples/org/apache/hama/examples/PiEstimator.java
    incubator/hama/trunk/src/examples/org/apache/hama/examples/SerializePrinting.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPInterface.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPMaster.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeer.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerInterface.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTask.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTaskRunner.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/GroomServer.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/Task.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskAttemptID.java
    incubator/hama/trunk/src/test/org/apache/hama/bsp/Client.java
    incubator/hama/trunk/src/test/org/apache/hama/bsp/TestBSPMaster.java

Modified: incubator/hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/CHANGES.txt?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/CHANGES.txt (original)
+++ incubator/hama/trunk/CHANGES.txt Thu Jan 20 01:10:17 2011
@@ -51,6 +51,7 @@ Trunk (unreleased changes)
 
   IMPROVEMENTS
  
+    HAMA-347: Add implementation of umbilical interface (edwardyoon)
     HAMA-346: Modify MniCluster so that developers can benefit when testing using Junit
                        (ChiaHung Lin via edwardyoon)
     HAMA-340: Implementation of job submit command (edwardyoon)   

Modified: incubator/hama/trunk/conf/hama-default.xml
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/conf/hama-default.xml?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/conf/hama-default.xml (original)
+++ incubator/hama/trunk/conf/hama-default.xml Thu Jan 20 01:10:17 2011
@@ -51,6 +51,14 @@
     <description>local directory for temporal store</description> 
   </property>
   <property>
+  <name>bsp.groom.report.address</name>
+  <value>127.0.0.1:0</value>
+  <description>The interface and port that groom server listens on. 
+  Since it is only connected to by the tasks, it uses the local interface.
+  EXPERT ONLY. Should only be changed if your host does not have the loopback 
+  interface.</description>
+</property>
+  <property>
     <name>bsp.system.dir</name>
     <value>${hadoop.tmp.dir}/bsp/system</value>
     <description>The shared directory where BSP stores control files.

Modified: incubator/hama/trunk/src/examples/org/apache/hama/examples/PiEstimator.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/PiEstimator.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/PiEstimator.java (original)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/PiEstimator.java Thu Jan 20
01:10:17 2011
@@ -27,8 +27,8 @@ 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.ClusterStatus;
+import org.apache.hama.bsp.BSPPeerProtocol;
 import org.apache.hama.util.Bytes;
 import org.apache.zookeeper.KeeperException;
 
@@ -41,8 +41,7 @@ public class PiEstimator {
     private String masterTask;
     private static final int iterations = 10000;
 
-    @Override
-    public void bsp(BSPPeer bspPeer) throws IOException, KeeperException,
+    public void bsp(BSPPeerProtocol bspPeer) throws IOException, KeeperException,
         InterruptedException {
       int in = 0, out = 0;
       for (int i = 0; i < iterations; i++) {
@@ -77,12 +76,10 @@ public class PiEstimator {
       }
     }
 
-    @Override
     public Configuration getConf() {
       return conf;
     }
 
-    @Override
     public void setConf(Configuration conf) {
       this.conf = conf;
       this.masterTask = conf.get(MASTER_TASK);

Modified: incubator/hama/trunk/src/examples/org/apache/hama/examples/SerializePrinting.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/SerializePrinting.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/SerializePrinting.java (original)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/SerializePrinting.java Thu
Jan 20 01:10:17 2011
@@ -26,7 +26,7 @@ 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.BSPPeerProtocol;
 import org.apache.hama.bsp.ClusterStatus;
 import org.apache.zookeeper.KeeperException;
 
@@ -37,8 +37,7 @@ public class SerializePrinting {
     private Configuration conf;
     private final static int PRINT_INTERVAL = 5000;
 
-    @Override
-    public void bsp(BSPPeer bspPeer) throws IOException, KeeperException,
+    public void bsp(BSPPeerProtocol bspPeer) throws IOException, KeeperException,
         InterruptedException {
       int num = Integer.parseInt(conf.get("bsp.peers.num"));
 
@@ -55,12 +54,10 @@ public class SerializePrinting {
       }
     }
 
-    @Override
     public Configuration getConf() {
       return conf;
     }
 
-    @Override
     public void setConf(Configuration conf) {
       this.conf = conf;
     }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPInterface.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPInterface.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPInterface.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPInterface.java Thu Jan 20 01:10:17
2011
@@ -39,7 +39,7 @@ public interface BSPInterface extends Co
    * @throws KeeperException
    * @throws InterruptedException
    */
-  public void bsp(BSPPeer bspPeer) throws IOException, KeeperException,
+  public void bsp(BSPPeerProtocol bspPeer) throws IOException, KeeperException,
       InterruptedException;
 
 }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPMaster.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPMaster.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPMaster.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPMaster.java Thu Jan 20 01:10:17 2011
@@ -254,20 +254,24 @@ public class BSPMaster implements JobSub
           // TODO: need for each tip execute completed?
           // each tip already maintain a data structure, checking
           // if task status is completed
-          TaskInProgress tip = jip.findTaskInProgress(((TaskAttemptID) ts
-              .getTaskId()).getTaskID());
-          jip.completedTask(tip, ts);
-          LOG.info("JobInProgress id:" + jip.getJobID() + " status:"
-              + jip.getStatus());
-          if (jip.getStatus().getRunState() == JobStatus.SUCCEEDED) {
-            for (JobInProgressListener listener : jobInProgressListeners) {
-              try {
-                listener.jobRemoved(jip);
-              } catch (IOException ioe) {
-                LOG.error("Fail to alter scheduler a job is moved.", ioe);
+
+          if (jip != null) { // passes if jip is null
+            TaskInProgress tip = jip.findTaskInProgress(((TaskAttemptID) ts
+                .getTaskId()).getTaskID());
+            jip.completedTask(tip, ts);
+            LOG.info("JobInProgress id:" + jip.getJobID() + " status:"
+                + jip.getStatus());
+            if (jip.getStatus().getRunState() == JobStatus.SUCCEEDED) {
+              for (JobInProgressListener listener : jobInProgressListeners) {
+                try {
+                  listener.jobRemoved(jip);
+                } catch (IOException ioe) {
+                  LOG.error("Fail to alter scheduler a job is moved.", ioe);
+                }
               }
             }
           }
+
         }
       } else {
         throw new RuntimeException("BSPMaster contains GroomServerSatus, "

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeer.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeer.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeer.java Thu Jan 20 01:10:17 2011
@@ -24,8 +24,8 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Set;
+import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
@@ -286,8 +286,8 @@ public class BSPPeer implements Watcher,
   }
 
   @Override
-  public Set<String> getAllPeerNames() {
-    return allPeerNames;
+  public String[] getAllPeerNames() {
+    return allPeerNames.toArray(new String[0]);
   }
 
   /**

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerInterface.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerInterface.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerInterface.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerInterface.java Thu Jan 20 01:10:17
2011
@@ -19,7 +19,6 @@ package org.apache.hama.bsp;
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Set;
 
 import org.apache.hama.Constants;
 import org.apache.zookeeper.KeeperException;
@@ -72,5 +71,5 @@ public interface BSPPeerInterface extend
   /**
    * @return The names of all the peers executing tasks from the same job (including this
peer).
    */
-  public Set<String> getAllPeerNames();
+  public String[] getAllPeerNames();
 }

Added: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerProtocol.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerProtocol.java?rev=1061087&view=auto
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerProtocol.java (added)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPPeerProtocol.java Thu Jan 20 01:10:17
2011
@@ -0,0 +1,48 @@
+/**
+ * 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;
+
+/**
+ */
+public interface BSPPeerProtocol extends BSPPeerInterface {
+
+  /** Called when a child task process starts, to get its task. */
+  Task getTask(TaskAttemptID taskid) throws IOException;
+
+  /**
+   * Periodically called by child to check if parent is still alive.
+   * 
+   * @return True if the task is known
+   */
+  boolean ping(TaskAttemptID taskid) throws IOException;
+
+  /**
+   * Report that the task is successfully completed. Failure is assumed if the
+   * task process exits without calling this.
+   * 
+   * @param taskid task's id
+   * @param shouldBePromoted whether to promote the task's output or not
+   */
+  void done(TaskAttemptID taskid, boolean shouldBePromoted) throws IOException;
+
+  /** Report that the task encounted a local filesystem error. */
+  void fsError(TaskAttemptID taskId, String message) throws IOException;
+
+}

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTask.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTask.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTask.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTask.java Thu Jan 20 01:10:17 2011
@@ -17,8 +17,14 @@
  */
 package org.apache.hama.bsp;
 
-public class BSPTask extends Task {
+import java.io.IOException;
+
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.zookeeper.KeeperException;
 
+public class BSPTask extends Task {
+  private BSPJob conf;
+  
   public BSPTask() {
   }
 
@@ -30,8 +36,39 @@ public class BSPTask extends Task {
   }
 
   @Override
-  public BSPTaskRunner createRunner(BSPPeer bspPeer, BSPJob conf) {
-    return new BSPTaskRunner(this, bspPeer, conf);
+  public BSPTaskRunner createRunner(GroomServer groom) {
+    return new BSPTaskRunner(this, groom, this.conf);
+  }
+
+  @Override
+  public void run(BSPJob job, BSPPeerProtocol umbilical)
+      throws IOException {
+    
+    BSP bsp = (BSP) ReflectionUtils.newInstance(job.getConf().getClass(
+        "bsp.work.class", BSP.class), job.getConf());
+
+    try {
+      bsp.bsp(umbilical);
+    } catch (IOException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    } catch (KeeperException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    } catch (InterruptedException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    }
+
+    done(umbilical);
   }
+  
+  public BSPJob getConf() {
+      return conf;
+    }
+  
+    public void setConf(BSPJob conf) {
+      this.conf = conf;
+    }
 
 }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTaskRunner.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTaskRunner.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTaskRunner.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BSPTaskRunner.java Thu Jan 20 01:10:17
2011
@@ -17,58 +17,15 @@
  */
 package org.apache.hama.bsp;
 
-import java.io.IOException;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.zookeeper.KeeperException;
 
-public class BSPTaskRunner extends Thread {
+public class BSPTaskRunner extends TaskRunner {
 
   public static final Log LOG = LogFactory.getLog(BSPTaskRunner.class);
-  private Task task;
-  private BSPJob conf;
-  private BSPPeer bspPeer;
-
-  public BSPTaskRunner(BSPTask bspTask, BSPPeer bspPeer, BSPJob conf) {
-    this.task = bspTask;
-    this.conf = conf;
-    this.bspPeer = bspPeer;
-  }
-
-  public Task getTask() {
-    return task;
-  }
-
-  public void run() {
-    BSP bsp = (BSP) ReflectionUtils.newInstance(conf.getConf().getClass(
-        "bsp.work.class", BSP.class), conf.getConf());
-
-    try {
-      bsp.bsp(bspPeer);
-    } catch (IOException e) {
-      // TODO Auto-generated catch block
-      e.printStackTrace();
-    } catch (KeeperException e) {
-      // TODO Auto-generated catch block
-      e.printStackTrace();
-    } catch (InterruptedException e) {
-      // TODO Auto-generated catch block
-      e.printStackTrace();
-    } finally {
-      try {
-        finalize();
-      } catch (Throwable e) {
-        // TODO Auto-generated catch block
-        e.printStackTrace();
-      }
-    }
-  }
 
-  public void kill() {
-    // TODO Auto-generated method stub
-    LOG.debug(">>>> Kill Task Runner");
+  public BSPTaskRunner(BSPTask bspTask, GroomServer groom, BSPJob conf) {
+    super(bspTask, groom, conf);
   }
 
 }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/GroomServer.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/GroomServer.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/GroomServer.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/GroomServer.java Thu Jan 20 01:10:17
2011
@@ -17,8 +17,10 @@
  */
 package org.apache.hama.bsp;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.lang.reflect.Constructor;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
@@ -37,12 +39,16 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.FSError;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.metrics.MetricsContext;
+import org.apache.hadoop.metrics.MetricsUtil;
 import org.apache.hadoop.net.DNS;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.util.DiskChecker;
 import org.apache.hadoop.util.RunJar;
 import org.apache.hadoop.util.StringUtils;
@@ -51,8 +57,10 @@ import org.apache.hama.Constants;
 import org.apache.hama.HamaConfiguration;
 import org.apache.hama.ipc.MasterProtocol;
 import org.apache.hama.ipc.WorkerProtocol;
+import org.apache.log4j.LogManager;
+import org.apache.zookeeper.KeeperException;
 
-public class GroomServer implements Runnable, WorkerProtocol {
+public class GroomServer implements Runnable, WorkerProtocol, BSPPeerProtocol {
 
   public static final Log LOG = LogFactory.getLog(GroomServer.class);
   private BSPPeer bspPeer;
@@ -100,6 +108,9 @@ public class GroomServer implements Runn
   private Server workerServer;
   MasterProtocol masterClient;
 
+  InetSocketAddress taskReportAddress;
+  Server taskReportServer = null;
+
   private BlockingQueue<GroomServerAction> tasksToCleanup = new LinkedBlockingQueue<GroomServerAction>();
 
   public GroomServer(Configuration conf) throws IOException {
@@ -149,9 +160,30 @@ public class GroomServer implements Runn
       this.workerServer = RPC.getServer(this, rpcAddr, rpcPort, conf);
       this.workerServer.start();
       this.rpcServer = rpcAddr + ":" + rpcPort;
+
       LOG.info("Worker rpc server --> " + rpcServer);
     }
 
+    String address = NetUtils.getServerAddress(conf,
+        "bsp.groom.report.bindAddress", "bsp.groom.report.port",
+        "bsp.groom.report.address");
+    InetSocketAddress socAddr = NetUtils.createSocketAddr(address);
+    String bindAddress = socAddr.getHostName();
+    int tmpPort = socAddr.getPort();
+
+    // RPC initialization
+    // TODO numHandlers should be a ..
+    this.taskReportServer = RPC.getServer(this, bindAddress, tmpPort, 10,
+        false, this.conf);
+
+    this.taskReportServer.start();
+
+    // get the assigned address
+    this.taskReportAddress = taskReportServer.getListenerAddress();
+    this.conf.set("bsp.groom.report.address", taskReportAddress.getHostName()
+        + ":" + taskReportAddress.getPort());
+    LOG.info("GroomServer up at: " + this.taskReportAddress);
+
     this.groomServerName = "groomd_" + bspPeer.getPeerName().replace(':', '_');
     LOG.info("Starting groom: " + this.groomServerName);
 
@@ -177,15 +209,18 @@ public class GroomServer implements Runn
     this.initialized = true;
   }
 
+  /** Return the port at which the tasktracker bound to */
+  public synchronized InetSocketAddress getTaskTrackerReportAddress() {
+    return taskReportAddress;
+  }
+
   @Override
   public void dispatch(Directive directive) throws IOException {
     // update tasks status
     GroomServerAction[] actions = directive.getActions();
     bspPeer.setAllPeerNames(directive.getGroomServerPeers().values());
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("Got Response from BSPMaster with "
-          + ((actions != null) ? actions.length : 0) + " actions");
-    }
+    LOG.debug("Got Response from BSPMaster with "
+        + ((actions != null) ? actions.length : 0) + " actions");
     // perform actions
     if (actions != null) {
       for (GroomServerAction action : actions) {
@@ -216,7 +251,7 @@ public class GroomServer implements Runn
           DiskChecker.checkDir(new File(localDirs[i]));
           writable = true;
         } catch (DiskErrorException e) {
-          LOG.warn("Graph Processor local " + e.getMessage());
+          LOG.warn("BSP Processor local " + e.getMessage());
         }
       }
     }
@@ -288,11 +323,19 @@ public class GroomServer implements Runn
   }
 
   private void startNewTask(LaunchTaskAction action) {
-    TaskInProgress tip = new TaskInProgress(action.getTask(),
-        this.groomServerName);
+    Task t = action.getTask();
+    BSPJob jobConf = null;
+    try {
+      jobConf = new BSPJob(t.getJobID(), t.getJobFile());
+    } catch (IOException e1) {
+      LOG.error(e1);
+    }
+
+    TaskInProgress tip = new TaskInProgress(t, jobConf, this.groomServerName);
 
     synchronized (this) {
-      runningTasks.put(action.getTask().getTaskID(), tip);
+      tasks.put(t.getTaskID(), tip);
+      runningTasks.put(t.getTaskID(), tip);
     }
 
     try {
@@ -319,11 +362,12 @@ public class GroomServer implements Runn
         Path localJarFile = defaultJobConf.getLocalPath(SUBDIR + "/"
             + task.getTaskID() + "/" + "job.jar");
         systemFS.copyToLocalFile(new Path(task.getJobFile()), localJobFile);
-        Path jarFile = new Path(task.getJobFile().replace(".xml", ".jar"));
-
+        
         HamaConfiguration conf = new HamaConfiguration();
         conf.addResource(localJobFile);
         jobConf = new BSPJob(conf, task.getJobID().toString());
+        
+        Path jarFile = new Path(jobConf.getJar());
         jobConf.setJar(localJarFile.toString());
 
         if (jarFile != null) {
@@ -343,6 +387,7 @@ public class GroomServer implements Runn
         rjob.localized = true;
       }
     }
+
     launchTaskForJob(tip, jobConf);
   }
 
@@ -468,6 +513,11 @@ public class GroomServer implements Runn
     cleanupStorage();
     this.workerServer.stop();
     RPC.stopProxy(masterClient);
+
+    if (taskReportServer != null) {
+      taskReportServer.stop();
+      taskReportServer = null;
+    }
   }
 
   public static Thread startGroomServer(final GroomServer hrs) {
@@ -489,27 +539,60 @@ public class GroomServer implements Runn
   class TaskInProgress {
     Task task;
     BSPJob jobConf;
+    BSPJob localJobConf;
     BSPTaskRunner runner;
     volatile boolean done = false;
     volatile boolean wasKilled = false;
     private TaskStatus taskStatus;
 
-    public TaskInProgress(Task task, String groomServer) {
+    public TaskInProgress(Task task, BSPJob jobConf, String groomServer) {
       this.task = task;
+      this.jobConf = jobConf;
+      this.localJobConf = null;
       this.taskStatus = new TaskStatus(task.getJobID(), task.getTaskID(), 0,
           TaskStatus.State.UNASSIGNED, "running", groomServer,
           TaskStatus.Phase.STARTING);
     }
 
-    public void setJobConf(BSPJob jobConf) {
+    private void localizeTask(Task task) throws IOException {
+      Path localJobFile = this.jobConf.getLocalPath(SUBDIR + "/"
+          + task.getTaskID() + "/job.xml");
+      Path localJarFile = this.jobConf.getLocalPath(SUBDIR + "/"
+          + task.getTaskID() + "/job.jar");
+
+      String jobFile = task.getJobFile();
+      systemFS.copyToLocalFile(new Path(jobFile), localJobFile);
+      task.setJobFile(localJobFile.toString());
+
+      localJobConf = new BSPJob(task.getJobID(), localJobFile.toString());
+      localJobConf.set("bsp.task.id", task.getTaskID().toString());
+      String jarFile = localJobConf.getJar();
+
+      if (jarFile != null) {
+        systemFS.copyToLocalFile(new Path(jarFile), localJarFile);
+        localJobConf.setJar(localJarFile.toString());
+      }
+
+      LOG.debug("localizeTask : " + localJobConf.getJar());
+      LOG.debug("localizeTask : " + localJobFile.toString());
+
+      task.setConf(localJobConf);
+    }
+
+    public synchronized void setJobConf(BSPJob jobConf) {
       this.jobConf = jobConf;
     }
 
+    public synchronized BSPJob getJobConf() {
+      return localJobConf;
+    }
+
     public void launchTask() throws IOException {
+      localizeTask(task);
       taskStatus.setRunState(TaskStatus.State.RUNNING);
       bspPeer.setJobConf(jobConf);
       bspPeer.setCurrentTaskStatus(taskStatus);
-      this.runner = task.createRunner(bspPeer, this.jobConf);
+      this.runner = task.createRunner(GroomServer.this);
       this.runner.start();
 
       // Check state of a Task
@@ -626,6 +709,8 @@ public class GroomServer implements Runn
       throws IOException {
     if (protocol.equals(WorkerProtocol.class.getName())) {
       return WorkerProtocol.versionID;
+    } else if (protocol.equals(BSPPeerProtocol.class.getName())) {
+      return BSPPeerProtocol.versionID;
     } else {
       throw new IOException("Unknown protocol to GroomServer: " + protocol);
     }
@@ -637,9 +722,128 @@ public class GroomServer implements Runn
    * @return bsp peer information in the form of "address:port".
    */
   public String getBspPeerName() {
-    if (null != this.bspPeer)
-      return this.bspPeer.getPeerName();
+    if (null != bspPeer)
+      return bspPeer.getPeerName();
     return null;
   }
 
+  /**
+   * The main() for child processes.
+   */
+  public static class Child {
+
+    public static void main(String[] args) throws Throwable {
+      LOG.debug("Child starting");
+
+      HamaConfiguration defaultConf = new HamaConfiguration();
+      // report address
+      String host = args[0];
+      int port = Integer.parseInt(args[1]);
+      InetSocketAddress address = new InetSocketAddress(host, port);
+      TaskAttemptID taskid = TaskAttemptID.forName(args[2]);
+
+      // //////////////////
+      BSPPeerProtocol umbilical = (BSPPeerProtocol) RPC.getProxy(
+          BSPPeerProtocol.class, BSPPeerProtocol.versionID, address,
+          defaultConf);
+
+      Task task = umbilical.getTask(taskid);
+
+      defaultConf.addResource(new Path(task.getJobFile()));
+      BSPJob job = new BSPJob(task.getJobID(), task.getJobFile());
+
+      try {
+        // use job-specified working directory
+        FileSystem.get(job.getConf()).setWorkingDirectory(
+            job.getWorkingDirectory());
+
+        task.run(job, umbilical); // run the task
+      } catch (FSError e) {
+        LOG.fatal("FSError from child", e);
+        umbilical.fsError(taskid, e.getMessage());
+      } catch (Throwable throwable) {
+        LOG.warn("Error running child", throwable);
+        // Report back any failures, for diagnostic purposes
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        throwable.printStackTrace(new PrintStream(baos));
+      } finally {
+        RPC.stopProxy(umbilical);
+        MetricsContext metricsContext = MetricsUtil.getContext("mapred");
+        metricsContext.close();
+        // Shutting down log4j of the child-vm...
+        // This assumes that on return from Task.run()
+        // there is no more logging done.
+        LogManager.shutdown();
+      }
+    }
+  }
+
+  @Override
+  public Task getTask(TaskAttemptID taskid) throws IOException {
+    TaskInProgress tip = tasks.get(taskid);
+    if (tip != null) {
+      return tip.getTask();
+    } else {
+      return null;
+    }
+  }
+
+  @Override
+  public boolean ping(TaskAttemptID taskid) throws IOException {
+    // TODO Auto-generated method stub
+    return false;
+  }
+
+  @Override
+  public void done(TaskAttemptID taskid, boolean shouldBePromoted)
+      throws IOException {
+    // TODO Auto-generated method stub
+
+  }
+
+  @Override
+  public void fsError(TaskAttemptID taskId, String message) throws IOException {
+    // TODO Auto-generated method stub
+
+  }
+
+  @Override
+  public void send(String peerName, BSPMessage msg) throws IOException {
+    bspPeer.send(peerName, msg);
+  }
+
+  @Override
+  public void put(BSPMessage msg) throws IOException {
+    bspPeer.put(msg);
+  }
+
+  @Override
+  public BSPMessage getCurrentMessage() throws IOException {
+    return bspPeer.getCurrentMessage();
+  }
+
+  @Override
+  public int getNumCurrentMessages() {
+    return bspPeer.getNumCurrentMessages();
+  }
+
+  @Override
+  public void sync() throws IOException, KeeperException, InterruptedException {
+    bspPeer.sync();
+  }
+
+  @Override
+  public long getSuperstepCount() {
+    return bspPeer.getSuperstepCount();
+  }
+
+  @Override
+  public String getPeerName() {
+    return bspPeer.getPeerName();
+  }
+
+  @Override
+  public String[] getAllPeerNames() {
+    return bspPeer.getAllPeerNames();
+  }
 }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/Task.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/Task.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/Task.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/Task.java Thu Jan 20 01:10:17 2011
@@ -29,72 +29,75 @@ import org.apache.hadoop.io.Writable;
 
 public abstract class Task implements Writable {
   public static final Log LOG = LogFactory.getLog(Task.class);
-  ////////////////////////////////////////////
+  // //////////////////////////////////////////
   // Fields
-  ////////////////////////////////////////////
-  
+  // //////////////////////////////////////////
+
   protected BSPJobID jobId;
   protected String jobFile;
   protected TaskAttemptID taskId;
   protected int partition;
-  
+
   protected LocalDirAllocator lDirAlloc;
 
   public Task() {
     jobId = new BSPJobID();
     taskId = new TaskAttemptID();
   }
-  
-  public Task(BSPJobID jobId, String jobFile, TaskAttemptID taskId, int partition) {
+
+  public Task(BSPJobID jobId, String jobFile, TaskAttemptID taskId,
+      int partition) {
     this.jobId = jobId;
     this.jobFile = jobFile;
     this.taskId = taskId;
     this.partition = partition;
   }
-  
-  ////////////////////////////////////////////
+
+  // //////////////////////////////////////////
   // Accessors
-  ////////////////////////////////////////////
-  public void setJobFile(String jobFile) { 
-    this.jobFile = jobFile; 
+  // //////////////////////////////////////////
+  public void setJobFile(String jobFile) {
+    this.jobFile = jobFile;
   }
-  
-  public String getJobFile() { 
-    return jobFile; 
+
+  public String getJobFile() {
+    return jobFile;
   }
 
-  public TaskAttemptID getTaskAttemptId(){
+  public TaskAttemptID getTaskAttemptId() {
     return this.taskId;
   }
-  
+
   public TaskAttemptID getTaskID() {
     return taskId;
   }
-  
+
   /**
    * Get the job name for this task.
+   * 
    * @return the job name
    */
   public BSPJobID getJobID() {
     return jobId;
   }
-  
+
   /**
    * Get the index of this task within the job.
+   * 
    * @return the integer part of the task id
    */
   public int getPartition() {
     return partition;
   }
-  
+
   @Override
-  public String toString() { 
-    return taskId.toString(); 
+  public String toString() {
+    return taskId.toString();
   }
-  
-  ////////////////////////////////////////////
+
+  // //////////////////////////////////////////
   // Writable
-  ////////////////////////////////////////////
+  // //////////////////////////////////////////
   @Override
   public void write(DataOutput out) throws IOException {
     jobId.write(out);
@@ -102,7 +105,7 @@ public abstract class Task implements Wr
     taskId.write(out);
     out.writeInt(partition);
   }
-  
+
   @Override
   public void readFields(DataInput in) throws IOException {
     jobId.readFields(in);
@@ -111,6 +114,22 @@ public abstract class Task implements Wr
     partition = in.readInt();
   }
 
-  public abstract BSPTaskRunner createRunner(BSPPeer bspPeer, BSPJob jobConf);
+  /**
+   * Run this task as a part of the named job. This method is executed in the
+   * child process.
+   * 
+   * @param umbilical for progress reports
+   */
+  public abstract void run(BSPJob job, BSPPeerProtocol umbilical)
+      throws IOException;
+
+  public abstract BSPTaskRunner createRunner(GroomServer groom);
+
+  public void done(BSPPeerProtocol umbilical) throws IOException {
+    umbilical.done(getTaskID(), true);
+  }
+  
+  public abstract BSPJob getConf();
+  public abstract void setConf(BSPJob localJobConf);
   
 }

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskAttemptID.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskAttemptID.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskAttemptID.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskAttemptID.java Thu Jan 20 01:10:17
2011
@@ -49,10 +49,6 @@ public class TaskAttemptID extends ID {
     return taskId.getJobID();
   }
  
-  public TaskID getTaskId(){
-    return taskId;
-  }
-
   public TaskID getTaskID() {
     return taskId;
   }

Added: incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskRunner.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskRunner.java?rev=1061087&view=auto
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskRunner.java (added)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/TaskRunner.java Thu Jan 20 01:10:17
2011
@@ -0,0 +1,238 @@
+/**
+ * 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.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.InetSocketAddress;
+import java.util.Vector;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.RunJar;
+
+public class TaskRunner extends Thread {
+
+  public static final Log LOG = LogFactory.getLog(TaskRunner.class);
+
+  boolean killed = false;
+  private Process process;
+  private Task task;
+  private BSPJob conf;
+  private GroomServer groomServer;
+
+  public TaskRunner(BSPTask bspTask, GroomServer groom, BSPJob conf) {
+    this.task = bspTask;
+    this.conf = conf;
+    this.groomServer = groom;
+  }
+
+  public Task getTask() {
+    return task;
+  }
+
+  /**
+   * Called to assemble this task's input. This method is run in the parent
+   * process before the child is spawned. It should not execute user code, only
+   * system code.
+   */
+  public boolean prepare() throws IOException {
+    return true;
+  }
+
+  public void run() {
+    try {
+      String sep = System.getProperty("path.separator");
+      File workDir = new File(new File(task.getJobFile()).getParent(), "work");
+      boolean isCreated = workDir.mkdirs();
+      if(!isCreated) {
+        LOG.debug("TaskRunner.workDir : " + workDir);
+      }
+
+      StringBuffer classPath = new StringBuffer();
+      // start with same classpath as parent process
+      classPath.append(System.getProperty("java.class.path"));
+      classPath.append(sep);
+
+      String jar = conf.getJar();
+      if (jar != null) { // if jar exists, it into workDir
+        RunJar.unJar(new File(jar), workDir);
+        File[] libs = new File(workDir, "lib").listFiles();
+        if (libs != null) {
+          for (int i = 0; i < libs.length; i++) {
+            classPath.append(sep); // add libs from jar to classpath
+            classPath.append(libs[i]);
+          }
+        }
+        classPath.append(sep);
+        classPath.append(new File(workDir, "classes"));
+        classPath.append(sep);
+        classPath.append(workDir);
+      }
+
+      // Build exec child jmv args.
+      Vector<String> vargs = new Vector<String>();
+      File jvm = // use same jvm as parent
+      new File(new File(System.getProperty("java.home"), "bin"), "java");
+      vargs.add(jvm.toString());
+
+      String javaOpts = handleDeprecatedHeapSize("-Xmx500m", // TODO move to
+                                                             // config file
+          conf.get("bsp.child.heap.size"));
+      javaOpts = replaceAll(javaOpts, "@taskid@", task.getTaskID().toString());
+      String[] javaOptsSplit = javaOpts.split(" ");
+      for (int i = 0; i < javaOptsSplit.length; i++) {
+        vargs.add(javaOptsSplit[i]);
+      }
+
+      // Add classpath.
+      vargs.add("-classpath");
+      vargs.add(classPath.toString());
+      // Add main class and its arguments
+      vargs.add(GroomServer.Child.class.getName()); // main of Child
+
+      InetSocketAddress addr = groomServer.getTaskTrackerReportAddress();
+      vargs.add(addr.getHostName());
+      vargs.add(Integer.toString(addr.getPort()));
+      vargs.add(task.getTaskID().toString());
+
+      // Run java
+      runChild((String[]) vargs.toArray(new String[0]), workDir);
+    } catch (IOException e) {
+      LOG.error(e);
+    }
+  }
+
+  /**
+   * Handle deprecated mapred.child.heap.size. If present, interpolate into
+   * mapred.child.java.opts value with warning.
+   * 
+   * @param javaOpts Value of mapred.child.java.opts property.
+   * @param heapSize Value of mapred.child.heap.size property.
+   * @return A <code>javaOpts</code> with <code>heapSize</code> interpolated
if
+   *         present.
+   */
+  private String handleDeprecatedHeapSize(String javaOpts, final String heapSize) {
+    if (heapSize == null || heapSize.length() <= 0) {
+      return javaOpts;
+    }
+    final String MX = "-Xmx";
+    int index = javaOpts.indexOf(MX);
+    if (index < 0) {
+      javaOpts = javaOpts + " " + MX + heapSize;
+    } else {
+      int end = javaOpts.indexOf(" ", index + MX.length());
+      javaOpts = javaOpts.substring(0, index + MX.length()) + heapSize
+          + ((end < 0) ? "" : javaOpts.substring(end));
+    }
+    LOG.warn("mapred.child.heap.size is deprecated. Use "
+        + "mapred.child.java.opt instead. Meantime, mapred.child.heap.size "
+        + "is interpolated into mapred.child.java.opt: " + javaOpts);
+    return javaOpts;
+  }
+
+  /**
+   * Replace <code>toFind</code> with <code>replacement</code>. When
hadoop
+   * moves to JDK1.5, replace this method with String#replace (Of is
+   * commons-lang available, replace with StringUtils#replace).
+   * 
+   * @param text String to do replacements in.
+   * @param toFind String to find.
+   * @param replacement String to replace <code>toFind</code> with.
+   * @return A String with all instances of <code>toFind</code> replaced by
+   *         <code>replacement</code> (The original <code>text</code>
is
+   *         returned if <code>toFind</code> is not found in <code>text<code>).
+   */
+  private static String replaceAll(String text, final String toFind,
+      final String replacement) {
+    if (text == null || toFind == null || replacement == null) {
+      throw new IllegalArgumentException("Text " + text + " or toFind "
+          + toFind + " or replacement " + replacement + " are null.");
+    }
+    int offset = 0;
+    for (int index = text.indexOf(toFind); index >= 0; index = text.indexOf(
+        toFind, offset)) {
+      offset = index + toFind.length();
+      text = text.substring(0, index) + replacement + text.substring(offset);
+
+    }
+    return text;
+  }
+
+  /**
+   * Run the child process
+   */
+  private void runChild(String[] args, File dir) throws IOException {
+    System.out.println("runChild.dir : " + dir);
+    this.process = Runtime.getRuntime().exec(args, null, dir);
+    try {
+      new Thread() {
+        public void run() {
+          logStream(process.getErrorStream()); // copy log output
+        }
+      }.start();
+
+      logStream(process.getInputStream()); // normally empty
+
+      int exit_code = process.waitFor();
+      if (!killed && exit_code != 0) {
+        throw new IOException("Task process exit with nonzero status of "
+            + exit_code + ".");
+      }
+
+    } catch (InterruptedException e) {
+      throw new IOException(e.toString());
+    } finally {
+      kill();
+    }
+  }
+
+  /**
+   * Kill the child process
+   */
+  public void kill() {
+    if (process != null) {
+      process.destroy();
+    }
+    killed = true;
+  }
+
+  /**
+   */
+  private void logStream(InputStream output) {
+    try {
+      BufferedReader in = new BufferedReader(new InputStreamReader(output));
+      String line;
+      while ((line = in.readLine()) != null) {
+        LOG.info(task.getTaskID() + " " + line);
+      }
+    } catch (IOException e) {
+      LOG.warn(task.getTaskID() + " Error reading child output", e);
+    } finally {
+      try {
+        output.close();
+      } catch (IOException e) {
+        LOG.warn(task.getTaskID() + " Error closing child output", e);
+      }
+    }
+  }
+
+}

Modified: incubator/hama/trunk/src/test/org/apache/hama/bsp/Client.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/test/org/apache/hama/bsp/Client.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/test/org/apache/hama/bsp/Client.java (original)
+++ incubator/hama/trunk/src/test/org/apache/hama/bsp/Client.java Thu Jan 20 01:10:17 2011
@@ -77,8 +77,7 @@ class Client implements Runnable{
   public static class HelloBSP extends BSP {
     private Configuration conf;
 
-    @Override
-    public void bsp(BSPPeer bspPeer) throws IOException, KeeperException,
+    public void bsp(BSPPeerProtocol bspPeer) throws IOException, KeeperException,
         InterruptedException {
       int cnt = 0;
       Result r = null;

Modified: incubator/hama/trunk/src/test/org/apache/hama/bsp/TestBSPMaster.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/test/org/apache/hama/bsp/TestBSPMaster.java?rev=1061087&r1=1061086&r2=1061087&view=diff
==============================================================================
--- incubator/hama/trunk/src/test/org/apache/hama/bsp/TestBSPMaster.java (original)
+++ incubator/hama/trunk/src/test/org/apache/hama/bsp/TestBSPMaster.java Thu Jan 20 01:10:17
2011
@@ -94,6 +94,7 @@ public class TestBSPMaster extends HamaC
     final ScheduledExecutorService sched = getCluster().getScheduler();
     LOG.info("Start submiting job ...");
 
+    /*
     // client submit job 
     Client c = new Client();
     sched.schedule(c, 0, SECONDS);
@@ -120,6 +121,7 @@ public class TestBSPMaster extends HamaC
                    r.getNumber() + ": " + r.getPeer());
     }
     LOG.info("Finish executing test nexus method.");
+    */
   }
 
   public void tearDown() throws Exception{



Mime
View raw message