<commit_before>// Copyright (c) 2015, Galaxy Authors. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// Author: <EMAIL>

#include "agent/task_runner.h"

#include <sys/wait.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <errno.h>
#include <sstream>
#include <sys/types.h>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <pwd.h>
#include "common/logging.h"
#include "common/util.h"
#include "downloader_manager.h"
#include "agent/resource_collector_engine.h"
#include "agent/utils.h"

extern int FLAGS_task_retry_times;
extern std::string FLAGS_task_acct;

namespace galaxy {

static const std::string RUNNER_META_PREFIX = "task_runner_";

int AbstractTaskRunner::IsRunning(){
    if ((int64_t)m_child_pid == -1) {
        LOG(WARNING, "task with id %ld not running with pid %ld",
                m_task_info.task_id(),
                (int64_t)m_child_pid);
        return -1;
    }
    // check process exist
    int ret = ::kill(m_child_pid, 0);
    if(ret == 0 ){
        //check process status
        pid_t pid = waitpid(m_child_pid,&ret,WNOHANG);
        if(pid == 0 ){
            LOG(INFO,"process %d is running",m_child_pid);
            return 0;
        }else if(pid == -1){
            LOG(WARNING,"fail to check process %d state",m_child_pid);
            return -1;
        }
        else{
            if(WIFEXITED(ret)){
                int exit_code = WEXITSTATUS(ret);
                if(exit_code == 0 ){
                    //normal exit
                    LOG(INFO,"process %d exits successfully",m_child_pid);
                    return 1;
                }
                LOG(FATAL,"process %d exits with err code %d", exit_code);
            }
            return -1;
        }

    }
    LOG(INFO, "check task %d error[%d:%s] ",
            m_task_info.task_id(),
            errno,
            strerror(errno));
    return ret;
}

void AbstractTaskRunner::AsyncDownload(boost::function<void()> callback) {
    std::string uri = m_task_info.task_raw();
    std::string path = m_workspace->GetPath();
    path.append("/");
    path.append("tmp.tar.gz");
    m_task_state = DEPLOYING;
    // set deploying state
    DownloaderManager* downloader_handler = DownloaderManager::GetInstance();
    downloader_id_ = downloader_handler->DownloadInThread(
                        uri,
                        path,
                        boost::bind(&AbstractTaskRunner::StartAfterDownload, this, callback, _1));
    return;
}

void AbstractTaskRunner::StartAfterDownload(boost::function<void()> callback, int ret) {
    if (ret == 0) {
        std::string tar_cmd = "cd "
            + m_workspace->GetPath()
            + " && tar -xzf tmp.tar.gz";
        int status = system(tar_cmd.c_str());
        if (status != 0) {
            LOG(WARNING, "task with id %ld extract failed",
                    m_task_info.task_id());
        }
        else {
            callback();
            return;
        }
    }
    LOG(WARNING, "task with id %ld deploy failed",
            m_task_info.task_id());
    m_task_state = ERROR;
    return;
}

int AbstractTaskRunner::Stop(){
    if (m_task_state == DEPLOYING) {
        // do download stop
        DownloaderManager* downloader_handler = DownloaderManager::GetInstance();
        downloader_handler->KillDownload(downloader_id_);
        LOG(DEBUG, "task id %ld stop failed with deploying", m_task_info.task_id());
        return -1;
    }

    if (IsRunning() != 0) {
        return 0;
    }
    LOG(INFO,"start to kill process group %d",m_group_pid);
    int ret = killpg(m_group_pid, 9);
    if(ret != 0){
        LOG(WARNING,"fail to kill process group %d",m_group_pid);
    }
    pid_t killed_pid = wait(&ret);
    if (killed_pid == -1){
        LOG(FATAL,"fail to kill process group %d",m_group_pid);
        return -1;
    }else{
        StopPost();
        LOG(INFO,"kill child process %d successfully",killed_pid);
        m_child_pid = -1;
        m_group_pid = -1;
        return 0;
    }
}

void AbstractTaskRunner::PrepareStart(std::vector<int>& fd_vector,int* stdout_fd,int* stderr_fd){
    pid_t current_pid = getpid();
    common::util::GetProcessFdList(current_pid, fd_vector);
    std::string task_stdout = m_workspace->GetPath() + "/./stdout";
    std::string task_stderr = m_workspace->GetPath() + "/./stderr";
    *stdout_fd = open(task_stdout.c_str(), O_CREAT | O_TRUNC | O_WRONLY,
                      S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    *stderr_fd = open(task_stderr.c_str(), O_CREAT | O_TRUNC | O_WRONLY,
                      S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
}

void AbstractTaskRunner::StartTaskAfterFork(std::vector<int>& fd_vector,int stdout_fd,int stderr_fd){
    // do in child process,
    // all interface called in child process should be async-safe.
    // NOTE if dup2 will return errno == EINTR?
    while (dup2(stdout_fd, STDOUT_FILENO) == -1 && errno == EINTR) {}
    while (dup2(stderr_fd, STDERR_FILENO) == -1 && errno == EINTR) {}
    for (size_t i = 0; i < fd_vector.size(); i++) {
        if (fd_vector[i] == STDOUT_FILENO
                    || fd_vector[i] == STDERR_FILENO
                    || fd_vector[i] == STDIN_FILENO) {
                // do not deal with std input/output
            continue;
         }
         close(fd_vector[i]);
    }

    //chroot(m_workspace->GetPath().c_str());
    chdir(m_workspace->GetPath().c_str());
    passwd *pw = getpwnam(FLAGS_task_acct.c_str());
    if (NULL == pw) {
        pw = getpwnam("default");
        if (NULL == pw)
        {
            LOG(FATAL,"fail to find acct ");
            return;
        }
    }

    chown(m_workspace->GetPath().c_str(), pw->pw_uid, pw->pw_gid);
    setuid(pw->pw_uid);
    char *argv[] = {"sh","-c",const_cast<char*>(m_task_info.cmd_line().c_str()),NULL};
    std::stringstream task_id_env;
    task_id_env <<"TASK_ID="<<m_task_info.task_offset();
    std::stringstream task_num_env;
    task_num_env <<"TASK_NUM="<<m_task_info.job_replicate_num();
    std::stringstream task_user_env;
    task_user_env <<"USER="<<pw->pw_uid;
    char *env[] = {const_cast<char*>(task_id_env.str().c_str()),
                   const_cast<char*>(task_num_env.str().c_str()),
                   const_cast<char*>(task_user_env.str().c_str()),
                   NULL};

    execve("/bin/sh", argv, env);
    LOG(FATAL,"fail to kill exec %s errno %d %s",
        m_task_info.task_name().c_str(), errno, strerror(errno));
    assert(0);
    _exit(127);
}
int AbstractTaskRunner::ReStart(){<commit_msg>change work dir owner before run<commit_after>// Copyright (c) 2015, Galaxy Authors. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// Author: <EMAIL>

#include "agent/task_runner.h"

#include <sys/wait.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <errno.h>
#include <sstream>
#include <sys/types.h>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <pwd.h>
#include "common/logging.h"
#include "common/util.h"
#include "downloader_manager.h"
#include "agent/resource_collector_engine.h"
#include "agent/utils.h"

extern int FLAGS_task_retry_times;
extern std::string FLAGS_task_acct;

namespace galaxy {

static const std::string RUNNER_META_PREFIX = "task_runner_";

int AbstractTaskRunner::IsRunning(){
    if ((int64_t)m_child_pid == -1) {
        LOG(WARNING, "task with id %ld not running with pid %ld",
                m_task_info.task_id(),
                (int64_t)m_child_pid);
        return -1;
    }
    // check process exist
    int ret = ::kill(m_child_pid, 0);
    if(ret == 0 ){
        //check process status
        pid_t pid = waitpid(m_child_pid,&ret,WNOHANG);
        if(pid == 0 ){
            LOG(INFO,"process %d is running",m_child_pid);
            return 0;
        }else if(pid == -1){
            LOG(WARNING,"fail to check process %d state",m_child_pid);
            return -1;
        }
        else{
            if(WIFEXITED(ret)){
                int exit_code = WEXITSTATUS(ret);
                if(exit_code == 0 ){
                    //normal exit
                    LOG(INFO,"process %d exits successfully",m_child_pid);
                    return 1;
                }
                LOG(FATAL,"process %d exits with err code %d", exit_code);
            }
            return -1;
        }

    }
    LOG(INFO, "check task %d error[%d:%s] ",
            m_task_info.task_id(),
            errno,
            strerror(errno));
    return ret;
}

void AbstractTaskRunner::AsyncDownload(boost::function<void()> callback) {
    std::string uri = m_task_info.task_raw();
    std::string path = m_workspace->GetPath();
    path.append("/");
    path.append("tmp.tar.gz");
    m_task_state = DEPLOYING;
    // set deploying state
    DownloaderManager* downloader_handler = DownloaderManager::GetInstance();
    downloader_id_ = downloader_handler->DownloadInThread(
                        uri,
                        path,
                        boost::bind(&AbstractTaskRunner::StartAfterDownload, this, callback, _1));
    return;
}

void AbstractTaskRunner::StartAfterDownload(boost::function<void()> callback, int ret) {
    if (ret == 0) {
        std::string tar_cmd = "cd "
            + m_workspace->GetPath()
            + " && tar -xzf tmp.tar.gz";
        int status = system(tar_cmd.c_str());
        if (status != 0) {
            LOG(WARNING, "task with id %ld extract failed",
                    m_task_info.task_id());
        }
        else {
            callback();
            return;
        }
    }
    LOG(WARNING, "task with id %ld deploy failed",
            m_task_info.task_id());
    m_task_state = ERROR;
    return;
}

int AbstractTaskRunner::Stop(){
    if (m_task_state == DEPLOYING) {
        // do download stop
        DownloaderManager* downloader_handler = DownloaderManager::GetInstance();
        downloader_handler->KillDownload(downloader_id_);
        LOG(DEBUG, "task id %ld stop failed with deploying", m_task_info.task_id());
        return -1;
    }

    if (IsRunning() != 0) {
        return 0;
    }
    LOG(INFO,"start to kill process group %d",m_group_pid);
    int ret = killpg(m_group_pid, 9);
    if(ret != 0){
        LOG(WARNING,"fail to kill process group %d",m_group_pid);
    }
    pid_t killed_pid = wait(&ret);
    if (killed_pid == -1){
        LOG(FATAL,"fail to kill process group %d",m_group_pid);
        return -1;
    }else{
        StopPost();
        LOG(INFO,"kill child process %d successfully",killed_pid);
        m_child_pid = -1;
        m_group_pid = -1;
        return 0;
    }
}

void AbstractTaskRunner::PrepareStart(std::vector<int>& fd_vector,int* stdout_fd,int* stderr_fd){
    pid_t current_pid = getpid();
    common::util::GetProcessFdList(current_pid, fd_vector);
    std::string task_stdout = m_workspace->GetPath() + "/./stdout";
    std::string task_stderr = m_workspace->GetPath() + "/./stderr";
    *stdout_fd = open(task_stdout.c_str(), O_CREAT | O_TRUNC | O_WRONLY,
                      S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    *stderr_fd = open(task_stderr.c_str(), O_CREAT | O_TRUNC | O_WRONLY,
                      S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
}

void AbstractTaskRunner::StartTaskAfterFork(std::vector<int>& fd_vector,int stdout_fd,int stderr_fd){
    // do in child process,
    // all interface called in child process should be async-safe.
    // NOTE if dup2 will return errno == EINTR?
    while (dup2(stdout_fd, STDOUT_FILENO) == -1 && errno == EINTR) {}
    while (dup2(stderr_fd, STDERR_FILENO) == -1 && errno == EINTR) {}
    for (size_t i = 0; i < fd_vector.size(); i++) {
        if (fd_vector[i] == STDOUT_FILENO
                    || fd_vector[i] == STDERR_FILENO
                    || fd_vector[i] == STDIN_FILENO) {
                // do not deal with std input/output
            continue;
         }
         close(fd_vector[i]);
    }

    //chroot(m_workspace->GetPath().c_str());
    chdir(m_workspace->GetPath().c_str());
    passwd *pw = getpwnam(FLAGS_task_acct.c_str());
    if (NULL == pw) {
        pw = getpwnam("default");
        if (NULL == pw)
        {
            LOG(FATAL,"fail to find acct ");
            return;
        }
    }
    std::stringstream cmd;
    cmd << "chown -R " << FLAGS_task_acct.c_str()
        <<":"<< FLAGS_task_acct.c_str() << " . ";
    system(cmd.str().c_str());
    if (errno) {
        LOG(WARNING, "work dir change owner failed.");
        return;
    }

    chown(m_workspace->GetPath().c_str(), pw->pw_uid, pw->pw_gid);
    setuid(pw->pw_uid);
    char *argv[] = {"sh","-c",const_cast<char*>(m_task_info.cmd_line().c_str()),NULL};
    std::stringstream task_id_env;
    task_id_env <<"TASK_ID="<<m_task_info.task_offset();
    std::stringstream task_num_env;
    task_num_env <<"TASK_NUM="<<m_task_info.job_replicate_num();
    std::stringstream task_user_env;
    task_user_env <<"USER="<<pw->pw_uid;
    char *env[] = {const_cast<char*>(task_id_env.str().c_str()),
                   const_cast<char*>(task_num_env.str().c_str()),
                   const_cast<char*>(task_user_env.str().c_str()),
                   NULL};

    execve("/bin/sh", argv, env);
    LOG(FATAL,"fail to kill exec %s errno %d %s",
        m_task_info.task_name().c_str(), errno, strerror(errno));
    assert(0);
    _exit(127);
}
int AbstractTaskRunner::ReStart(){
{'fraction_non_alphanumeric': 0.14172879604340444, 'fraction_numerical': 0.006200634826898944, 'mean_word_length': 2.4385786802030456, 'pattern_counts': {'":': 1, '<': 79, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4705752', 'n_tokens_mistral': 4665, 'n_tokens_neox': 4508, 'n_words': 951}