Index: src/java/org/apache/hadoop/mapred/JobTracker.java
===================================================================
--- src/java/org/apache/hadoop/mapred/JobTracker.java	(revision 599245)
+++ src/java/org/apache/hadoop/mapred/JobTracker.java	(working copy)
@@ -1915,27 +1915,46 @@
        
     public void run() {
       while (!isInterrupted()) {
-        try {
-          JobInProgress.JobWithTaskContext j = queue.take();
-          JobInProgress job = j.getJob();
-          TaskInProgress tip = j.getTIP();
-          String taskid = j.getTaskId();
-          JobTrackerMetrics metrics = j.getJobTrackerMetrics();
-          Task t;
-          TaskStatus status;
-          boolean isTipComplete = false;
-          TaskStatus.State state;
-          synchronized (JobTracker.this) {
-            synchronized (job) {
-              synchronized (tip) {
-                status = tip.getTaskStatus(taskid);
-                t = tip.getTaskObject(taskid);
-                state = status.getRunState();
-                isTipComplete = tip.isComplete();
-              }
+            ArrayList <JobInProgress.JobWithTaskContext> jlist=new ArrayList<JobInProgress.JobWithTaskContext>(queue.size());
+            try {
+                queue.drainTo(jlist, queue.size());
+            } catch (Exception ie) {
+                return;
             }
-          }
-          try {
+            JobInProgress[] joblist=new JobInProgress[jlist.size()];
+            TaskInProgress[] tiplist = new TaskInProgress[jlist.size()];
+            String[] taskidlist = new String[jlist.size()];
+            JobTrackerMetrics[] metricslist =new JobTrackerMetrics[jlist.size()];
+
+            Iterator<JobInProgress.JobWithTaskContext> jiter=jlist.iterator();
+            int count=0;
+
+            while (jiter.hasNext()) {
+                JobInProgress.JobWithTaskContext j=jiter.next();
+                joblist[count] = j.getJob();
+                tiplist[count] = j.getTIP();
+                taskidlist[count]= j.getTaskId();
+                metricslist[count] = j.getJobTrackerMetrics();
+                count++;
+            }
+            Task[] tlist=new Task[jlist.size()];
+            TaskStatus[] statuslist=new TaskStatus[jlist.size()];
+            boolean[] isTipCompletelist =new boolean[jlist.size()];
+
+
+            TaskStatus.State[] statelist=new TaskStatus.State[jlist.size()];
+            synchronized (JobTracker.this) {
+                for(count=0;count<jlist.size();count++) {
+                    synchronized (joblist[count]) {
+                        synchronized (tiplist[count]) {
+                            statuslist[count] = tiplist[count].getTaskStatus(taskidlist[count]);
+                            tlist[count] = tiplist[count].getTaskObject(taskidlist[count]);
+                            statelist[count] = statuslist[count].getRunState();
+                            isTipCompletelist[count] = tiplist[count].isComplete();
+                        }
+                    }
+                }
+            }
             //For COMMIT_PENDING tasks, we save the task output in the dfs
             //as well as manipulate the JT datastructures to reflect a
             //successful task. This guarantees that we don't declare a task
@@ -1945,84 +1964,93 @@
             //datastructures updates is done earlier as soon as the failure
             //is detected so that the JT can immediately schedule another
             //attempt for that task.
-            if (state == TaskStatus.State.COMMIT_PENDING) {
-              if (!isTipComplete) {
-                t.saveTaskOutput();
-              }
-              synchronized (JobTracker.this) {
+
+            /*This loop can be further parallelized handling more promotions
+              at any given time, provided the namenode can handle them*/
+            for(count=0;count<jlist.size();count++) {
+   try{
+                    if (statelist[count]== TaskStatus.State.COMMIT_PENDING) {
+                        if (!isTipCompletelist[count]) {
+                            tlist[count].saveTaskOutput();
+                        }
+                    }
+                } catch (IOException ioe) {
+                    // Oops! Failed to copy the task's output to its final place;
+                    // fail the task!
+                    statelist[count] = TaskStatus.State.FAILED;
+                    synchronized (JobTracker.this) {
+                        joblist[count].failedTask(tiplist[count], statuslist[count].getTaskId(),
+                                "Failed to rename output with the exception: " +
+                                StringUtils.stringifyException(ioe),
+                                (tiplist[count].isMapTask() ?
+                                 TaskStatus.Phase.MAP :
+                                 TaskStatus.Phase.REDUCE),
+                                TaskStatus.State.FAILED,
+                                statuslist[count].getTaskTracker(), null);
+                    }
+                    LOG.info("Failed to rename the output of " + statuslist[count].getTaskId() +
+                            " with: " + StringUtils.stringifyException(ioe));
+
+                }
+            }
+
+
+            synchronized (JobTracker.this) {
                 //do a check for the case where after the task went to
                 //COMMIT_PENDING, it was lost. So although we would have
                 //saved the task output, we cannot declare it a SUCCESS.
-                TaskStatus newStatus = null;
-                synchronized (job) {
-                  synchronized (tip) {
-                    status = tip.getTaskStatus(taskid);
-                    if (!isTipComplete) {
-                      if (status.getRunState() != 
-                        TaskStatus.State.COMMIT_PENDING) {
-                        state = TaskStatus.State.KILLED;
-                      } else {
-                        state = TaskStatus.State.SUCCEEDED;
-                      }
-                    } else {
-                      tip.addDiagnosticInfo(t.getTaskId(),"Already completed " +
-                      "TIP");
-                      state = TaskStatus.State.KILLED;
+                TaskStatus[] newStatuslist = new TaskStatus[jlist.size()];
+                for(count=0;count<jlist.size();count++) {
+                    if(statelist[count]==TaskStatus.State.FAILED)
+                        continue;
 
+                    newStatuslist[count]=null;
+                    synchronized (joblist[count]) {
+                        synchronized (tiplist[count]) {
+                            statuslist[count] = tiplist[count].getTaskStatus(taskidlist[count]);
+                            if (!isTipCompletelist[count]) {
+                                if (statuslist[count].getRunState() !=
+                                        TaskStatus.State.COMMIT_PENDING) {
+                                    statelist[count] = TaskStatus.State.KILLED;
+                                } else {
+                                    statelist[count] = TaskStatus.State.SUCCEEDED;
+                                }
+                            } else {
+                                tiplist[count].addDiagnosticInfo(tlist[count].getTaskId(),"Already completed " +
+                                        "TIP");
+                                statelist[count] = TaskStatus.State.KILLED;
+
+                            }
+                            //create new status if required. If the state changed from
+                            //COMMIT_PENDING to KILLED in the JobTracker, while we were
+                            //saving the output,the JT would have called updateTaskStatus
+                            //and we don't need to call it again
+                            if (statuslist[count].getRunState() == TaskStatus.State.COMMIT_PENDING){
+    newStatuslist[count] = (TaskStatus)statuslist[count].clone();
+                                newStatuslist[count].setRunState(statelist[count]);
+                                newStatuslist[count].setProgress((statelist[count] == TaskStatus.State.SUCCEEDED) ? 1.0f : 0.0f);
+                            }
+                        }
+                        if (newStatuslist[count] != null) {
+                            joblist[count].updateTaskStatus(tiplist[count], newStatuslist[count], metricslist[count]);
+                        }
                     }
-                    //create new status if required. If the state changed from
-                    //COMMIT_PENDING to KILLED in the JobTracker, while we were
-                    //saving the output,the JT would have called updateTaskStatus
-                    //and we don't need to call it again
-                    if (status.getRunState() == TaskStatus.State.COMMIT_PENDING){
-                      newStatus = (TaskStatus)status.clone();
-                      newStatus.setRunState(state);
-                      newStatus.setProgress((state == TaskStatus.State.SUCCEEDED) ? 1.0f : 0.0f);
+                }
+            }
+            for(count=0;count<jlist.size();count++) {
+
+                if (statelist[count] == TaskStatus.State.FAILED ||
+                        statelist[count] == TaskStatus.State.KILLED) {
+                    try {
+                        tlist[count].discardTaskOutput();
+                    } catch (IOException ioe) {
+                        LOG.info("Failed to discard the output of task " +
+                                statuslist[count].getTaskId() + " with: " +
+                                StringUtils.stringifyException(ioe));
                     }
-                  }
-                  if (newStatus != null) {
-                    job.updateTaskStatus(tip, newStatus, metrics);
-                  }
                 }
-              }
             }
-          } catch (IOException ioe) {
-            // Oops! Failed to copy the task's output to its final place;
-            // fail the task!
-            state = TaskStatus.State.FAILED;
-            synchronized (JobTracker.this) {
-              job.failedTask(tip, status.getTaskId(), 
-                  "Failed to rename output with the exception: " + 
-                  StringUtils.stringifyException(ioe), 
-                  (tip.isMapTask() ? 
-                      TaskStatus.Phase.MAP : 
-                        TaskStatus.Phase.REDUCE), 
-                        TaskStatus.State.FAILED,  
-                        status.getTaskTracker(), null);
-            }
-            LOG.info("Failed to rename the output of " + status.getTaskId() + 
-                " with: " + StringUtils.stringifyException(ioe));
-          }
-          if (state == TaskStatus.State.FAILED || 
-              state == TaskStatus.State.KILLED) {
-            try {
-              t.discardTaskOutput();
-            } catch (IOException ioe) { 
-              LOG.info("Failed to discard the output of task " + 
-                  status.getTaskId() + " with: " + 
-                  StringUtils.stringifyException(ioe));
-            }
-          }
-        } catch (InterruptedException ie) {
-          LOG.warn(getName() + " exiting, got interrupted: " + 
-                   StringUtils.stringifyException(ie));
-          return;
         }
-        catch (Throwable t) {
-          LOG.error(getName() + " got an exception: " +
-                    StringUtils.stringifyException(t));
-        }
-      }
       
       LOG.warn(getName() + " exiting..."); 
     }