incubator-hama-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From edwardy...@apache.org
Subject svn commit: r1033838 - in /incubator/hama/trunk: ./ src/examples/org/apache/hama/examples/ src/java/org/apache/hama/bsp/
Date Thu, 11 Nov 2010 08:44:57 GMT
Author: edwardyoon
Date: Thu Nov 11 08:44:57 2010
New Revision: 1033838

URL: http://svn.apache.org/viewvc?rev=1033838&view=rev
Log:
BSP program doesn't work when the desired tasks smaller than cluster size

Modified:
    incubator/hama/trunk/CHANGES.txt
    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/BSPMaster.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/JobInProgress.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/JobProfile.java
    incubator/hama/trunk/src/java/org/apache/hama/bsp/SimpleTaskScheduler.java

Modified: incubator/hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/CHANGES.txt?rev=1033838&r1=1033837&r2=1033838&view=diff
==============================================================================
--- incubator/hama/trunk/CHANGES.txt (original)
+++ incubator/hama/trunk/CHANGES.txt Thu Nov 11 08:44:57 2010
@@ -181,6 +181,8 @@ Trunk (unreleased changes)
 
   BUG FIXES
   
+    HAMA-330: BSP program doesn't work when the desired tasks smaller than 
+                       cluster size (edwardyoon)
     HAMA-326: GroomServer should wait until a job finishes (edwardyoon)
     HAMA-324: Removing javadoc warnings and useless libraries (edwardyoon)
     HAMA-322: Make sure failed assertions on test threads are reported

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=1033838&r1=1033837&r2=1033838&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 Nov 11
08:44:57 2010
@@ -99,7 +99,7 @@ public class PiEstimator {
     // Set the job name
     bsp.setJobName("pi estimation example");
     bsp.setBspClass(MyEstimator.class);
-    bsp.setNumBspTask(10);
+    bsp.setNumBspTask(1);
     
     BSPJobClient jobClient = new BSPJobClient(conf);
     ClusterStatus cluster = jobClient.getClusterStatus(true);

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=1033838&r1=1033837&r2=1033838&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
Nov 11 08:44:57 2010
@@ -27,6 +27,7 @@ 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 {
@@ -44,7 +45,7 @@ public class SerializePrinting {
       int i = 0;
       for (String otherPeer : bspPeer.getAllPeerNames()) {
         if (bspPeer.getPeerName().equals(otherPeer)) {
-          LOG.info("Hello BSP from " + i + " of " + num + ": "
+          LOG.info("Hello BSP from " + (i + 1) + " of " + num + ": "
               + bspPeer.getPeerName());
         }
         
@@ -75,7 +76,11 @@ public class SerializePrinting {
     // Set the job name
     bsp.setJobName("serialize printing");
     bsp.setBspClass(HelloBSP.class);
-    bsp.setNumBspTask(10);
+    
+    // Set the task size as a number of GroomServer
+    BSPJobClient jobClient = new BSPJobClient(conf);
+    ClusterStatus cluster = jobClient.getClusterStatus(false);
+    bsp.setNumBspTask(cluster.getGroomServers());
     
     BSPJobClient.runJob(bsp);
   }

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=1033838&r1=1033837&r2=1033838&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 Nov 11 08:44:57 2010
@@ -381,13 +381,7 @@ public class BSPMaster implements JobSub
 
         for (Task task : taskList) {
           if (task != null) {
-
-            if (!jobs.get(task.getJobID()).getStatus().isJobComplete()) {
-              if (jobs.get(task.getJobID()).getStatus().getRunState() != JobStatus.RUNNING)
{
                 actions.add(new LaunchTaskAction(task));
-              }
-            }
-
           }
         }
       }
@@ -463,7 +457,6 @@ public class BSPMaster implements JobSub
     LOG.debug("Removing task '" + taskid + "'");
   }
 
-  /*
   private List<GroomServerAction> getTasksToKill(String groomName) {
     Set<String> taskIds = (TreeSet<String>) trackerToTaskMap.get(groomName);
     if (taskIds != null) {
@@ -498,7 +491,6 @@ public class BSPMaster implements JobSub
     return null;
 
   }
-  */
   
   /**
    * Process incoming heartbeat messages from the groom.
@@ -584,6 +576,16 @@ public class BSPMaster implements JobSub
       jobs.put(job.getProfile().getJobID(), job);
       taskScheduler.addJob(job);
     }
+    
+    
+    // TODO Later, we should use the JobInProgressListener -- edwardyoon
+    try {
+      job.initTasks();
+    } catch (IOException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    }
+    
     return job.getStatus();
   }
 

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/JobInProgress.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/JobInProgress.java?rev=1033838&r1=1033837&r2=1033838&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/JobInProgress.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/JobInProgress.java Thu Nov 11 08:44:57
2010
@@ -46,6 +46,7 @@ class JobInProgress {
   }
 
   static final Log LOG = LogFactory.getLog(JobInProgress.class);
+  boolean tasksInited = false;
 
   Configuration conf;
   JobProfile profile;
@@ -62,15 +63,16 @@ class JobInProgress {
   // private LocalFileSystem localFs;
   private BSPJobID jobId;
   final BSPMaster master;
-  List<TaskInProgress> tasks;
+  TaskInProgress tasks[] = new TaskInProgress[0];
   private long superstepCounter;
 
+  int numBSPTasks = 0;
+  int clusterSize;
+
   public JobInProgress(BSPJobID jobId, BSPMaster master, Configuration conf)
       throws IOException {
     this.conf = conf;
     this.jobId = jobId;
-
-    this.tasks = new ArrayList<TaskInProgress>();
     this.localFs = FileSystem.getLocal(conf);
 
     this.master = master;
@@ -88,7 +90,8 @@ class JobInProgress {
     FileSystem fs = jobDir.getFileSystem(conf);
     jobFile = new Path(jobDir, "job.xml");
     fs.copyToLocalFile(jobFile, localJobFile);
-    BSPJobContext job = new BSPJobContext(localJobFile, jobId);
+    BSPJob job = new BSPJob(jobId, localJobFile.toString());
+    this.numBSPTasks = job.getNumBspTask();
 
     this.profile = new JobProfile(job.getUser(), jobId, jobFile.toString(), job
         .getJobName());
@@ -100,9 +103,6 @@ class JobInProgress {
 
   }
 
-  // ///////////////////////////////////////////////////
-  // Accessors for the JobInProgress
-  // ///////////////////////////////////////////////////
   public JobProfile getProfile() {
     return profile;
   }
@@ -124,6 +124,13 @@ class JobInProgress {
   }
 
   /**
+   * @return the number of desired tasks.
+   */
+  public int desiredBSPTasks() {
+    return numBSPTasks;
+  }
+
+  /**
    * @return The JobID of this JobInProgress.
    */
   public BSPJobID getJobID() {
@@ -139,16 +146,45 @@ class JobInProgress {
   // ///////////////////////////////////////////////////
   // Create/manage tasks
   // ///////////////////////////////////////////////////
+
+  public synchronized void initTasks() throws IOException {
+    if (tasksInited) {
+      return;
+    }
+
+    // adjust number of map tasks to actual number of splits
+    this.tasks = new TaskInProgress[numBSPTasks];
+    for (int i = 0; i < numBSPTasks; i++) {
+      tasks[i] = new TaskInProgress(getJobID(), this.jobFile.toString(),
+          this.master, this.conf, this, i);
+    }
+
+    // Update job status
+    this.status = new JobStatus(this.status.getJobID(), 1.0f, 1.0f,
+       JobStatus.RUNNING);
+
+    tasksInited = true;
+    LOG.debug("Job is initialized.");
+  }
+
   public synchronized Task obtainNewTask(GroomServerStatus status,
       int clusterSize, int numUniqueHosts) {
-    LOG.debug("clusterSize: " + clusterSize);
-
+    this.clusterSize = clusterSize;
+    
+    if (this.status.getRunState() != JobStatus.RUNNING) {
+      LOG.info("Cannot create task split for " + profile.getJobID());
+      return null;
+    }
+    
     Task result = null;
     try {
-      TaskInProgress tip = new TaskInProgress(getJobID(), this.jobFile
-          .toString(), this.master, this.conf, this, numUniqueHosts);
-      tasks.add(tip);
-      result = tip.getTaskToRun(status);
+      for (int i = 0; i < tasks.length; i++) {
+        if(!tasks[i].isRunning()) {
+          result = tasks[i].getTaskToRun(status);
+          break;
+        }
+      }
+      
     } catch (IOException e) {
       e.printStackTrace();
     }
@@ -156,7 +192,7 @@ class JobInProgress {
     return result;
   }
 
-  public void completedTask(TaskInProgress tip, TaskStatus status) {
+  public synchronized void completedTask(TaskInProgress tip, TaskStatus status) {
     String taskid = status.getTaskId();
     updateTaskStatus(tip, status);
     LOG.info("Taskid '" + taskid + "' has finished successfully.");
@@ -174,9 +210,6 @@ class JobInProgress {
       }
     }
 
-    this.status = new JobStatus(this.status.getJobID(), 1.0f, 1.0f,
-        JobStatus.RUNNING);
-
     if (allDone) {
       LOG.debug("Job successfully done.");
 
@@ -186,13 +219,14 @@ class JobInProgress {
     }
   }
 
-  public synchronized void updateTaskStatus(TaskInProgress tip, TaskStatus taskStatus) {
+  public synchronized void updateTaskStatus(TaskInProgress tip,
+      TaskStatus taskStatus) {
     tip.updateStatus(taskStatus); // update tip
 
     if (superstepCounter < taskStatus.getSuperstepCount()) {
       superstepCounter = taskStatus.getSuperstepCount();
       // TODO Later, we have to update JobInProgress status here
-      
+
     }
   }
 

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/JobProfile.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/JobProfile.java?rev=1033838&r1=1033837&r2=1033838&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/JobProfile.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/JobProfile.java Thu Nov 11 08:44:57
2010
@@ -26,10 +26,9 @@ import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableFactories;
 import org.apache.hadoop.io.WritableFactory;
 
-/**************************************************
+/*
  * A JobProfile tracks job's status
- * 
- **************************************************/
+ */
 public class JobProfile implements Writable {
 
   static { // register a ctor
@@ -96,9 +95,6 @@ public class JobProfile implements Writa
     return name;
   }
 
-  // /////////////////////////////////////
-  // Writable
-  // /////////////////////////////////////
   public void write(DataOutput out) throws IOException {
     jobid.write(out);
     Text.writeString(out, jobFile);

Modified: incubator/hama/trunk/src/java/org/apache/hama/bsp/SimpleTaskScheduler.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/SimpleTaskScheduler.java?rev=1033838&r1=1033837&r2=1033838&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/SimpleTaskScheduler.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/SimpleTaskScheduler.java Thu Nov 11
08:44:57 2010
@@ -76,28 +76,24 @@ class SimpleTaskScheduler extends TaskSc
       // instance to the scheduler.
       synchronized (jobQueue) {
         for (JobInProgress job : jobQueue) {
+          if (job.getStatus().getRunState() != JobStatus.RUNNING) {
+            continue;
+          }
 
-          if (!job.getStatus().isJobComplete() && job.getStatus().getRunState() !=
JobStatus.RUNNING) {
-            /*
-             * if (job.getStatus().getRunState() != JobStatus.RUNNING) {
-             * continue; }
-             */
-
-            Task t = null;
+          Task t = null;
 
-            t = job.obtainNewTask(groomStatus, numGroomServers,
-                groomServerManager.getNumberOfUniqueHosts());
+          t = job.obtainNewTask(groomStatus, numGroomServers,
+              groomServerManager.getNumberOfUniqueHosts());
 
-            if (t != null) {
-              assignedTasks.add(t);
-              break; // TODO - Now, simple scheduler assigns only one task to
-              // each groom. Later, it will be improved for scheduler to
-              // assign one or more tasks to each groom according to
-              // its capacity.
-            }
+          if (t != null) {
+            assignedTasks.add(t);
+            break; // TODO - Now, simple scheduler assigns only one task to
+            // each groom. Later, it will be improved for scheduler to
+            // assign one or more tasks to each groom according to
+            // its capacity.
           }
-          
         }
+
       }
     }
 



Mime
View raw message