hce-node application  1.4.3
HCE Hierarchical Cluster Engine node application
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros
DRCECommonTask.hpp
Go to the documentation of this file.
1 
14 #ifndef DRCE_COMMON_TASK_HPP
15 #define DRCE_COMMON_TASK_HPP
16 
17 #include <Poco/Task.h>
18 #include <Poco/Timestamp.h>
19 #include <vector>
20 
21 #include "DRCENodeOptions.hpp"
22 #include "CustomMessage.hpp"
23 #include "CommandExecutor.hpp"
24 #include "DRCEFilesList.hpp"
25 #include "DRCEResultData.hpp"
26 #include "DRCETaskRequestSetExecute.hpp"
27 #include "DRCEAsyncTasksQueue.hpp"
28 #include "DRCEResourceMonitor.hpp"
29 #include "Process.hpp"
30 #include "DRCEError.hpp"
31 #include "DRCEInputJsonMessage.hpp"
32 
33 namespace HCE
34 {
35 namespace drce
36 {
37 class DRCEAsyncTasksQueue;
38 //-----------------------------------------------------------------------------
39 class DRCECommonTask : public Poco::Task
40 {
41 public:
42  typedef std::vector<std::pair<unsigned int, SessionOptions::CleanupFlag> > SubtasksList;
43 public:
44  class ApplyPidHandler
45  {
46  public:
47  ApplyPidHandler(DRCEAsyncTasksQueue& asyncTasksQueue_,
48  AsyncTasks& asyncTask_,
49  DRCETaskRequest::TaskState allowedTaskState_,
50  SessionOptions::ThreadMode threadMode_);
51  ~ApplyPidHandler(void) {}
52 
53  void operator()(pid_t pid);
54  private:
55  DRCEAsyncTasksQueue& asyncTasksQueue;
56  AsyncTasks& asyncTask;
57  DRCETaskRequest::TaskState allowedTaskState;
58  SessionOptions::ThreadMode threadMode;
59  };
60 public:
61  DRCECommonTask(const std::string& taskName_, DRCENodeOptions& nodeOptions_, CustomMessage& message_,
62  const DRCETaskRequestSetExecute& taskRequestSetExecute_, DRCEAsyncTasksQueue& asyncTasksQueue_,
63  DRCEResourceMonitor& resourceMonitor_, const DRCEInputJsonMessage& inputJsonMessage, unsigned int parentTaskId_);
64 
65  virtual void runTask(void) = 0;
66  virtual DRCEResultDataItem execute(Process::CallbackFunction fn) throw (Poco::Exception) = 0;
67 
68 
69  void updateTasks(DRCEResultDataItem& resultDataItem, unsigned int& progressCount);
70  static void updateTasks(DRCEResultDataItem& resultDataItem,
71  DRCENodeOptions& nodeOptions,
72  CustomMessage& message,
73  DRCEAsyncTasksQueue& asyncTasksQueue,
74  unsigned int& progressCount,
75  DRCECommonTask* pCommonTask=nullptr);
76 
77  void makeFilesBefore(DRCEFilesList& filesList) throw (Poco::Exception);
78  void makeFilesAfter(DRCEFilesList& filesList, DRCEResultDataItem& resultDataItem);
79 
80  static size_t getElapsedTimeMsec(const Poco::Timestamp& tsStart); // return msec
81 
82  unsigned int getParentTaskId(void) const {return parentTaskId;}
83 
84  void saveRequestData(void);
85  void saveResultData(DRCEResultDataItem& resultDataItem);
86 protected:
87  DRCENodeOptions& nodeOptions;
88  CustomMessage& message;
89  DRCETaskRequestSetExecute taskRequestSetExecute;
90  DRCEAsyncTasksQueue& asyncTasksQueue;
91  DRCEResourceMonitor& resourceMonitor;
92  DRCEInputJsonMessage inputJsonMessage;
93  unsigned int parentTaskId;
94 public:
95  void executeSubtasks(DRCEResultDataItem& resultDataItem);
96  void cleanup(bool needCleanup=false);
97  static void getSubtasksList(DRCEInputJsonMessage& inputJsonMessage, SubtasksList& subtasks);
98 
99  static DRCEInputJsonMessage getRequestData(unsigned int taskId,
100  DRCENodeOptions& nodeOptions,
101  CustomMessage& message);
102 
103  static void waitUpdateAllTasks(unsigned int taskId,
104  DRCENodeOptions& nodeOptions,
105  CustomMessage& message,
106  DRCEAsyncTasksQueue& asyncTasksQueue,
107  Poco::Task* pTask=nullptr);
108 
109  static void waitEndAllTask(DRCEInputJsonMessage& inputJsonMessage,
110  DRCEAsyncTasksQueue& asyncTasksQueue,
111  Poco::Task* pTask=nullptr);
112 
113  static void cleanup(unsigned int taskId,
114  DRCENodeOptions& nodeOptions,
115  CustomMessage& message,
116  bool needCleanup=false);
117 
118  static void cleanup(DRCEInputJsonMessage& inputJsonMessage,
119  DRCENodeOptions& nodeOptions,
120  CustomMessage& message,
121  bool needCleanup=false);
122 
123  static void saveRequestData(DRCEInputJsonMessage& inputJsonMessage,
124  DRCENodeOptions& nodeOptions,
125  CustomMessage& message);
126 
127  static void saveRequest(unsigned int taskId,
128  DRCENodeOptions& nodeOptions,
129  CustomMessage& message);
130 
131  static void saveRequest(DRCEInputJsonMessage& inputJsonMessage,
132  DRCENodeOptions& nodeOptions,
133  CustomMessage& message);
134 
135  static void saveResultData(DRCEResultDataItem& resultDataItem,
136  DRCENodeOptions& nodeOptions,
137  CustomMessage& message,
138  DRCEAsyncTasksQueue& asyncTasksQueue);
139 
140  static DRCEResultDataItem makeResultDataItem(DRCEInputJsonMessage& inputJsonMessage,
141  DRCENodeOptions& nodeOptions,
142  CustomMessage& message,
143  DRCETaskRequest::RequestType requestType,
144  DRCETaskRequest::TaskState taskState,
145  unsigned int errorCode=NO_ERROR,
146  const std::string& errorMessage="",
147  unsigned int pid=0,
148  size_t timeElapsed=0,
149  unsigned int exitStatus=0);
150 
151  static unsigned int getHostParentTaskId(unsigned int taskId,
152  DRCEAsyncTasksQueue& asyncTasksQueue);
153 
154  static DRCEResultDataItem extractStatusResultDataItem(unsigned int taskId,
155  DRCENodeOptions& nodeOptions,
156  DRCEAsyncTasksQueue& asyncTasksQueue);
157 
158  static DRCEResultDataItem extractDataResultDataItem(unsigned int taskId,
159  DRCENodeOptions& nodeOptions,
160  DRCEAsyncTasksQueue& asyncTasksQueue);
161 
162  static void safeCleanup(const std::string& path);
163  static DRCEResultDataItem extractResultDataItem(const std::string& fileName);
164 private:
165  typedef Poco::Task inherited;
166 };
167 //-----------------------------------------------------------------------------
168 //-----------------------------------------------------------------------------
169 } // end namespace drce
170 } // end namespace HCE
171 
172 #endif // DRCE_COMMON_TASK_HPP