Changes between Version 12 and Version 13 of schedule

Timestamp:

10/19/2012 11:32:15 AM (14 years ago)

Author:

liaojiaohe

Comment:

--

schedule

@@ -16 +16 @@
 修改JobQueueManager，jobAdded 函数
 
-
+修改前
 {{{
   @Override
@@ -39 +39 @@
   }
 }}}
+
+修改后
+
+{{{
+ @Override
+  public void jobAdded(JobInProgress job) throws IOException {
+    LOG.info("Job " + job.getJobID() + " submitted to queue " + 
+        job.getProfile().getQueueName());
+    
+    // add job to the right queue
+    CapacitySchedulerQueue queue = getQueue(job.getProfile().getQueueName());
+    if (null == queue) {
+      // job was submitted to a queue we're not aware of
+      LOG.warn("Invalid queue " + job.getProfile().getQueueName() + 
+          " specified for job" + job.getProfile().getJobID() + 
+          ". Ignoring job.");
+      return;
+    }
+    
+    if ((queue.getNumSlotsOccupied(org.apache.hadoop.mapreduce.TaskType.MAP)/queue.getCapacity(org.apache.hadoop.mapreduce.TaskType.MAP) >= 1
+            || queue.getNumSlotsOccupied(org.apache.hadoop.mapreduce.TaskType.REDUCE)/queue.getCapacity(org.apache.hadoop.mapreduce.TaskType.REDUCE) >= 1)
+            && (job.getPriority() == JobPriority.HIGH || job.getPriority() == JobPriority.VERY_HIGH )){
+        //如果任务的级别比一般的高，一般是要求短时间完成的压力比较少的定时任务，分配到另外一条queue上面执行
+        
+        CapacitySchedulerQueue highPriorityQueue = getQueue("highpriority");
+        if (highPriorityQueue != null ){
+            highPriorityQueue.addWaitingJob(job);
+        }else{
+            queue.addWaitingJob(job);
+        }
+            
+    }else{
+            // add job to waiting queue. It will end up in the right place, 
+            // based on priority. 
+        queue.addWaitingJob(job);
+    }
+    
+
+    // let scheduler know. 
+    scheduler.jobAdded(job);
+  }
+
+}}}