DRCEAsyncTasksQueue.cpp

Go to the documentation of this file.

#include <iostream>
#include <fstream>
#include <sstream>
#include <utility>
#include <iterator>
#include <signal.h>
#include <dirent.h>
#include <sys/types.h>
#include <Poco/TaskManager.h>
#include <Poco/Timestamp.h>
#include <Poco/File.h>
#include <Poco/Logger.h>

#include "Process.hpp"
#include "DRCEAsyncTasksQueue.hpp"
#include "DRCEFileStream.hpp"
#include "DRCEOutputJsonMessage.hpp"
#include "DRCEMessageConst.hpp"
#include "DRCEError.hpp"
#include "DRCEResultData.hpp"
#include "DRCEFileBuilder.hpp"
#include "DRCECommonTask.hpp"
#include "DRCEFileExtractor.hpp"
#include "DRCECleanupTask.hpp"
#include "DRCETasksQueue.hpp"
#include "DRCEReadProcessData.hpp"
#include "DRCETaskRequestTerminateExecutor.hpp"
#include "DRCEResourceLimitsChecker.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
void ProgressHandler::onStarted(Poco::TaskStartedNotification* pNf)
{
  if (pNf->task())
    this->log("Started task: "+pNf->task()->name());
  pNf->release();
}

//-----------------------------------------------------------------------------
void ProgressHandler::onCancelled(Poco::TaskCancelledNotification* pNf)
{
  if (pNf->task())
    this->log("Cancelled task: "+pNf->task()->name());
  pNf->release();
}
//-----------------------------------------------------------------------------
void ProgressHandler::onFinished(Poco::TaskFinishedNotification* pNf)
{
  if (pNf->task())
    this->log("Stopped task: "+pNf->task()->name());
  pNf->release();
}
//-----------------------------------------------------------------------------
void ProgressHandler::log(const std::string& msg)
{
  try
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, msg, Poco::Message::Priority::PRIO_INFORMATION));
  }
  catch(Poco::Exception& e)
  {
    std::cerr << "ProgressHandler: " << e.displayText() << std::endl;
  }
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
AsyncTasks::AsyncTasks(void)
: taskId(0), parentTaskId(0), pid(0), timeMax(0), timeStart(0), timeElapsed(0),
  state(DRCETaskRequest::TaskState::FINISHED), locked(false), usageLimits()
{
}
//-----------------------------------------------------------------------------
AsyncTasks::AsyncTasks(unsigned int taskId_, unsigned int parentTaskId_, pid_t pid_, unsigned int timeMax_, size_t timeStart_,
    size_t timeElapsed_, DRCETaskRequest::TaskState state_)
: taskId(taskId_), parentTaskId(parentTaskId_), pid(pid_), timeMax(timeMax_), timeStart(timeStart_), timeElapsed(timeElapsed_),
  state(state_), locked(false), usageLimits()
{
}
//-----------------------------------------------------------------------------
AsyncTasks::AsyncTasks(const AsyncTasks& rhs)
: taskId(0), parentTaskId(0), pid(0), timeMax(0), timeStart(0), timeElapsed(0),
  state(DRCETaskRequest::TaskState::FINISHED), locked(false), usageLimits()
{
  (*this) = rhs;
}
//-----------------------------------------------------------------------------
AsyncTasks::AsyncTasks(AsyncTasks&& rhs)
: taskId(0), parentTaskId(0), pid(0), timeMax(0), timeStart(0), timeElapsed(0),
  state(DRCETaskRequest::TaskState::FINISHED), locked(false), usageLimits()
{
  (*this) = std::forward<AsyncTasks>(rhs);
}
//-----------------------------------------------------------------------------
AsyncTasks& AsyncTasks::operator=(const AsyncTasks& rhs)
{
  if (this!=&rhs)
  {
    taskId = rhs.taskId;
    parentTaskId = rhs.parentTaskId;
    pid = rhs.pid;
    timeMax = rhs.timeMax;
    timeStart = rhs.timeStart;
    timeElapsed = rhs.timeElapsed;
    state = rhs.state;
    locked = rhs.locked;
    usageLimits = rhs.usageLimits;
  }
  return *this;
}
//-----------------------------------------------------------------------------
AsyncTasks& AsyncTasks::operator=(AsyncTasks&& rhs)
{
  if (this!=&rhs)
  {
    taskId = std::move(rhs.taskId);
    parentTaskId = std::move(rhs.parentTaskId);
    pid = std::move(rhs.pid);
    timeMax = std::move(rhs.timeMax);
    timeStart = std::move(rhs.timeStart);
    timeElapsed = std::move(rhs.timeElapsed);
    state = std::move(rhs.state);
    locked = std::move(rhs.locked);
    usageLimits = std::move(rhs.usageLimits);
  }
  return *this;
}
//-----------------------------------------------------------------------------
void AsyncTasks::clear(void)
{
  taskId = 0;
  parentTaskId = 0;
  pid = 0;
  timeMax = 0;
  timeStart = 0;
  timeElapsed = 0;
  state = DRCETaskRequest::TaskState::UNDEFINED;
  locked = false;
  usageLimits.clear();
}
//-----------------------------------------------------------------------------
std::ostream& operator<<(std::ostream& os, const AsyncTasks& rhs)
{
  os << rhs.taskId << " " << rhs.parentTaskId << " " << rhs.pid << " " << rhs.timeMax << " " << rhs.timeStart << " " << rhs.timeElapsed << " "
      << static_cast<unsigned int>(rhs.state);
  return os;
}
//-----------------------------------------------------------------------------
std::istream& operator>>(std::istream& is, AsyncTasks& rhs)
{
  is >> rhs.taskId;
  is >> rhs.parentTaskId;
  is >> rhs.pid;
  is >> rhs.timeMax;
  is >> rhs.timeStart;
  is >> rhs.timeElapsed;
  unsigned int state = 0;
  is >> state;
  rhs.state = static_cast<DRCETaskRequest::TaskState>(state);

  return is;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
DRCEAsyncTasksQueue::DRCEAsyncTasksQueue(DRCENodeOptions& nodeOptions_,
    DRCENotificationExecutor& notificationExecutor_,
    CustomMessage& message_)
: maxThreadCount(DEFAULT_MAX_CAPACITY), threadPool(), taskManager(threadPool), progressHandler(*this), thread(),
  nodeOptions(nodeOptions_), notificationExecutor(notificationExecutor_), message(message_),
  tasks(), subtasks(), mutex(), terminated(false), tasksQueueDumpPeriod(DEFAULT_QUEUE_DUMP_PERIOD), cleanupTaskManager(), syncTask(), resetErrorCodeStateNotification(false)
{
  taskManager.addObserver(Poco::Observer<ProgressHandler, Poco::TaskStartedNotification>(progressHandler, &ProgressHandler::onStarted));
  taskManager.addObserver(Poco::Observer<ProgressHandler, Poco::TaskFinishedNotification>(progressHandler, &ProgressHandler::onFinished));
  taskManager.addObserver(Poco::Observer<ProgressHandler, Poco::TaskCancelledNotification>(progressHandler, &ProgressHandler::onCancelled));
  loadTasksQueue();
  thread.start(*this);
}
//-----------------------------------------------------------------------------
DRCEAsyncTasksQueue::~DRCEAsyncTasksQueue(void)
{
  cancelAll();
  joinAll();

  terminate();    // terminate 'watch dogs' thread
  thread.join();
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::cancelAll(void)
{
  taskManager.cancelAll();
  makeCanceledTasks();
  cleanupTaskManager.cancelAll();
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::joinAll(void)
{
  taskManager.joinAll();
  cleanupTaskManager.joinAll();
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::startTask(Poco::Task* pTask) throw (Poco::Exception)
{
  if (pTask)
  {
    unsigned int taskId = 0;
    try
    {
      taskId = std::stoul(pTask->name());
    }
    catch(std::exception& e)
    {
      throw Poco::Exception(message(message_const::BAD_TASK_ID, e.what()), ERROR_CREATE_NEW_TASK);
    }

    if (isExistAsyncTask(taskId))
      throw Poco::Exception(message(message_const::TASK_CONTINUES_TO_RUN), ERROR_CREATE_NEW_TASK);

    taskManager.start(pTask);
  }
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
bool DRCEAsyncTasksQueue::hasTask(unsigned int taskId)
{
  bool result = false;
  try
  {
    const std::string taskName = std::to_string(taskId);

    Poco::TaskManager::TaskList taskList = taskManager.taskList();
    for (Poco::TaskManager::TaskList::iterator iter=taskList.begin();iter!=taskList.end();++iter)
    {
      if ((*iter)->name() == taskName)
      {
        result = true;
        break;
      }
    }
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::SEARCH_TASK, e.what()), Poco::Message::Priority::PRIO_ERROR));
  }
  return result;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::cancelTask(unsigned int taskId)
{
  try
  {
    const std::string taskName = std::to_string(taskId);
    Poco::TaskManager::TaskList taskList = taskManager.taskList();
    for (Poco::TaskManager::TaskList::iterator iter=taskList.begin();iter!=taskList.end();++iter)
    {
      if ((*iter)->name() == taskName)
      {
        (*iter)->cancel();
        break;
      }
    }
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::CANCEL_TASK, e.what()), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setMaxThreadCount(unsigned int threadCount)
{
  if (threadCount < maxThreadCount)
    threadPool.addCapacity(maxThreadCount - threadPool.capacity());
  else
    threadPool.addCapacity(static_cast<int>(threadCount) - threadPool.capacity());
}
//-----------------------------------------------------------------------------
unsigned int DRCEAsyncTasksQueue::getMaxThreadCount(void) const
{
  return static_cast<unsigned int>(threadPool.capacity());
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::changeState(AsyncTasks&& asyncTask, DRCEResultDataItem* pResultDataItem,
      const std::string& errorMessage, unsigned int errorCode) throw (Poco::Exception)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCEAsyncTasksQueue::changeState enter", Poco::Message::Priority::PRIO_TRACE));

  if (!asyncTask.timeStart)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
        "!!! DRCEAsyncTasksQueue::changeState recieved empty 'asyncTask.timeStart' !!!",
        Poco::Message::Priority::PRIO_TRACE));
    return;
  }

  AsyncTaskLocker(*this, asyncTask);

  DRCEResultData resultData;
  DRCEResultDataItem resultDataItem;
  std::stringstream outMsg;

  Poco::File dataFile(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), asyncTask.taskId));
  if (dataFile.exists())
  {
    try
    {
      DataFileExtractor dataFileExtractor(message);
      resultDataItem = dataFileExtractor.extract(dataFile.path());
    }
    catch(Poco::Exception& e)
    {
      outMsg.str("");
      outMsg << "!!!!! DataFileExtractor::extract(" << dataFile.path() << "): " << e.displayText();
      outMsg << "  SET STATE TO IN_PROGRESS";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));

      resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::IN_PROGRESS));
    }
  }

  if (pResultDataItem)
  {
    resultDataItem = *pResultDataItem;
  }

  outMsg.str("");
  outMsg << "DRCEAsyncTasksQueue::changeState  taskId: " << asyncTask.taskId
      << " asyncTasks.state: " << (int)asyncTask.state
      << " resultDataItem.state: " << resultDataItem.getState()
      << " asyncTask.parentTaskId: " << asyncTask.parentTaskId;
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

  resultDataItem.setNodeName(nodeOptions.getNodeName());
  resultDataItem.setNodeHost(nodeOptions.getNodeHost());
  resultDataItem.setNodePort(nodeOptions.getNodePort());

  resultDataItem.setRequestId(asyncTask.taskId);
  resultDataItem.setPid(asyncTask.pid);
  resultDataItem.setState(static_cast<unsigned int>(asyncTask.state));
  resultDataItem.setTime(asyncTask.timeElapsed);

  if (!errorMessage.empty())
    resultDataItem.setErrorMessage(errorMessage);
  if (errorCode!=NO_ERROR)
    resultDataItem.setErrorCode(errorCode);

  resultData.addDataItem(std::forward<DRCEResultDataItem>(resultDataItem));

  // execute notification only for `main` task and only if tasks has been changed state
  if (!asyncTask.parentTaskId)
  {
    AsyncTasks prevStateAsyncTask = getAsyncTask(asyncTask.taskId);
    if (prevStateAsyncTask.state != asyncTask.state || prevStateAsyncTask.pid != asyncTask.pid)
    {
      // change information about state and other property async task
      setAsyncTask(asyncTask);
      // execute notification use new state
      executeNotification(resultData, DRCETaskRequest::RequestType::rtTaskStateNotification);
    }
  }

  if ((asyncTask.state == DRCETaskRequest::TaskState::FINISHED) ||
      (asyncTask.state == DRCETaskRequest::TaskState::TERMINATED) ||
      (asyncTask.state == DRCETaskRequest::TaskState::CRASHED) ||
      (asyncTask.state == DRCETaskRequest::TaskState::DELETED) ||
      (asyncTask.state == DRCETaskRequest::TaskState::UNDEFINED) ||
      (asyncTask.state == DRCETaskRequest::TaskState::TERMINATED_BY_TTL))
  {
    bool ret = removeAsyncTask(asyncTask.taskId);
    outMsg.str("");
    outMsg << "TASK (ID: " << asyncTask.taskId << ", PID: " << asyncTask.pid << ", TIME: " << asyncTask.timeElapsed
        << ", STATE: " << static_cast<unsigned int>(asyncTask.state) <<  ") WAS " << ((ret)?"SUCCESS":"FAILED") << " ERASED FROM QUEUE !!!";
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));


    // write to tasks log information about finished state of task
  }

  if (asyncTask.state != DRCETaskRequest::TaskState::DELETED)
  {
    // rewrite content of `data` and `status` files
    DataFileBuilder dataFileBuilder(message, nodeOptions.getTasksDataDir());
    dataFileBuilder.build(resultData);

    StatusFileBuilder statusFileBuilder(message, nodeOptions.getTasksStatusDir());
    statusFileBuilder.build(resultData);
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::safeChangeState(AsyncTasks&& asyncTask, DRCEResultDataItem* pResultDataItem,
    const std::string& errorMessage, unsigned int errorCode)
{
  try
  {
    changeState(std::forward<AsyncTasks>(asyncTask), pResultDataItem, errorMessage, errorCode);
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::CHANGE_STATE, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::CHANGE_STATE, e.what()), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
AsyncTasks DRCEAsyncTasksQueue::getAsyncTask(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  AsyncTasks asyncTask;
  Tasks::iterator iter = tasks.find(taskId);
  if (iter != tasks.end())
  {
    asyncTask = (*iter).second;
  }
  return asyncTask;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setAsyncTask(const AsyncTasks& asyncTask)
{
  Poco::Mutex::ScopedLock lock(mutex);
  tasks[asyncTask.taskId] = asyncTask;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setAsyncTask(AsyncTasks&& asyncTask)
{
  Poco::Mutex::ScopedLock lock(mutex);
  tasks[asyncTask.taskId] = std::forward<AsyncTasks>(asyncTask);
}
//-----------------------------------------------------------------------------
size_t DRCEAsyncTasksQueue::getAsyncTasksCount(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return tasks.size();
}
//-----------------------------------------------------------------------------
bool DRCEAsyncTasksQueue::removeAsyncTask(unsigned int taskId)
{
  bool result = false;
  Poco::Mutex::ScopedLock lock(mutex);
  Tasks::iterator iter = tasks.find(taskId);
  if (iter != tasks.end())
  {
    tasks.erase(iter);
    result = true;
  }
  return result;
}
//-----------------------------------------------------------------------------
bool DRCEAsyncTasksQueue::isExistAsyncTask(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return (tasks.find(taskId)!=tasks.end());
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::lockAsyncTask(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  Tasks::iterator iter = tasks.find(taskId);
  if (iter != tasks.end())
  {
    (*iter).second.locked = true;
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::lockAsyncTask(const AsyncTasks& asyncTask)
{
  lockAsyncTask(asyncTask.taskId);
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::unlockAsyncTask(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  Tasks::iterator iter = tasks.find(taskId);
  if (iter != tasks.end())
  {
    (*iter).second.locked = false;
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::unlockAsyncTask(const AsyncTasks& asyncTask)
{
  unlockAsyncTask(asyncTask.taskId);
}
//-----------------------------------------------------------------------------
bool DRCEAsyncTasksQueue::isLockedAsyncTask(unsigned int taskId)
{
  bool result = false;
  Poco::Mutex::ScopedLock lock(mutex);
  Tasks::iterator iter = tasks.find(taskId);
  if (iter != tasks.end())
  {
    result = (*iter).second.locked;
  }
  return result;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::terminate(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  terminated = true;
}
//-----------------------------------------------------------------------------
bool DRCEAsyncTasksQueue::isTerminated(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return terminated;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setTasksQueueDumpPeriod(unsigned int tasksQueueDumpPeriod_)
{
  Poco::Mutex::ScopedLock lock(mutex);
  tasksQueueDumpPeriod = tasksQueueDumpPeriod_;
}
//-----------------------------------------------------------------------------
unsigned int DRCEAsyncTasksQueue::getTasksQueueDumpPeriod(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return tasksQueueDumpPeriod;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::loadTasksQueue(void)
{
  Poco::Mutex::ScopedLock lock(mutex);

  const std::string dumpFileName = FileStream::getDumpFileName(nodeOptions.getTasksStatusDir(), nodeOptions.getNodeName());
  FileStream fileStream(dumpFileName, FileStream::read);
  if (fileStream.isOpen())
  {
    std::string fileContent;
    fileStream >> fileContent;
    std::istringstream istr(fileContent);
    std::string line;
    while(std::getline(istr, line))
    {
      AsyncTasks asyncTask;
      std::stringstream str(line);
      str >> asyncTask;

      std::stringstream outMsg;
      outMsg << ">>> loadTasksQueue\ttaskId = " << asyncTask.taskId << " parentTaskId = " << asyncTask.parentTaskId << " pid = " << asyncTask.pid
          << " timeMax = " << asyncTask.timeMax << " timeStart = " << asyncTask.timeStart << " state = " << (unsigned int)(asyncTask.state);
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));

      unsigned int taskId = asyncTask.taskId;
      if (HCE::Process::isExistProcess(asyncTask.pid))
      {
        tasks[taskId] = std::forward<AsyncTasks>(asyncTask); // add task to queue
      }
      else
      {
        asyncTask.state = DRCETaskRequest::TaskState::UNDEFINED;
        tasks.insert(Tasks::value_type(asyncTask.taskId, asyncTask));
      }
    }
    fileStream.close();
    try
    {
      Poco::File dumpFile(dumpFileName);
      if (dumpFile.exists())
        dumpFile.remove();
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, dumpFileName+", "+message(message_const::ERROR, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
    }
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::saveTasksQueue(void)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCEAsyncTasksQueue::saveTasksQueue enter", Poco::Message::Priority::PRIO_TRACE));

  Poco::Mutex::ScopedLock lock(mutex);

  FileStream fileStream(FileStream::getDumpFileName(nodeOptions.getTasksStatusDir(), nodeOptions.getNodeName()), FileStream::write);

  for (Tasks::iterator iter=tasks.begin();iter!=tasks.end();++iter)
  {
    std::stringstream ostr;
    ostr << ((*iter).second) << std::endl;
    fileStream << ostr.str();
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::executeNotification(const std::string& json)
{
  Poco::Mutex::ScopedLock lock(mutex);
  notificationExecutor.execute(json);
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::executeNotification(const DRCEResultData& resultData, DRCETaskRequest::RequestType requestType)
{
  try
  {
    DRCEResultData localResultData(resultData);
    const size_t itemCount = resultData.getItemsCount();

    if (resetErrorCodeStateNotification)
    {
      for (size_t i=0;i<itemCount;++i)
      {
        DRCEResultDataItem resultDataItem = localResultData.getDataItem(i);
        if (resultDataItem.getErrorCode() != ERROR_TERMINATE_EXPIRED_TASK)
        {
          resultDataItem.setErrorCode(NO_ERROR);
          resultDataItem.setErrorMessage("");
          localResultData.setDataItem(i, std::forward<DRCEResultDataItem>(resultDataItem));
        }
      }
    }
    for (size_t i=0;i<itemCount;++i)
    {
      DRCEResultDataItem resultDataItem = localResultData.getDataItem(i);
      resultDataItem.setRequestType(requestType);
      localResultData.setDataItem(i, std::forward<DRCEResultDataItem>(resultDataItem));
    }

    if (itemCount)
    {
      if (requestType == DRCETaskRequest::RequestType::rtTaskStateNotification)
        executeNotification(StatusFileBuilder::getStatusJson(localResultData));
      else
        executeNotification(DataFileBuilder::makeJson(localResultData));
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR_EXECUTE_NOTIFICATION, e.message()), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR_EXECUTE_NOTIFICATION, e.what()), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
DRCEResultData DRCEAsyncTasksQueue::getTasksAsResultData(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  DRCEResultData resultData;

  for (Tasks::iterator iter=tasks.begin();iter!=tasks.end();++iter)
  {
    DataFileExtractor dataFileExtractor(message);
    Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), (*iter).second.taskId));
    if (statusFile.exists())
    {
      try
      {
        DRCEResultDataItem resultDataItem = dataFileExtractor.extract(statusFile.path());
        AsyncTasks asyncTask = getAsyncTask((*iter).second.taskId);
        asyncTask.state = static_cast<HCE::drce::DRCETaskRequest::TaskState>(resultDataItem.getState());
        asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
        setAsyncTask(std::forward<AsyncTasks>(asyncTask));

        if (asyncTask.state!=HCE::drce::DRCETaskRequest::TaskState::FINISHED)
          resultDataItem.setTime(asyncTask.timeElapsed);

        resultData.addDataItem(std::forward<DRCEResultDataItem>(resultDataItem));
      }
      catch(Poco::Exception& e)
      {
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.message(), Poco::Message::Priority::PRIO_DEBUG));
      }
    }
  }
  return resultData;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::startCleanupTask(unsigned int taskId)
{
  try
  {
    cleanupTaskManager.start(new DRCECleanupTask(taskId, nodeOptions, message, *this));
  }
  catch(Poco::NoThreadAvailableException& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::START_CLEANUP_TASK_ERROR, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::START_CLEANUP_TASK_ERROR, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
DRCETasksQueue DRCEAsyncTasksQueue::getCurrentTasksQueue(void)
{
  DRCETasksQueue taskQueue;
  Poco::Mutex::ScopedLock lock(mutex);
  for (Tasks::iterator iter=tasks.begin();iter!=tasks.end();++iter)
  {
    taskQueue.addItem(std::forward<TaskStatusData>(TaskStatusData((*iter).second.taskId,
        (*iter).second.timeStart,
        (*iter).second.state,
        SessionOptions::ThreadMode::tmAsync)));
  }
  return taskQueue;
}
//-----------------------------------------------------------------------------
std::string DRCEAsyncTasksQueue::getCurrentTasksQueueAsString(void)
{
  std::string resultJson;
  DRCETasksQueue taskQueue = std::forward<DRCETasksQueue>(getCurrentTasksQueue());
  if (!taskQueue.serialize(resultJson))
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
        message(message_const::GET_CURRENT_TASKS_QUEUE_ERROR, taskQueue.getErrorMsg()),
        Poco::Message::Priority::PRIO_ERROR));
  }
  return resultJson;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::writeToTasksLog(const std::string& delimiter, const std::string& taskStatus, const AsyncTasks& asyncTask)
{
  std::stringstream outMsg;
  outMsg << taskStatus << delimiter
      << asyncTask.taskId << delimiter
      << asyncTask.timeStart << delimiter
      << static_cast<unsigned int>(asyncTask.state) << delimiter
      << static_cast<unsigned int>(SessionOptions::ThreadMode::tmAsync);
  Poco::Logger::get(drce_const::DRCE_TASKS_LOGGER).log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_INFORMATION));
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setSyncTaskAsTerminated(const std::string& errorMessage, unsigned int errorCode)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCEAsyncTasksQueue::setSyncTaskAsTerminated enter", Poco::Message::Priority::PRIO_TRACE));

  if (!syncTask.timeStart)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
        "!!! DRCEAsyncTasksQueue::setSyncTaskAsTerminated recieved empty 'syncTask.timeStart' !!!",
        Poco::Message::Priority::PRIO_TRACE));
    return;
  }

  syncTask.state = DRCETaskRequest::TaskState::TERMINATED_BY_TTL;

  DRCEResultData resultData;
  DRCEResultDataItem resultDataItem;

  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_TRACE));
    }

    std::stringstream outMsg;
    outMsg << ">>>>> DataFileExtractor::extract Size of " << syncTask.taskId << ".data: " << dataFile.getSize();
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));
  }

  resultDataItem.setNodeName(nodeOptions.getNodeName());
  resultDataItem.setNodeHost(nodeOptions.getNodeHost());
  resultDataItem.setNodePort(nodeOptions.getNodePort());

  resultDataItem.setRequestId(syncTask.taskId);
  resultDataItem.setPid(syncTask.pid);
  resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::TERMINATED_BY_TTL));
  resultDataItem.setTime(Poco::Timestamp::fromEpochTime(syncTask.timeStart).elapsed()/1000);
  resultDataItem.setErrorMessage(errorMessage);
  resultDataItem.setErrorCode(errorCode);

  resultData.addDataItem(std::forward<DRCEResultDataItem>(resultDataItem));

  try
  {
    // rewrite content of `data` and `status` files
    DataFileBuilder dataFileBuilder(message, nodeOptions.getTasksDataDir());
    dataFileBuilder.build(resultData);

    StatusFileBuilder statusFileBuilder(message, nodeOptions.getTasksStatusDir());
    statusFileBuilder.build(resultData);
  }
  catch(Poco::Exception& e)
  {
    std::stringstream outMsg;
    outMsg << "!!!!!FileBuilder::build: " << e.displayText();
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    std::stringstream outMsg;
    outMsg << "!!!!!FileBuilder::build: " << e.what();
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::checkExpiredTime(void)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCEAsyncTasksQueue::checkExpiredTime enter", Poco::Message::Priority::PRIO_TRACE));

  Poco::Mutex::ScopedLock lock(mutex);

  for (Tasks::iterator iter=tasks.begin();iter!=tasks.end();)
  {
    std::stringstream outMsg;
    if (!(*iter).second.timeStart || !(*iter).second.timeMax || !(*iter).second.taskId)
    {
      outMsg << "taskId: " << (*iter).second.taskId
          << " pid: " << (*iter).second.pid
          << " timeMax: " << (*iter).second.timeMax
          << " timeStart: " << (*iter).second.timeStart
          << " state: " << (*iter).second.state << " WAS FORCE ERASE";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));
      iter = tasks.erase(iter);
    }
    else
      ++iter;
  }

  Tasks taskList = tasks;
  for (Tasks::iterator iter=taskList.begin();iter!=taskList.end();++iter)
  {
    std::stringstream outMsg;
    outMsg << "taskId: " << (*iter).second.taskId
        << " pid: " << (*iter).second.pid
        << " timeMax: " << (*iter).second.timeMax
        << " timeStart: " << (*iter).second.timeStart
        << " state: " << (*iter).second.state;

    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

    Poco::Timestamp::TimeDiff diff = Poco::Timestamp::fromEpochTime((*iter).second.timeStart).elapsed();
    (*iter).second.timeElapsed = static_cast<size_t>(diff/1000);

    outMsg.str("");
    outMsg << "diff = " << (diff/1000);
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

    if ((*iter).second.timeMax &&
        ((*iter).second.timeMax < (diff/1000)))
    {
      AsyncTasks asyncTask((*iter).second);
      DRCETaskRequestTerminateExecutor::terminateExpiredTask((*iter).second, *this, nodeOptions, message);

      std::stringstream outMsg;
      outMsg << "TASK (ID: " << asyncTask.taskId << ", PID: " << asyncTask.pid << ", TIME: " << asyncTask.timeElapsed << ") WAS TERMINATED AS EXPIRED !!!";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_INFORMATION));

      outMsg.str("");
      outMsg << "DRCEAsyncTasksQueue::checkExpiredTime  getSubtaskCleanup(" << asyncTask.taskId << ") = " << std::boolalpha <<  getSubtaskCleanup(asyncTask.taskId);
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));
    }
  }

  if (syncTask.timeStart && syncTask.taskId)
  {
    std::stringstream outMsg;
    Poco::Timestamp::TimeDiff diff = Poco::Timestamp::fromEpochTime(syncTask.timeStart).elapsed();

    outMsg << "taskId: " << syncTask.taskId
        << " pid: " << syncTask.pid
        << " timeMax: " << syncTask.timeMax
        << " timeStart: " << syncTask.timeStart
        << " state: " << syncTask.state;
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

    outMsg.str("");
    outMsg << "diff = " << (diff/1000);
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

    if (syncTask.timeMax && (syncTask.timeMax < (diff/1000)))
    {
      if (syncTask.pid)
      {
        // make sync task to expired state
        setSyncTaskAsTerminated(message(message_const::TASK_TIME_EXPIRED, syncTask.timeMax) , ERROR_TERMINATE_EXPIRED_TASK);
        // kill all processes of sync task
        std::vector<pid_t> descendants = DRCEReadProcessData::getAllProcessIds(syncTask.pid);
        for (size_t i=0;i<descendants.size();++i)
          kill(descendants[i], SIGKILL);
        // log termination
        outMsg.str("");
        outMsg << "TASK (ID: " << syncTask.taskId << ", PID: " << syncTask.pid << ", TIME: " << (unsigned int)(diff/1000) << ") WAS TERMINATED AS EXPIRED !!!";
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_INFORMATION));
      }
    }
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::checkUsageResources(void)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCEAsyncTasksQueue::checkUsageResources enter", Poco::Message::Priority::PRIO_TRACE));

  Poco::Mutex::ScopedLock lock(mutex);

  DRCEResourceLimitsChecker resourceLimitsChecker(message);

  Tasks taskList = tasks;
  for (Tasks::iterator iter=taskList.begin();iter!=taskList.end();++iter)
  {
    try
    {
      resourceLimitsChecker.checkLimits((*iter).second, (*iter).second.usageLimits);
    }
    catch(Poco::Exception& e)
    {
      std::stringstream outMsg;
      outMsg << "TASK (ID: " << (*iter).second.taskId << ", PID: " << (*iter).second.pid << ", TIME: " << (*iter).second.timeElapsed << ") WAS TERMINATED AS EXCEEDED USAGE RESOURCES !!!";
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_INFORMATION));

      DRCETaskRequestTerminateExecutor::executeTerminate((*iter).second.taskId, *this, nodeOptions, message, e.message(), e.code());
    }
  }

  if (syncTask.timeStart && syncTask.taskId)
  {
    try
    {
      resourceLimitsChecker.checkLimits(syncTask, syncTask.usageLimits);
    }
    catch(Poco::Exception& e)
    {
      if (syncTask.pid)
      {
        Poco::Timestamp::TimeDiff diff = Poco::Timestamp::fromEpochTime(syncTask.timeStart).elapsed();
        // log termination
        std::stringstream outMsg;
        outMsg << "TASK (ID: " << syncTask.taskId << ", PID: " << syncTask.pid << ", TIME: " << (unsigned int)(diff/1000) << ") WAS TERMINATED AS EXCEEDED USAGE RESOURCES !!!";
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_INFORMATION));

        // make sync task to expired state
        setSyncTaskAsTerminated(e.message(), e.code());
        // kill all processes of sync task
        std::vector<pid_t> descendants = DRCEReadProcessData::getAllProcessIds(syncTask.pid);
        for (size_t i=0;i<descendants.size();++i)
          kill(descendants[i], SIGKILL);
      }
    }
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::makeCanceledTasks(void)
{
  Poco::Mutex::ScopedLock lock(mutex);

  for (Tasks::iterator iter=tasks.begin();iter!=tasks.end(); ++iter)
  {
    if ((*iter).second.state == DRCETaskRequest::TaskState::IN_PROGRESS ||
        (*iter).second.state == DRCETaskRequest::TaskState::SET_AS_NEW ||
        (*iter).second.state == DRCETaskRequest::TaskState::QUEUED_TO_RUN)
    {
      AsyncTasks asyncTask((*iter).second);
      asyncTask.state = DRCETaskRequest::TaskState::TERMINATED;
      asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
      setAsyncTask(std::forward<AsyncTasks>(asyncTask));
    }
  }
  saveTasksQueue();

  if (syncTask.pid)
    kill(syncTask.pid, SIGKILL);
  syncTask.clear();
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCEAsyncTasksQueue::run(void)
{
  const unsigned int periodSleep = 1000; // msec

  while(!isTerminated())
  {
    unsigned int periodMs = getTasksQueueDumpPeriod();
    if (periodMs > periodSleep)
    {
      unsigned int repeat = periodMs/periodSleep; // msec to sec
      for (unsigned int i=0;i<repeat;++i)
      {
        Poco::Thread::sleep(periodSleep);
        if (isTerminated())
          return;
      }
    }
    else
    {
      Poco::Thread::sleep(periodMs);
    }
    try
    {
      checkExpiredTime();
      checkUsageResources();

      if (isTerminated())
        return;

      saveTasksQueue();
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
    }
    catch(std::exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR, e.what()), Poco::Message::Priority::PRIO_ERROR));
    }
  }
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::addSubtask(unsigned int taskId, unsigned int subtaskId, bool needCleanup)
{
  Poco::Mutex::ScopedLock lock(mutex);
  subtasks[taskId][subtaskId] = needCleanup;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setSubtaskCleanup(unsigned int subtaskId, bool needCleanup)
{
  Poco::Mutex::ScopedLock lock(mutex);
  for (SubTasks::iterator iter=subtasks.begin();iter!=subtasks.end();++iter)
  {
    std::map<unsigned int, bool>::iterator it = (*iter).second.find(subtaskId);
    if (it!=(*iter).second.end())
    {
      (*it).second = needCleanup;
      break;
    }
  }
}
//-----------------------------------------------------------------------------
bool DRCEAsyncTasksQueue::getSubtaskCleanup(unsigned int subtaskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  bool needCleanup = false;
  for (SubTasks::iterator iter=subtasks.begin();iter!=subtasks.end();++iter)
  {
    std::map<unsigned int, bool>::iterator it = (*iter).second.find(subtaskId);
    if (it!=(*iter).second.end())
    {
      needCleanup = (*it).second;
      break;
    }
  }
  return needCleanup;
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::removeSubtasks(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  subtasks.erase(taskId);
}
//-----------------------------------------------------------------------------
void DRCEAsyncTasksQueue::setAllSubtaskCleanup(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  SubTasks::iterator iter = subtasks.find(taskId);
  if (iter!=subtasks.end())
  {
    for (std::map<unsigned int, bool>::iterator iT=(*iter).second.begin();iT!=(*iter).second.end();++iT)
    {
      (*iT).second = true;
    }
  }
}
//-----------------------------------------------------------------------------
DRCEAsyncTasksQueue::SubTasks::const_iterator DRCEAsyncTasksQueue::getSubtasksBegin(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return subtasks.begin();
}
//-----------------------------------------------------------------------------
DRCEAsyncTasksQueue::SubTasks::const_iterator DRCEAsyncTasksQueue::getSubtasksEnd(void)
{
  Poco::Mutex::ScopedLock lock(mutex);
  return subtasks.end();
}
//-----------------------------------------------------------------------------
std::map<unsigned int, bool> DRCEAsyncTasksQueue::getSubtasks(unsigned int taskId)
{
  Poco::Mutex::ScopedLock lock(mutex);
  std::map<unsigned int, bool> sub;
  SubTasks::iterator iter = subtasks.find(taskId);
  if (iter!=subtasks.end())
  {
    sub = (*iter).second;
  }
  return sub;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
AsyncTaskLocker::AsyncTaskLocker(DRCEAsyncTasksQueue& asyncTasksQueue_, unsigned int taskId_)
: asyncTasksQueue(asyncTasksQueue_), taskId(taskId_)
{
  asyncTasksQueue.lockAsyncTask(taskId);
}
//-----------------------------------------------------------------------------
AsyncTaskLocker::AsyncTaskLocker(DRCEAsyncTasksQueue& asyncTasksQueue_, const AsyncTasks& asyncTask)
: asyncTasksQueue(asyncTasksQueue_), taskId(asyncTask.taskId)
{
  asyncTasksQueue.lockAsyncTask(asyncTask);
}
//-----------------------------------------------------------------------------
AsyncTaskLocker::~AsyncTaskLocker(void)
{
  asyncTasksQueue.unlockAsyncTask(taskId);
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE