Java 类org.apache.hadoop.mapred.JobQueueJobInProgressListener.JobSchedulingInfo 实例源码

public CapacitySchedulerQueue(String queueName, CapacitySchedulerConf conf) {
  this.queueName = queueName;

  // Do not allow changes to 'supportsPriorities'
  supportsPriorities = conf.isPrioritySupported(queueName);

  initializeQueue(conf);

  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  this.waitingJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.initializingJobs =
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.runningJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);

  this.mapSlots = new SlotsUsage();
  this.reduceSlots = new SlotsUsage();    
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  JobInProgress waitingJob = queue.removeWaitingJob(oldInfo, runState);
  JobInProgress initializingJob = 
    queue.removeInitializingJob(oldInfo, runState);
  JobInProgress runningJob = queue.removeRunningJob(oldInfo, runState);

  // let scheduler know if necessary
  // sometimes this isn't necessary if the job was rejected during submission
  if (runningJob != null || initializingJob != null || waitingJob != null) {
    scheduler.jobCompleted(job);
  }
}

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  if(queue.removeWaitingJob(oldInfo, runState) != null) {
    try {
      queue.addWaitingJob(job);
    } catch (IOException ioe) {
      // Ignore, cannot happen
      LOG.warn("Couldn't change priority!");
      return;
    }
  }
  if (queue.removeInitializingJob(oldInfo, runState) != null) {
    queue.addInitializingJob(job);
  }
  if(queue.removeRunningJob(oldInfo, runState) != null) {
    queue.addRunningJob(job);
  }
}

private void jobStateChanged(JobStatusChangeEvent event, QueueInfo qi) {
  JobInProgress job = event.getJobInProgress();
  JobSchedulingInfo oldJobStateInfo = 
    new JobSchedulingInfo(event.getOldStatus());
  // Check if the ordering of the job has changed
  // For now priority and start-time can change the job ordering
  if (event.getEventType() == EventType.PRIORITY_CHANGED 
      || event.getEventType() == EventType.START_TIME_CHANGED) {
    // Make a priority change
    reorderJobs(job, oldJobStateInfo, qi);
  } else if (event.getEventType() == EventType.RUN_STATE_CHANGED) {
    // Check if the job is complete
    int runState = job.getStatus().getRunState();
    if (runState == JobStatus.SUCCEEDED
        || runState == JobStatus.FAILED
        || runState == JobStatus.KILLED) {
      jobCompleted(job, oldJobStateInfo, qi);
    } else if (runState == JobStatus.RUNNING) {
      makeJobRunning(job, oldJobStateInfo, qi);
    }
  }
}

public CapacitySchedulerQueue(String queueName, CapacitySchedulerConf conf) {
  this.queueName = queueName;

  // Do not allow changes to 'supportsPriorities'
  supportsPriorities = conf.isPrioritySupported(queueName);

  initializeQueue(conf);

  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  this.waitingJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.initializingJobs =
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.runningJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);

  this.mapSlots = new SlotsUsage();
  this.reduceSlots = new SlotsUsage();    
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  JobInProgress waitingJob = queue.removeWaitingJob(oldInfo, runState);
  JobInProgress initializingJob = 
    queue.removeInitializingJob(oldInfo, runState);
  JobInProgress runningJob = queue.removeRunningJob(oldInfo, runState);

  // let scheduler know if necessary
  // sometimes this isn't necessary if the job was rejected during submission
  if (runningJob != null || initializingJob != null || waitingJob != null) {
    scheduler.jobCompleted(job);
  }
}

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  if(queue.removeWaitingJob(oldInfo, runState) != null) {
    try {
      queue.addWaitingJob(job);
    } catch (IOException ioe) {
      // Ignore, cannot happen
      LOG.warn("Couldn't change priority!");
      return;
    }
  }
  if (queue.removeInitializingJob(oldInfo, runState) != null) {
    queue.addInitializingJob(job);
  }
  if(queue.removeRunningJob(oldInfo, runState) != null) {
    queue.addRunningJob(job);
  }
}

private void jobStateChanged(JobStatusChangeEvent event, QueueInfo qi) {
  JobInProgress job = event.getJobInProgress();
  JobSchedulingInfo oldJobStateInfo = 
    new JobSchedulingInfo(event.getOldStatus());
  // Check if the ordering of the job has changed
  // For now priority and start-time can change the job ordering
  if (event.getEventType() == EventType.PRIORITY_CHANGED 
      || event.getEventType() == EventType.START_TIME_CHANGED) {
    // Make a priority change
    reorderJobs(job, oldJobStateInfo, qi);
  } else if (event.getEventType() == EventType.RUN_STATE_CHANGED) {
    // Check if the job is complete
    int runState = job.getStatus().getRunState();
    if (runState == JobStatus.SUCCEEDED
        || runState == JobStatus.FAILED
        || runState == JobStatus.KILLED) {
      jobCompleted(job, oldJobStateInfo, qi);
    } else if (runState == JobStatus.RUNNING) {
      makeJobRunning(job, oldJobStateInfo, qi);
    }
  }
}

private void jobStateChanged(JobStatusChangeEvent event, QueueInfo qi) {
  JobInProgress job = event.getJobInProgress();
  JobSchedulingInfo oldJobStateInfo = 
    new JobSchedulingInfo(event.getOldStatus());
  // Check if the ordering of the job has changed
  // For now priority and start-time can change the job ordering
  if (event.getEventType() == EventType.PRIORITY_CHANGED 
      || event.getEventType() == EventType.START_TIME_CHANGED) {
    // Make a priority change
    reorderJobs(job, oldJobStateInfo, qi);
  } else if (event.getEventType() == EventType.RUN_STATE_CHANGED) {
    // Check if the job is complete
    int runState = job.getStatus().getRunState();
    if (runState == JobStatus.SUCCEEDED
        || runState == JobStatus.FAILED
        || runState == JobStatus.KILLED) {
      jobCompleted(job, oldJobStateInfo, qi);
    } else if (runState == JobStatus.RUNNING) {
      makeJobRunning(job, oldJobStateInfo, qi);
    }
  }
}

public CapacitySchedulerQueue(String queueName, CapacitySchedulerConf conf) {
  this.queueName = queueName;

  // Do not allow changes to 'supportsPriorities'
  supportsPriorities = conf.isPrioritySupported(queueName);

  initializeQueue(conf);

  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  this.waitingJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.initializingJobs =
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.runningJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);

  this.mapSlots = new SlotsUsage();
  this.reduceSlots = new SlotsUsage();    
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  JobInProgress waitingJob = queue.removeWaitingJob(oldInfo, runState);
  JobInProgress initializingJob = 
    queue.removeInitializingJob(oldInfo, runState);
  JobInProgress runningJob = queue.removeRunningJob(oldInfo, runState);

  // let scheduler know if necessary
  // sometimes this isn't necessary if the job was rejected during submission
  if (runningJob != null || initializingJob != null || waitingJob != null) {
    scheduler.jobCompleted(job);
  }
}

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  if(queue.removeWaitingJob(oldInfo, runState) != null) {
    try {
      queue.addWaitingJob(job);
    } catch (IOException ioe) {
      // Ignore, cannot happen
      LOG.warn("Couldn't change priority!");
      return;
    }
  }
  if (queue.removeInitializingJob(oldInfo, runState) != null) {
    queue.addInitializingJob(job);
  }
  if(queue.removeRunningJob(oldInfo, runState) != null) {
    queue.addRunningJob(job);
  }
}

public CapacitySchedulerQueue(String queueName, CapacitySchedulerConf conf) {
  this.queueName = queueName;

  // Do not allow changes to 'supportsPriorities'
  supportsPriorities = conf.isPrioritySupported(queueName);

  initializeQueue(conf);

  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  this.waitingJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.initializingJobs =
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.runningJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);

  this.mapSlots = new SlotsUsage();
  this.reduceSlots = new SlotsUsage();    
}

private void jobStateChanged(JobStatusChangeEvent event, QueueInfo qi) {
  JobInProgress job = event.getJobInProgress();
  JobSchedulingInfo oldJobStateInfo = 
    new JobSchedulingInfo(event.getOldStatus());
  // Check if the ordering of the job has changed
  // For now priority and start-time can change the job ordering
  if (event.getEventType() == EventType.PRIORITY_CHANGED 
      || event.getEventType() == EventType.START_TIME_CHANGED) {
    // Make a priority change
    reorderJobs(job, oldJobStateInfo, qi);
  } else if (event.getEventType() == EventType.RUN_STATE_CHANGED) {
    // Check if the job is complete
    int runState = job.getStatus().getRunState();
    if (runState == JobStatus.SUCCEEDED
        || runState == JobStatus.FAILED
        || runState == JobStatus.KILLED) {
      jobCompleted(job, oldJobStateInfo, qi);
    } else if (runState == JobStatus.RUNNING) {
      makeJobRunning(job, oldJobStateInfo, qi);
    }
  }
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  JobInProgress waitingJob = queue.removeWaitingJob(oldInfo, runState);
  JobInProgress initializingJob = 
    queue.removeInitializingJob(oldInfo, runState);
  JobInProgress runningJob = queue.removeRunningJob(oldInfo, runState);

  // let scheduler know if necessary
  // sometimes this isn't necessary if the job was rejected during submission
  if (runningJob != null || initializingJob != null || waitingJob != null) {
    scheduler.jobCompleted(job);
  }
}

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  if(queue.removeWaitingJob(oldInfo, runState) != null) {
    try {
      queue.addWaitingJob(job);
    } catch (IOException ioe) {
      // Ignore, cannot happen
      LOG.warn("Couldn't change priority!");
      return;
    }
  }
  if (queue.removeInitializingJob(oldInfo, runState) != null) {
    queue.addInitializingJob(job);
  }
  if(queue.removeRunningJob(oldInfo, runState) != null) {
    queue.addRunningJob(job);
  }
}

public CapacitySchedulerQueue(String queueName, CapacitySchedulerConf conf) {
  this.queueName = queueName;

  // Do not allow changes to 'supportsPriorities'
  supportsPriorities = conf.isPrioritySupported(queueName);

  initializeQueue(conf);

  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  this.waitingJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.initializingJobs =
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  this.runningJobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);

  this.mapSlots = new SlotsUsage();
  this.reduceSlots = new SlotsUsage();    
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
    CapacitySchedulerQueue queue, int runState) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  JobInProgress waitingJob = queue.removeWaitingJob(oldInfo, runState);
  JobInProgress initializingJob = 
    queue.removeInitializingJob(oldInfo, runState);
  JobInProgress runningJob = queue.removeRunningJob(oldInfo, runState);

  // let scheduler know if necessary
  // sometimes this isn't necessary if the job was rejected during submission
  if (runningJob != null || initializingJob != null || waitingJob != null) {
    scheduler.jobCompleted(job);
  }
}

/**
 * This method returns the first job in the queue and removes the same.
 * 
 * @param queue
 *          queue name
 * @return First job in the queue and removes it.
 */
private JobInProgress getFirstJobInQueue(String queue) {
  Map<JobSchedulingInfo, JobInProgress> jobsList = jobsPerQueue.get(queue);
  synchronized (jobsList) {
    if (jobsList.isEmpty()) {
      return null;
    }
    Iterator<JobInProgress> jobIterator = jobsList.values().iterator();
    JobInProgress job = jobIterator.next();
    jobIterator.remove();
    currentJobCount.getAndDecrement();
    return job;
  }
}

void addJobsToQueue(String queue, JobInProgress job) {
  Map<JobSchedulingInfo, JobInProgress> jobs = jobsPerQueue.get(queue);
  if (jobs == null) {
    LOG.error("Invalid queue passed to the thread : " + queue
        + " For job :: " + job.getJobID());
  }
  synchronized (jobs) {
    JobSchedulingInfo schedInfo = new JobSchedulingInfo(job);
    jobs.put(schedInfo, job);
    currentJobCount.getAndIncrement();
  }
}

void addQueue(String queueName) {
  CapacitySchedulerQueue queue = jobQueueManager.getQueue(queueName);

  TreeMap<JobSchedulingInfo, JobInProgress> jobs = 
    new TreeMap<JobSchedulingInfo, JobInProgress>(queue.getComparator());
  jobsPerQueue.put(queueName, jobs);
}

public int compare(JobSchedulingInfo o1, JobSchedulingInfo o2) {
  // the job that started earlier wins
  if (o1.getStartTime() < o2.getStartTime()) {
    return -1;
  } else {
    return (o1.getStartTime() == o2.getStartTime() 
            ? o1.getJobID().compareTo(o2.getJobID()) 
            : 1);
  }
}

public void jobInitializing(JobSchedulingInfo jobSchedInfo, 
    JobInProgress job) {
  if (!initializingJobs.containsKey(jobSchedInfo)) {
    initializingJobs.put(jobSchedInfo, job);
    activeTasks += job.desiredTasks();
  }
}

synchronized JobInProgress removeInitializingJob(
    JobSchedulingInfo jobSchedInfo, int runState) {
  JobInProgress job = initializingJobs.remove(jobSchedInfo);

  if (job != null) {
    String user = job.getProfile().getUser();
    UserInfo userInfo = users.get(user);
    userInfo.removeInitializingJob(jobSchedInfo);

    // Decrement counts if the job is killed _while_ it was selected for
    // initialization, but aborted
    // NOTE: addRunningJob calls removeInitializingJob with runState==RUNNING
    if (runState != JobStatus.RUNNING) {
      finishJob(jobSchedInfo, job);
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("removeInitializingJob:" +
          " job=" + job.getJobID() +
          " user=" + user + 
          " queue=" + queueName +
          " qWaitJobs=" +  getNumWaitingJobs() +
          " qInitJobs=" +  getNumInitializingJobs()+
          " qRunJobs=" +  getNumRunningJobs() +
          " qActiveTasks=" +  getNumActiveTasks() +
          " uWaitJobs=" +  getNumWaitingJobsByUser(user) +
          " uInitJobs=" +  getNumInitializingJobsByUser(user) +
          " uRunJobs=" +  getNumRunningJobsByUser(user) +
          " uActiveTasks=" +  getNumActiveTasksByUser(user)
      );
    }
  }

  return job;
}

synchronized void addRunningJob(JobInProgress job) {
  JobSchedulingInfo jobSchedInfo = new JobSchedulingInfo(job);

  if (runningJobs.containsKey(jobSchedInfo)) {
    LOG.info("job " + job.getJobID() + " already running in queue'" +
        queueName + "'!");
    return;
  }

  // Mark the job as running
  runningJobs.put(jobSchedInfo,job);

  // Update user stats
  String user = job.getProfile().getUser();
  UserInfo userInfo = users.get(user);
  userInfo.jobInitialized(jobSchedInfo, job);

  if (LOG.isDebugEnabled()) {
    LOG.debug("addRunningJob:" +
        " job=" + job.getJobID() +
        " user=" + user + 
        " queue=" + queueName +
        " qWaitJobs=" +  getNumWaitingJobs() +
        " qInitJobs=" +  getNumInitializingJobs()+
        " qRunJobs=" +  getNumRunningJobs() +
        " qActiveTasks=" +  getNumActiveTasks() +
        " uWaitJobs=" +  getNumWaitingJobsByUser(user) +
        " uInitJobs=" +  getNumInitializingJobsByUser(user) +
        " uRunJobs=" +  getNumRunningJobsByUser(user) +
        " uActiveTasks=" +  getNumActiveTasksByUser(user)
    );
  }

  // Remove from 'initializing' list
  // Note that at this point job.status.state != RUNNING, 
  // however, logically it is a reasonable state to pass in to ensure
  // that removeInitializingJob doesn't double-decrement  
  // the relevant queue/user counters
  removeInitializingJob(jobSchedInfo, JobStatus.RUNNING);
}

synchronized private void addJob(JobSchedulingInfo jobSchedInfo,
    JobInProgress job) {
  // Update queue stats
  activeTasks += job.desiredTasks();

  // Update user stats
  String user = job.getProfile().getUser();
  UserInfo userInfo = users.get(user);
  userInfo.jobInitializing(jobSchedInfo, job);
}

synchronized private void finishJob(JobSchedulingInfo jobSchedInfo,
    JobInProgress job) {
  // Update user stats
  String user = job.getProfile().getUser();
  UserInfo userInfo = users.get(user);
  userInfo.jobCompleted(jobSchedInfo, job);

  if (userInfo.isInactive()) {
    users.remove(userInfo);
  }

  // Update queue stats
  activeTasks -= job.desiredTasks();
}

synchronized JobInProgress removeRunningJob(JobSchedulingInfo jobSchedInfo, 
    int runState) {
  JobInProgress job = runningJobs.remove(jobSchedInfo); 

  // We have to be careful, we might be trying to remove a job  
  // which might not have been initialized
  if (job != null) {
    String user = job.getProfile().getUser();
    finishJob(jobSchedInfo, job);

    if (LOG.isDebugEnabled()) {
      LOG.debug("removeRunningJob:" +
          " job=" + job.getJobID() +
          " user=" + user + 
          " queue=" + queueName +
          " qWaitJobs=" +  getNumWaitingJobs() +
          " qInitJobs=" +  getNumInitializingJobs()+
          " qRunJobs=" +  getNumRunningJobs() +
          " qActiveTasks=" +  getNumActiveTasks() +
          " uWaitJobs=" +  getNumWaitingJobsByUser(user) +
          " uInitJobs=" +  getNumInitializingJobsByUser(user) +
          " uRunJobs=" +  getNumRunningJobsByUser(user) +
          " uActiveTasks=" +  getNumActiveTasksByUser(user)
      );
    }
  }

  return job;
}

synchronized void addWaitingJob(JobInProgress job) throws IOException {
  JobSchedulingInfo jobSchedInfo = new JobSchedulingInfo(job);
  if (waitingJobs.containsKey(jobSchedInfo)) {
    LOG.info("job " + job.getJobID() + " already waiting in queue '" + 
        queueName + "'!");
    return;
  }

  String user = job.getProfile().getUser();

  // Check acceptance limits
  checkJobSubmissionLimits(job, user);

  waitingJobs.put(jobSchedInfo, job);

  // Update user stats
  UserInfo userInfo = users.get(user);
  if (userInfo == null) {
    userInfo = new UserInfo(comparator);
    users.put(user, userInfo);
  }
  userInfo.jobAdded(jobSchedInfo, job);

  if (LOG.isDebugEnabled()) {
    LOG.debug("addWaitingJob:" +
        " job=" + job.getJobID() +
        " user=" + user + 
        " queue=" + queueName +
        " qWaitJobs=" +  getNumWaitingJobs() +
        " qInitJobs=" +  getNumInitializingJobs()+
        " qRunJobs=" +  getNumRunningJobs() +
        " qActiveTasks=" +  getNumActiveTasks() +
        " uWaitJobs=" +  getNumWaitingJobsByUser(user) +
        " uInitJobs=" +  getNumInitializingJobsByUser(user) +
        " uRunJobs=" +  getNumRunningJobsByUser(user) +
        " uActiveTasks=" +  getNumActiveTasksByUser(user)
    );
  }
}

synchronized JobInProgress removeWaitingJob(JobSchedulingInfo jobSchedInfo, 
    int unused) {
  JobInProgress job = waitingJobs.remove(jobSchedInfo);
  if (job != null) {
    String user = job.getProfile().getUser();
    UserInfo userInfo = users.get(user);
    userInfo.removeWaitingJob(jobSchedInfo);

    if (LOG.isDebugEnabled()) {
      LOG.debug("removeWaitingJob:" +
          " job=" + job.getJobID() +
          " user=" + user + 
          " queue=" + queueName +
          " qWaitJobs=" +  getNumWaitingJobs() +
          " qInitJobs=" +  getNumInitializingJobs()+
          " qRunJobs=" +  getNumRunningJobs() +
          " qActiveTasks=" +  getNumActiveTasks() +
          " uWaitJobs=" +  getNumWaitingJobsByUser(user) +
          " uInitJobs=" +  getNumInitializingJobsByUser(user) +
          " uRunJobs=" +  getNumRunningJobsByUser(user) +
          " uActiveTasks=" +  getNumActiveTasksByUser(user)
      );
    }
  }

  return job;
}

private void makeJobRunning(JobInProgress job, JobSchedulingInfo oldInfo, 
                            CapacitySchedulerQueue queue) {
  // Removing of the job from job list is responsibility of the
  //initialization poller.
  // Add the job to the running queue
  queue.addRunningJob(job);
}

/**
 * This method returns the first job in the queue and removes the same.
 * 
 * @param queue
 *          queue name
 * @return First job in the queue and removes it.
 */
private JobInProgress getFirstJobInQueue(String queue) {
  TreeMap<JobSchedulingInfo, JobInProgress> jobsList = jobsPerQueue
      .get(queue);
  synchronized (jobsList) {
    if (jobsList.isEmpty()) {
      return null;
    }
    Iterator<JobInProgress> jobIterator = jobsList.values().iterator();
    JobInProgress job = jobIterator.next();
    jobIterator.remove();
    currentJobCount.getAndDecrement();
    return job;
  }
}

void addJobsToQueue(String queue, JobInProgress job) {
  TreeMap<JobSchedulingInfo, JobInProgress> jobs = jobsPerQueue
      .get(queue);
  if (jobs == null) {
    LOG.error("Invalid queue passed to the thread : " + queue
        + " For job :: " + job.getJobID());
  }
  synchronized (jobs) {
    JobSchedulingInfo schedInfo = new JobSchedulingInfo(job);
    jobs.put(schedInfo, job);
    currentJobCount.getAndIncrement();
  }
}

public int compare(JobSchedulingInfo o1, JobSchedulingInfo o2) {
  // the job that started earlier wins
  if (o1.getStartTime() < o2.getStartTime()) {
    return -1;
  } else {
    return (o1.getStartTime() == o2.getStartTime() 
            ? o1.getJobID().compareTo(o2.getJobID()) 
            : 1);
  }
}

QueueInfo(boolean prio) {
  this.supportsPriorities = prio;
  if (supportsPriorities) {
    // use the default priority-aware comparator
    comparator = JobQueueJobInProgressListener.FIFO_JOB_QUEUE_COMPARATOR;
  }
  else {
    comparator = STARTTIME_JOB_COMPARATOR;
  }
  waitingJobs = new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
  runningJobs = new TreeMap<JobSchedulingInfo, JobInProgress>(comparator);
}

private void jobCompleted(JobInProgress job, JobSchedulingInfo oldInfo, 
                          QueueInfo qi) {
  LOG.info("Job " + job.getJobID().toString() + " submitted to queue " 
      + job.getProfile().getQueueName() + " has completed");
  //remove jobs from both queue's a job can be in
  //running and waiting queue at the same time.
  qi.removeRunningJob(oldInfo);
  qi.removeWaitingJob(oldInfo);
  // let scheduler know
  scheduler.jobCompleted(job);
}

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo, 
                         QueueInfo qi) {

  if(qi.removeWaitingJob(oldInfo) != null) {
    qi.addWaitingJob(job);
  }
  if(qi.removeRunningJob(oldInfo) != null) {
    qi.addRunningJob(job);
  }
}

private void makeJobRunning(JobInProgress job, JobSchedulingInfo oldInfo, 
                            QueueInfo qi) {
  // Removing of the job from job list is responsibility of the
  //initialization poller.
  // Add the job to the running queue
  qi.addRunningJob(job);
}

/**
 * This method returns the first job in the queue and removes the same.
 * 
 * @param queue
 *          queue name
 * @return First job in the queue and removes it.
 */
private JobInProgress getFirstJobInQueue(String queue) {
  Map<JobSchedulingInfo, JobInProgress> jobsList = jobsPerQueue.get(queue);
  synchronized (jobsList) {
    if (jobsList.isEmpty()) {
      return null;
    }
    Iterator<JobInProgress> jobIterator = jobsList.values().iterator();
    JobInProgress job = jobIterator.next();
    jobIterator.remove();
    currentJobCount.getAndDecrement();
    return job;
  }
}

void addJobsToQueue(String queue, JobInProgress job) {
  Map<JobSchedulingInfo, JobInProgress> jobs = jobsPerQueue.get(queue);
  if (jobs == null) {
    LOG.error("Invalid queue passed to the thread : " + queue
        + " For job :: " + job.getJobID());
  }
  synchronized (jobs) {
    JobSchedulingInfo schedInfo = new JobSchedulingInfo(job);
    jobs.put(schedInfo, job);
    currentJobCount.getAndIncrement();
  }
}