DRCEAsyncTasksQueue.hpp

Go to the documentation of this file.

#ifndef DRCE_ASYNC_TASKS_QUEUE_HPP
#define DRCE_ASYNC_TASKS_QUEUE_HPP

#include <map>
#include <list>
#include <Poco/SharedPtr.h>
#include <Poco/Observer.h>
#include <Poco/Task.h>
#include <Poco/Thread.h>
#include <Poco/Mutex.h>
#include <Poco/ThreadPool.h>
#include <Poco/TaskManager.h>
#include <Poco/TaskNotification.h>

#include "DRCETaskRequest.hpp"
#include "DRCENodeOptions.hpp"
#include "DRCENotificationExecutor.hpp"
#include "CustomMessage.hpp"
#include "DRCEResourceExtractor.hpp"
#include "DRCEResourceLimits.hpp"

namespace HCE
{
namespace drce
{
class DRCEAsyncTasksQueue;
class DRCEResultDataItem;
class DRCEResultData;
class DRCECommonTask;
class DRCETasksQueue;
//-----------------------------------------------------------------------------
class ProgressHandler
{
public:
  ProgressHandler(DRCEAsyncTasksQueue& asyncTasksQueue_) : asyncTasksQueue(asyncTasksQueue_) {};

  void onStarted(Poco::TaskStartedNotification* pNf);
  void onFinished(Poco::TaskFinishedNotification* pNf);
  void onCancelled(Poco::TaskCancelledNotification* pNf);
private:
  static void log(const std::string& msg);

  DRCEAsyncTasksQueue& asyncTasksQueue;
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
struct AsyncTasks
{
public:
  AsyncTasks(void);
  AsyncTasks(unsigned int taskId_,
      unsigned int parentTaskId_,
      pid_t pid_,
      unsigned int timeMax_,
      size_t timeStart_,
      size_t timeElapsed_,
      DRCETaskRequest::TaskState state_);
  AsyncTasks(const AsyncTasks& rhs);
  AsyncTasks(AsyncTasks&& rhs);

  AsyncTasks& operator=(const AsyncTasks& rhs);
  AsyncTasks& operator=(AsyncTasks&& rhs);

  void clear(void);

  unsigned int taskId;              // task ID
  unsigned int parentTaskId;        // parent task ID
  unsigned int pid;                 // task process ID
  unsigned int timeMax;             // max time allowed for execution
  size_t timeStart;                 // time start of task for execution
  size_t timeElapsed;               // time elapsed for task
  DRCETaskRequest::TaskState state; // task state
  bool locked;                      // flag for locked access to structure on disk
  ResourceUsageLimits usageLimits;  // resource limits during execution of task

  friend std::ostream& operator<<(std::ostream& os, const AsyncTasks& rhs);
  friend std::istream& operator>>(std::istream& is, AsyncTasks& rhs);
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
class DRCEAsyncTasksQueue : Poco::Runnable
{
  enum {DEFAULT_MAX_CAPACITY=16};
  enum {DEFAULT_QUEUE_DUMP_PERIOD=10000};
public:
  typedef std::map<unsigned int, AsyncTasks> Tasks;
  typedef std::map<unsigned int, std::map<unsigned int, bool> > SubTasks;
  typedef AsyncTasks SyncTasks;

  DRCEAsyncTasksQueue(DRCENodeOptions& nodeOptions_,
      DRCENotificationExecutor& notificationExecutor_,
      CustomMessage& message_);
  ~DRCEAsyncTasksQueue(void);

  void cancelAll(void);
  void joinAll(void);
  void startTask(Poco::Task* pTask) throw (Poco::Exception);
  bool hasTask(unsigned int taskId);
  void cancelTask(unsigned int taskId);

  void setMaxThreadCount(unsigned int threadCount);
  unsigned int getMaxThreadCount(void) const;

  void makeCanceledTasks(void);
  void checkExpiredTime(void);
  void checkUsageResources(void);

  AsyncTasks getAsyncTask(unsigned int taskId);
  void setAsyncTask(const AsyncTasks& asyncTask);
  void setAsyncTask(AsyncTasks&& asyncTask);
  size_t getAsyncTasksCount(void);
  bool isExistAsyncTask(unsigned int taskId);

  bool removeAsyncTask(unsigned int taskId);

  void lockAsyncTask(unsigned int taskId);
  void lockAsyncTask(const AsyncTasks& asyncTask);
  void unlockAsyncTask(unsigned int taskId);
  void unlockAsyncTask(const AsyncTasks& asyncTask);
  bool isLockedAsyncTask(unsigned int taskId);

  void run(void);

  void terminate(void);
  bool isTerminated(void);

  void changeState(AsyncTasks&& asyncTask, DRCEResultDataItem* pResultDataItem=nullptr,
        const std::string& errorMessage="", unsigned int errorCode=0) throw (Poco::Exception);

  void safeChangeState(AsyncTasks&& asyncTask, DRCEResultDataItem* pResultDataItem=nullptr,
      const std::string& errorMessage="", unsigned int errorCode=0);

  void setTasksQueueDumpPeriod(unsigned int tasksQueueDumpPeriod_); // period in msec
  unsigned int getTasksQueueDumpPeriod(void);

  void loadTasksQueue(void);
  void saveTasksQueue(void);

  void executeNotification(const std::string& json);
  void executeNotification(const DRCEResultData& resultData, DRCETaskRequest::RequestType requestType);

  void setResetErrorCodeStateNotification(bool resetErrorCodeStateNotification_) {resetErrorCodeStateNotification=resetErrorCodeStateNotification_;}
  bool getResetErrorCodeStateNotification(void) const {return resetErrorCodeStateNotification;}

  DRCEResultData getTasksAsResultData(void);
  void startCleanupTask(unsigned int taskId);

  DRCETasksQueue getCurrentTasksQueue(void);
  std::string getCurrentTasksQueueAsString(void);

  void writeToTasksLog(const std::string& delimiter, const std::string& taskStatus, const AsyncTasks& asyncTask);

  AsyncTasks& getSyncTasks(void) {return syncTask;}

  void addSubtask(unsigned int taskId, unsigned int subtaskId, bool needCleanup);
  void setSubtaskCleanup(unsigned int subtaskId, bool needCleanup);
  bool getSubtaskCleanup(unsigned int subtaskId);
  void removeSubtasks(unsigned int taskId);
  void setAllSubtaskCleanup(unsigned int taskId);
  SubTasks::const_iterator getSubtasksBegin(void);
  SubTasks::const_iterator getSubtasksEnd(void);
  std::map<unsigned int, bool> getSubtasks(unsigned int taskId);
private:
  unsigned int maxThreadCount;
  Poco::ThreadPool threadPool;
  Poco::TaskManager taskManager;
  ProgressHandler progressHandler;
  Poco::Thread thread;
  DRCENodeOptions& nodeOptions;
  DRCENotificationExecutor& notificationExecutor;
  CustomMessage& message;

  Tasks tasks;
  SubTasks subtasks;
  Poco::Mutex mutex;
  volatile bool terminated;
  unsigned int tasksQueueDumpPeriod;

  Poco::TaskManager cleanupTaskManager;
  SyncTasks syncTask; // task data use in sync mode

  bool resetErrorCodeStateNotification;
private:
  void setSyncTaskAsTerminated(const std::string& errorMessage, unsigned int errorCode);
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
class AsyncTaskLocker
{
public:
  AsyncTaskLocker(DRCEAsyncTasksQueue& asyncTasksQueue_, unsigned int taskId_);
  AsyncTaskLocker(DRCEAsyncTasksQueue& asyncTasksQueue_, const AsyncTasks& asyncTask);
  ~AsyncTaskLocker(void);
private:
  DRCEAsyncTasksQueue& asyncTasksQueue;
  unsigned taskId;
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE 

#endif // DRCE_ASYNC_TASKS_QUEUE_HPP