/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.mapred;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType;
import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;

/**
 * A {@link JobInProgressListener} which initializes the tasks for a job as soon
 * as the job is added (using the {@link #jobAdded(JobInProgress)} method).
 */
class EagerTaskInitializationListener extends JobInProgressListener {
  
  private static final int DEFAULT_NUM_THREADS = 4;
  private static final Log LOG = LogFactory.getLog(
      EagerTaskInitializationListener.class.getName());
  
  /////////////////////////////////////////////////////////////////
  //  Used to init new jobs that have just been created
  /////////////////////////////////////////////////////////////////
  class JobInitManager implements Runnable {
   
    public void run() {
      JobInProgress job = null;
      while (true) {
        try {
          synchronized (jobInitQueue) {
            while (jobInitQueue.isEmpty()) {
              jobInitQueue.wait();
            }
            job = jobInitQueue.remove(0);
          }
          threadPool.execute(new InitJob(job));
        } catch (InterruptedException t) {
          LOG.info("JobInitManagerThread interrupted.");
          break;
        } 
      }
      LOG.info("Shutting down thread pool");
      threadPool.shutdownNow();
    }
  }
  
  class InitJob implements Runnable {
  
    private JobInProgress job;
    
    public InitJob(JobInProgress job) {
      this.job = job;
    }
    
    public void run() {
      ttm.initJob(job);
    }
  }
  
  private JobInitManager jobInitManager = new JobInitManager();
  private Thread jobInitManagerThread;
  private List<JobInProgress> jobInitQueue = new ArrayList<JobInProgress>();
  private ExecutorService threadPool;
  private int numThreads;
  TaskTrackerManager ttm;
  
  public EagerTaskInitializationListener(Configuration conf) {
    numThreads = 
      conf.getInt(JTConfig.JT_JOBINIT_THREADS, DEFAULT_NUM_THREADS);
    threadPool = Executors.newFixedThreadPool(numThreads);
  }
  
  public void setTaskTrackerManager(TaskTrackerManager ttm) {
    this.ttm = ttm;
  }
  
  public void start() throws IOException {
    this.jobInitManagerThread = new Thread(jobInitManager, "jobInitManager");
    jobInitManagerThread.setDaemon(true);
    this.jobInitManagerThread.start();
  }
  
  public void terminate() throws IOException {
    if (jobInitManagerThread != null && jobInitManagerThread.isAlive()) {
      LOG.info("Stopping Job Init Manager thread");
      jobInitManagerThread.interrupt();
      try {
        jobInitManagerThread.join();
      } catch (InterruptedException ex) {
        ex.printStackTrace();
      }
    }
  }

  /**
   * We add the JIP to the jobInitQueue, which is processed 
   * asynchronously to handle split-computation and build up
   * the right TaskTracker/Block mapping.
   */
  @Override
  public void jobAdded(JobInProgress job) {
    synchronized (jobInitQueue) {
      jobInitQueue.add(job);
      resortInitQueue();
      jobInitQueue.notifyAll();
    }

  }
  
  /**
   * Sort jobs by priority and then by start time.
   */
  private synchronized void resortInitQueue() {
    Comparator<JobInProgress> comp = new Comparator<JobInProgress>() {
      public int compare(JobInProgress o1, JobInProgress o2) {
        int res = o1.getPriority().compareTo(o2.getPriority());
        if(res == 0) {
          if(o1.getStartTime() < o2.getStartTime())
            res = -1;
          else
            res = (o1.getStartTime()==o2.getStartTime() ? 0 : 1);
        }
          
        return res;
      }
    };
    
    synchronized (jobInitQueue) {
      Collections.sort(jobInitQueue, comp);
    }
  }

  @Override
  public void jobRemoved(JobInProgress job) {
    synchronized (jobInitQueue) {
      jobInitQueue.remove(job);
    }
  }

  @Override
  public void jobUpdated(JobChangeEvent event) {
    if (event instanceof JobStatusChangeEvent) {
      jobStateChanged((JobStatusChangeEvent)event);
    }
  }
  
  // called when the job's status is changed
  private void jobStateChanged(JobStatusChangeEvent event) {
    // Resort the job queue if the job-start-time or job-priority changes
    if (event.getEventType() == EventType.START_TIME_CHANGED
        || event.getEventType() == EventType.PRIORITY_CHANGED) {
      synchronized (jobInitQueue) {
        resortInitQueue();
      }
    }
  }

}