<commit_before>#include "executor.h"
#include <unistd.h>
#include <boost/lexical_cast.hpp>
#include <boost/scoped_ptr.hpp>

namespace baidu {
namespace shuttle {

const int sLineBufferSize = 40960;
const int sKeyLimit = 65536;

Executor::Executor() {
    line_buf_ = (char*)malloc(sLineBufferSize);
}

Executor::~Executor() {
    free(line_buf_);
}

void Executor::Stop(int32_t task_id) {
    MutexLock locker(&mu_);
    stop_task_ids_.insert(task_id);
}

bool Executor::ShouldStop(int32_t task_id) {
    MutexLock locker(&mu_);
    if (stop_task_ids_.find(task_id) != stop_task_ids_.end()) {
        return true;
    }
    return false;
}

Executor* Executor::GetExecutor(WorkMode mode) {
    Executor* executor;
    switch(mode) {
    case kMap:
        executor = new MapExecutor();
        break;
    case kReduce:
        executor = new ReduceExecutor();
        break;
    case kMapOnly:
        executor = new MapOnlyExecutor();
        break;
    default:
        LOG(FATAL, "unkonw work mode: %d", mode);
        abort();
        break;
    }
    return executor;
}

void Executor::SetEnv(const std::string& jobid, const TaskInfo& task) {
    ::setenv("mapred_job_id", jobid.c_str(), 1);
    ::setenv("mapred_job_name", task.job().name().c_str(), 1);
    ::setenv("mapred_output_dir", task.job().output().c_str(), 1);
    ::setenv("map_input_file", task.input().input_file().c_str(), 1);
    ::setenv("map_input_start", 
             boost::lexical_cast<std::string>(task.input().input_offset()).c_str(), 
             1);
    ::setenv("map_input_length", 
             boost::lexical_cast<std::string>(task.input().input_size()).c_str(),
             1);

    ::setenv("mapred_map_tasks", 
             boost::lexical_cast<std::string>(task.job().map_total()).c_str(), 
            1);
    ::setenv("mapred_reduce_tasks", 
             boost::lexical_cast<std::string>(task.job().reduce_total()).c_str(), 
            1);
    ::setenv("mapred_task_partition", 
             boost::lexical_cast<std::string>(task.task_id()).c_str(), 
            1);
    ::setenv("mapred_task_id",
             boost::lexical_cast<std::string>(task.task_id()).c_str(), 
            1);

    ::setenv("mapred_attempt_id",
             boost::lexical_cast<std::string>(task.attempt_id()).c_str(),
             1);
    ::setenv("minion_shuffle_work_dir", GetShuffleWorkDir(task).c_str(), 1);
    ::setenv("minion_input_dfs_host", task.job().input_dfs().host().c_str(), 1);
    ::setenv("minion_input_dfs_port", task.job().input_dfs().port().c_str(), 1);
    ::setenv("minion_input_dfs_user", task.job().input_dfs().user().c_str(), 1);
    ::setenv("minion_input_dfs_password", task.job().input_dfs().password().c_str(), 1);
    ::setenv("minion_output_dfs_host", task.job().output_dfs().host().c_str(), 1);
    ::setenv("minion_output_dfs_port", task.job().output_dfs().port().c_str(), 1);
    ::setenv("minion_output_dfs_user", task.job().output_dfs().user().c_str(), 1);
    ::setenv("minion_output_dfs_password", task.job().output_dfs().password().c_str(), 1);
    if (task.job().input_format() == kTextInput) {
        ::setenv("minion_input_format", "text", 1);
    } else if (task.job().input_format() == kBinaryInput) {
        ::setenv("minion_input_format", "binary", 1);
    } else if (task.job().input_format() == kNLineInput) {
        ::setenv("minion_input_format", "text", 1);
        ::setenv("minion_input_is_nline", "true", 1);
    }
    if (task.job().output_format() == kTextOutput) {<commit_msg>bug fix: task_id is a full id<commit_after>#include "executor.h"
#include <unistd.h>
#include <boost/lexical_cast.hpp>
#include <boost/scoped_ptr.hpp>

namespace baidu {
namespace shuttle {

const int sLineBufferSize = 40960;
const int sKeyLimit = 65536;

Executor::Executor() {
    line_buf_ = (char*)malloc(sLineBufferSize);
}

Executor::~Executor() {
    free(line_buf_);
}

void Executor::Stop(int32_t task_id) {
    MutexLock locker(&mu_);
    stop_task_ids_.insert(task_id);
}

bool Executor::ShouldStop(int32_t task_id) {
    MutexLock locker(&mu_);
    if (stop_task_ids_.find(task_id) != stop_task_ids_.end()) {
        return true;
    }
    return false;
}

Executor* Executor::GetExecutor(WorkMode mode) {
    Executor* executor;
    switch(mode) {
    case kMap:
        executor = new MapExecutor();
        break;
    case kReduce:
        executor = new ReduceExecutor();
        break;
    case kMapOnly:
        executor = new MapOnlyExecutor();
        break;
    default:
        LOG(FATAL, "unkonw work mode: %d", mode);
        abort();
        break;
    }
    return executor;
}

void Executor::SetEnv(const std::string& jobid, const TaskInfo& task) {
    ::setenv("mapred_job_id", jobid.c_str(), 1);
    ::setenv("mapred_job_name", task.job().name().c_str(), 1);
    ::setenv("mapred_output_dir", task.job().output().c_str(), 1);
    ::setenv("map_input_file", task.input().input_file().c_str(), 1);
    ::setenv("map_input_start", 
             boost::lexical_cast<std::string>(task.input().input_offset()).c_str(), 
             1);
    ::setenv("map_input_length", 
             boost::lexical_cast<std::string>(task.input().input_size()).c_str(),
             1);

    ::setenv("mapred_map_tasks", 
             boost::lexical_cast<std::string>(task.job().map_total()).c_str(), 
            1);
    ::setenv("mapred_reduce_tasks", 
             boost::lexical_cast<std::string>(task.job().reduce_total()).c_str(), 
            1);
    ::setenv("mapred_task_partition", 
             boost::lexical_cast<std::string>(task.task_id()).c_str(), 
            1);
    std::stringstream ss;
    ss << "attempt_" << jobid << "_" << task.task_id() << "_" << task.attempt_id();
    ::setenv("mapred_task_id",
             ss.str().c_str(), 
            1);

    ::setenv("mapred_attempt_id",
             boost::lexical_cast<std::string>(task.attempt_id()).c_str(),
             1);
    ::setenv("minion_shuffle_work_dir", GetShuffleWorkDir(task).c_str(), 1);
    ::setenv("minion_input_dfs_host", task.job().input_dfs().host().c_str(), 1);
    ::setenv("minion_input_dfs_port", task.job().input_dfs().port().c_str(), 1);
    ::setenv("minion_input_dfs_user", task.job().input_dfs().user().c_str(), 1);
    ::setenv("minion_input_dfs_password", task.job().input_dfs().password().c_str(), 1);
    ::setenv("minion_output_dfs_host", task.job().output_dfs().host().c_str(), 1);
    ::setenv("minion_output_dfs_port", task.job().output_dfs().port().c_str(), 1);
    ::setenv("minion_output_dfs_user", task.job().output_dfs().user().c_str(), 1);
    ::setenv("minion_output_dfs_password", task.job().output_dfs().password().c_str(), 1);
    if (task.job().input_format() == kTextInput) {
        ::setenv("minion_input_format", "text", 1);
    } else if (task.job().input_format() == kBinaryInput) {
        ::setenv("minion_input_format", "binary", 1);
    } else if (task.job().input_format() == kNLineInput) {
        ::setenv("minion_input_format", "text", 1);
        ::setenv("minion_input_is_nline", "true", 1);
    }
    if (task.job().output_format() == kTextOutput) {
{'fraction_non_alphanumeric': 0.1878523111612176, 'fraction_numerical': 0.010710259301014656, 'mean_word_length': 3.5905562742561448, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 22, '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': '3800596', 'n_tokens_mistral': 2584, 'n_tokens_neox': 2494, 'n_words': 355}