<commit_before>#include "allocator.hpp"
#include "master.hpp"
#include "master_webui.hpp"
#include "allocator_factory.hpp"
#include "ft_messaging.hpp"

#include <glog/logging.h>

using std::endl;
using std::max;
using std::min;
using std::pair;
using std::make_pair;
using std::ostringstream;
using std::map;
using std::set;
using std::string;
using std::vector;

using boost::lexical_cast;
using boost::unordered_map;
using boost::unordered_set;

using namespace nexus;
using namespace nexus::internal;
using namespace nexus::internal::master;


namespace {

// A process that periodically pings the master to check filter expiries, etc
class AllocatorTimer : public Tuple<Process>
{
private:
  PID master;

protected:
  void operator () ()
  {
    link(master);
    do {
      // TODO: Make timer interval configurable, and hopefully less than 1 sec
      switch (receive(1)) {
      case PROCESS_TIMEOUT:
	send(master, pack<M2M_TIMER_TICK>());
	break;
      case PROCESS_EXIT:
	return;
      }
    } while (true);
  }

public:
  AllocatorTimer(const PID &_master) : master(_master) {}
};


// A process that periodically prints frameworks' shares to a file
class SharesPrinter : public Tuple<Process>
{
protected:
  PID master;

  void operator () ()
  {
    int tick = 0;

    std::ofstream file ("/mnt/shares");
    if (!file.is_open())
      LOG(FATAL) << "Could not open /mnt/shares";

    while (true) {
      pause(1);
      send(master, pack<M2M_GET_STATE>());
      receive();
      state::MasterState *state;
      unpack<M2M_GET_STATE_REPLY>(*((int64_t *) &state));

      uint32_t total_cpus = 0;
      uint64_t total_mem = 0;

      foreach (state::Slave *s, state->slaves) {
        total_cpus += s->cpus;
        total_mem += s->mem;
      }
      
      if (state->frameworks.empty()) {
        file << "--------------------------------" << std::endl;
      } else {
        foreach (state::Framework *f, state->frameworks) {
          double cpu_share = f->cpus / (double) total_cpus;
          double mem_share = f->mem / (double) total_mem;
          double max_share = max(cpu_share, mem_share);
          file << tick << "#" << f->id << "#" << f->name << "#" 
               << f->cpus << "#" << f->mem << "#"
               << cpu_share << "#" << mem_share << "#" << max_share << endl;
        }
      }
      delete state;
      tick++;
    }
    file.close();
  }

public:
  SharesPrinter(const PID &_master) : master(_master) {}
  ~SharesPrinter() {}
};

}


Master::Master(const string &zk)
  : masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0), 
    nextSlotOfferId(0), allocatorType("simple"), masterId(0)
{
  if (zk != "") {
    pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
    if (urlPair.first == UrlProcessor::ZOO) {
      isFT = true;
      zkServers = urlPair.second;
    } else {
      LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
      exit(1);
    }
  }
  ftMsg = FTMessaging::getInstance();
}


Master::Master(const string& _allocatorType, const string &zk)
  : masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0), 
    nextSlotOfferId(0), allocatorType(_allocatorType), masterId(0)
{
  if (zk != "") {
    pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
    if (urlPair.first == UrlProcessor::ZOO) {
      isFT = true;
      zkServers = urlPair.second;
    } else {
      LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
      exit(1);
    }
  }
  ftMsg = FTMessaging::getInstance();
}
                   

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

  if (masterDetector != NULL) {
    delete masterDetector;
    masterDetector = NULL;
  }

  delete allocator;

  foreachpair (_, Framework *framework, frameworks) {
    foreachpair(_, TaskInfo *task, framework->tasks)
      delete task;
    delete framework;
  }

  foreachpair (_, Slave *slave, slaves) {
    delete slave;
  }

  foreachpair (_, SlotOffer *offer, slotOffers) {
    delete offer;
  }
}


state::MasterState * Master::getState()
{
  std::ostringstream oss;
  oss << self();
  state::MasterState *state =
    new state::MasterState(BUILD_DATE, BUILD_USER, oss.str(), isFT);

  foreachpair (_, Slave *s, slaves) {
    state::Slave *slave = new state::Slave(s->id, s->hostname, s->publicDns,
          s->resources.cpus, s->resources.mem, s->connectTime);
    state->slaves.push_back(slave);
  }

  foreachpair (_, Framework *f, frameworks) {
    state::Framework *framework = new state::Framework(f->id, f->name, 
       f->executorInfo.uri, f->resources.cpus, f->resources.mem,
       f->connectTime);
    state->frameworks.push_back(framework);
    foreachpair (_, TaskInfo *t, f->tasks) {
      state::Task *task = new state::Task(t->id, t->name, t->frameworkId,
          t->slaveId, t->state, t->resources.cpus, t->resources.mem);
      framework->tasks.push_back(task);
    }
    foreach (SlotOffer *o, f->slotOffers) {
      state::SlotOffer *offer = new state::SlotOffer(o->id, o->frameworkId);
      foreach (SlaveResources &r, o->resources) {
        state::SlaveResources *resources = new state::SlaveResources(
            r.slave->id, r.resources.cpus, r.resources.mem);
        offer->resources.push_back(resources);
      }
      framework->offers.push_back(offer);
    }
  }
  
  return state;
}


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


// Return connected slaves that are not in the process of being removed
vector<Slave *> Master::getActiveSlaves()
{
  vector <Slave *> result;
  foreachpair(_, Slave *slave, slaves)
    if (slave->active)
      result.push_back(slave);
  return result;
}


Framework * Master::lookupFramework(FrameworkID fid)
{
  unordered_map<FrameworkID, Framework *>::iterator it =
    frameworks.find(fid);
  if (it != frameworks.end())
    return it->second;
  else
    return NULL;
}


Slave * Master::lookupSlave(SlaveID sid)
{
  unordered_map<SlaveID, Slave *>::iterator it =
    slaves.find(sid);
  if (it != slaves.end())
    return it->second;
  else
    return NULL;
}


SlotOffer * Master::lookupSlotOffer(OfferID oid)
{
  unordered_map<OfferID, SlotOffer *>::iterator it =
    slotOffers.find(oid);
  if (it != slotOffers.end()) 
    return it->second;
  else
    return NULL;
}

void Master::updateFrameworkTasks() {
  foreachpair (SlaveID sid, Slave *slave, slaves) {
    foreachpair (_, TaskInfo *task, slave->tasks) {
      updateFrameworkTasks(task);
    }
  }
}

void Master::updateFrameworkTasks(TaskInfo *task) {
  Framework *fwrk = lookupFramework(task->frameworkId);
  if (fwrk != NULL) {
    if (fwrk->tasks.find(task->id) == fwrk->tasks.end()) {
      fwrk->tasks[task->id] = task;
      fwrk->resources += task->resources;
    }
  }
}


void Master::operator () ()
{
  LOG(INFO) << "Master started at nexus://" << self();

  if (isFT) {
    LOG(INFO) << "Connecting to ZooKeeper at " << zkServers;
    masterDetector = new MasterDetector(zkServers, ZNODE, self(), true);
  } else {
    send(self(), pack<GOT_MASTER_SEQ>("0"));
  }

  // Don't do anything until we get a sequence identifier.
  bool waitingForSeq = true;
  do {
    switch (receive()) {
      case GOT_MASTER_SEQ: {
	string mySeq;
	unpack<GOT_MASTER_SEQ>(mySeq);
	masterId = lexical_cast<long>(mySeq);
	LOG(INFO) << "Master ID:" << masterId;
	break;
      }
    }
  } while (waitingForSeq);
<commit_msg>Making the master testable requires at least setting isFT to false, not sure if this code ever got tested via alltests before, but it's working now!<commit_after>#include "allocator.hpp"
#include "master.hpp"
#include "master_webui.hpp"
#include "allocator_factory.hpp"
#include "ft_messaging.hpp"

#include <glog/logging.h>

using std::endl;
using std::max;
using std::min;
using std::pair;
using std::make_pair;
using std::ostringstream;
using std::map;
using std::set;
using std::string;
using std::vector;

using boost::lexical_cast;
using boost::unordered_map;
using boost::unordered_set;

using namespace nexus;
using namespace nexus::internal;
using namespace nexus::internal::master;


namespace {

// A process that periodically pings the master to check filter expiries, etc
class AllocatorTimer : public Tuple<Process>
{
private:
  PID master;

protected:
  void operator () ()
  {
    link(master);
    do {
      // TODO: Make timer interval configurable, and hopefully less than 1 sec
      switch (receive(1)) {
      case PROCESS_TIMEOUT:
	send(master, pack<M2M_TIMER_TICK>());
	break;
      case PROCESS_EXIT:
	return;
      }
    } while (true);
  }

public:
  AllocatorTimer(const PID &_master) : master(_master) {}
};


// A process that periodically prints frameworks' shares to a file
class SharesPrinter : public Tuple<Process>
{
protected:
  PID master;

  void operator () ()
  {
    int tick = 0;

    std::ofstream file ("/mnt/shares");
    if (!file.is_open())
      LOG(FATAL) << "Could not open /mnt/shares";

    while (true) {
      pause(1);
      send(master, pack<M2M_GET_STATE>());
      receive();
      state::MasterState *state;
      unpack<M2M_GET_STATE_REPLY>(*((int64_t *) &state));

      uint32_t total_cpus = 0;
      uint64_t total_mem = 0;

      foreach (state::Slave *s, state->slaves) {
        total_cpus += s->cpus;
        total_mem += s->mem;
      }
      
      if (state->frameworks.empty()) {
        file << "--------------------------------" << std::endl;
      } else {
        foreach (state::Framework *f, state->frameworks) {
          double cpu_share = f->cpus / (double) total_cpus;
          double mem_share = f->mem / (double) total_mem;
          double max_share = max(cpu_share, mem_share);
          file << tick << "#" << f->id << "#" << f->name << "#" 
               << f->cpus << "#" << f->mem << "#"
               << cpu_share << "#" << mem_share << "#" << max_share << endl;
        }
      }
      delete state;
      tick++;
    }
    file.close();
  }

public:
  SharesPrinter(const PID &_master) : master(_master) {}
  ~SharesPrinter() {}
};

}


Master::Master(const string &zk)
  : isFT(false), masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0), 
    nextSlotOfferId(0), allocatorType("simple"), masterId(0)
{
  if (zk != "") {
    pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
    if (urlPair.first == UrlProcessor::ZOO) {
      isFT = true;
      zkServers = urlPair.second;
    } else {
      LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
      exit(1);
    }
  }
  ftMsg = FTMessaging::getInstance();
}


Master::Master(const string& _allocatorType, const string &zk)
  : isFT(false), masterDetector(NULL), nextFrameworkId(0), nextSlaveId(0), 
    nextSlotOfferId(0), allocatorType(_allocatorType), masterId(0)
{
  if (zk != "") {
    pair<UrlProcessor::URLType, string> urlPair = UrlProcessor::process(zk);
    if (urlPair.first == UrlProcessor::ZOO) {
      isFT = true;
      zkServers = urlPair.second;
    } else {
      LOG(ERROR) << "Failed to parse URL for ZooKeeper servers. URL must start with zoo:// or zoofile://";
      exit(1);
    }
  }
  ftMsg = FTMessaging::getInstance();
}
                   

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

  if (masterDetector != NULL) {
    delete masterDetector;
    masterDetector = NULL;
  }

  delete allocator;

  foreachpair (_, Framework *framework, frameworks) {
    foreachpair(_, TaskInfo *task, framework->tasks)
      delete task;
    delete framework;
  }

  foreachpair (_, Slave *slave, slaves) {
    delete slave;
  }

  foreachpair (_, SlotOffer *offer, slotOffers) {
    delete offer;
  }
}


state::MasterState * Master::getState()
{
  std::ostringstream oss;
  oss << self();
  state::MasterState *state =
    new state::MasterState(BUILD_DATE, BUILD_USER, oss.str(), isFT);

  foreachpair (_, Slave *s, slaves) {
    state::Slave *slave = new state::Slave(s->id, s->hostname, s->publicDns,
          s->resources.cpus, s->resources.mem, s->connectTime);
    state->slaves.push_back(slave);
  }

  foreachpair (_, Framework *f, frameworks) {
    state::Framework *framework = new state::Framework(f->id, f->name, 
       f->executorInfo.uri, f->resources.cpus, f->resources.mem,
       f->connectTime);
    state->frameworks.push_back(framework);
    foreachpair (_, TaskInfo *t, f->tasks) {
      state::Task *task = new state::Task(t->id, t->name, t->frameworkId,
          t->slaveId, t->state, t->resources.cpus, t->resources.mem);
      framework->tasks.push_back(task);
    }
    foreach (SlotOffer *o, f->slotOffers) {
      state::SlotOffer *offer = new state::SlotOffer(o->id, o->frameworkId);
      foreach (SlaveResources &r, o->resources) {
        state::SlaveResources *resources = new state::SlaveResources(
            r.slave->id, r.resources.cpus, r.resources.mem);
        offer->resources.push_back(resources);
      }
      framework->offers.push_back(offer);
    }
  }
  
  return state;
}


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


// Return connected slaves that are not in the process of being removed
vector<Slave *> Master::getActiveSlaves()
{
  vector <Slave *> result;
  foreachpair(_, Slave *slave, slaves)
    if (slave->active)
      result.push_back(slave);
  return result;
}


Framework * Master::lookupFramework(FrameworkID fid)
{
  unordered_map<FrameworkID, Framework *>::iterator it =
    frameworks.find(fid);
  if (it != frameworks.end())
    return it->second;
  else
    return NULL;
}


Slave * Master::lookupSlave(SlaveID sid)
{
  unordered_map<SlaveID, Slave *>::iterator it =
    slaves.find(sid);
  if (it != slaves.end())
    return it->second;
  else
    return NULL;
}


SlotOffer * Master::lookupSlotOffer(OfferID oid)
{
  unordered_map<OfferID, SlotOffer *>::iterator it =
    slotOffers.find(oid);
  if (it != slotOffers.end()) 
    return it->second;
  else
    return NULL;
}

void Master::updateFrameworkTasks() {
  foreachpair (SlaveID sid, Slave *slave, slaves) {
    foreachpair (_, TaskInfo *task, slave->tasks) {
      updateFrameworkTasks(task);
    }
  }
}

void Master::updateFrameworkTasks(TaskInfo *task) {
  Framework *fwrk = lookupFramework(task->frameworkId);
  if (fwrk != NULL) {
    if (fwrk->tasks.find(task->id) == fwrk->tasks.end()) {
      fwrk->tasks[task->id] = task;
      fwrk->resources += task->resources;
    }
  }
}


void Master::operator () ()
{
  LOG(INFO) << "Master started at nexus://" << self();

  if (isFT) {
    LOG(INFO) << "Connecting to ZooKeeper at " << zkServers;
    masterDetector = new MasterDetector(zkServers, ZNODE, self(), true);
  } else {
    send(self(), pack<GOT_MASTER_SEQ>("0"));
  }

  // Don't do anything until we get a sequence identifier.
  bool waitingForSeq = true;
  do {
    switch (receive()) {
      case GOT_MASTER_SEQ: {
	string mySeq;
	unpack<GOT_MASTER_SEQ>(mySeq);
	masterId = lexical_cast<long>(mySeq);
	LOG(INFO) << "Master ID:" << masterId;
	waitingForSeq = false;
	break;
      }
    }
  } while (waitingForSeq);

{'fraction_non_alphanumeric': 0.15152099813492828, 'fraction_numerical': 0.0033442665123159044, 'mean_word_length': 3.22784121805329, 'pattern_counts': {'":': 0, '<': 155, '<?xml version=': 0, '>': 151, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3253575', 'n_tokens_mistral': 5320, 'n_tokens_neox': 5161, 'n_words': 1199}