DRCETaskShellHost.cpp

Go to the documentation of this file.

#include <fstream>
#include <list>
#include <sys/wait.h>
#include <Poco/SharedPtr.h>
#include <Poco/Timestamp.h>
#include <Poco/Logger.h>
#include <Poco/Thread.h>
#include <Poco/TaskManager.h>
#include <Poco/File.h>

#include "CommandExecutor.hpp"
#include "DRCEMessageConst.hpp"
#include "DRCEError.hpp"
#include "DRCEFileStream.hpp"
#include "DRCEAsyncTasksQueue.hpp"
#include "DRCETaskShellHost.hpp"
#include "DRCETaskRequestSetExecuteExecutor.hpp"
#include "DRCEFileExtractor.hpp"
#include "DRCETaskRequestFactory.hpp"
#include "DRCECleanupTask.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
DRCETaskShellHost::DRCETaskShellHost(const std::string& taskName_, DRCENodeOptions& nodeOptions_, CustomMessage& message_,
    const DRCETaskRequestSetExecute& taskRequestSetExecute_, DRCEAsyncTasksQueue& asyncTasksQueue_,
    DRCEResourceMonitor& resourceMonitor_, const DRCEInputJsonMessage& inputJsonMessage_, unsigned int parentTaskId_)
: inherited(taskName_, nodeOptions_, message_, taskRequestSetExecute_, asyncTasksQueue_, resourceMonitor_, inputJsonMessage_, parentTaskId_)
{
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCETaskShellHost::runTask(void)
{
  if (isCancelled())
    return;

  Poco::Timestamp now;
  std::time_t timeStart = now.epochTime();

  DRCEResultDataItem resultDataItem = DRCETaskRequestSetExecuteExecutor::getResultDataItem(inputJsonMessage, nodeOptions, DRCETaskRequest::TaskState::SET_AS_NEW);
  AsyncTasks asyncTask(0, parentTaskId, 0, taskRequestSetExecute.getSessionOptions().timeMax, timeStart, 0, DRCETaskRequest::TaskState::SET_AS_NEW);
  asyncTask.usageLimits = taskRequestSetExecute.getResourceLimits().usageLimits;

  try
  {
    unsigned int taskId = std::stoul(name());

    DRCECommonTask::SubtasksList subtasks;
    DRCECommonTask::getSubtasksList(inputJsonMessage, subtasks);
    for (size_t i=0;i<subtasks.size();++i)
    {
      asyncTasksQueue.addSubtask(taskId, subtasks[i].first, (subtasks[i].second==SessionOptions::CleanupFlag::cfDelete));
    }

    asyncTask.taskId = taskId;
    asyncTask.parentTaskId = parentTaskId;
    asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), &resultDataItem);

    // write to tasks log information about appear new task
    asyncTasksQueue.writeToTasksLog(drce_const::DRCE_TASKS_LOGGER_DELIMITER, drce_const::DRCE_TASKS_LOGGER_CREATED, asyncTask);

    asyncTask.state = DRCETaskRequest::TaskState::IN_PROGRESS;
    asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), &resultDataItem);

    ApplyPidHandler applyPidHandler(asyncTasksQueue, asyncTask, DRCETaskRequest::TaskState::IN_PROGRESS, SessionOptions::ThreadMode::tmAsync);

    resultDataItem = execute(applyPidHandler);

    if (!isCancelled())
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCETaskShellHost::runTask  before executeSubtasks()", Poco::Message::Priority::PRIO_TRACE));
      // block running subtasks
      executeSubtasks(resultDataItem);

      if (!isCancelled())
      {
        for (size_t i=0;i<resultDataItem.getSubtasksCount();++i)
        {
          DRCECommonTask::waitUpdateAllTasks(resultDataItem.getSubtaskItem(i).getRequestId(), nodeOptions, message, asyncTasksQueue, this);
        }
      }
    }
    // save values lenght of stdout and time execution (for resource monitor)
    resourceMonitor.addSizeOutputBufferAsyncTasks(resultDataItem.getStdoutStream().length());
    resourceMonitor.addTimeAsyncTask(resultDataItem.getTime());
    // increment count of fail sync tasks
    if (resultDataItem.getErrorCode()!=NO_ERROR)
      resourceMonitor.incrementCountAsyncTasksFail();

    std::stringstream outMsg;
    outMsg << "DRCETaskShellHost::runTask  isCancelled = " << std::boolalpha << isCancelled();
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

    if (!isCancelled())
    {
      asyncTask.state = DRCETaskRequest::TaskState::FINISHED;
      asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
      asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), &resultDataItem);

      outMsg.str("");
      outMsg << ">>>>>  FINISHED (" << taskId << ")";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));
    }
    else
    {
      asyncTask.state = DRCETaskRequest::TaskState::TERMINATED;
      asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));

      outMsg.str("");
      outMsg << ">>>>>  TERMINATED (" << taskId << ")\n";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

      Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), taskId));
      if (statusFile.exists())
      {
        unsigned int errorCode = NO_ERROR;
        unsigned int errorCodeMainTask = NO_ERROR;
        std::string errorMessage;
        DRCETaskRequest::TaskState taskState = DRCETaskRequest::TaskState::TERMINATED;

        DRCEResultDataItem resultItem = DRCECommonTask::extractStatusResultDataItem(taskId, nodeOptions, asyncTasksQueue);
        if (resultItem.getErrorCode() == ERROR_TERMINATE_EXPIRED_TASK)
        {
          errorCode = ERROR_TERMINATE_SUBTASK_AFTER_EXPIRED_MAIN_TASK;
          errorCodeMainTask = ERROR_TERMINATE_EXPIRED_TASK;
          errorMessage = message(message_const::TASK_TIME_EXPIRED, asyncTask.timeMax);
          taskState = DRCETaskRequest::TaskState::TERMINATED_BY_TTL;
        }
        else
        {
          errorCode = resultItem.getErrorCode();
          errorCodeMainTask = resultItem.getErrorCode();
          errorMessage = resultItem.getErrorMessage();
          taskState = static_cast<DRCETaskRequest::TaskState>(resultItem.getState());
        }

        resultDataItem = makeResultDataItem(inputJsonMessage, nodeOptions, message,
            DRCETaskRequest::RequestType::rtSetTaskExecute, taskState,
            errorCode, errorMessage, asyncTask.pid, asyncTask.timeElapsed);
        resultDataItem.setErrorCode(errorCodeMainTask);
        resultDataItem.setTime(asyncTask.timeElapsed);
      }
    }

    DRCECommonTask::waitUpdateAllTasks(taskId, nodeOptions, message, asyncTasksQueue);
    DRCECommonTask::waitEndAllTask(inputJsonMessage, asyncTasksQueue, this);

    if (!getParentTaskId())
    {

      outMsg.str("");
      outMsg << "DRCETaskShellHost::runTask  getSubtaskCleanup(" << taskId << ") = " << std::boolalpha <<  asyncTasksQueue.getSubtaskCleanup(taskId);
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

      std::map<unsigned int, bool> sub = asyncTasksQueue.getSubtasks(taskId);

      for (std::map<unsigned int, bool>::iterator iter=sub.begin();iter!=sub.end();++iter)
      {
        outMsg.str("");
        outMsg << "DRCETaskShellHost::runTask  cleanup()  taskId: " << (*iter).first << "  cleanup: " << std::boolalpha << (*iter).second;
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

        DRCECommonTask::cleanup((*iter).first, nodeOptions, message, (*iter).second);
      }
    }
  }
  catch(Poco::Exception& e)
  {
    resultDataItem.setErrorCode(e.code());
    resultDataItem.setErrorMessage(e.message());
    asyncTask.state = DRCETaskRequest::TaskState::CRASHED;
    asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
    if (asyncTasksQueue.isExistAsyncTask(asyncTask.taskId))
      asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), &resultDataItem, e.message(), e.code());
    resourceMonitor.incrementCountAsyncTasksFail();
  }
  catch(std::exception& e)
  {
    resultDataItem.setErrorCode(ERROR_EXECUTE_TASK);
    resultDataItem.setErrorMessage(e.what());
    asyncTask.state = DRCETaskRequest::TaskState::CRASHED;
    asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
    if (asyncTasksQueue.isExistAsyncTask(asyncTask.taskId))
    asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), &resultDataItem, e.what(), ERROR_COMMAND_EXECUTION);
      resourceMonitor.incrementCountAsyncTasksFail();
  }

  if (asyncTask.state != DRCETaskRequest::TaskState::TERMINATED)
  {
    // write to tasks log information about finished state of task
    asyncTasksQueue.writeToTasksLog(drce_const::DRCE_TASKS_LOGGER_DELIMITER, drce_const::DRCE_TASKS_LOGGER_DELETED, asyncTask);
  }
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCETaskShellHost::execute(Process::CallbackFunction fn) throw (Poco::Exception)
{
  saveRequestData();

  Poco::Timestamp tsStart;
  makeFilesBefore(taskRequestSetExecute);

  DRCEResultDataItem resultDataItem;
  Poco::SharedPtr<CommandResultData> pCommandResultData = nullptr;
  if (!taskRequestSetExecute.getCommandLine().empty())
  {
    Command command(taskRequestSetExecute.getCommandLine(),
        taskRequestSetExecute.getSessionOptions().homeDir,
        taskRequestSetExecute.getInputStream(),
        taskRequestSetExecute.getSessionOptions().environments,
        taskRequestSetExecute.getSessionOptions().shellName);

    CommandExecutor commandExecutor;
    pCommandResultData = commandExecutor.execute(command, fn, taskRequestSetExecute.getSessionOptions().timeMax);

    if (pCommandResultData.isNull())
      throw Poco::Exception(message(message_const::EXECUTE_SHELL_HOST_COMMAND_FAIL));

    resultDataItem.setStdoutStream(pCommandResultData->getOutStream().str());
    resultDataItem.setStderrStream(pCommandResultData->getErrStream().str());
    resultDataItem.setExitStatus(pCommandResultData->getExitStatus());
    resultDataItem.setPid(pCommandResultData->getProcessId());

    if (commandExecutor.getStatusExecution()==StarterExecutor::StatusExecution::seTerminated)
    {
      resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::TERMINATED));
      resultDataItem.setErrorMessage(message(message_const::TASK_TIME_EXPIRED));
      resultDataItem.setErrorCode(ERROR_TERMINATE_EXPIRED_TASK);
      this->cancel();
    }
    else
      resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::FINISHED));
  }
  else
    resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::FINISHED));

  resultDataItem.setRequestId(taskRequestSetExecute.getTaskId());
  resultDataItem.setRequestType(DRCETaskRequest::RequestType::rtSetTaskExecute);
  resultDataItem.setNodeName(nodeOptions.getNodeName());
  resultDataItem.setNodeHost(nodeOptions.getNodeHost());
  resultDataItem.setNodePort(nodeOptions.getNodePort());

  makeFilesAfter(taskRequestSetExecute, resultDataItem);

  if (!pCommandResultData.isNull())
  {
    if (pCommandResultData->getIsError())
      throw Poco::Exception(message(message_const::SHELL_HOST_COMMAND_RETURN_ERROR)+
          message(message_const::ERROR, pCommandResultData->getErrStream().str()), ERROR_COMMAND_EXECUTION);
  }

  // fill spent time value
  resultDataItem.setTime(getElapsedTimeMsec(tsStart));

  return resultDataItem;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE