<commit_before>#include "gru.h"

#include <sstream>
#include <gflags/gflags.h>
#include <boost/algorithm/string.hpp>
#include <vector>
#include "logging.h"
#include "util.h"

DECLARE_int32(galaxy_deploy_step);
DECLARE_string(minion_path);
DECLARE_string(nexus_server_list);
DECLARE_string(nexus_root_path);
DECLARE_string(master_path);
DECLARE_bool(enable_cpu_soft_limit);
DECLARE_bool(enable_memory_soft_limit);
DECLARE_string(galaxy_node_label);
DECLARE_string(galaxy_user);
DECLARE_string(galaxy_token);
DECLARE_string(galaxy_pool);
DECLARE_int32(max_minions_per_host);

namespace baidu {
namespace shuttle {

static const int64_t default_additional_map_memory = 1024l * 1024 * 1024;
static const int64_t default_additional_reduce_memory = 2048l * 1024 * 1024;
static const int default_map_additional_millicores = 0;
static const int default_reduce_additional_millicores = 500;

int Gru::additional_map_millicores = default_map_additional_millicores;
int Gru::additional_reduce_millicores = default_reduce_additional_millicores;
int64_t Gru::additional_map_memory = default_additional_map_memory;
int64_t Gru::additional_reduce_memory = default_additional_reduce_memory;

Gru::Gru(::baidu::galaxy::sdk::AppMaster* galaxy, JobDescriptor* job,
         const std::string& job_id, WorkMode mode) :
        galaxy_(galaxy), job_(job), job_id_(job_id), mode_(mode) {
    mode_str_ = ((mode == kReduce) ? "reduce" : "map");
    minion_name_ = job->name() + "_" + mode_str_;
}

Status Gru::Start() {
    ::baidu::galaxy::sdk::SubmitJobRequest galaxy_job;
    galaxy_job.user.user = FLAGS_galaxy_user;
    galaxy_job.user.token = FLAGS_galaxy_token;
    galaxy_job.hostname = ::baidu::common::util::GetLocalHostName();
    galaxy_job.job.deploy.pools.push_back(FLAGS_galaxy_pool);
    galaxy_job.job.name = minion_name_ + "@minion";
    galaxy_job.job.type = ::baidu::galaxy::sdk::kJobBatch;
    galaxy_job.job.deploy.replica = (mode_ == kReduce) ? job_->reduce_capacity() : job_->map_capacity();
    galaxy_job.job.deploy.step = FLAGS_galaxy_deploy_step;
    galaxy_job.job.deploy.interval = 1;
    galaxy_job.job.deploy.max_per_host = FLAGS_max_minions_per_host;
    galaxy_job.job.deploy.update_break_count = 0;
    galaxy_job.job.version = "1.0.0";
    galaxy_job.job.run_user = "galaxy";
    if (!FLAGS_galaxy_node_label.empty()) {
        galaxy_job.job.deploy.tag = FLAGS_galaxy_node_label;
    }
    ::baidu::galaxy::sdk::PodDescription & pod_desc = galaxy_job.job.pod;
    pod_desc.workspace_volum.size = (3L << 30);
    pod_desc.workspace_volum.medium = ::baidu::galaxy::sdk::kDisk;
    pod_desc.workspace_volum.exclusive = false;
    pod_desc.workspace_volum.readonly = false;
    pod_desc.workspace_volum.use_symlink = false;
    pod_desc.workspace_volum.dest_path = "/home/shuttle";
    pod_desc.workspace_volum.type = ::baidu::galaxy::sdk::kEmptyDir;

    ::baidu::galaxy::sdk::TaskDescription task_desc;
    if (mode_str_ == "map") {
        task_desc.cpu.milli_core = job_->millicores() + additional_map_millicores;<commit_msg>set deploy step<commit_after>#include "gru.h"

#include <sstream>
#include <gflags/gflags.h>
#include <boost/algorithm/string.hpp>
#include <vector>
#include "logging.h"
#include "util.h"

DECLARE_int32(galaxy_deploy_step);
DECLARE_string(minion_path);
DECLARE_string(nexus_server_list);
DECLARE_string(nexus_root_path);
DECLARE_string(master_path);
DECLARE_bool(enable_cpu_soft_limit);
DECLARE_bool(enable_memory_soft_limit);
DECLARE_string(galaxy_node_label);
DECLARE_string(galaxy_user);
DECLARE_string(galaxy_token);
DECLARE_string(galaxy_pool);
DECLARE_int32(max_minions_per_host);

namespace baidu {
namespace shuttle {

static const int64_t default_additional_map_memory = 1024l * 1024 * 1024;
static const int64_t default_additional_reduce_memory = 2048l * 1024 * 1024;
static const int default_map_additional_millicores = 0;
static const int default_reduce_additional_millicores = 500;

int Gru::additional_map_millicores = default_map_additional_millicores;
int Gru::additional_reduce_millicores = default_reduce_additional_millicores;
int64_t Gru::additional_map_memory = default_additional_map_memory;
int64_t Gru::additional_reduce_memory = default_additional_reduce_memory;

Gru::Gru(::baidu::galaxy::sdk::AppMaster* galaxy, JobDescriptor* job,
         const std::string& job_id, WorkMode mode) :
        galaxy_(galaxy), job_(job), job_id_(job_id), mode_(mode) {
    mode_str_ = ((mode == kReduce) ? "reduce" : "map");
    minion_name_ = job->name() + "_" + mode_str_;
}

Status Gru::Start() {
    ::baidu::galaxy::sdk::SubmitJobRequest galaxy_job;
    galaxy_job.user.user = FLAGS_galaxy_user;
    galaxy_job.user.token = FLAGS_galaxy_token;
    galaxy_job.hostname = ::baidu::common::util::GetLocalHostName();
    galaxy_job.job.deploy.pools.push_back(FLAGS_galaxy_pool);
    galaxy_job.job.name = minion_name_ + "@minion";
    galaxy_job.job.type = ::baidu::galaxy::sdk::kJobBatch;
    galaxy_job.job.deploy.replica = (mode_ == kReduce) ? job_->reduce_capacity() : job_->map_capacity();
    galaxy_job.job.deploy.step = std::min((int)FLAGS_galaxy_deploy_step, (int)galaxy_job.job.deploy.replica);
    galaxy_job.job.deploy.interval = 1;
    galaxy_job.job.deploy.max_per_host = FLAGS_max_minions_per_host;
    galaxy_job.job.deploy.update_break_count = 0;
    galaxy_job.job.version = "1.0.0";
    galaxy_job.job.run_user = "galaxy";
    if (!FLAGS_galaxy_node_label.empty()) {
        galaxy_job.job.deploy.tag = FLAGS_galaxy_node_label;
    }
    ::baidu::galaxy::sdk::PodDescription & pod_desc = galaxy_job.job.pod;
    pod_desc.workspace_volum.size = (3L << 30);
    pod_desc.workspace_volum.medium = ::baidu::galaxy::sdk::kDisk;
    pod_desc.workspace_volum.exclusive = false;
    pod_desc.workspace_volum.readonly = false;
    pod_desc.workspace_volum.use_symlink = false;
    pod_desc.workspace_volum.dest_path = "/home/shuttle";
    pod_desc.workspace_volum.type = ::baidu::galaxy::sdk::kEmptyDir;

    ::baidu::galaxy::sdk::TaskDescription task_desc;
    if (mode_str_ == "map") {
        task_desc.cpu.milli_core = job_->millicores() + additional_map_millicores;
{'fraction_non_alphanumeric': 0.12400065263501386, 'fraction_numerical': 0.01566324033284386, 'mean_word_length': 6.054085155350978, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 19, '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': '6841485', 'n_tokens_mistral': 2333, 'n_tokens_neox': 2208, 'n_words': 302}