/*
 * 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.reef.runtime.common.driver.evaluator;

import org.apache.reef.annotations.audience.DriverSide;
import org.apache.reef.annotations.audience.Private;
import org.apache.reef.driver.evaluator.FailedEvaluator;
import org.apache.reef.driver.parameters.EvaluatorConfigurationProviders;
import org.apache.reef.driver.restart.DriverRestartManager;
import org.apache.reef.driver.restart.EvaluatorRestartState;
import org.apache.reef.exception.NonSerializableException;
import org.apache.reef.runtime.common.driver.api.*;
import org.apache.reef.runtime.common.driver.evaluator.pojos.ContextStatusPOJO;
import org.apache.reef.runtime.common.driver.evaluator.pojos.EvaluatorStatusPOJO;
import org.apache.reef.runtime.common.driver.evaluator.pojos.State;
import org.apache.reef.runtime.common.driver.evaluator.pojos.TaskStatusPOJO;
import org.apache.reef.tang.ConfigurationProvider;
import org.apache.reef.driver.context.ActiveContext;
import org.apache.reef.driver.context.FailedContext;
import org.apache.reef.driver.evaluator.AllocatedEvaluator;
import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
import org.apache.reef.driver.task.FailedTask;
import org.apache.reef.exception.EvaluatorException;
import org.apache.reef.exception.EvaluatorKilledByResourceManagerException;
import org.apache.reef.io.naming.Identifiable;
import org.apache.reef.proto.EvaluatorRuntimeProtocol;
import org.apache.reef.proto.ReefServiceProtos;
import org.apache.reef.driver.evaluator.EvaluatorProcess;
import org.apache.reef.runtime.common.driver.context.ContextControlHandler;
import org.apache.reef.runtime.common.driver.context.ContextRepresenters;
import org.apache.reef.runtime.common.driver.idle.EventHandlerIdlenessSource;
import org.apache.reef.runtime.common.driver.resourcemanager.ResourceStatusEvent;
import org.apache.reef.runtime.common.driver.task.TaskRepresenter;
import org.apache.reef.runtime.common.utils.ExceptionCodec;
import org.apache.reef.runtime.common.utils.RemoteManager;
import org.apache.reef.tang.annotations.Name;
import org.apache.reef.tang.annotations.NamedParameter;
import org.apache.reef.tang.annotations.Parameter;
import org.apache.reef.tang.formats.ConfigurationSerializer;
import org.apache.reef.util.Optional;
import org.apache.reef.util.logging.LoggingScopeFactory;
import org.apache.reef.wake.EventHandler;
import org.apache.reef.wake.remote.RemoteMessage;
import org.apache.reef.wake.time.Clock;
import org.apache.reef.wake.time.event.Alarm;

import javax.inject.Inject;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Manages a single Evaluator instance including all lifecycle instances:
 * (AllocatedEvaluator, CompletedEvaluator, FailedEvaluator).
 * <p>
 * A (periodic) heartbeat channel is established from EvaluatorRuntime to EvaluatorManager.
 * The EvaluatorRuntime will (periodically) send (status) messages to the EvaluatorManager using this
 * heartbeat channel.
 * <p>
 * A (push-based) EventHandler channel is established from EvaluatorManager to EvaluatorRuntime.
 * The EvaluatorManager uses this to forward Driver messages, launch Tasks, and initiate
 * control information (e.g., shutdown, suspend).
 */
@Private
@DriverSide
public final class EvaluatorManager implements Identifiable, AutoCloseable {

  private static final Logger LOG = Logger.getLogger(EvaluatorManager.class.getName());

  private final EvaluatorHeartBeatSanityChecker sanityChecker = new EvaluatorHeartBeatSanityChecker();
  private final Clock clock;
  private final ResourceReleaseHandler resourceReleaseHandler;
  private final ResourceLaunchHandler resourceLaunchHandler;
  private final String evaluatorId;
  private final EvaluatorDescriptorImpl evaluatorDescriptor;
  private final ContextRepresenters contextRepresenters;
  private final EvaluatorMessageDispatcher messageDispatcher;
  private final EvaluatorControlHandler evaluatorControlHandler;
  private final ContextControlHandler contextControlHandler;
  private final EvaluatorStatusManager stateManager;
  private final ExceptionCodec exceptionCodec;
  private final EventHandlerIdlenessSource idlenessSource;
  private final RemoteManager remoteManager;
  private final ConfigurationSerializer configurationSerializer;
  private final LoggingScopeFactory loggingScopeFactory;
  private final Set<ConfigurationProvider> evaluatorConfigurationProviders;
  private final DriverRestartManager driverRestartManager;
  private final EvaluatorIdlenessThreadPool idlenessThreadPool;

  // Mutable fields
  private Optional<TaskRepresenter> task = Optional.empty();
  private boolean resourceNotReleased = true;
  private boolean allocationNotFired = true;

  @Inject
  private EvaluatorManager(
      @Parameter(EvaluatorIdentifier.class) final String evaluatorId,
      @Parameter(EvaluatorDescriptorName.class) final EvaluatorDescriptorImpl evaluatorDescriptor,
      @Parameter(EvaluatorConfigurationProviders.class)
        final Set<ConfigurationProvider> evaluatorConfigurationProviders,
      final Clock clock,
      final RemoteManager remoteManager,
      final ResourceReleaseHandler resourceReleaseHandler,
      final ResourceLaunchHandler resourceLaunchHandler,
      final ContextRepresenters contextRepresenters,
      final ConfigurationSerializer configurationSerializer,
      final EvaluatorMessageDispatcher messageDispatcher,
      final EvaluatorControlHandler evaluatorControlHandler,
      final ContextControlHandler contextControlHandler,
      final EvaluatorStatusManager stateManager,
      final ExceptionCodec exceptionCodec,
      final EventHandlerIdlenessSource idlenessSource,
      final LoggingScopeFactory loggingScopeFactory,
      final DriverRestartManager driverRestartManager,
      final EvaluatorIdlenessThreadPool idlenessThreadPool) {

    LOG.log(Level.FINEST, "Instantiating 'EvaluatorManager' for evaluator: {0}", evaluatorId);

    this.evaluatorId = evaluatorId;
    this.evaluatorDescriptor = evaluatorDescriptor;
    this.evaluatorConfigurationProviders = evaluatorConfigurationProviders;

    this.clock = clock;
    this.contextRepresenters = contextRepresenters;
    this.idlenessSource = idlenessSource;
    this.resourceReleaseHandler = resourceReleaseHandler;
    this.resourceLaunchHandler = resourceLaunchHandler;

    this.messageDispatcher = messageDispatcher;
    this.evaluatorControlHandler = evaluatorControlHandler;
    this.contextControlHandler = contextControlHandler;
    this.stateManager = stateManager;
    this.exceptionCodec = exceptionCodec;

    this.remoteManager = remoteManager;
    this.configurationSerializer = configurationSerializer;
    this.loggingScopeFactory = loggingScopeFactory;
    this.driverRestartManager = driverRestartManager;
    this.idlenessThreadPool = idlenessThreadPool;

    LOG.log(Level.FINEST, "Instantiated 'EvaluatorManager' for evaluator: [{0}]", this.getId());
  }

  /**
   * Get the id of current job/application.
   */
  public static String getJobIdentifier() {
    // TODO[JIRA REEF-818]: currently we obtain the job id directly by parsing execution (container) directory path
    // #845 is open to get the id from RM properly
    for (File directory = new File(System.getProperty("user.dir"));
         directory != null; directory = directory.getParentFile()) {
      final String currentDirectoryName = directory.getName();
      if (currentDirectoryName.toLowerCase().contains("application_")) {
        return currentDirectoryName;
      }
    }
    // cannot find a directory that contains application_, presumably we are on local runtime
    // again, this is a hack for now, we need #845 as a proper solution
    return "REEF_LOCAL_RUNTIME";
  }

  /**
   * Fires the EvaluatorAllocatedEvent to the handlers. Can only be done once.
   */
  public synchronized void fireEvaluatorAllocatedEvent() {

    if (this.stateManager.isAllocated() && this.allocationNotFired) {

      final AllocatedEvaluator allocatedEvaluator =
          new AllocatedEvaluatorImpl(this,
              this.remoteManager.getMyIdentifier(),
              this.configurationSerializer,
              getJobIdentifier(),
              this.loggingScopeFactory,
              this.evaluatorConfigurationProviders);

      LOG.log(Level.FINEST, "Firing AllocatedEvaluator event for Evaluator with ID [{0}]", this.evaluatorId);

      this.messageDispatcher.onEvaluatorAllocated(allocatedEvaluator);
      this.allocationNotFired = false;

    } else {
      LOG.log(Level.WARNING, "AllocatedEvaluator event fired more than once.");
    }
  }

  @Override
  public String getId() {
    return this.evaluatorId;
  }

  public void setProcess(final EvaluatorProcess process) {
    this.evaluatorDescriptor.setProcess(process);
  }

  public EvaluatorDescriptor getEvaluatorDescriptor() {
    return this.evaluatorDescriptor;
  }

  @Override
  public void close() {

    LOG.log(Level.FINER, "Close EvaluatorManager {0} - begin", this.evaluatorId);

    synchronized (this.evaluatorDescriptor) {

      if (this.stateManager.isAvailable()) {

        LOG.log(Level.WARNING, "Dirty shutdown of running evaluator id[{0}]", getId());

        try {

          if (this.stateManager.isRunning()) {

            // Killing the evaluator means that it doesn't need to send a confirmation; it just dies.
            this.sendEvaluatorControlMessage(
                EvaluatorRuntimeProtocol.EvaluatorControlProto.newBuilder()
                    .setTimestamp(System.currentTimeMillis())
                    .setIdentifier(getId())
                    .setKillEvaluator(EvaluatorRuntimeProtocol.KillEvaluatorProto.newBuilder().build())
                    .build());

            this.stateManager.setClosing();

          } else {
            this.stateManager.setKilled();
          }

        } catch (Exception e) {
          LOG.log(Level.WARNING, "Exception occurred when manager sends killing message to task.", e);
          this.stateManager.setKilled();
        }
      }

      if (this.resourceNotReleased) {

        this.resourceNotReleased = false;

        final ResourceReleaseEvent releaseEvent = ResourceReleaseEventImpl.newBuilder()
            .setIdentifier(this.evaluatorId)
            .setRuntimeName(this.getEvaluatorDescriptor().getRuntimeName())
            .build();

        try {
          // We need to wait awhile before returning the container to the RM
          // in order to give the EvaluatorRuntime (and Launcher) time to cleanly exit.
          this.clock.scheduleAlarm(200, new EventHandler<Alarm>() {
            @Override
            public void onNext(final Alarm alarm) {
              LOG.log(Level.FINER, "Close EvaluatorManager {0} - release to RM", evaluatorId);
              resourceReleaseHandler.onNext(releaseEvent);
              shutdown();
            }
          });
        } catch (final IllegalStateException e) {
          LOG.log(Level.WARNING, "Force resource release because the client closed the clock.", e);
          this.resourceReleaseHandler.onNext(releaseEvent);
          this.shutdown();
        }
      }
    }

    this.idlenessThreadPool.runCheckAsync(this);

    LOG.log(Level.FINER, "Close EvaluatorManager {0} - end", this.evaluatorId);
  }

  /**
   * Close message dispatcher for the evaluator.
   */
  public void shutdown() {
    LOG.log(Level.FINEST, "Shutdown EvaluatorManager: {0}", this.evaluatorId);
    this.messageDispatcher.close();
  }

  /**
   * Return true if the state is DONE, FAILED, or KILLED,
   * <em>and</em> there are no messages queued or in processing.
   */
  public boolean isClosed() {
    return this.messageDispatcher.isEmpty() && this.stateManager.isCompleted();
  }

  /**
   * Return true if the state is CLOSING.
   */
  public boolean isClosing() {
    return this.stateManager.isClosing();
  }

  /**
   * Return true if the state is DONE, FAILED, KILLED, or CLOSING.
   */
  public boolean isClosedOrClosing() {
    return isClosed() || isClosing();
  }

  /**
   * Triggers a call to check the idleness of the Evaluator.
   */
  void checkIdlenessSource() {
    this.idlenessSource.check();
  }

  /**
   * EvaluatorException will trigger is FailedEvaluator and state transition to FAILED.
   *
   * @param exception on the EvaluatorRuntime
   */
  public void onEvaluatorException(final EvaluatorException exception) {
    synchronized (this.evaluatorDescriptor) {

      if (this.stateManager.isCompleted()) {
        LOG.log(Level.FINE,
            "Ignoring an exception received for Evaluator {0} which is already in state {1}.",
            new Object[] {this.getId(), this.stateManager});
        return;
      }

      LOG.log(Level.WARNING, "Failed evaluator: " + getId(), exception);

      try {

        final List<FailedContext> failedContextList = this.contextRepresenters.getFailedContextsForEvaluatorFailure();

        final Optional<FailedTask> failedTaskOptional;
        if (this.task.isPresent()) {

          final String taskId = this.task.get().getId();
          final Optional<ActiveContext> evaluatorContext = Optional.empty();
          final Optional<byte[]> bytes = Optional.empty();
          final Optional<Throwable> taskException = Optional.<Throwable>of(new Exception("Evaluator crash"));
          final String message = "Evaluator crash";
          final Optional<String> description = Optional.empty();
          final FailedTask failedTask =
              new FailedTask(taskId, message, description, taskException, bytes, evaluatorContext);

          failedTaskOptional = Optional.of(failedTask);

        } else {
          failedTaskOptional = Optional.empty();
        }

        final FailedEvaluator failedEvaluator = new FailedEvaluatorImpl(
            exception, failedContextList, failedTaskOptional, this.evaluatorId);

        if (driverRestartManager.getEvaluatorRestartState(evaluatorId).isFailedOrExpired()) {
          this.messageDispatcher.onDriverRestartEvaluatorFailed(failedEvaluator);
        } else {
          this.messageDispatcher.onEvaluatorFailed(failedEvaluator);
        }
      } catch (final Exception e) {
        LOG.log(Level.SEVERE, "Exception while handling FailedEvaluator", e);
      } finally {
        this.stateManager.setFailed();
        this.close();
      }
    }
  }

  /**
   * Process an evaluator heartbeat message.
   */
  public void onEvaluatorHeartbeatMessage(
      final RemoteMessage<EvaluatorRuntimeProtocol.EvaluatorHeartbeatProto> evaluatorHeartbeatProtoRemoteMessage) {

    final EvaluatorRuntimeProtocol.EvaluatorHeartbeatProto evaluatorHeartbeatProto =
        evaluatorHeartbeatProtoRemoteMessage.getMessage();

    LOG.log(Level.FINEST, "Evaluator heartbeat: {0}", evaluatorHeartbeatProto);

    synchronized (this.evaluatorDescriptor) {

      if (this.stateManager.isCompleted()) {

        LOG.log(Level.FINE,
            "Ignoring a heartbeat received for Evaluator {0} which is already in state {1}.",
            new Object[] {this.getId(), this.stateManager});

        return;

      } else if (this.stateManager.isAvailable()) {

        this.sanityChecker.check(this.evaluatorId, evaluatorHeartbeatProto.getTimestamp());
        final String evaluatorRID = evaluatorHeartbeatProtoRemoteMessage.getIdentifier().toString();

        final EvaluatorRestartState evaluatorRestartState =
            this.driverRestartManager.getEvaluatorRestartState(this.evaluatorId);

        /*
         * First message from a running evaluator. The evaluator can be a new evaluator or be a previous evaluator
         * from a separate application attempt. In the case of a previous evaluator, if the restart period has not
         * yet expired, we should register it and trigger context active and task events. If the restart period has
         * expired, we should return immediately after setting its remote ID in order to close it.
         */
        if (this.stateManager.isSubmitted() ||
            evaluatorRestartState == EvaluatorRestartState.REPORTED ||
            evaluatorRestartState == EvaluatorRestartState.EXPIRED) {

          this.evaluatorControlHandler.setRemoteID(evaluatorRID);

          if (evaluatorRestartState == EvaluatorRestartState.EXPIRED) {
            // Don't do anything if evaluator has expired. Close it immediately upon exit of this method.
            return;
          }

          this.stateManager.setRunning();
          LOG.log(Level.FINEST, "Evaluator {0} is running", this.evaluatorId);

          if (evaluatorRestartState == EvaluatorRestartState.REPORTED) {
            this.driverRestartManager.setEvaluatorReregistered(this.evaluatorId);
          }
        }
      }

      // All messages from a heartbeat receive the heartbeat timestamp as a sequence number. See REEF-806.
      final long messageSequenceNumber = evaluatorHeartbeatProto.getTimestamp();

      // Process the Evaluator status message
      if (evaluatorHeartbeatProto.hasEvaluatorStatus()) {
        this.onEvaluatorStatusMessage(new EvaluatorStatusPOJO(evaluatorHeartbeatProto.getEvaluatorStatus()));
      }

      // Process the Context status message(s)
      final boolean informClientOfNewContexts = !evaluatorHeartbeatProto.hasTaskStatus();
      final List<ContextStatusPOJO> contextStatusList = new ArrayList<>();
      for (ReefServiceProtos.ContextStatusProto proto : evaluatorHeartbeatProto.getContextStatusList()) {
        contextStatusList.add(new ContextStatusPOJO(proto, messageSequenceNumber));
      }

      this.contextRepresenters.onContextStatusMessages(contextStatusList, informClientOfNewContexts);

      // Process the Task status message
      if (evaluatorHeartbeatProto.hasTaskStatus()) {
        this.onTaskStatusMessage(new TaskStatusPOJO(evaluatorHeartbeatProto.getTaskStatus(), messageSequenceNumber));
      }

      LOG.log(Level.FINE, "DONE with evaluator heartbeat from Evaluator {0}", this.getId());
    }
  }

  /**
   * Process a evaluator status message.
   *
   * @param message
   */
  private synchronized void onEvaluatorStatusMessage(final EvaluatorStatusPOJO message) {

    switch (message.getState()) {
    case DONE:
      this.onEvaluatorDone(message);
      break;
    case FAILED:
      this.onEvaluatorFailed(message);
      break;
    case KILLED:
      this.onEvaluatorKilled(message);
      break;
    case INIT:
    case RUNNING:
    case SUSPEND:
      break;
    default:
      throw new RuntimeException("Unknown state: " + message.getState());
    }
  }

  /**
   * Process an evaluator message that indicates that the evaluator shut down cleanly.
   *
   * @param message
   */
  private synchronized void onEvaluatorDone(final EvaluatorStatusPOJO message) {

    assert message.getState() == State.DONE;

    LOG.log(Level.FINEST, "Evaluator {0} done.", getId());

    // Send an ACK to the Evaluator.
    sendEvaluatorControlMessage(
        EvaluatorRuntimeProtocol.EvaluatorControlProto.newBuilder()
            .setTimestamp(System.currentTimeMillis())
            .setIdentifier(getId())
            .setDoneEvaluator(EvaluatorRuntimeProtocol.DoneEvaluatorProto.newBuilder().build())
            .build());

    this.stateManager.setDone();
    this.messageDispatcher.onEvaluatorCompleted(new CompletedEvaluatorImpl(this.evaluatorId));

    this.close();
  }

  /**
   * Process an evaluator message that indicates a crash.
   *
   * @param evaluatorStatus
   */
  private synchronized void onEvaluatorFailed(final EvaluatorStatusPOJO evaluatorStatus) {

    assert evaluatorStatus.getState() == State.FAILED;

    final EvaluatorException evaluatorException;

    if (evaluatorStatus.hasError()) {

      final Optional<Throwable> exception =
          this.exceptionCodec.fromBytes(evaluatorStatus.getError());

      evaluatorException = new EvaluatorException(getId(), exception.isPresent() ? exception.get() :
          new NonSerializableException("Exception sent, but can't be deserialized", evaluatorStatus.getError()));

    } else {
      evaluatorException = new EvaluatorException(getId(), new Exception("No exception sent"));
    }

    this.onEvaluatorException(evaluatorException);
  }

  /**
   * Process an evaluator message that indicates that the evaluator completed the unclean shut down request.
   *
   * @param message
   */
  private synchronized void onEvaluatorKilled(final EvaluatorStatusPOJO message) {

    assert message.getState() == State.KILLED;
    assert this.stateManager.isClosing();

    LOG.log(Level.WARNING, "Evaluator {0} killed completely.", getId());

    this.stateManager.setKilled();
  }

  public void onResourceLaunch(final ResourceLaunchEvent resourceLaunchEvent) {
    synchronized (this.evaluatorDescriptor) {
      if (this.stateManager.isAllocated()) {
        this.stateManager.setSubmitted();
        this.resourceLaunchHandler.onNext(resourceLaunchEvent);
      } else if (this.stateManager.isCompletedAbnormally()) {
        LOG.log(Level.WARNING, "Evaluator manager expected {0} state but instead is in state {1}",
            new Object[] {EvaluatorState.ALLOCATED, this.stateManager});
      } else {
        throw new RuntimeException("Evaluator manager expected " + EvaluatorState.ALLOCATED +
            " state but instead is in state " + this.stateManager);
      }
    }
  }

  /**
   * Packages the ContextControlProto in an EvaluatorControlProto and forward it to the EvaluatorRuntime.
   *
   * @param contextControlProto message contains context control info.
   */
  public void sendContextControlMessage(final EvaluatorRuntimeProtocol.ContextControlProto contextControlProto) {
    synchronized (this.evaluatorDescriptor) {
      LOG.log(Level.FINEST, "Context control message to {0}", this.evaluatorId);
      this.contextControlHandler.send(contextControlProto);
    }
  }

  /**
   * Forward the EvaluatorControlProto to the EvaluatorRuntime.
   *
   * @param evaluatorControlProto message contains evaluator control information.
   */
  void sendEvaluatorControlMessage(final EvaluatorRuntimeProtocol.EvaluatorControlProto evaluatorControlProto) {
    synchronized (this.evaluatorDescriptor) {
      this.evaluatorControlHandler.send(evaluatorControlProto);
    }
  }

  /**
   * Handle task status messages.
   *
   * @param taskStatus message contains the current task status.
   */
  private void onTaskStatusMessage(final TaskStatusPOJO taskStatus) {

    if (!(this.task.isPresent() && this.task.get().getId().equals(taskStatus.getTaskId()))) {

      final State state = taskStatus.getState();
      if (state.isRestartable() ||
          this.driverRestartManager.getEvaluatorRestartState(this.evaluatorId).isReregistered()) {

        // [REEF-308] exposes a bug where the .NET evaluator does not send its states in the right order
        // [REEF-289] is a related item which may fix the issue
        if (state.isRunning()) {
          LOG.log(Level.WARNING,
              "Received a message of state {0} for Task {1} before receiving its {2} state",
              new Object[] {State.RUNNING, taskStatus.getTaskId(), State.INIT});
        }

        // FAILED is a legal first state of a Task as it could have failed during construction.
        this.task = Optional.of(
            new TaskRepresenter(taskStatus.getTaskId(),
                this.contextRepresenters.getContext(taskStatus.getContextId()),
                this.messageDispatcher,
                this,
                this.exceptionCodec,
                this.driverRestartManager));
      } else {
        throw new RuntimeException("Received a message of state " + state +
            ", not INIT, RUNNING, or FAILED for Task " + taskStatus.getTaskId() +
            " which we haven't heard from before.");
      }
    }

    this.task.get().onTaskStatusMessage(taskStatus);

    if (this.task.get().isNotRunning()) {
      LOG.log(Level.FINEST, "Task no longer running. De-registering it.");
      this.task = Optional.empty();
    }
  }

  /**
   * Resource status information from the (actual) resource manager.
   */
  public void onResourceStatusMessage(final ResourceStatusEvent resourceStatusEvent) {

    synchronized (this.evaluatorDescriptor) {

      final State state = resourceStatusEvent.getState();
      LOG.log(Level.FINEST, "Resource manager state update: {0}", state);

      if (!this.stateManager.isAvailable()) {

        LOG.log(Level.FINE,
            "Ignoring resource status update for Evaluator {0} which is already in state {1}.",
            new Object[] {this.getId(), this.stateManager});

      } else if (state.isCompleted() && this.stateManager.isAvailable()) {

        // Something is wrong. The resource manager reports that the Evaluator is done or failed,
        // but the Driver assumes it to be alive.
        final StringBuilder messageBuilder = new StringBuilder("Evaluator [")
            .append(this.evaluatorId)
            .append("] is assumed to be in state [")
            .append(this.stateManager.toString())
            .append("]. But the resource manager reports it to be in state [")
            .append(state)
            .append("].");

        if (this.stateManager.isSubmitted()) {
          messageBuilder
              .append(" This most likely means that the Evaluator suffered a failure before establishing " +
                  "a communications link to the driver.");
        } else if (this.stateManager.isAllocated()) {
          messageBuilder.append(" This most likely means that the Evaluator suffered a failure before being used.");
        } else if (this.stateManager.isRunning()) {
          messageBuilder.append(" This means that the Evaluator failed but wasn't able to send an error message " +
              "back to the driver.");
        }

        if (this.task.isPresent()) {
          messageBuilder.append(" Task [")
              .append(this.task.get().getId())
              .append("] was running when the Evaluator crashed.");
        }

        if (resourceStatusEvent.getState() == State.KILLED) {
          this.onEvaluatorException(
              new EvaluatorKilledByResourceManagerException(this.evaluatorId, messageBuilder.toString()));
        } else {
          this.onEvaluatorException(new EvaluatorException(this.evaluatorId, messageBuilder.toString()));
        }
      }
    }
  }

  @Override
  public String toString() {
    return "EvaluatorManager:"
        + " id=" + this.evaluatorId
        + " state=" + this.stateManager
        + " task=" + this.task;
  }

  // Dynamic Parameters

  /**
   * The Evaluator Identifier.
   */
  @NamedParameter(doc = "The Evaluator Identifier.")
  public static final class EvaluatorIdentifier implements Name<String> {
  }

  /**
   * The Evaluator Host.
   */
  @NamedParameter(doc = "The Evaluator Host.")
  public static final class EvaluatorDescriptorName implements Name<EvaluatorDescriptorImpl> {
  }
}