DRCETaskRequestSetExecuteExecutor.cpp

Go to the documentation of this file.

#include <fstream>
#include <sys/wait.h>
#include <iostream>
#include <Poco/AutoPtr.h>
#include <Poco/Logger.h>
#include <Poco/File.h>

#include "DRCEError.hpp"
#include "DRCEMessageConst.hpp"
#include "DRCEResourceLimitsChecker.hpp"
#include "DRCETaskRequestSetExecuteExecutor.hpp"
#include "DRCETaskFactory.hpp"
#include "DRCETaskRequest.hpp"
#include "DRCETaskShellHost.hpp"
#include "DRCETaskSsh.hpp"
#include "DRCEFileBuilder.hpp"
#include "StarterExecutor.hpp"
#include "DRCEFileExtractor.hpp"
#include "DRCEFileStream.hpp"

#include "CommandExecutor.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
DRCETaskRequestSetExecuteExecutor::DRCETaskRequestSetExecuteExecutor(DRCEAsyncTasksQueue& asyncTaskQueue_,
    DRCENodeOptions& nodeOptions_, CustomMessage& message_, DRCEResourceMonitor& resourceMonitor_)
: inherited(asyncTaskQueue_, nodeOptions_, message_, resourceMonitor_)
{
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCETaskRequestSetExecuteExecutor::execute(DRCETaskRequest* pTaskRequest, DRCEInputJsonMessage& inputJsonMessage)
{
  DRCEResultDataItem resultDataItem;

  try
  {
    DRCETaskRequestSetExecute* pTaskReqiestSetExecute = dynamic_cast<DRCETaskRequestSetExecute*>(pTaskRequest);
    if (!pTaskReqiestSetExecute)
      throw Poco::Exception(message(message_const::WRONG_TYPE_TASK_REQUEST), ERROR_BAD_COMMAND_PARAMS);

    if (!pTaskReqiestSetExecute->getInputStream().empty())
      if (pTaskReqiestSetExecute->getCommandLine().empty())
        throw Poco::Exception(message(message_const::EMPTY_COMMAND_LINE), ERROR_BAD_COMMAND_PARAMS);

    if (pTaskReqiestSetExecute->getSessionOptions().homeDir.empty())
    {
      SessionOptions sessionOptions = pTaskReqiestSetExecute->getSessionOptions();
      sessionOptions.homeDir = this->nodeOptions.getHomeDir();
      pTaskReqiestSetExecute->setSessionOptions(std::forward<SessionOptions>(sessionOptions));
    }

    if (!pTaskReqiestSetExecute->getSessionOptions().shellName.empty())
    {
      if (!HCE::StarterExecutor::isExistShellScript(pTaskReqiestSetExecute->getSessionOptions().homeDir, pTaskReqiestSetExecute->getSessionOptions().shellName))
        throw Poco::Exception(message(message_const::SHELL_SCRIPT_NOT_EXIST), ERROR_SHELL_SCRIPT_NOT_EXIST);

      if (!HCE::StarterExecutor::canExecuteShellScript(pTaskReqiestSetExecute->getSessionOptions().homeDir, pTaskReqiestSetExecute->getSessionOptions().shellName))
        throw Poco::Exception(message(message_const::SHELL_SCRIPT_CANNOT_EXECUTE), ERROR_SHELL_SCRIPT_CANNOT_EXECUTE);
    }

    DRCEResourceLimitsChecker resourceLimitsChecker(message);
    resourceLimitsChecker.checkLimits(pTaskReqiestSetExecute->getResourceLimits());

    DRCETaskFactory taskFactory;
    DRCECommonTask* pCommonTask = taskFactory.create(pTaskReqiestSetExecute->getSessionOptions().sessionType,
        std::to_string(pTaskReqiestSetExecute->getTaskId()),
        nodeOptions,
        message,
        (*pTaskReqiestSetExecute),
        asyncTaskQueue,
        resourceMonitor,
        inputJsonMessage,
        pTaskRequest->getParentTaskId());

    resultDataItem = executeTask(pTaskReqiestSetExecute->getSessionOptions().tmode,
        pCommonTask,
        pTaskReqiestSetExecute,
        inputJsonMessage);
  }
  catch(Poco::NoThreadAvailableException& e)
  {
    resultDataItem = getResultDataItem(*pTaskRequest, DRCETaskRequest::TaskState::NOT_SET_AS_NEW);
    resultDataItem.setErrorCode(ERROR_CREATE_NEW_TASK);
    resultDataItem.setErrorMessage(e.displayText());
  }
  catch(Poco::Exception& e)
  {
    resultDataItem = getResultDataItem(*pTaskRequest, DRCETaskRequest::TaskState::NOT_SET_AS_NEW);
    resultDataItem.setErrorCode(e.code());
    resultDataItem.setErrorMessage(e.message());
  }
  catch(std::exception& e)
  {
    resultDataItem = getResultDataItem(*pTaskRequest, DRCETaskRequest::TaskState::NOT_SET_AS_NEW);
    resultDataItem.setErrorCode(ERROR_EXECUTE_TASK);
    resultDataItem.setErrorMessage(e.what());
  }
  return resultDataItem;
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCETaskRequestSetExecuteExecutor::executeTask(SessionOptions::ThreadMode threadMode,
    DRCECommonTask* pCommonTask,
    DRCETaskRequestSetExecute* pTaskReqiestSetExecute,
    DRCEInputJsonMessage& inputJsonMessage) throw (Poco::Exception)
{
  DRCEResultDataItem resultDataItem;

  if (pCommonTask==nullptr)
    throw Poco::Exception(message(message_const::EMPTY_OBJECT_INSTANCE), ERROR_EXECUTE_TASK);

  if (pTaskReqiestSetExecute==nullptr)
    throw Poco::Exception(message(message_const::EMPTY_OBJECT_INSTANCE), ERROR_EXECUTE_TASK);

  pCommonTask->cleanup(pTaskReqiestSetExecute->getTaskId(), nodeOptions, message, true);

  if (pTaskReqiestSetExecute->getSessionOptions().cleanup != SessionOptions::CleanupFlag::cfNotDelete &&
      pTaskReqiestSetExecute->getSessionOptions().cleanup != SessionOptions::CleanupFlag::cfDelete)
    throw Poco::Exception(message(message_const::CLEANUP_BAD_FLAG,
        static_cast<unsigned int>(pTaskReqiestSetExecute->getSessionOptions().cleanup)),
        ERROR_CLEANUP_TASK_DATA);

  if (threadMode != SessionOptions::ThreadMode::tmSync &&
      threadMode != SessionOptions::ThreadMode::tmAsync)
    throw Poco::Exception(message(message_const::WRONG_TYPE_THREAD_MODE, static_cast<unsigned int>(threadMode)), ERROR_EXECUTE_TASK);

  pCommonTask->saveRequest(inputJsonMessage, nodeOptions, message);

  switch(threadMode)
  {
    case SessionOptions::ThreadMode::tmSync:
      {
        // save value lenght of stdin
        resourceMonitor.addSizeInputBufferSyncTasks(pTaskReqiestSetExecute->getInputStream().length());

        // running task in sync mode
        DRCEAsyncTasksQueue::SyncTasks& syncTask = asyncTaskQueue.getSyncTasks();
        Poco::Timestamp now;
        syncTask.timeStart = now.epochTime();
        syncTask.timeMax = pTaskReqiestSetExecute->getSessionOptions().timeMax;
        syncTask.taskId = pTaskReqiestSetExecute->getTaskId();
        syncTask.state = DRCETaskRequest::TaskState::IN_PROGRESS;
        syncTask.usageLimits = pTaskReqiestSetExecute->getResourceLimits().usageLimits;

        DRCECommonTask::ApplyPidHandler applyPidHandler(asyncTaskQueue, syncTask, DRCETaskRequest::TaskState::IN_PROGRESS, SessionOptions::ThreadMode::tmSync);
        resultDataItem = pCommonTask->execute(applyPidHandler);

        if (!pCommonTask->isCancelled())
        {
          Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
              "DRCETaskRequestSetExecuteExecutor::executeTask  before executeSubtasks()  SessionOptions::ThreadMode::tmSync",
              Poco::Message::Priority::PRIO_TRACE));
          // block running subtasks
          pCommonTask->executeSubtasks(resultDataItem);

          if (!pCommonTask->isCancelled())
          {
            // update all subtasks
            for (size_t i=0;i<resultDataItem.getSubtasksCount();++i)
            {
              pCommonTask->waitUpdateAllTasks(resultDataItem.getSubtaskItem(i).getRequestId(), nodeOptions, message, asyncTaskQueue, pCommonTask);
            }
          }
        }

        if (pCommonTask->isCancelled())
        {
          unsigned int errorCode = NO_ERROR;
          unsigned int errorCodeMainTask = NO_ERROR;
          std::string errorMessage;
          DRCETaskRequest::TaskState taskState = DRCETaskRequest::TaskState::TERMINATED;

          DRCEResultDataItem resultItem = DRCECommonTask::extractStatusResultDataItem(pTaskReqiestSetExecute->getTaskId(), nodeOptions, asyncTaskQueue);
          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, syncTask.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 = pCommonTask->makeResultDataItem(inputJsonMessage, nodeOptions, message,
              DRCETaskRequest::RequestType::rtSetTaskExecute, taskState,
              errorCode, errorMessage, syncTask.pid, syncTask.timeElapsed);
          resultDataItem.setErrorCode(errorCodeMainTask);
          pCommonTask->saveResultData(resultDataItem);
        }

        // start task for cleanup tasks data if necessary
        if (!pCommonTask->getParentTaskId())
        {
          Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
              "DRCETaskRequestSetExecuteExecutor::executeTask  asyncTaskQueue.startCleanupTask",
              Poco::Message::Priority::PRIO_TRACE));
          asyncTaskQueue.startCleanupTask(inputJsonMessage.getRequestId());
        }

        Poco::File dataFile(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), syncTask.taskId));
        if (dataFile.exists())
        {
          try
          {
            DataFileExtractor dataFileExtractor(message);
            resultDataItem = dataFileExtractor.extract(dataFile.path());
          }
          catch(Poco::Exception& e)
          {
            std::stringstream outMsg;
            outMsg << "!!!!! DataFileExtractor::extract(" << dataFile.path() << "): " << e.displayText();
            Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));
          }
        }
        // free resources
        pCommonTask->release();
        syncTask.clear();

        // save values lenght of stdout and time execution
        resourceMonitor.addSizeOutputBufferSyncTasks(resultDataItem.getStdoutStream().length());
        resourceMonitor.addTimeSyncTask(resultDataItem.getTime());
        // increment count of sync tasks
        resourceMonitor.incrementCountSyncTasks();
        if (resultDataItem.getErrorCode()!=NO_ERROR)
          resourceMonitor.incrementCountSyncTasksFail();
      }
      break;
    case SessionOptions::ThreadMode::tmAsync:
      {
        // save value lenght of stdin
        resourceMonitor.addSizeInputBufferAsyncTasks(pTaskReqiestSetExecute->getInputStream().length());

        // running task in async mode
        asyncTaskQueue.startTask(pCommonTask);
        resultDataItem = getResultDataItem(inputJsonMessage, DRCETaskRequest::TaskState::SET_AS_NEW);

        // increment count of async tasks
        resourceMonitor.incrementCountAsyncTasks();
      }
      break;
    default:;
  }
  return resultDataItem;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE