/**
 * 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.
 */

#include <fstream>
#include <iomanip>
#include <list>
#include <sstream>

#include <process/defer.hpp>
#include <process/delay.hpp>
#include <process/id.hpp>
#include <process/run.hpp>

#include <stout/check.hpp>
#include <stout/lambda.hpp>
#include <stout/multihashmap.hpp>
#include <stout/nothing.hpp>
#include <stout/os.hpp>
#include <stout/owned.hpp>
#include <stout/path.hpp>
#include <stout/utils.hpp>
#include <stout/uuid.hpp>

#include "sasl/authenticator.hpp"

#include "common/build.hpp"
#include "common/date_utils.hpp"
#include "common/protobuf_utils.hpp"

#include "logging/flags.hpp"
#include "logging/logging.hpp"

#include "master/allocator.hpp"
#include "master/flags.hpp"
#include "master/master.hpp"

using std::list;
using std::string;
using std::vector;

using process::wait; // Necessary on some OS's to disambiguate.

using std::tr1::shared_ptr;

namespace mesos {
namespace internal {
namespace master {

using allocator::Allocator;


class WhitelistWatcher : public Process<WhitelistWatcher> {
public:
  WhitelistWatcher(const string& _path, Allocator* _allocator)
  : ProcessBase(ID::generate("whitelist")),
    path(_path),
    allocator(_allocator) {}

protected:
  virtual void initialize()
  {
    watch();
  }

  void watch()
  {
    // Get the list of white listed slaves.
    Option<hashset<string> > whitelist;
    if (path == "*") { // Accept all slaves.
      VLOG(1) << "No whitelist given. Advertising offers for all slaves";
    } else {
      // Read from local file.
      // TODO(vinod): Add support for reading from ZooKeeper.
      // TODO(vinod): Ensure this read is atomic w.r.t external
      // writes/updates to this file.
      Try<string> read = os::read(
          strings::remove(path, "file://", strings::PREFIX));
      if (read.isError()) {
        LOG(ERROR) << "Error reading whitelist file: " << read.error() << ". "
                   << "Retrying";
        whitelist = lastWhitelist;
      } else if (read.get().empty()) {
        LOG(WARNING) << "Empty whitelist file " << path << ". "
                     << "No offers will be made!";
        whitelist = Option<hashset<string> >::some(hashset<string>());
      } else {
        hashset<string> hostnames;
        vector<string> lines = strings::tokenize(read.get(), "\n");
        foreach (const string& hostname, lines) {
          hostnames.insert(hostname);
        }
        whitelist = Option<hashset<string> >::some(hostnames);
      }
    }

    // Send the whitelist to allocator, if necessary.
    if (whitelist != lastWhitelist) {
      allocator->updateWhitelist(whitelist);
    }

    // Check again.
    lastWhitelist = whitelist;
    delay(WHITELIST_WATCH_INTERVAL, self(), &WhitelistWatcher::watch);
  }

private:
  const string path;
  Allocator* allocator;
  Option<hashset<string> > lastWhitelist;
};


class SlaveObserver : public Process<SlaveObserver>
{
public:
  SlaveObserver(const UPID& _slave,
                const SlaveInfo& _slaveInfo,
                const SlaveID& _slaveId,
                const PID<Master>& _master)
    : ProcessBase(ID::generate("slave-observer")),
      slave(_slave),
      slaveInfo(_slaveInfo),
      slaveId(_slaveId),
      master(_master),
      timeouts(0),
      pinged(false)
  {
    install("PONG", &SlaveObserver::pong);
  }

protected:
  virtual void initialize()
  {
    send(slave, "PING");
    pinged = true;
    delay(SLAVE_PING_TIMEOUT, self(), &SlaveObserver::timeout);
  }

  void pong(const UPID& from, const string& body)
  {
    timeouts = 0;
    pinged = false;
  }

  void timeout()
  {
    if (pinged) { // So we haven't got back a pong yet ...
      if (++timeouts >= MAX_SLAVE_PING_TIMEOUTS) {
        deactivate();
        return;
      }
    }

    send(slave, "PING");
    pinged = true;
    delay(SLAVE_PING_TIMEOUT, self(), &SlaveObserver::timeout);
  }

  void deactivate()
  {
    dispatch(master, &Master::deactivateSlave, slaveId);
  }

private:
  const UPID slave;
  const SlaveInfo slaveInfo;
  const SlaveID slaveId;
  const PID<Master> master;
  uint32_t timeouts;
  bool pinged;
};


Master::Master(Allocator* _allocator, Files* _files)
  : ProcessBase("master"),
    http(*this),
    flags(),
    allocator(_allocator),
    files(_files),
    completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}


Master::Master(Allocator* _allocator, Files* _files, const Flags& _flags)
  : ProcessBase("master"),
    http(*this),
    flags(_flags),
    allocator(_allocator),
    files(_files),
    completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}


Master::~Master()
{
  LOG(INFO) << "Shutting down master";

  // Remove the frameworks.
  // Note we are not deleting the pointers to the frameworks from the
  // allocator or the roles because it is unnecessary bookkeeping at
  // this point since we are shutting down.
  foreachvalue (Framework* framework, frameworks) {
    // Remove pointers to the framework's tasks in slaves.
    foreachvalue (Task* task, utils::copy(framework->tasks)) {
      Slave* slave = getSlave(task->slave_id());
      // Since we only find out about tasks when the slave re-registers,
      // it must be the case that the slave exists!
      CHECK(slave != NULL)
        << "Unknown slave " << task->slave_id()
        << " in the task " << task->task_id();

      removeTask(task);
    }

    // Remove the framework's offers (if they weren't removed before).
    foreach (Offer* offer, utils::copy(framework->offers)) {
      removeOffer(offer);
    }

    delete framework;
  }
  frameworks.clear();

  foreachvalue (Future<Nothing> future, authenticating) {
    // NOTE: This is necessary during tests because a copy of
    // this future is used to setup authentication timeout. If a
    // test doesn't discard this future, authentication timeout might
    // fire in a different test and any associated callbacks
    // (e.g., '_authenticate()') would be called. This is because the
    // master pid doesn't change across the tests.
    // TODO(vinod): This seems to be a bug in libprocess or the
    // testing infrastructure.
    future.discard();
  }

  CHECK_EQ(offers.size(), 0UL);

  foreachvalue (Slave* slave, slaves) {
    LOG(INFO) << "Removing slave " << slave->id
              << " (" << slave->info.hostname() << ")";

    // Remove tasks that are in the slave but not in any framework.
    // This could happen when the framework has yet to reregister
    // after master failover.
    // NOTE: keys() and values() are used because slave->tasks is
    //       modified by removeTask()!
    foreach (const FrameworkID& frameworkId, slave->tasks.keys()) {
      foreach (Task* task, slave->tasks[frameworkId].values()) {
        removeTask(task);
      }
    }

    // Kill the slave observer.
    terminate(slave->observer);
    wait(slave->observer);

    delete slave->observer;
    delete slave;
  }
  slaves.clear();

  terminate(whitelistWatcher);
  wait(whitelistWatcher);

  delete whitelistWatcher;
}


void Master::initialize()
{
  LOG(INFO) << "Master started on " << string(self()).substr(7);

  // The master ID is currently comprised of the current date, the IP
  // address and port from self() and the OS PID.

  Try<string> id =
    strings::format("%s-%u-%u-%d", DateUtils::currentDate(),
                    self().ip, self().port, getpid());

  CHECK(!id.isError()) << id.error();

  info.set_id(id.get());
  info.set_ip(self().ip);
  info.set_port(self().port);

  LOG(INFO) << "Master ID: " << info.id();

  if (flags.authenticate) {
    LOG(INFO) << "Master only allowing authenticated frameworks to register!";

    if (flags.credentials.isNone()) {
      EXIT(1) << "Authentication requires a credentials file"
              << " (see --credentials flag)";
    }
  } else {
    LOG(INFO) << "Master allowing unauthenticated frameworks to register!!";
  }


  if (flags.credentials.isSome()) {
    vector<Credential> credentials;

    const std::string& path = flags.credentials.get();

    // TODO(vinod): Warn if the credentials file has bad file
    // permissions (e.g., world readable!).
    Try<string> read = os::read(
        strings::remove(path, "file://", strings::PREFIX));
    if (read.isError()) {
      EXIT(1) << "Failed to read credentials file '" << path
              << "': " << read.error();
    } else if (read.get().empty()) {
      LOG(WARNING) << "Empty credentials file '" << path << "'. "
                   << "!!No frameworks will be allowed to register!!";
    } else {
      foreach (const string& line, strings::tokenize(read.get(), "\n")) {
        const vector<string>& pairs = strings::tokenize(line, " ");
        if (pairs.size() != 2) {
          EXIT(1)
            << "Invalid credential format at line: " << (credentials.size() + 1)
            << " (see --credentials flag)";
        }

        // Add the credential.
        Credential credential;
        credential.set_principal(pairs[0]);
        credential.set_secret(pairs[1]);

        credentials.push_back(credential);
      }
    }
    // Give Authenticator access to credentials.
    sasl::secrets::load(credentials);
  }

  hashmap<string, RoleInfo> roleInfos;

  // Add the default role.
  RoleInfo roleInfo;
  roleInfo.set_name("*");
  roleInfos["*"] = roleInfo;

  // Add other roles.
  if (flags.roles.isSome()) {
    vector<string> tokens = strings::tokenize(flags.roles.get(), ",");

    foreach (const std::string& role, tokens) {
      RoleInfo roleInfo;
      roleInfo.set_name(role);
      roleInfos[role] = roleInfo;
    }
  }

  // Add role weights.
  if (flags.weights.isSome()) {
    vector<string> tokens = strings::tokenize(flags.weights.get(), ",");

    foreach (const std::string& token, tokens) {
      vector<string> pair = strings::tokenize(token, "=");
      if (pair.size() != 2) {
        EXIT(1) << "Invalid weight: '" << token << "'. --weights should"
          "be of the form 'role=weight,role=weight'\n";
      } else if (!roleInfos.contains(pair[0])) {
        EXIT(1) << "Invalid weight: '" << token << "'. " << pair[0]
                << " is not a valid role.";
      }

      double weight = atof(pair[1].c_str());
      if (weight <= 0) {
        EXIT(1) << "Invalid weight: '" << token
                << "'. Weights must be positive.";
      }

      roleInfos[pair[0]].set_weight(weight);
    }
  }

  foreachpair (const std::string& role,
               const RoleInfo& roleInfo,
               roleInfos) {
    roles[role] = new Role(roleInfo);
  }

  // Initialize the allocator.
  allocator->initialize(flags, self(), roleInfos);

  // Parse the white list
  whitelistWatcher = new WhitelistWatcher(flags.whitelist, allocator);
  spawn(whitelistWatcher);

  elected = false;

  nextFrameworkId = 0;
  nextSlaveId = 0;
  nextOfferId = 0;

  // Start all the statistics at 0.
  stats.tasks[TASK_STAGING] = 0;
  stats.tasks[TASK_STARTING] = 0;
  stats.tasks[TASK_RUNNING] = 0;
  stats.tasks[TASK_FINISHED] = 0;
  stats.tasks[TASK_FAILED] = 0;
  stats.tasks[TASK_KILLED] = 0;
  stats.tasks[TASK_LOST] = 0;
  stats.validStatusUpdates = 0;
  stats.invalidStatusUpdates = 0;
  stats.validFrameworkMessages = 0;
  stats.invalidFrameworkMessages = 0;

  startTime = Clock::now();

  // Install handler functions for certain messages.
  install<SubmitSchedulerRequest>(
      &Master::submitScheduler,
      &SubmitSchedulerRequest::name);

  install<NewMasterDetectedMessage>(
      &Master::newMasterDetected,
      &NewMasterDetectedMessage::pid);

  install<NoMasterDetectedMessage>(
      &Master::noMasterDetected);

  install<RegisterFrameworkMessage>(
      &Master::registerFramework,
      &RegisterFrameworkMessage::framework);

  install<ReregisterFrameworkMessage>(
      &Master::reregisterFramework,
      &ReregisterFrameworkMessage::framework,
      &ReregisterFrameworkMessage::failover);

  install<UnregisterFrameworkMessage>(
      &Master::unregisterFramework,
      &UnregisterFrameworkMessage::framework_id);

  install<DeactivateFrameworkMessage>(
        &Master::deactivateFramework,
        &DeactivateFrameworkMessage::framework_id);

  install<ResourceRequestMessage>(
      &Master::resourceRequest,
      &ResourceRequestMessage::framework_id,
      &ResourceRequestMessage::requests);

  install<LaunchTasksMessage>(
      &Master::launchTasks,
      &LaunchTasksMessage::framework_id,
      &LaunchTasksMessage::offer_id,
      &LaunchTasksMessage::tasks,
      &LaunchTasksMessage::filters);

  install<ReviveOffersMessage>(
      &Master::reviveOffers,
      &ReviveOffersMessage::framework_id);

  install<KillTaskMessage>(
      &Master::killTask,
      &KillTaskMessage::framework_id,
      &KillTaskMessage::task_id);

  install<FrameworkToExecutorMessage>(
      &Master::schedulerMessage,
      &FrameworkToExecutorMessage::slave_id,
      &FrameworkToExecutorMessage::framework_id,
      &FrameworkToExecutorMessage::executor_id,
      &FrameworkToExecutorMessage::data);

  install<RegisterSlaveMessage>(
      &Master::registerSlave,
      &RegisterSlaveMessage::slave);

  install<ReregisterSlaveMessage>(
      &Master::reregisterSlave,
      &ReregisterSlaveMessage::slave_id,
      &ReregisterSlaveMessage::slave,
      &ReregisterSlaveMessage::executor_infos,
      &ReregisterSlaveMessage::tasks);

  install<UnregisterSlaveMessage>(
      &Master::unregisterSlave,
      &UnregisterSlaveMessage::slave_id);

  install<StatusUpdateMessage>(
      &Master::statusUpdate,
      &StatusUpdateMessage::update,
      &StatusUpdateMessage::pid);

  install<ReconcileTasksMessage>(
      &Master::reconcileTasks,
      &ReconcileTasksMessage::framework_id,
      &ReconcileTasksMessage::statuses);

  install<ExitedExecutorMessage>(
      &Master::exitedExecutor,
      &ExitedExecutorMessage::slave_id,
      &ExitedExecutorMessage::framework_id,
      &ExitedExecutorMessage::executor_id,
      &ExitedExecutorMessage::status);

  install<AuthenticateMessage>(
      &Master::authenticate,
      &AuthenticateMessage::pid);

  // Setup HTTP routes.
  route("/health",
        Http::HEALTH_HELP,
        lambda::bind(&Http::health, http, lambda::_1));
  route("/redirect",
        Http::REDIRECT_HELP,
        lambda::bind(&Http::redirect, http, lambda::_1));
  route("/stats.json",
        None(),
        lambda::bind(&Http::stats, http, lambda::_1));
  route("/state.json",
        None(),
        lambda::bind(&Http::state, http, lambda::_1));
  route("/roles.json",
        None(),
        lambda::bind(&Http::roles, http, lambda::_1));

  // Provide HTTP assets from a "webui" directory. This is either
  // specified via flags (which is necessary for running out of the
  // build directory before 'make install') or determined at build
  // time via the preprocessor macro '-DMESOS_WEBUI_DIR' set in the
  // Makefile.
  provide("", path::join(flags.webui_dir, "master/static/index.html"));
  provide("static", path::join(flags.webui_dir, "master/static"));

  if (flags.log_dir.isSome()) {
    Try<string> log = logging::getLogFile(google::INFO);
    if (log.isError()) {
      LOG(ERROR) << "Master log file cannot be found: " << log.error();
    } else {
      files->attach(log.get(), "/master/log")
        .onAny(defer(self(), &Self::fileAttached, lambda::_1, log.get()));
    }
  }
}


void Master::finalize()
{
  LOG(INFO) << "Master terminating";
}


void Master::exited(const UPID& pid)
{
  foreachvalue (Framework* framework, frameworks) {
    if (framework->pid == pid) {
      LOG(INFO) << "Framework " << framework->id << " disconnected";

      // Deactivate framework.
      deactivate(framework);

      // Set 'failoverTimeout' to the default and update only if the
      // input is valid.
      Try<Duration> failoverTimeout_ =
        Duration::create(FrameworkInfo().failover_timeout());
      CHECK_SOME(failoverTimeout_);
      Duration failoverTimeout = failoverTimeout_.get();

      failoverTimeout_ =
        Duration::create(framework->info.failover_timeout());
      if (failoverTimeout_.isSome()) {
        failoverTimeout = failoverTimeout_.get();
      } else {
        LOG(WARNING) << "Using the default value for 'failover_timeout' because"
                     << "the input value is invalid: "
                     << failoverTimeout_.error();
      }

      LOG(INFO) << "Giving framework " << framework->id << " "
                << failoverTimeout << " to failover";

      // Delay dispatching a message to ourselves for the timeout.
      delay(failoverTimeout,
          self(),
          &Master::frameworkFailoverTimeout,
          framework->id,
          framework->reregisteredTime);

      return;
    }
  }

  // The semantics when a slave gets disconnected are as follows:
  // 1) If the slave is not checkpointing, the slave is immediately
  //    removed and all tasks running on it are transitioned to LOST.
  //    No resources are recovered, because the slave is removed.
  // 2) If the slave is checkpointing, the frameworks running on it
  //    fall into one of the 2 cases:
  //    2.1) Framework is checkpointing: No immediate action is taken.
  //         The slave is given a chance to reconnect until the slave
  //         observer times out (75s) and removes the slave (Case 1).
  //    2.2) Framework is not-checkpointing: The slave is not removed
  //         but the framework is removed from the slave's structs,
  //         its tasks transitioned to LOST and resources recovered.
  foreachvalue (Slave* slave, slaves) {
    if (slave->pid == pid) {
      LOG(INFO) << "Slave " << slave->id << " (" << slave->info.hostname()
                << ") disconnected";

      // Remove the slave, if it is not checkpointing.
      if (!slave->info.checkpoint()) {
        LOG(INFO) << "Removing disconnected slave " << slave->id
                  << "(" << slave->info.hostname() << ") "
                  << "because it is not checkpointing!";
        removeSlave(slave);
        return;
      } else if (!slave->disconnected) {
        // Mark the slave as disconnected and remove it from the allocator.
        slave->disconnected = true;

        allocator->slaveDisconnected(slave->id);

        // If a slave is checkpointing, remove all non-checkpointing
        // frameworks from the slave.
        // First, collect all the frameworks running on this slave.
        hashset<FrameworkID> frameworkIds =
          slave->tasks.keys() | slave->executors.keys();

        // Now, remove all the non-checkpointing frameworks.
        foreach (const FrameworkID& frameworkId, frameworkIds) {
          Framework* framework = getFramework(frameworkId);
          if (framework != NULL && !framework->info.checkpoint()) {
            LOG(INFO) << "Removing non-checkpointing framework " << frameworkId
                      << " from disconnected slave " << slave->id
                      << "(" << slave->info.hostname() << ")";

            removeFramework(slave, framework);
          }
        }

        foreach (Offer* offer, utils::copy(slave->offers)) {
          allocator->resourcesRecovered(
              offer->framework_id(), slave->id, offer->resources());

          // Remove and rescind offers.
          removeOffer(offer, true); // Rescind!
        }
      } else {
        LOG(WARNING) << "Ignoring duplicate exited() notification for "
                     << "checkpointing slave " << slave->id
                     << " (" << slave->info.hostname() << ")";
      }
    }
  }
}


void Master::fileAttached(const Future<Nothing>& result, const string& path)
{
  if (result.isReady()) {
    LOG(INFO) << "Successfully attached file '" << path << "'";
  } else {
    LOG(ERROR) << "Failed to attach file '" << path << "': "
               << result.isFailed() ? result.failure() : "discarded";
  }
}


void Master::submitScheduler(const string& name)
{
  LOG(INFO) << "Scheduler submit request for " << name;
  SubmitSchedulerResponse response;
  response.set_okay(false);
  reply(response);
}


void Master::newMasterDetected(const UPID& pid)
{
  // Check and see if we are (1) still waiting to be the elected
  // master, (2) newly elected master, (3) no longer elected master,
  // or (4) still elected master.

  leader = pid;

  if (leader != self() && !elected) {
    LOG(INFO) << "Waiting to be master!";
  } else if (leader == self() && !elected) {
    LOG(INFO) << "Elected as master!";
    elected = true;
  } else if (leader != self() && elected) {
    LOG(FATAL) << "No longer elected master ... committing suicide!";
  } else if (leader == self() && elected) {
    LOG(INFO) << "Still acting as master!";
  }
}


void Master::noMasterDetected()
{
  if (elected) {
    LOG(FATAL) << "No longer elected master ... committing suicide!";
  } else {
    LOG(FATAL) << "No master detected (?) ... committing suicide!";
  }
}


void Master::registerFramework(
    const UPID& from,
    const FrameworkInfo& frameworkInfo)
{
  if (authenticating.contains(from)) {
    LOG(INFO) << "Queuing up registration request from " << from
              << " because authentication is still in progress";

    authenticating[from]
      .onReady(defer(self(), &Self::registerFramework, from, frameworkInfo));
    return;
  }

  if (!elected) {
    LOG(WARNING) << "Ignoring register framework message from " << from
                 << " since not elected yet";
    return;
  }

  if (flags.authenticate && !authenticated.contains(from)) {
    // This could happen if another authentication request came
    // through before we are here or if a framework tried to register
    // without authentication.
    LOG(WARNING) << "Refusing registration of framework at " << from
                 << " because it is not authenticated";
    FrameworkErrorMessage message;
    message.set_message("Framework at " + stringify(from) +
                        " is not authenticated.");
    send(from, message);
    return;
  }

  if (!roles.contains(frameworkInfo.role())) {
    FrameworkErrorMessage message;
    message.set_message("Role '" + frameworkInfo.role() + "' is not valid.");
    send(from, message);
    return;
  }

  LOG(INFO) << "Received registration request from " << from;

  // Check if this framework is already registered (because it retries).
  foreachvalue (Framework* framework, frameworks) {
    if (framework->pid == from) {
      LOG(INFO) << "Framework " << framework->id << " (" << framework->pid
                << ") already registered, resending acknowledgement";
      FrameworkRegisteredMessage message;
      message.mutable_framework_id()->MergeFrom(framework->id);
      message.mutable_master_info()->MergeFrom(info);
      send(from, message);
      return;
    }
  }

  Framework* framework =
    new Framework(frameworkInfo, newFrameworkId(), from, Clock::now());

  LOG(INFO) << "Registering framework " << framework->id << " at " << from;

  bool rootSubmissions = flags.root_submissions;

  if (framework->info.user() == "root" && rootSubmissions == false) {
    LOG(INFO) << framework << " registering as root, but "
              << "root submissions are disabled on this cluster";
    FrameworkErrorMessage message;
    message.set_message("User 'root' is not allowed to run frameworks");
    send(from, message);
    delete framework;
    return;
  }

  addFramework(framework);
}


void Master::reregisterFramework(
    const UPID& from,
    const FrameworkInfo& frameworkInfo,
    bool failover)
{
  if (authenticating.contains(from)) {
    LOG(INFO) << "Queuing up re-registration request from " << from
              << " because authentication is still in progress";

    authenticating[from]
      .onReady(defer(self(),
                     &Self::reregisterFramework,
                     from,
                     frameworkInfo,
                     failover));
    return;
  }

  if (!elected) {
    LOG(WARNING) << "Ignoring re-register framework message from " << from
                 << " since not elected yet";
    return;
  }

  if (!frameworkInfo.has_id() || frameworkInfo.id() == "") {
    LOG(ERROR) << "Framework re-registering without an id!";
    FrameworkErrorMessage message;
    message.set_message("Framework reregistered without a framework id");
    send(from, message);
    return;
  }

  if (flags.authenticate && !authenticated.contains(from)) {
    // This could happen if another authentication request came
    // through before we are here or if a framework tried to
    // re-register without authentication.
    LOG(WARNING) << "Refusing registration of framework at " << from
                  << " because it is not authenticated";
    FrameworkErrorMessage message;
    message.set_message("Framework '" + frameworkInfo.id().value() + "' at " +
                        stringify(from) + " is not authenticated.");
    send(from, message);
    return;
  }

  if (!roles.contains(frameworkInfo.role())) {
    FrameworkErrorMessage message;
    message.set_message("Role '" + frameworkInfo.role() + "' is not valid.");
    send(from, message);
    return;
  }

  LOG(INFO) << "Re-registering framework " << frameworkInfo.id()
            << " at " << from;

  if (frameworks.count(frameworkInfo.id()) > 0) {
    // Using the "failover" of the scheduler allows us to keep a
    // scheduler that got partitioned but didn't die (in ZooKeeper
    // speak this means didn't lose their session) and then
    // eventually tried to connect to this master even though
    // another instance of their scheduler has reconnected. This
    // might not be an issue in the future when the
    // master/allocator launches the scheduler can get restarted
    // (if necessary) by the master and the master will always
    // know which scheduler is the correct one.

    Framework* framework = frameworks[frameworkInfo.id()];
    framework->reregisteredTime = Clock::now();

    if (failover) {
      // We do not attempt to detect a duplicate re-registration
      // message here because it is impossible to distinguish between
      // a duplicate message, and a scheduler failover to the same
      // pid, given the existing libprocess primitives (PID does not
      // identify the libprocess Process instance).

      // TODO: Should we check whether the new scheduler has given
      // us a different framework name, user name or executor info?
      LOG(INFO) << "Framework " << frameworkInfo.id() << " failed over";
      failoverFramework(framework, from);
    } else if (from != framework->pid) {
      LOG(ERROR)
        << "Framework " << frameworkInfo.id() << " at " << from
        << " attempted to re-register while a framework at " << framework->pid
        << " is already registered";
      FrameworkErrorMessage message;
      message.set_message("Framework failed over");
      send(from, message);
      return;
    } else {
      LOG(INFO) << "Allowing the Framework " << frameworkInfo.id()
                << " to re-register with an already used id";

      // Remove any offers sent to this framework.
      // NOTE: We need to do this because the scheduler might have
      // replied to the offers but the driver might have dropped
      // those messages since it wasn't connected to the master.
      foreach (Offer* offer, utils::copy(framework->offers)) {
        allocator->resourcesRecovered(
            offer->framework_id(), offer->slave_id(), offer->resources());
        removeOffer(offer);
      }

      FrameworkReregisteredMessage message;
      message.mutable_framework_id()->MergeFrom(frameworkInfo.id());
      message.mutable_master_info()->MergeFrom(info);
      send(from, message);
      return;
    }
  } else {
    // We don't have a framework with this ID, so we must be a newly
    // elected Mesos master to which either an existing scheduler or a
    // failed-over one is connecting. Create a Framework object and add
    // any tasks it has that have been reported by reconnecting slaves.
    Framework* framework =
      new Framework(frameworkInfo, frameworkInfo.id(), from, Clock::now());
    framework->reregisteredTime = Clock::now();

    // TODO(benh): Check for root submissions like above!

    // Add any running tasks reported by slaves for this framework.
    foreachvalue (Slave* slave, slaves) {
      foreachkey (const FrameworkID& frameworkId, slave->tasks) {
        foreachvalue (Task* task, slave->tasks[frameworkId]) {
          if (framework->id == task->framework_id()) {
            framework->addTask(task);

            // Also add the task's executor for resource accounting
            // if it's still alive on the slave and we've not yet
            // added it to the framework.
            if (task->has_executor_id() &&
                slave->hasExecutor(framework->id, task->executor_id()) &&
                !framework->hasExecutor(slave->id, task->executor_id())) {
              const ExecutorInfo& executorInfo =
                slave->executors[framework->id][task->executor_id()];
              framework->addExecutor(slave->id, executorInfo);
            }
          }
        }
      }
    }

    // N.B. Need to add the framwwork _after_ we add it's tasks
    // (above) so that we can properly determine the resources it's
    // currently using!
    addFramework(framework);
  }

  CHECK(frameworks.contains(frameworkInfo.id()))
    << "Unknown framework " << frameworkInfo.id();

  // Broadcast the new framework pid to all the slaves. We have to
  // broadcast because an executor might be running on a slave but
  // it currently isn't running any tasks. This could be a
  // potential scalability issue ...
  foreachvalue (Slave* slave, slaves) {
    UpdateFrameworkMessage message;
    message.mutable_framework_id()->MergeFrom(frameworkInfo.id());
    message.set_pid(from);
    send(slave->pid, message);
  }

  return;
}


void Master::unregisterFramework(
    const UPID& from,
    const FrameworkID& frameworkId)
{
  LOG(INFO) << "Asked to unregister framework " << frameworkId;

  Framework* framework = getFramework(frameworkId);
  if (framework != NULL) {
    if (framework->pid == from) {
      removeFramework(framework);
    } else {
      LOG(WARNING)
        << "Ignoring unregister framework message for framework " << frameworkId
        << " from " << from << " because it is not from the registered"
        << " framework " << framework->pid;
    }
  }
}


void Master::deactivateFramework(
    const UPID& from,
    const FrameworkID& frameworkId)
{
  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
      << "Ignoring deactivate framework message for framework " << frameworkId
      << " because the framework cannot be found";
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring deactivate framework message for framework " << frameworkId
      << " from '" << from << "' because it is not from the registered"
      << " framework '" << framework->pid << "'";
    return;
  }

  deactivate(framework);
}


void Master::deactivate(Framework* framework)
{
  CHECK_NOTNULL(framework);

  LOG(INFO) << "Deactivating framework " << framework->id;

  // Stop sending offers here for now.
  framework->active = false;

  // Tell the allocator to stop allocating resources to this framework.
  allocator->frameworkDeactivated(framework->id);

  // Remove the framework from authenticated. This is safe because
  // a framework will always reauthenticate before (re-)registering.
  authenticated.erase(framework->pid);

  // Remove the framework's offers.
  foreach (Offer* offer, utils::copy(framework->offers)) {
    allocator->resourcesRecovered(
        offer->framework_id(),
        offer->slave_id(),
        Resources(offer->resources()));

    removeOffer(offer);
  }
}


void Master::resourceRequest(
    const UPID& from,
    const FrameworkID& frameworkId,
    const vector<Request>& requests)
{
  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
        << "Ignoring resource request message from framework " << frameworkId
        << " because the framework cannot be found";
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring resource request message from framework " << frameworkId
      << " from '" << from << "' because it is not from the registered "
      << " framework '" << framework->pid << "'";
    return;
  }

  LOG(INFO) << "Requesting resources for framework " << frameworkId;
  allocator->resourcesRequested(frameworkId, requests);
}


void Master::launchTasks(
    const UPID& from,
    const FrameworkID& frameworkId,
    const OfferID& offerId,
    const vector<TaskInfo>& tasks,
    const Filters& filters)
{
  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
      << "Ignoring launch tasks message for offer " << offerId
      << " of framework " << frameworkId
      << " because the framework cannot be found";
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring launch tasks message for offer " << offerId
      << " of framework " << frameworkId << " from '" << from
      << "' because it is not from the registered framework '"
      << framework->pid << "'";
    return;
  }

  // TODO(benh): Support offer "hoarding" and allow multiple offers
  // *from the same slave* to be used to launch tasks. This can be
  // accomplished rather easily by collecting and merging all offers
  // into a mega-offer and passing that offer to
  // Master::processTasks.
  Offer* offer = getOffer(offerId);
  if (offer != NULL) {
    CHECK_EQ(offer->framework_id(), frameworkId)
        << "Offer " << offerId
        << " has invalid frameworkId " << offer->framework_id();

    Slave* slave = getSlave(offer->slave_id());
    CHECK(slave != NULL)
      << "Offer " << offerId << " outlived  slave "
      << slave->id << " (" << slave->info.hostname() << ")";

    // If a slave is disconnected we should've removed its offers.
    CHECK(!slave->disconnected)
      << "Offer " << offerId << " outlived disconnected slave "
      << slave->id << " (" << slave->info.hostname() << ")";

    processTasks(offer, framework, slave, tasks, filters);
  } else {
    // The offer is gone (possibly rescinded, lost slave, re-reply
    // to same offer, etc). Report all tasks in it as failed.
    // TODO: Consider adding a new task state TASK_INVALID for
    // situations like these.
    LOG(WARNING) << "Offer " << offerId << " is no longer valid";
    foreach (const TaskInfo& task, tasks) {
      StatusUpdateMessage message;
      StatusUpdate* update = message.mutable_update();
      update->mutable_framework_id()->MergeFrom(frameworkId);
      TaskStatus* status = update->mutable_status();
      status->mutable_task_id()->MergeFrom(task.task_id());
      status->set_state(TASK_LOST);
      status->set_message("Task launched with invalid offer");
      update->set_timestamp(Clock::now().secs());
      update->set_uuid(UUID::random().toBytes());

      LOG(INFO) << "Sending status update " << *update
                << " for launch task attempt on invalid offer " << offerId;
      send(framework->pid, message);
    }
  }
}


void Master::reviveOffers(const UPID& from, const FrameworkID& frameworkId)
{
  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
      << "Ignoring revive offers message for framework " << frameworkId
      << " because the framework cannot be found";
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring revive offers message for framework " << frameworkId
      << " from '" << from << "' because it is not from the registered"
      << " framework '" << framework->pid << "'";
    return;
  }

  LOG(INFO) << "Reviving offers for framework " << framework->id;
  allocator->offersRevived(framework->id);
}


void Master::killTask(
    const UPID& from,
    const FrameworkID& frameworkId,
    const TaskID& taskId)
{
  LOG(INFO) << "Asked to kill task " << taskId
            << " of framework " << frameworkId;

  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
      << "Ignoring kill task message for task " << taskId << " of framework "
      << frameworkId << " because the framework cannot be found";
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring kill task message for task " << taskId
      << " of framework " << frameworkId << " from '" << from
      << "' because it is not from the registered framework '"
      << framework->pid << "'";
    return;
  }

  Task* task = framework->getTask(taskId);
  if (task != NULL) {
    Slave* slave = getSlave(task->slave_id());
    CHECK(slave != NULL) << "Unknown slave " << task->slave_id();

    // We add the task to 'killedTasks' here because the slave
    // might be partitioned or disconnected but the master
    // doesn't know it yet.
    slave->killedTasks.put(frameworkId, taskId);

    // NOTE: This task will be properly reconciled when the
    // disconnected slave re-registers with the master.
    if (!slave->disconnected) {
      LOG(INFO) << "Telling slave " << slave->id << " ("
                << slave->info.hostname() << ")"
                << " to kill task " << taskId
                << " of framework " << frameworkId;

      KillTaskMessage message;
      message.mutable_framework_id()->MergeFrom(frameworkId);
      message.mutable_task_id()->MergeFrom(taskId);
      send(slave->pid, message);
    }
  } else {
    // TODO(benh): Once the scheduler has persistance and
    // high-availability of it's tasks, it will be the one that
    // determines that this invocation of 'killTask' is silly, and
    // can just return "locally" (i.e., after hitting only the other
    // replicas). Unfortunately, it still won't know the slave id.

    LOG(WARNING) << "Cannot kill task " << taskId
                 << " of framework " << frameworkId
                 << " because the task cannot be found";
    StatusUpdateMessage message;
    StatusUpdate* update = message.mutable_update();
    update->mutable_framework_id()->MergeFrom(frameworkId);
    TaskStatus* status = update->mutable_status();
    status->mutable_task_id()->MergeFrom(taskId);
    status->set_state(TASK_LOST);
    status->set_message("Task not found");
    update->set_timestamp(Clock::now().secs());
    update->set_uuid(UUID::random().toBytes());
    send(framework->pid, message);
  }
}


void Master::schedulerMessage(
    const UPID& from,
    const SlaveID& slaveId,
    const FrameworkID& frameworkId,
    const ExecutorID& executorId,
    const string& data)
{
  Framework* framework = getFramework(frameworkId);

  if (framework == NULL) {
    LOG(WARNING)
      << "Ignoring framework message for executor " << executorId
      << " of framework " << frameworkId
      << " because the framework cannot be found";
    stats.invalidFrameworkMessages++;
    return;
  }

  if (from != framework->pid) {
    LOG(WARNING)
      << "Ignoring framework message for executor " << executorId
      << " of framework " << frameworkId << " from " << from
      << " because it is not from the registered framework "
      << framework->pid;
    stats.invalidFrameworkMessages++;
    return;
  }

  Slave* slave = getSlave(slaveId);
  if (slave != NULL) {
    if (!slave->disconnected) {
      LOG(INFO) << "Sending framework message for framework "
                << frameworkId << " to slave " << slaveId
                << " (" << slave->info.hostname() << ")";

      FrameworkToExecutorMessage message;
      message.mutable_slave_id()->MergeFrom(slaveId);
      message.mutable_framework_id()->MergeFrom(frameworkId);
      message.mutable_executor_id()->MergeFrom(executorId);
      message.set_data(data);
      send(slave->pid, message);

      stats.validFrameworkMessages++;
    } else {
      LOG(WARNING) << "Cannot send framework message for framework "
                   << frameworkId << " to slave " << slaveId
                   << " (" << slave->info.hostname() << ")"
                   << " because slave is disconnected";
      stats.invalidFrameworkMessages++;
    }
  } else {
    LOG(WARNING) << "Cannot send framework message for framework "
                 << frameworkId << " to slave " << slaveId
                 << " because slave does not exist";
    stats.invalidFrameworkMessages++;
  }
}


void Master::registerSlave(const UPID& from, const SlaveInfo& slaveInfo)
{
  if (!elected) {
    LOG(WARNING) << "Ignoring register slave message from "
                 << slaveInfo.hostname() << " since not elected yet";
    return;
  }

  // Check if this slave is already registered (because it retries).
  foreachvalue (Slave* slave, slaves) {
    if (slave->pid == from) {
      if (slave->disconnected) {
        // The slave was previously disconnected but it is now trying
        // to register as a new slave. This could happen if the slave
        // failed recovery and hence registering as a new slave before
        // the master removed the old slave from its map.
        LOG(INFO) << "Removing old disconnected slave " << slave->id << " ("
                  << slave->info.hostname() << ") because a registration"
                  << " attempt is being made from " << from;
        removeSlave(slave);
        break;
      } else {
        LOG(INFO) << "Slave " << slave->id << " (" << slave->info.hostname()
                  << ") already registered, resending acknowledgement";
        SlaveRegisteredMessage message;
        message.mutable_slave_id()->MergeFrom(slave->id);
        reply(message);
        return;
      }
    }
  }

  Slave* slave = new Slave(slaveInfo, newSlaveId(), from, Clock::now());

  LOG(INFO) << "Attempting to register slave on " << slave->info.hostname()
            << " at " << slave->pid;

  // TODO(benh): We assume all slaves can register for now.
  CHECK_EQ(flags.slaves, "*");
  addSlave(slave);

//   // Checks if this slave, or if all slaves, can be accepted.
//   if (slaveHostnamePorts.contains(slaveInfo.hostname(), from.port)) {
//     run(&SlaveRegistrar::run, slave, self());
//   } else if (flags.slaves == "*") {
//     run(&SlaveRegistrar::run, slave, self(), slavesManager->self());
//   } else {
//     LOG(WARNING) << "Cannot register slave at "
//                  << slaveInfo.hostname() << ":" << from.port
//                  << " because not in allocated set of slaves!";
//     reply(ShutdownMessage());
//   }
}


void Master::reregisterSlave(
    const UPID& from,
    const SlaveID& slaveId,
    const SlaveInfo& slaveInfo,
    const vector<ExecutorInfo>& executorInfos,
    const vector<Task>& tasks)
{
  if (!elected) {
    LOG(WARNING) << "Ignoring re-register slave message from "
                 << slaveInfo.hostname() << " since not elected yet";
    return;
  }

  if (slaveId == "") {
    LOG(ERROR) << "Slave " << from << " re-registered without an id!";
    reply(ShutdownMessage());
  } else if (deactivatedSlaves.contains(from)) {
    // We disallow deactivated slaves from re-registering. This is
    // to ensure that when a master deactivates a slave that was
    // partitioned, we don't allow the slave to re-register, as we've
    // already informed frameworks that the tasks were lost.
    LOG(ERROR) << "Slave " << slaveId << " at " << from
               << " attempted to re-register after deactivation";
    reply(ShutdownMessage());
  } else {
    Slave* slave = getSlave(slaveId);

    if (slave != NULL) {
      slave->reregisteredTime = Clock::now();

      // NOTE: This handles the case where a slave tries to
      // re-register with an existing master (e.g. because of a
      // spurious Zookeeper session expiration or after the slave
      // recovers after a restart).
      // For now, we assume this slave is not nefarious (eventually
      // this will be handled by orthogonal security measures like key
      // based authentication).
      LOG(WARNING) << "Slave at " << from << " (" << slave->info.hostname()
                   << ") is being allowed to re-register with an already"
                   << " in use id (" << slaveId << ")";

      // TODO(bmahler): There's an implicit assumption here that when
      // the master already knows about this slave, the slave cannot
      // have tasks unknown to the master. This _should_ be the case
      // since the causal relationship is:
      //   slave removes task -> master removes task
      // We should enforce this via a CHECK (dangerous), or by shutting
      // down slaves that are found to violate this assumption.

      SlaveReregisteredMessage message;
      message.mutable_slave_id()->MergeFrom(slave->id);
      reply(message);

      // Update the slave pid and relink to it.
      // NOTE: Re-linking the slave here always rather than only when
      // the slave is disconnected can lead to multiple exited events
      // in succession for a disconnected slave. As a result, we
      // ignore duplicate exited events for disconnected checkpointing
      // slaves.
      // See: https://issues.apache.org/jira/browse/MESOS-675
      slave->pid = from;
      link(slave->pid);

      // Reconcile tasks between master and the slave.
      // NOTE: This needs to be done after the registration message is
      // sent to the slave and the new pid is linked.
      reconcile(slave, executorInfos, tasks);

      // If this is a disconnected slave, add it back to the allocator.
      // This is done after reconciliation to ensure the allocator's
      // offers include the recovered resources initially on this
      // slave.
      if (slave->disconnected) {
        slave->disconnected = false; // Reset the flag.
        allocator->slaveReconnected(slaveId);
      }
    } else {
      // NOTE: This handles the case when the slave tries to
      // re-register with a failed over master.
      slave = new Slave(slaveInfo, slaveId, from, Clock::now());
      slave->reregisteredTime = Clock::now();

      LOG(INFO) << "Attempting to re-register slave " << slave->id << " at "
                << slave->pid << " (" << slave->info.hostname() << ")";

      // TODO(benh): We assume all slaves can register for now.
      CHECK_EQ(flags.slaves, "*");
      readdSlave(slave, executorInfos, tasks);
    }

    // Send the latest framework pids to the slave.
    CHECK_NOTNULL(slave);
    hashset<UPID> pids;
    foreach (const Task& task, tasks) {
      Framework* framework = getFramework(task.framework_id());
      if (framework != NULL && !pids.contains(framework->pid)) {
        UpdateFrameworkMessage message;
        message.mutable_framework_id()->MergeFrom(framework->id);
        message.set_pid(framework->pid);
        send(slave->pid, message);

        pids.insert(framework->pid);
      }
    }
  }
}


void Master::unregisterSlave(const SlaveID& slaveId)
{
  LOG(INFO) << "Asked to unregister slave " << slaveId;

  // TODO(benh): Check that only the slave is asking to unregister?

  Slave* slave = getSlave(slaveId);
  if (slave != NULL) {
    removeSlave(slave);
  }
}


// NOTE: We cannot use 'from' here to identify the slave as this is
// now sent by the StatusUpdateManagerProcess and master itself when
// it generates TASK_LOST messages. Only 'pid' can be used to identify
// the slave.
void Master::statusUpdate(const StatusUpdate& update, const UPID& pid)
{
  const TaskStatus& status = update.status();

  Slave* slave = getSlave(update.slave_id());
  if (slave == NULL) {
    if (deactivatedSlaves.contains(pid)) {
      // If the slave is deactivated, we have already informed
      // frameworks that its tasks were LOST, so the slave should
      // shut down.
      LOG(WARNING) << "Ignoring status update " << update
                   << " from deactivated slave " << pid
                   << " with id " << update.slave_id() << " ; asking slave "
                   << " to shutdown";
      send(pid, ShutdownMessage());
    } else {
      LOG(WARNING) << "Ignoring status update " << update
                   << " from unknown slave " << pid
                   << " with id " << update.slave_id();
    }
    stats.invalidStatusUpdates++;
    return;
  }

  CHECK(!deactivatedSlaves.contains(pid))
    << "Received status update " << update << " from " << pid
    << " which is deactivated slave " << update.slave_id()
    << "(" << slave->info.hostname() << ")";

  Framework* framework = getFramework(update.framework_id());
  if (framework == NULL) {
    LOG(WARNING) << "Ignoring status update " << update
                 << " from " << pid << " ("
                 << slave->info.hostname() << "): error, couldn't lookup "
                 << "framework " << update.framework_id();
    stats.invalidStatusUpdates++;
    return;
  }

  // Pass on the (transformed) status update to the framework.
  StatusUpdateMessage message;
  message.mutable_update()->MergeFrom(update);
  message.set_pid(pid);
  send(framework->pid, message);

  // Lookup the task and see if we need to update anything locally.
  Task* task = slave->getTask(update.framework_id(), status.task_id());
  if (task == NULL) {
    LOG(WARNING) << "Status update " << update
                 << " from " << pid << " ("
                 << slave->info.hostname() << "): error, couldn't lookup task";
    stats.invalidStatusUpdates++;
    return;
  }

  LOG(INFO) << "Status update " << update << " from " << pid;

  task->set_state(status.state());

  // Handle the task appropriately if it's terminated.
  if (protobuf::isTerminalState(status.state())) {
    removeTask(task);
  }

  stats.tasks[status.state()]++;
  stats.validStatusUpdates++;
}


void Master::exitedExecutor(
    const UPID& from,
    const SlaveID& slaveId,
    const FrameworkID& frameworkId,
    const ExecutorID& executorId,
    int32_t status)
{
  // Only update master's internal data structures here for properly accounting.
  // The TASK_LOST updates are handled by the slave.
  Slave* slave = getSlave(slaveId);
  if (slave == NULL) {
    if (deactivatedSlaves.contains(from)) {
      // If the slave is deactivated, we have already informed
      // frameworks that its tasks were LOST, so the slave should
      // shut down.
      LOG(WARNING) << "Ignoring exited executor '" << executorId
                   << "' of framework " << frameworkId
                   << " on deactivated slave " << slaveId
                   << " ; asking slave to shutdown";
      reply(ShutdownMessage());
    } else {
      LOG(WARNING) << "Ignoring exited executor '" << executorId
                   << "' of framework " << frameworkId
                   << " on unknown slave " << slaveId;
    }
    return;
  }

  CHECK(!deactivatedSlaves.contains(from))
    << "Received exited message for executor " << executorId << " from " << from
    << " which is the deactivated slave " << slaveId
    << "(" << slave->info.hostname() << ")";

  // Tell the allocator about the recovered resources.
  if (slave->hasExecutor(frameworkId, executorId)) {
    ExecutorInfo executor = slave->executors[frameworkId][executorId];

    LOG(INFO) << "Executor " << executorId
              << " of framework " << frameworkId
              << " on slave " << slaveId
              << " (" << slave->info.hostname() << ")"
              << " exited with status " << status;

    allocator->resourcesRecovered(frameworkId,
        slaveId,
        Resources(executor.resources()));

    // Remove executor from slave and framework.
    slave->removeExecutor(frameworkId, executorId);
  } else {
    LOG(WARNING) << "Ignoring unknown exited executor "
                 << executorId << " on slave " << slaveId
                 << " (" << slave->info.hostname() << ")";
  }

  Framework* framework = getFramework(frameworkId);
  if (framework != NULL) {
    framework->removeExecutor(slave->id, executorId);

    // TODO(benh): Send the framework its executor's exit status?
    // Or maybe at least have something like
    // Scheduler::executorLost?
  }
}


void Master::deactivateSlave(const SlaveID& slaveId)
{
  if (!slaves.contains(slaveId)) {
    // Possible when the SlaveObserver dispatched to deactivate a slave,
    // but exited() was already called for this slave.
    LOG(WARNING) << "Unable to deactivate unknown slave " << slaveId;
    return;
  }

  Slave* slave = slaves[slaveId];
  CHECK_NOTNULL(slave);

  LOG(WARNING) << "Removing slave " << slave->id << " at " << slave->pid
               << " because it has been deactivated";

  send(slave->pid, ShutdownMessage());
  removeSlave(slave);
}


void Master::reconcileTasks(
    const UPID& from,
    const FrameworkID& frameworkId,
    const std::vector<TaskStatus>& statuses)
{
  Framework* framework = getFramework(frameworkId);
  if (framework == NULL) {
    LOG(WARNING) << "Unknown framework " << frameworkId << " at " << from
                 << " attempted to reconcile tasks";
    return;
  }

  LOG(INFO) << "Performing task state reconciliation for framework "
            << frameworkId;

  // Verify expected task states and send status updates whenever expectations
  // are not met. When:
  //   1) Slave is unknown.*
  //   2) Task is unknown.*
  //   3) Task state has changed.
  //
  // *) TODO(nnielsen): Missing slaves and tasks are currently treated silently
  //                    i.e. nothing is sent. To give accurate responses in
  //                    these cases during master fail-over, we need to leverage
  //                    the registrar.
  foreach (const TaskStatus& status, statuses) {
    if (!status.has_slave_id()) {
      LOG(WARNING) << "Status from task " << status.task_id()
                   << " does not include slave id";
      continue;
    }

    Slave* slave = getSlave(status.slave_id());
    if (slave != NULL) {
      Task* task = slave->getTask(frameworkId, status.task_id());
      if (task != NULL && task->state() != status.state()) {
        const StatusUpdate& update = protobuf::createStatusUpdate(
          frameworkId,
          task->slave_id(),
          task->task_id(),
          task->state(),
          "Task state changed");

        statusUpdate(update, UPID());
      }
    }
  }
}


void Master::frameworkFailoverTimeout(const FrameworkID& frameworkId,
                                      const Time& reregisteredTime)
{
  Framework* framework = getFramework(frameworkId);

  if (framework != NULL && !framework->active) {
    // If the re-registration time has not changed, then the framework
    // has not re-registered within the failover timeout.
    if (framework->reregisteredTime == reregisteredTime) {
      LOG(INFO) << "Framework failover timeout, removing framework "
                << framework->id;
      removeFramework(framework);
    }
  }
}


void Master::offer(const FrameworkID& frameworkId,
                   const hashmap<SlaveID, Resources>& resources)
{
  if (!frameworks.contains(frameworkId) || !frameworks[frameworkId]->active) {
    LOG(WARNING) << "Master returning resources offered to framework "
                 << frameworkId << " because the framework"
                 << " has terminated or is inactive";

    foreachpair (const SlaveID& slaveId, const Resources& offered, resources) {
      allocator->resourcesRecovered(frameworkId, slaveId, offered);
    }
    return;
  }

  // Create an offer for each slave and add it to the message.
  ResourceOffersMessage message;

  Framework* framework = frameworks[frameworkId];
  foreachpair (const SlaveID& slaveId, const Resources& offered, resources) {
    if (!slaves.contains(slaveId)) {
      LOG(WARNING) << "Master returning resources offered to framework "
                   << frameworkId << " because slave " << slaveId
                   << " is not valid";

      allocator->resourcesRecovered(frameworkId, slaveId, offered);
      continue;
    }

    Slave* slave = slaves[slaveId];

    CHECK(slave->info.checkpoint() || !framework->info.checkpoint())
        << "Resources of non checkpointing slave " << slaveId
        << " (" << slave->info.hostname() << ") are being offered to"
        << " checkpointing framework " << frameworkId;

    // This could happen if the allocator dispatched 'Master::offer' before
    // it received 'Allocator::slaveRemoved' from the master.
    if (slave->disconnected) {
      LOG(WARNING) << "Master returning resources offered because slave "
                   << slaveId << " is disconnected";

      allocator->resourcesRecovered(frameworkId, slaveId, offered);
      continue;
    }

    Offer* offer = new Offer();
    offer->mutable_id()->MergeFrom(newOfferId());
    offer->mutable_framework_id()->MergeFrom(framework->id);
    offer->mutable_slave_id()->MergeFrom(slave->id);
    offer->set_hostname(slave->info.hostname());
    offer->mutable_resources()->MergeFrom(offered);
    offer->mutable_attributes()->MergeFrom(slave->info.attributes());

    // Add all framework's executors running on this slave.
    if (slave->executors.contains(framework->id)) {
      const hashmap<ExecutorID, ExecutorInfo>& executors =
        slave->executors[framework->id];
      foreachkey (const ExecutorID& executorId, executors) {
        offer->add_executor_ids()->MergeFrom(executorId);
      }
    }

    offers[offer->id()] = offer;

    framework->addOffer(offer);
    slave->addOffer(offer);

    // Add the offer *AND* the corresponding slave's PID.
    message.add_offers()->MergeFrom(*offer);
    message.add_pids(slave->pid);
  }

  if (message.offers().size() == 0) {
    return;
  }

  LOG(INFO) << "Sending " << message.offers().size()
            << " offers to framework " << framework->id;

  send(framework->pid, message);
}


// TODO(vinod): If due to network partition there are two instances
// of the framework that think they are leaders and try to
// authenticate with master they would be stepping on each other's
// toes. Currently it is tricky to detect this case because the
// 'authenticate' message doesn't contain the 'FrameworkID'.
void Master::authenticate(const UPID& from, const UPID& pid)
{
  // Deactivate the framework if it's already registered.
  foreachvalue (Framework* framework, frameworks) {
    if (framework->pid == pid) {
      deactivate(framework);
      break;
    }
  }

  authenticated.erase(pid);

  if (authenticating.contains(pid)) {
    LOG(INFO) << "Queuing up authentication request from " << pid
              << " because authentication is still in progress";

    // Try to cancel the in progress authentication by deleting
    // the authenticator.
    authenticators.erase(pid);

    // Retry after the current authenticator finishes.
    authenticating[pid]
      .onAny(defer(self(), &Self::authenticate, from, pid));

    return;
  }

  LOG(INFO) << "Authenticating framework at " << pid;

  // Create a promise to capture the entire "authenticating"
  // procedure. We'll set this _after_ we finish _authenticate.
  Owned<Promise<Nothing> > promise = new Promise<Nothing>();

  // Create the authenticator.
  Owned<sasl::Authenticator> authenticator = new sasl::Authenticator(from);

  // Start authentication.
  const Future<bool>& future = authenticator->authenticate()
    .onAny(defer(self(), &Self::_authenticate, pid, promise, lambda::_1));

  // Don't wait for authentication to happen for ever.
  delay(Seconds(5),
        self(),
        &Self::authenticationTimeout,
        future);

  // Save our state.
  authenticating[pid] = promise->future();
  authenticators.put(pid, authenticator);
}


void Master::_authenticate(
    const UPID& pid,
    const Owned<Promise<Nothing> >& promise,
    const Future<bool>& future)
{
  if (!future.isReady() || !future.get()) {
    const string& error = future.isReady()
        ? "Refused authentication"
        : (future.isFailed() ? future.failure() : "future discarded");

    LOG(WARNING) << "Failed to authenticate framework at " << pid
                 << ": " << error;

    promise->fail(error);
  } else {
    LOG(INFO) << "Successfully authenticated framework at " << pid;

    promise->set(Nothing());
    authenticated.insert(pid);
  }

  authenticators.erase(pid);
  authenticating.erase(pid);
}


void Master::authenticationTimeout(Future<bool> future)
{
  // Note that a 'discard' here is safe even if another
  // authenticator is in progress because this copy of the future
  // corresponds to the original authenticator that started the timer.
  if (future.discard()) { // This is a no-op if the future is already ready.
    LOG(WARNING) << "Authentication timed out";
  }
}


// Return connected frameworks that are not in the process of being removed
vector<Framework*> Master::getActiveFrameworks() const
{
  vector <Framework*> result;
  foreachvalue (Framework* framework, frameworks) {
    if (framework->active) {
      result.push_back(framework);
    }
  }
  return result;
}


// We use the visitor pattern to abstract the process of performing
// any validations, aggregations, etc. of tasks that a framework
// attempts to run within the resources provided by an offer. A
// visitor can return an optional error (typedef'ed as an option of a
// string) which will cause the master to send a failed status update
// back to the framework for only that task description. An instance
// will be reused for each task description from same offer, but not
// for task descriptions from different offers.
typedef Option<string> TaskInfoError;

struct TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave) = 0;

  virtual ~TaskInfoVisitor() {}
};


// Checks that the slave ID used by a task is correct.
struct SlaveIDChecker : TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave)
  {
    if (!(task.slave_id() == slave->id)) {
      return TaskInfoError::some(
          "Task uses invalid slave: " + task.slave_id().value());
    }

    return TaskInfoError::none();
  }
};


// Checks that each task uses a unique ID. Regardless of whether a
// task actually gets launched (for example, another checker may
// return an error for a task), we always consider it an error when a
// task tries to re-use an ID.
struct UniqueTaskIDChecker : TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave)
  {
    const TaskID& taskId = task.task_id();

    if (ids.contains(taskId) || framework->tasks.contains(taskId)) {
      return TaskInfoError::some(
          "Task has duplicate ID: " + taskId.value());
    }

    ids.insert(taskId);

    return TaskInfoError::none();
  }

  hashset<TaskID> ids;
};


// Checks that the used resources by a task (and executor if
// necessary) on each slave does not exceed the total resources
// offered on that slave
struct ResourceUsageChecker : TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave)
  {
    if (task.resources().size() == 0) {
      return TaskInfoError::some("Task uses no resources");
    }

    foreach (const Resource& resource, task.resources()) {
      if (!Resources::isAllocatable(resource)) {
        // TODO(benh): Send back the invalid resources?
        return TaskInfoError::some("Task uses invalid resources");
      }
    }

    // Check if this task uses more resources than offered.
    Resources taskResources = task.resources();

    if (!((usedResources + taskResources) <= offer->resources())) {
      return TaskInfoError::some(
          "Task " + stringify(task.task_id()) + " attempted to use " +
          stringify(taskResources) + " combined with already used " +
          stringify(usedResources) + " is greater than offered " +
          stringify(offer->resources()));
    }

    // Check this task's executor's resources.
    if (task.has_executor()) {
      // TODO(benh): Check that the executor uses some resources.

      foreach (const Resource& resource, task.executor().resources()) {
        if (!Resources::isAllocatable(resource)) {
          // TODO(benh): Send back the invalid resources?
          return TaskInfoError::some(
              "Executor for task " + stringify(task.task_id()) +
              " uses invalid resources " + stringify(resource));
        }
      }

      // Check if this task's executor is running, and if not check if
      // the task + the executor use more resources than offered.
      if (!executors.contains(task.executor().executor_id())) {
        if (!slave->hasExecutor(framework->id, task.executor().executor_id())) {
          taskResources += task.executor().resources();
          if (!((usedResources + taskResources) <= offer->resources())) {
            return TaskInfoError::some(
                "Task " + stringify(task.task_id()) + " + executor attempted" +
                " to use " + stringify(taskResources) + " combined with" +
                " already used " + stringify(usedResources) + " is greater" +
                " than offered " + stringify(offer->resources()));
          }
        }
        executors.insert(task.executor().executor_id());
      }
    }

    usedResources += taskResources;

    return TaskInfoError::none();
  }

  Resources usedResources;
  hashset<ExecutorID> executors;
};


// Checks that tasks that use the "same" executor (i.e., same
// ExecutorID) have an identical ExecutorInfo.
struct ExecutorInfoChecker : TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave)
  {
    if (task.has_executor() == task.has_command()) {
      return TaskInfoError::some(
          "Task should have at least one (but not both) of CommandInfo or"
          " ExecutorInfo present");
    }

    if (task.has_executor()) {
      if (slave->hasExecutor(framework->id, task.executor().executor_id())) {
        const ExecutorInfo& executorInfo =
          slave->executors[framework->id][task.executor().executor_id()];
        if (!(task.executor() == executorInfo)) {
          return TaskInfoError::some(
              "Task has invalid ExecutorInfo (existing ExecutorInfo"
              " with same ExecutorID is not compatible).\n"
              "------------------------------------------------------------\n"
              "Existing ExecutorInfo:\n" +
              stringify(executorInfo) + "\n"
              "------------------------------------------------------------\n"
              "Task's ExecutorInfo:\n" +
              stringify(task.executor()) + "\n"
              "------------------------------------------------------------\n");
        }
      }
    }

    return TaskInfoError::none();
  }
};


// Checks that a task that asks for checkpointing is not being
// launched on a slave that has not enabled checkpointing.
// TODO(vinod): Consider not offering resources for non-checkpointing
// slaves to frameworks that need checkpointing.
struct CheckpointChecker : TaskInfoVisitor
{
  virtual TaskInfoError operator () (
      const TaskInfo& task,
      Offer* offer,
      Framework* framework,
      Slave* slave)
  {
    if (framework->info.checkpoint() && !slave->info.checkpoint()) {
      return TaskInfoError::some(
          "Task asked to be checkpointed but the slave "
          "has checkpointing disabled");
    }
    return TaskInfoError::none();
  }
};


// Process a resource offer reply (for a non-cancelled offer) by
// launching the desired tasks (if the offer contains a valid set of
// tasks) and reporting used resources to the allocator.
void Master::processTasks(Offer* offer,
                          Framework* framework,
                          Slave* slave,
                          const vector<TaskInfo>& tasks,
                          const Filters& filters)
{
  CHECK_NOTNULL(offer);
  CHECK_NOTNULL(framework);
  CHECK_NOTNULL(slave);

  LOG(INFO) << "Processing reply for offer " << offer->id()
            << " on slave " << slave->id
            << " (" << slave->info.hostname() << ")"
            << " for framework " << framework->id;

  Resources usedResources; // Accumulated resources used from this offer.

  // Create task visitors.
  list<TaskInfoVisitor*> visitors;
  visitors.push_back(new SlaveIDChecker());
  visitors.push_back(new UniqueTaskIDChecker());
  visitors.push_back(new ResourceUsageChecker());
  visitors.push_back(new ExecutorInfoChecker());
  visitors.push_back(new CheckpointChecker());

  // Loop through each task and check it's validity.
  foreach (const TaskInfo& task, tasks) {
    // Possible error found while checking task's validity.
    TaskInfoError error = TaskInfoError::none();

    // Invoke each visitor.
    foreach (TaskInfoVisitor* visitor, visitors) {
      error = (*visitor)(task, offer, framework, slave);
      if (error.isSome()) {
        break;
      }
    }

    if (error.isNone()) {
      // Task looks good, get it running!
      usedResources += launchTask(task, framework, slave);
    } else {
      // Error validating task, send a failed status update.
      LOG(WARNING) << "Failed to validate task " << task.task_id()
                   << " : " << error.get();

      const StatusUpdate& update = protobuf::createStatusUpdate(
          framework->id,
          slave->id,
          task.task_id(),
          TASK_LOST,
          error.get());

      LOG(INFO) << "Sending status update " << update << " for invalid task";
      StatusUpdateMessage message;
      message.mutable_update()->CopyFrom(update);
      send(framework->pid, message);
    }
  }

  // Cleanup visitors.
  do {
    TaskInfoVisitor* visitor = visitors.front();
    visitors.pop_front();
    delete visitor;
  } while (!visitors.empty());

  // All used resources should be allocatable, enforced by our validators.
  CHECK_EQ(usedResources, usedResources.allocatable());

  // Calculate unused resources.
  Resources unusedResources = offer->resources() - usedResources;

  if (unusedResources.allocatable().size() > 0) {
    // Tell the allocator about the unused (e.g., refused) resources.
    allocator->resourcesUnused(offer->framework_id(),
                               offer->slave_id(),
                               unusedResources,
                               filters);
  }

  removeOffer(offer);
}


Resources Master::launchTask(const TaskInfo& task,
                             Framework* framework,
                             Slave* slave)
{
  CHECK_NOTNULL(framework);
  CHECK_NOTNULL(slave);

  Resources resources; // Total resources used on slave by launching this task.

  // Determine if this task launches an executor, and if so make sure
  // the slave and framework state has been updated accordingly.
  Option<ExecutorID> executorId;

  if (task.has_executor()) {
    // TODO(benh): Refactor this code into Slave::addTask.
    if (!slave->hasExecutor(framework->id, task.executor().executor_id())) {
      CHECK(!framework->hasExecutor(slave->id, task.executor().executor_id()))
        << "Executor " << task.executor().executor_id()
        << " known to the framework " << framework->id
        << " but unknown to the slave " << slave->id;

      slave->addExecutor(framework->id, task.executor());
      framework->addExecutor(slave->id, task.executor());
      resources += task.executor().resources();
    }

    executorId = Option<ExecutorID>::some(task.executor().executor_id());
  }

  // Add the task to the framework and slave.
  Task* t = new Task();
  t->mutable_framework_id()->MergeFrom(framework->id);
  t->set_state(TASK_STAGING);
  t->set_name(task.name());
  t->mutable_task_id()->MergeFrom(task.task_id());
  t->mutable_slave_id()->MergeFrom(task.slave_id());
  t->mutable_resources()->MergeFrom(task.resources());

  if (executorId.isSome()) {
    t->mutable_executor_id()->MergeFrom(executorId.get());
  }

  framework->addTask(t);

  slave->addTask(t);

  resources += task.resources();

  // Tell the slave to launch the task!
  LOG(INFO) << "Launching task " << task.task_id()
            << " of framework " << framework->id
            << " with resources " << task.resources() << " on slave "
            << slave->id << " (" << slave->info.hostname() << ")";

  RunTaskMessage message;
  message.mutable_framework()->MergeFrom(framework->info);
  message.mutable_framework_id()->MergeFrom(framework->id);
  message.set_pid(framework->pid);
  message.mutable_task()->MergeFrom(task);
  send(slave->pid, message);

  stats.tasks[TASK_STAGING]++;

  return resources;
}


// NOTE: This function is only called when the slave re-registers
// with a master that already knows about it (i.e., not a failed
// over master).
void Master::reconcile(
    Slave* slave,
    const vector<ExecutorInfo>& executors,
    const vector<Task>& tasks)
{
  CHECK_NOTNULL(slave);

  // We convert the 'tasks' into a map for easier lookup below.
  // TODO(vinod): Check if the tasks are known to the master.
  multihashmap<FrameworkID, TaskID> slaveTasks;
  foreach (const Task& task, tasks) {
    slaveTasks.put(task.framework_id(), task.task_id());
  }

  // Send TASK_LOST updates for tasks present in the master but
  // missing from the slave. This could happen if the task was
  // dropped by the slave (e.g., slave exited before getting the
  // task or the task was launched while slave was in recovery).
  // NOTE: keys() and values() are used since statusUpdate()
  //       modifies slave->tasks.
  foreach (const FrameworkID& frameworkId, slave->tasks.keys()) {
    foreach (Task* task, slave->tasks[frameworkId].values()) {
      if (!slaveTasks.contains(task->framework_id(), task->task_id())) {
        LOG(WARNING) << "Sending TASK_LOST for task " << task->task_id()
                     << " of framework " << task->framework_id()
                     << " unknown to the slave " << slave->id
                     << " (" << slave->info.hostname() << ")";

        const StatusUpdate& update = protobuf::createStatusUpdate(
            task->framework_id(),
            slave->id,
            task->task_id(),
            TASK_LOST,
            "Task is unknown to the slave");

        statusUpdate(update, UPID());
      }
    }
  }

  // Likewise, any executors that are present in the master but
  // not present in the slave must be removed to correctly account
  // for resources. First we index the executors for fast lookup below.
  multihashmap<FrameworkID, ExecutorID> slaveExecutors;
  foreach (const ExecutorInfo& executor, executors) {
    // TODO(bmahler): The slave ensures the framework id is set in the
    // framework info when re-registering. This can be killed in 0.15.0
    // as we've added code in 0.14.0 to ensure the framework id is set
    // in the scheduler driver.
    if (!executor.has_framework_id()) {
      LOG(ERROR) << "Slave " << slave->id
                 << " (" << slave->info.hostname() << ") "
                 << "re-registered with executor " << executor.executor_id()
                 << " without setting the framework id";
      continue;
    }
    slaveExecutors.put(executor.framework_id(), executor.executor_id());
  }

  // Now that we have the index for lookup, remove all the executors
  // in the master that are not known to the slave.
  foreachkey (const FrameworkID& frameworkId, utils::copy(slave->executors)) {
    foreachkey (const ExecutorID& executorId,
                utils::copy(slave->executors[frameworkId])) {
      if (!slaveExecutors.contains(frameworkId, executorId)) {
        LOG(WARNING) << "Removing executor " << executorId << " of framework "
                     << frameworkId << " as it is unknown to the slave "
                     << slave->id << " (" << slave->info.hostname() << ")";

        // TODO(bmahler): This is duplicated in several locations, we
        // may benefit from a method for removing an executor from
        // all the relevant data structures and the allocator, akin
        // to removeTask().
        allocator->resourcesRecovered(
            frameworkId,
            slave->id,
            slave->executors[frameworkId][executorId].resources());

        slave->removeExecutor(frameworkId, executorId);

        if (frameworks.contains(frameworkId)) {
          frameworks[frameworkId]->removeExecutor(slave->id, executorId);
        }
      }
    }
  }

  // Send KillTaskMessages for tasks in 'killedTasks' that are
  // still alive on the slave. This could happen if the slave
  // did not receive KillTaskMessage because of a partition or
  // disconnection.
  foreach (const Task& task, tasks) {
    if (!protobuf::isTerminalState(task.state()) &&
        slave->killedTasks.contains(task.framework_id(), task.task_id())) {
      LOG(WARNING) << " Slave " << slave->id << " (" << slave->info.hostname()
                   << ") has non-terminal task " << task.task_id()
                   << " that is supposed to be killed. Killing it now!";

      KillTaskMessage message;
      message.mutable_framework_id()->MergeFrom(task.framework_id());
      message.mutable_task_id()->MergeFrom(task.task_id());
      send(slave->pid, message);
    }
  }

  // Send ShutdownFrameworkMessages for frameworks that are completed.
  // This could happen if the message wasn't received by the slave
  // (e.g., slave was down, partitioned).
  // NOTE: This is a short-term hack because this information is lost
  // when the master fails over. Also, 'completedFrameworks' has a
  // limited capacity.
  // TODO(vinod): Revisit this when registrar is in place. It would
  // likely involve storing this information in the registrar.
  foreach (const shared_ptr<Framework>& framework, completedFrameworks) {
    if (slaveTasks.contains(framework->id)) {
      LOG(WARNING)
        << "Slave " << slave->id << " (" << slave->info.hostname()
        << ") re-registered with completed framework " << framework->id
        << ". Shutting down the framework on the slave";

      ShutdownFrameworkMessage message;
      message.mutable_framework_id()->MergeFrom(framework->id);
      send(slave->pid, message);
    }
  }
}


void Master::addFramework(Framework* framework)
{
  CHECK(!frameworks.contains(framework->id))
    << "Framework " << framework->id << "already exists!";

  frameworks[framework->id] = framework;

  link(framework->pid);

  // Enforced by Master::registerFramework.
  CHECK(roles.contains(framework->info.role()))
    << "Unknown role " << framework->info.role()
    << " of framework " << framework->id ;

  roles[framework->info.role()]->addFramework(framework);

  FrameworkRegisteredMessage message;
  message.mutable_framework_id()->MergeFrom(framework->id);
  message.mutable_master_info()->MergeFrom(info);
  send(framework->pid, message);

  allocator->frameworkAdded(
      framework->id, framework->info, framework->resources);
}


// Replace the scheduler for a framework with a new process ID, in the
// event of a scheduler failover.
void Master::failoverFramework(Framework* framework, const UPID& newPid)
{
  const UPID& oldPid = framework->pid;

  // There are a few failover cases to consider:
  //   1. The pid has changed. In this case we definitely want to
  //      send a FrameworkErrorMessage to shut down the older
  //      scheduler.
  //   2. The pid has not changed.
  //      2.1 The old scheduler on that pid failed over to a new
  //          instance on the same pid. No need to shut down the old
  //          scheduler as it is necessarily dead.
  //      2.2 This is a duplicate message. In this case, the scheduler
  //          has not failed over, so we do not want to shut it down.
  if (oldPid != newPid) {
    FrameworkErrorMessage message;
    message.set_message("Framework failed over");
    send(oldPid, message);
  }

  // TODO(benh): unlink(oldPid);

  framework->pid = newPid;
  link(newPid);

  // Make sure we can get offers again.
  if (!framework->active) {
    framework->active = true;
    allocator->frameworkActivated(framework->id, framework->info);
  }

  // The scheduler driver safely ignores any duplicate registration
  // messages, so we don't need to compare the old and new pids here.
  {
    FrameworkRegisteredMessage message;
    message.mutable_framework_id()->MergeFrom(framework->id);
    message.mutable_master_info()->MergeFrom(info);
    send(newPid, message);
  }

  // Remove the framework's offers (if they weren't removed before).
  // We do this after we have updated the pid and sent the framework
  // registered message so that the allocator can immediately re-offer
  // these resources to this framework if it wants.
  // TODO(benh): Consider just reoffering these to
  foreach (Offer* offer, utils::copy(framework->offers)) {
    allocator->resourcesRecovered(offer->framework_id(),
                                  offer->slave_id(),
                                  Resources(offer->resources()));
    removeOffer(offer);
  }
}


void Master::removeFramework(Framework* framework)
{
  CHECK_NOTNULL(framework);

  LOG(INFO) << "Removing framework " << framework->id;

  if (framework->active) {
    // Tell the allocator to stop allocating resources to this framework.
    allocator->frameworkDeactivated(framework->id);
  }

  // Tell slaves to shutdown the framework.
  foreachvalue (Slave* slave, slaves) {
    ShutdownFrameworkMessage message;
    message.mutable_framework_id()->MergeFrom(framework->id);
    send(slave->pid, message);
  }

  // Remove pointers to the framework's tasks in slaves.
  foreachvalue (Task* task, utils::copy(framework->tasks)) {
    Slave* slave = getSlave(task->slave_id());
    // Since we only find out about tasks when the slave re-registers,
    // it must be the case that the slave exists!
    CHECK(slave != NULL)
      << "Unknown slave " << task->slave_id()
      << " for task " << task->task_id();

    removeTask(task);
  }

  // Remove the framework's offers (if they weren't removed before).
  foreach (Offer* offer, utils::copy(framework->offers)) {
    allocator->resourcesRecovered(offer->framework_id(),
                                  offer->slave_id(),
                                  Resources(offer->resources()));
    removeOffer(offer);
  }

  // Remove the framework's executors for correct resource accounting.
  foreachkey (const SlaveID& slaveId, framework->executors) {
    Slave* slave = getSlave(slaveId);
    if (slave != NULL) {
      foreachpair (const ExecutorID& executorId,
                   const ExecutorInfo& executorInfo,
                   framework->executors[slaveId]) {
        allocator->resourcesRecovered(framework->id,
                                      slave->id,
                                      executorInfo.resources());
        slave->removeExecutor(framework->id, executorId);
      }
    }
  }

  // TODO(benh): Similar code between removeFramework and
  // failoverFramework needs to be shared!

  // TODO(benh): unlink(framework->pid);

  framework->unregisteredTime = Clock::now();

  // The completedFramework buffer now owns the framework pointer.
  completedFrameworks.push_back(std::tr1::shared_ptr<Framework>(framework));

  CHECK(roles.contains(framework->info.role()))
    << "Unknown role " << framework->info.role()
    << " of framework " << framework->id;

  roles[framework->info.role()]->removeFramework(framework);


  // Remove the framework from authenticated.
  authenticated.erase(framework->pid);

  // Remove it.
  frameworks.erase(framework->id);
  allocator->frameworkRemoved(framework->id);
}


void Master::removeFramework(Slave* slave, Framework* framework)
{
  CHECK_NOTNULL(slave);
  CHECK_NOTNULL(framework);

  LOG(INFO) << "Removing framework " << framework->id
            << " from slave " << slave->id
            << " (" << slave->info.hostname() << ")";

  // Remove pointers to framework's tasks in slaves, and send status
  // updates.
  // NOTE: values() is used because statusUpdate() modifies
  //       slave->tasks.
  foreach (Task* task, slave->tasks[framework->id].values()) {
    // Remove tasks that belong to this framework.
    if (task->framework_id() == framework->id) {
      // A framework might not actually exist because the master failed
      // over and the framework hasn't reconnected yet. For more info
      // please see the comments in 'removeFramework(Framework*)'.
      const StatusUpdate& update = protobuf::createStatusUpdate(
        task->framework_id(),
        task->slave_id(),
        task->task_id(),
        TASK_LOST,
        "Slave " + slave->info.hostname() + " disconnected",
        (task->has_executor_id()
            ? Option<ExecutorID>(task->executor_id()) : None()));

      statusUpdate(update, UPID());
    }
  }

  // Remove the framework's executors from the slave and framework
  // for proper resource accounting.
  if (slave->executors.contains(framework->id)) {
    foreachkey (const ExecutorID& executorId,
                utils::copy(slave->executors[framework->id])) {
      allocator->resourcesRecovered(
          framework->id,
          slave->id,
          slave->executors[framework->id][executorId].resources());

      framework->removeExecutor(slave->id, executorId);
      slave->removeExecutor(framework->id, executorId);
    }
  }
}


void Master::addSlave(Slave* slave, bool reregister)
{
  CHECK_NOTNULL(slave);

  LOG(INFO) << "Adding slave " << slave->id
            << " at " << slave->info.hostname()
            << " with " << slave->info.resources();

  deactivatedSlaves.erase(slave->pid);
  slaves[slave->id] = slave;

  link(slave->pid);

  if (!reregister) {
    SlaveRegisteredMessage message;
    message.mutable_slave_id()->MergeFrom(slave->id);
    send(slave->pid, message);
  } else {
    SlaveReregisteredMessage message;
    message.mutable_slave_id()->MergeFrom(slave->id);
    send(slave->pid, message);
  }

  // TODO(benh):
  //     // Ask the slaves manager to monitor this slave for us.
  //     dispatch(slavesManager->self(), &SlavesManager::monitor,
  //              slave->pid, slave->info, slave->id);

  // Set up an observer for the slave.
  slave->observer = new SlaveObserver(slave->pid, slave->info,
                                      slave->id, self());
  spawn(slave->observer);

  if (!reregister) {
    allocator->slaveAdded(slave->id,
                          slave->info,
                          hashmap<FrameworkID, Resources>());
  }
}


void Master::readdSlave(Slave* slave,
    const vector<ExecutorInfo>& executorInfos,
    const vector<Task>& tasks)
{
  CHECK_NOTNULL(slave);

  addSlave(slave, true);

  // Add the executors and tasks to the slave and framework state and
  // determine the resources that have been allocated to frameworks.
  hashmap<FrameworkID, Resources> resources;

  foreach (const ExecutorInfo& executorInfo, executorInfos) {
    // TODO(bmahler): ExecutorInfo.framework_id is set by the Scheduler
    // Driver in 0.14.0. Therefore, in 0.15.0, the slave no longer needs
    // to set it, and we could remove this CHECK if desired.
    CHECK(executorInfo.has_framework_id())
      << "Executor " << executorInfo.executor_id()
      << " doesn't have frameworkId set";

    if (!slave->hasExecutor(executorInfo.framework_id(),
                            executorInfo.executor_id())) {
      slave->addExecutor(executorInfo.framework_id(), executorInfo);
    }

    Framework* framework = getFramework(executorInfo.framework_id());
    if (framework != NULL) {
      if (!framework->hasExecutor(slave->id, executorInfo.executor_id())) {
        framework->addExecutor(slave->id, executorInfo);
      }
    }

    resources[executorInfo.framework_id()] += executorInfo.resources();
  }

  foreach (const Task& task, tasks) {
    // Ignore tasks that have reached terminal state.
    if (protobuf::isTerminalState(task.state())) {
      continue;
    }

    Task* t = new Task(task);

    // Add the task to the slave.
    slave->addTask(t);

    // Try and add the task to the framework too, but since the
    // framework might not yet be connected we won't be able to
    // add them. However, when the framework connects later we
    // will add them then. Again, we do the same thing
    // if a framework currently isn't registered.
    Framework* framework = getFramework(task.framework_id());
    if (framework != NULL) {
      framework->addTask(t);
    } else {
      // TODO(benh): We should really put a timeout on how long we
      // keep tasks running on a slave that never have frameworks
      // reregister and claim them.
      LOG(WARNING) << "Possibly orphaned task " << task.task_id()
                   << " of framework " << task.framework_id()
                   << " running on slave " << slave->id << " ("
                   << slave->info.hostname() << ")";
    }

    resources[task.framework_id()] += task.resources();
  }

  allocator->slaveAdded(slave->id, slave->info, resources);
}


// Lose all of a slave's tasks and delete the slave object.
void Master::removeSlave(Slave* slave)
{
  CHECK_NOTNULL(slave);

  LOG(INFO) << "Removing slave " << slave->id
            << " (" << slave->info.hostname() << ")";

  // We do this first, to make sure any of the resources recovered
  // below (e.g., removeTask()) are ignored by the allocator.
  if (!slave->disconnected) {
    allocator->slaveRemoved(slave->id);
  }

  // Remove pointers to slave's tasks in frameworks, and send status
  // updates.
  // NOTE: keys() and values() are used because statusUpdate()
  //       modifies slave->tasks.
  foreach (const FrameworkID& frameworkId, slave->tasks.keys()) {
    foreach (Task* task, slave->tasks[frameworkId].values()) {
      // A framework might not actually exist because the master failed
      // over and the framework hasn't reconnected. This can be a tricky
      // situation for frameworks that want to have high-availability,
      // because if they eventually do connect they won't ever get a
      // status update about this task.  Perhaps in the future what we
      // want to do is create a local Framework object to represent that
      // framework until it fails over. See the TODO above in
      // Master::reregisterSlave.
      const StatusUpdate& update = protobuf::createStatusUpdate(
          task->framework_id(),
          task->slave_id(),
          task->task_id(),
          TASK_LOST,
          "Slave " + slave->info.hostname() + " removed",
          (task->has_executor_id() ?
              Option<ExecutorID>(task->executor_id()) : None()));

      statusUpdate(update, UPID());
    }
  }

  foreach (Offer* offer, utils::copy(slave->offers)) {
    // TODO(vinod): We don't need to call 'Allocator::resourcesRecovered'
    // once MESOS-621 is fixed.
    allocator->resourcesRecovered(
        offer->framework_id(), slave->id, offer->resources());

    // Remove and rescind offers.
    removeOffer(offer, true); // Rescind!
  }

  // Remove executors from the slave for proper resource accounting.
  foreachkey (const FrameworkID& frameworkId, slave->executors) {
    Framework* framework = getFramework(frameworkId);
    if (framework != NULL) {
      foreachkey (const ExecutorID& executorId, slave->executors[frameworkId]) {
        // TODO(vinod): We don't need to call 'Allocator::resourcesRecovered'
        // once MESOS-621 is fixed.
        allocator->resourcesRecovered(
            frameworkId,
            slave->id,
            slave->executors[frameworkId][executorId].resources());

        framework->removeExecutor(slave->id, executorId);
      }
    }
  }

  // Send lost-slave message to all frameworks (this helps them re-run
  // previously finished tasks whose output was on the lost slave).
  foreachvalue (Framework* framework, frameworks) {
    LostSlaveMessage message;
    message.mutable_slave_id()->MergeFrom(slave->id);
    send(framework->pid, message);
  }

  // TODO(benh):
  //     // Tell the slaves manager to stop monitoring this slave for us.
  //     dispatch(slavesManager->self(), &SlavesManager::forget,
  //              slave->pid, slave->info, slave->id);

  // Kill the slave observer.
  terminate(slave->observer);
  wait(slave->observer);

  delete slave->observer;

  // TODO(benh): unlink(slave->pid);

  // Mark the slave as deactivated.
  deactivatedSlaves.insert(slave->pid);
  slaves.erase(slave->id);
  delete slave;
}


void Master::removeTask(Task* task)
{
  CHECK_NOTNULL(task);

  // Remove from framework.
  Framework* framework = getFramework(task->framework_id());
  if (framework != NULL) { // A framework might not be re-connected yet.
    framework->removeTask(task);
  }

  // Remove from slave.
  Slave* slave = getSlave(task->slave_id());
  CHECK_NOTNULL(slave);
  slave->removeTask(task);

  // Tell the allocator about the recovered resources.
  allocator->resourcesRecovered(
      task->framework_id(), task->slave_id(), Resources(task->resources()));

  delete task;
}


void Master::removeOffer(Offer* offer, bool rescind)
{
  // Remove from framework.
  Framework* framework = getFramework(offer->framework_id());
  CHECK(framework != NULL)
    << "Unknown framework " << offer->framework_id()
    << " in the offer " << offer->id();

  framework->removeOffer(offer);

  // Remove from slave.
  Slave* slave = getSlave(offer->slave_id());
  CHECK(slave != NULL)
    << "Unknown slave " << offer->slave_id()
    << " in the offer " << offer->id();

  slave->removeOffer(offer);

  if (rescind) {
    RescindResourceOfferMessage message;
    message.mutable_offer_id()->MergeFrom(offer->id());
    send(framework->pid, message);
  }

  // Delete it.
  offers.erase(offer->id());
  delete offer;
}


Framework* Master::getFramework(const FrameworkID& frameworkId)
{
  return frameworks.contains(frameworkId) ? frameworks[frameworkId] : NULL;
}


Slave* Master::getSlave(const SlaveID& slaveId)
{
  return slaves.contains(slaveId) ? slaves[slaveId] : NULL;
}


Offer* Master::getOffer(const OfferID& offerId)
{
  return offers.contains(offerId) ? offers[offerId] : NULL;
}


// Create a new framework ID. We format the ID as MASTERID-FWID, where
// MASTERID is the ID of the master (launch date plus fault tolerant ID)
// and FWID is an increasing integer.
FrameworkID Master::newFrameworkId()
{
  std::ostringstream out;

  out << info.id() << "-" << std::setw(4)
      << std::setfill('0') << nextFrameworkId++;

  FrameworkID frameworkId;
  frameworkId.set_value(out.str());

  return frameworkId;
}


OfferID Master::newOfferId()
{
  OfferID offerId;
  offerId.set_value(info.id() + "-" + stringify(nextOfferId++));
  return offerId;
}


SlaveID Master::newSlaveId()
{
  SlaveID slaveId;
  slaveId.set_value(info.id() + "-" + stringify(nextSlaveId++));
  return slaveId;
}

} // namespace master {
} // namespace internal {
} // namespace mesos {