/**
* 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.mapreduce.v2.app.rm;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptContainerAssignedEvent;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.exceptions.YarnException;

import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;

/**
 * Keeps the data for RMContainer's reuse.
 */
public class RMContainerReuseRequestor extends RMContainerRequestor
    implements EventHandler<ContainerAvailableEvent> {
  private static final Log LOG = LogFactory
      .getLog(RMContainerReuseRequestor.class);

  private Map<Container, HostInfo> containersToReuse =
      new ConcurrentHashMap<>();
  private Map<ContainerId, List<TaskAttemptId>> containerToTaskAttemptsMap =
      new HashMap<ContainerId, List<TaskAttemptId>>();
  private int containerReuseMaxMapTasks;
  private int containerReuseMaxReduceTasks;
  private int maxMapTaskContainers;
  private int maxReduceTaskContainers;
  private int noOfMapTaskContainersForReuse;
  private int noOfReduceTaskContainersForReuse;
  private final RMCommunicator rmCommunicator;
  @SuppressWarnings("rawtypes")
  private final EventHandler eventHandler;

  @SuppressWarnings("rawtypes")
  public RMContainerReuseRequestor(
      EventHandler eventHandler,
      RMCommunicator rmCommunicator) {
    super(eventHandler, rmCommunicator);
    this.rmCommunicator = rmCommunicator;
    this.eventHandler = eventHandler;
  }

  @Override
  protected void serviceInit(Configuration conf) throws Exception {
    super.serviceInit(conf);
    containerReuseMaxMapTasks = conf.getInt(
        MRJobConfig.MR_AM_CONTAINER_REUSE_MAX_MAPTASKS,
        MRJobConfig.DEFAULT_MR_AM_CONTAINER_REUSE_MAX_MAPTASKS);
    containerReuseMaxReduceTasks = conf.getInt(
        MRJobConfig.MR_AM_CONTAINER_REUSE_MAX_REDUCETASKS,
        MRJobConfig.DEFAULT_MR_AM_CONTAINER_REUSE_MAX_REDUCETASKS);
    maxMapTaskContainers = conf.getInt(
        MRJobConfig.MR_AM_CONTAINER_REUSE_MAX_MAPTASKCONTAINERS,
        MRJobConfig.DEFAULT_MR_AM_CONTAINER_REUSE_MAX_MAPTASKCONTAINERS);
    maxReduceTaskContainers = conf.getInt(
        MRJobConfig.MR_AM_CONTAINER_REUSE_MAX_REDUCETASKCONTAINERS,
        MRJobConfig.DEFAULT_MR_AM_CONTAINER_REUSE_MAX_REDUCETASKCONTAINERS);
  }

  @Override
  protected void serviceStart() throws Exception {
    super.serviceStart();
  }

  @Override
  public AllocateResponse makeRemoteRequest()
      throws YarnException, IOException {
    AllocateResponse amResponse = super.makeRemoteRequest();
    synchronized (containersToReuse) {
      List<Container> allocatedContainers = amResponse.getAllocatedContainers();
      allocatedContainers.addAll(containersToReuse.keySet());
      containersToReuse.clear();
    }
    return amResponse;
  }

  @Override
  public void containerFailedOnHost(String hostName) {
    super.containerFailedOnHost(hostName);
    boolean blacklisted = super.isNodeBlacklisted(hostName);
    if (blacklisted) {
      Set<Container> containersOnHost = new HashSet<Container>();
      for (Entry<Container, HostInfo> elem : containersToReuse.entrySet()) {
        if (elem.getValue().getHost().equals(hostName)) {
          containersOnHost.add(elem.getKey());
        }
      }
      for (Container container : containersOnHost) {
        containersToReuse.remove(container);
      }
    }
  }

  @Override
  public void handle(ContainerAvailableEvent event) {
    Container container = event.getContainer();
    ContainerId containerId = container.getId();
    String resourceName = container.getNodeId().getHost();
    boolean canReuse = false;
    Priority priority = container.getPriority();
    if (RMContainerAllocator.PRIORITY_MAP.equals(priority)
        || RMContainerAllocator.PRIORITY_REDUCE.equals(priority)) {
      List<TaskAttemptId> containerTaskAttempts = null;
      containerTaskAttempts = containerToTaskAttemptsMap.get(containerId);
      if (containerTaskAttempts == null) {
        containerTaskAttempts = new ArrayList<TaskAttemptId>();
        containerToTaskAttemptsMap.put(containerId, containerTaskAttempts);
      }
      TaskAttemptId taskAttemptId = event.getTaskAttemptId();
      if (checkMapContainerReuseConstraints(priority, containerTaskAttempts)
          || checkReduceContainerReuseConstraints(priority,
              containerTaskAttempts)) {
        Map<String, Map<Resource, ResourceRequest>> resourceRequests =
            remoteRequestsTable.get(priority);
        // If there are any eligible requests
        if (resourceRequests != null && !resourceRequests.isEmpty()) {
          canReuse = true;
          containerTaskAttempts.add(taskAttemptId);
        }
      }
      ((RMContainerAllocator) rmCommunicator)
          .resetContainerForReuse(container.getId());
      if (canReuse) {
        int shufflePort =
            rmCommunicator.getJob().getTask(taskAttemptId.getTaskId())
                .getAttempt(taskAttemptId).getShufflePort();
        containersToReuse.put(container,
            new HostInfo(resourceName, shufflePort));
        incrementRunningReuseContainers(priority);
        LOG.info("Adding the " + containerId + " for reuse.");
      } else {
        LOG.info("Releasing the container : " + containerId
            + " since it is not eligible for reuse or no pending requests.");
        containerComplete(container);
        pendingRelease.add(containerId);
        release(containerId);
      }
    }
  }

  private boolean checkMapContainerReuseConstraints(Priority priority,
      List<TaskAttemptId> containerTaskAttempts) {
    return RMContainerAllocator.PRIORITY_MAP.equals(priority)
        // Check for how many tasks can map task container run maximum
        && ((containerTaskAttempts.size() < containerReuseMaxMapTasks
            || containerReuseMaxMapTasks == -1)
            // Check for no of map task containers running
            && (noOfMapTaskContainersForReuse < maxMapTaskContainers
                || maxMapTaskContainers == -1));
  }

  private boolean checkReduceContainerReuseConstraints(Priority priority,
      List<TaskAttemptId> containerTaskAttempts) {
    return RMContainerAllocator.PRIORITY_REDUCE.equals(priority)
        // Check for how many tasks can reduce task container run maximum
        && ((containerTaskAttempts.size() < containerReuseMaxReduceTasks
            || containerReuseMaxReduceTasks == -1)
            // Check for no of reduce task containers running
            && (noOfReduceTaskContainersForReuse < maxReduceTaskContainers
                || maxReduceTaskContainers == -1));
  }

  private void containerComplete(Container container) {
    if (!containerToTaskAttemptsMap.containsKey(container.getId())) {
      return;
    }
    containerToTaskAttemptsMap.remove(container.getId());
    if (RMContainerAllocator.PRIORITY_MAP.equals(container.getPriority())) {
      noOfMapTaskContainersForReuse--;
    } else if (RMContainerAllocator.PRIORITY_REDUCE
        .equals(container.getPriority())) {
      noOfReduceTaskContainersForReuse--;
    }
  }

  private void incrementRunningReuseContainers(Priority priority) {
    if (RMContainerAllocator.PRIORITY_MAP.equals(priority)) {
      noOfMapTaskContainersForReuse++;
    } else if (RMContainerAllocator.PRIORITY_REDUCE.equals(priority)) {
      noOfReduceTaskContainersForReuse++;
    }
  }

  @Private
  @VisibleForTesting
  Map<Container, HostInfo> getContainersToReuse() {
    return containersToReuse;
  }

  /**
   * Container Available EventType.
   */
  public static enum EventType {
    CONTAINER_AVAILABLE
  }

  @SuppressWarnings("unchecked")
  @Override
  public void containerAssigned(Container allocated, ContainerRequest req,
      Map<ApplicationAccessType, String> applicationACLs) {
    if(containersToReuse.containsKey(allocated)){
      decContainerReq(req);
      // send the container-assigned event to task attempt
      eventHandler.handle(new TaskAttemptContainerAssignedEvent(
          req.attemptID, allocated, applicationACLs));
    } else {
      super.containerAssigned(allocated, req, applicationACLs);
    }
  }

  static class HostInfo {
    private String host;
    private int port;
    public HostInfo(String host, int port) {
      super();
      this.host = host;
      this.port = port;
    }
    public String getHost() {
      return host;
    }
    public int getPort() {
      return port;
    }
  }
}