You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by cu...@apache.org on 2006/05/24 00:23:39 UTC
svn commit: r409014 - in /lucene/hadoop/trunk: CHANGES.txt
src/java/org/apache/hadoop/mapred/JobInProgress.java
src/java/org/apache/hadoop/mapred/JobTracker.java
Author: cutting
Date: Tue May 23 15:23:35 2006
New Revision: 409014
URL: http://svn.apache.org/viewvc?rev=409014&view=rev
Log:
HADOOP-205. Incorporate pending tasksinto tasktracker load calculations. Contributed by Mahadev.
Modified:
lucene/hadoop/trunk/CHANGES.txt
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
Modified: lucene/hadoop/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/CHANGES.txt?rev=409014&r1=409013&r2=409014&view=diff
==============================================================================
--- lucene/hadoop/trunk/CHANGES.txt (original)
+++ lucene/hadoop/trunk/CHANGES.txt Tue May 23 15:23:35 2006
@@ -52,6 +52,10 @@
tasks and logs. Also add log access to job tracker web interface.
(omalley via cutting)
+14. HADOOP-205. Incorporate pending tasks into tasktracker load
+ calculations. (Mahadev Konar via cutting)
+
+
Release 0.2.1 - 2006-05-12
1. HADOOP-199. Fix reduce progress (broken by HADOOP-182).
Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java?rev=409014&r1=409013&r2=409014&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java Tue May 23 15:23:35 2006
@@ -42,7 +42,10 @@
TaskInProgress reduces[] = new TaskInProgress[0];
int numMapTasks = 0;
int numReduceTasks = 0;
-
+ int runningMapTasks = 0;
+ int runningReduceTasks = 0;
+ int finishedMapTasks = 0;
+ int finishedReduceTasks = 0;
JobTracker jobtracker = null;
HashMap hostToMaps = new HashMap();
@@ -201,27 +204,21 @@
return numMapTasks;
}
public int finishedMaps() {
- int finishedCount = 0;
- for (int i = 0; i < maps.length; i++) {
- if (maps[i].isComplete()) {
- finishedCount++;
- }
- }
- return finishedCount;
+ return finishedMapTasks;
}
public int desiredReduces() {
return numReduceTasks;
}
+ public synchronized int runningMaps() {
+ return runningMapTasks;
+ }
+ public synchronized int runningReduces() {
+ return runningReduceTasks;
+ }
public int finishedReduces() {
- int finishedCount = 0;
- for (int i = 0; i < reduces.length; i++) {
- if (reduces[i].isComplete()) {
- finishedCount++;
- }
- }
- return finishedCount;
+ return finishedReduceTasks;
}
-
+
/**
* Get the list of map tasks
* @return the raw array of maps for this job
@@ -377,12 +374,24 @@
//
if (cacheTarget >= 0) {
t = maps[cacheTarget].getTaskToRun(taskTracker, tts, avgProgress);
+ runningMapTasks += 1;
} else if (stdTarget >= 0) {
t = maps[stdTarget].getTaskToRun(taskTracker, tts, avgProgress);
- } else if (specTarget >= 0) {
+ runningMapTasks += 1;
+ } else if (specTarget >= 0) {
+ //should always be true, but being paranoid
+ boolean isRunning = maps[specTarget].isRunning();
t = maps[specTarget].getTaskToRun(taskTracker, tts, avgProgress);
+ if (!isRunning){
+ runningMapTasks += 1;
+ }
} else if (failedTarget >= 0) {
+ //should always be false, but being paranoid again
+ boolean isRunning = maps[failedTarget].isRunning();
t = maps[failedTarget].getTaskToRun(taskTracker, tts, avgProgress);
+ if (!isRunning) {
+ runningMapTasks += 1;
+ }
}
return t;
}
@@ -424,11 +433,21 @@
if (stdTarget >= 0) {
t = reduces[stdTarget].getTaskToRun(taskTracker, tts, avgProgress);
- } else if (specTarget >= 0) {
+ runningReduceTasks += 1;
+ } else if (specTarget >= 0) {
+ //should be false
+ boolean isRunning = reduces[specTarget].isRunning();
t = reduces[specTarget].getTaskToRun(taskTracker, tts, avgProgress);
+ if (!isRunning){
+ runningReduceTasks += 1;
+ }
} else if (failedTarget >= 0) {
+ boolean isRunning = reduces[failedTarget].isRunning();
t = reduces[failedTarget].getTaskToRun(taskTracker, tts,
avgProgress);
+ if (!isRunning){
+ runningReduceTasks += 1;
+ }
}
return t;
}
@@ -439,10 +458,25 @@
public synchronized void completedTask(TaskInProgress tip,
TaskStatus status) {
String taskid = status.getTaskId();
+ boolean oldDone = tip.isComplete();
updateTaskStatus(tip, status);
LOG.info("Taskid '" + taskid + "' has finished successfully.");
tip.completed(taskid);
-
+ boolean newDone = tip.isComplete();
+ // updating the running/finished map/reduce counts
+ if (oldDone != newDone) {
+ if (newDone) {
+ if (tip.isMapTask()){
+ runningMapTasks -= 1;
+ finishedMapTasks += 1;
+ }
+ else{
+ runningReduceTasks -= 1;
+ finishedReduceTasks += 1;
+ }
+ }
+ }
+
//
// Figure out whether the Job is done
//
@@ -480,7 +514,8 @@
if (status.getRunState() != JobStatus.FAILED) {
this.status = new JobStatus(status.getJobId(), 1.0f, 1.0f, JobStatus.FAILED);
this.finishTime = System.currentTimeMillis();
-
+ this.runningMapTasks = 0;
+ this.runningReduceTasks = 0;
//
// kill all TIPs.
//
@@ -509,9 +544,31 @@
*/
public synchronized void failedTask(TaskInProgress tip, String taskid,
TaskStatus status, String trackerName) {
+ boolean oldStatus = tip.isRunning();
+ boolean oldRun = tip.isComplete();
tip.failedSubTask(taskid, trackerName);
updateTaskStatus(tip, status);
-
+ boolean newStatus = tip.isRunning();
+ boolean newRun = tip.isComplete();
+ //update running count on task failure.
+ if (oldStatus != newStatus) {
+ if (!newStatus) {
+ if (tip.isMapTask()){
+ runningMapTasks -= 1;
+ }
+ else{
+ runningReduceTasks -= 1;
+ }
+ }
+ }
+ // the case when the map was complete but the task tracker went down.
+ if (oldRun != newRun) {
+ if (oldRun){
+ if (tip.isMapTask()){
+ finishedMapTasks -= 1;
+ }
+ }
+ }
// After this, try to assign tasks with the one after this, so that
// the failed task goes to the end of the list.
if (tip.isMapTask()) {
Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java?rev=409014&r1=409013&r2=409014&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java Tue May 23 15:23:35 2006
@@ -645,15 +645,34 @@
//
int avgMaps = 0;
int avgReduces = 0;
+ int remainingReduceLoad = 0;
+ int remainingMapLoad = 0;
int numTaskTrackers;
TaskTrackerStatus tts;
+ int avgMapLoad = 0;
+ int avgReduceLoad = 0;
+
synchronized (taskTrackers) {
numTaskTrackers = taskTrackers.size();
tts = (TaskTrackerStatus) taskTrackers.get(taskTracker);
}
+ synchronized(jobsByArrival){
+ for (Iterator it = jobsByArrival.iterator(); it.hasNext(); ) {
+ JobInProgress job = (JobInProgress) it.next();
+ if (job.getStatus().getRunState() == JobStatus.RUNNING) {
+ int totalMapTasks = job.desiredMaps();
+ int totalReduceTasks = job.desiredReduces();
+ remainingMapLoad += (totalMapTasks - job.finishedMaps());
+ remainingReduceLoad += (totalReduceTasks - job.finishedReduces());
+ }
+ }
+ }
+
if (numTaskTrackers > 0) {
avgMaps = totalMaps / numTaskTrackers;
avgReduces = totalReduces / numTaskTrackers;
+ avgMapLoad = remainingMapLoad / numTaskTrackers;
+ avgReduceLoad = remainingReduceLoad / numTaskTrackers;
}
int totalCapacity = numTaskTrackers * maxCurrentTasks;
//
@@ -676,15 +695,18 @@
//
// We hand a task to the current taskTracker if the given machine
- // has a workload that's equal to or less than the averageMaps
+ // has a workload that's equal to or less than the pendingMaps average.
+ // This way the maps are launched if the TaskTracker has running tasks
+ // less than the pending average
// +/- TASK_ALLOC_EPSILON. (That epsilon is in place in case
// there is an odd machine that is failing for some reason but
// has not yet been removed from the pool, making capacity seem
// larger than it really is.)
//
+
synchronized (jobsByArrival) {
if ((numMaps < maxCurrentTasks) &&
- (numMaps <= (avgMaps + TASK_ALLOC_EPSILON))) {
+ (numMaps <= avgMapLoad + 1 + TASK_ALLOC_EPSILON)) {
int totalNeededMaps = 0;
for (Iterator it = jobsByArrival.iterator(); it.hasNext(); ) {
@@ -719,7 +741,7 @@
// Same thing, but for reduce tasks
//
if ((numReduces < maxCurrentTasks) &&
- (numReduces <= (avgReduces + TASK_ALLOC_EPSILON))) {
+ (numReduces <= avgReduceLoad + 1 + TASK_ALLOC_EPSILON)) {
int totalNeededReduces = 0;
for (Iterator it = jobsByArrival.iterator(); it.hasNext(); ) {