1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
#include <QJsonArray>
#include <QJsonObject>
#include <QtDebug>
#include <QString>
#include <unistd.h>
#include <iostream>
#include <QtCore>
#include <responsemessage.h>
#include <callmessage.h>
#include <messagefactory.h>
#include <messageenginefactory.h>
#include "taskmanager.h"
TaskManager::TaskManager(QObject* parent) : QObject(parent), m_loop(nullptr) {
}
TaskManager::~TaskManager() {
}
void TaskManager::open(const QUrl &bindingAddress)
{
qRegisterMetaType<std::shared_ptr<Message>>("std::shared_ptr<Message>");
m_loop.reset();
m_loop = MessageEngineFactory::getInstance().getMessageEngine(bindingAddress);
QObject::connect(m_loop.get(), &MessageEngine::connected, this, &TaskManager::onConnected);
QObject::connect(m_loop.get(), &MessageEngine::messageReceived, this, &TaskManager::onMessageReceived, Qt::QueuedConnection);
}
void TaskManager::onConnected()
{
query(); // let's not wait 3 seconds and issue the first call directly
timer = new QTimer();
connect(timer, SIGNAL(timeout()), this, SLOT(query()));
timer->start(3000);
loadAvg();
loadAvgTimer = new QTimer();
connect(loadAvgTimer, SIGNAL(timeout()), this, SLOT(loadAvg()));
loadAvgTimer->start(1000);
}
void TaskManager::kill(int tid) {
callService("kill_process", QJsonValue(tid));
}
void TaskManager::getExtraInfo(int tid) {
callService("get_extra_info", QJsonValue(tid));
}
void TaskManager::loadAvg() {
callService("get_load_avg", QJsonValue());
}
void TaskManager::query() {
callService("get_process_list",
QJsonValue(QJsonObject({{"processes", QJsonValue()}})));
callService("get_netstat", QJsonValue());
}
void TaskManager::callService(const QString& command, QJsonValue value) {
std::unique_ptr<Message> message = MessageFactory::getInstance().createOutboundMessage(MessageId::Call);
if (!message)
return;
CallMessage* msg = static_cast<CallMessage*>(message.get());
msg->createRequest("taskmanager", command, value);
qDebug() << "sending message " << msg->serialize();
m_loop->sendMessage(std::move(message));
}
void TaskManager::onMessageReceived(std::shared_ptr<Message> message)
{
if (!message)
return;
if (message->isReply())
ProcessResponse(message);
}
void TaskManager::ProcessResponse(std::shared_ptr<Message> message)
{
std::shared_ptr<ResponseMessage> rmsg = std::static_pointer_cast<ResponseMessage>(message);
qDebug() << "got message ";
QString msgType = rmsg->replyData()["msgType"].toString();
if (msgType.isNull())
return; // no type supplied, ignoring
if (QString::compare(msgType, "processList") == 0) {
QJsonArray processes = rmsg->replyData()["processes"].toArray();
ProcessResponseTasklist(processes);
}
if (QString::compare(msgType, "extraInfo") == 0) {
QJsonObject info = rmsg->replyData()["info"].toObject();
ProcessResponseExtraInfo(info);
}
if (QString::compare(msgType, "loadAvgInfo") == 0) {
QJsonObject loadInfo = rmsg->replyData()["loadInfo"].toObject();
ProcessResponseLoadAvg(loadInfo);
}
if (QString::compare(msgType, "netStatInfo") == 0) {
QJsonObject netstat = rmsg->replyData()["netstat"].toObject();
ProcessResponseNetStat(netstat);
}
// more response types to follow
}
void TaskManager::ProcessResponseTasklist(QJsonArray& processes)
{
std::vector<ProcInfo> procs;
if (processes.size() == 0) {
// this is not a valid process list response
return;
}
for(auto it = processes.constBegin(); it != processes.constEnd(); ++it) {
ProcInfo Proc(it->toObject());
procs.push_back(Proc);
}
int flag;
if(m_procinfos.empty()){
for(auto it = procs.begin(); it != procs.end(); ++it){ // if m_procinfos is empty then this is first call
emit addProcess(it->cmd(), it->tid(), it->euid(), it->scpu(), it->ucpu(), it->resident_memory(), it->state());
}
} else {
for(auto it = procs.begin(); it != procs.end(); ++it){ // loop through procs, it = procs element (ProcInfo obj)
flag = 0;
for(auto it2 = m_procinfos.begin(); it2 != m_procinfos.end(); ++it2){ // loop through m_procinfos, it2 m_procinfos element (ProcInfo obj)
// if(*it == *it2){ // if the same ID exists in both vectors
if(it->tid() == it2->tid()){
if(it->cmd() == it2->cmd()){ // if the name is still the same
if(!(it == it2)){ // if the same process has new data
emit updateProcess(it->cmd(), it->tid(), it->euid(), it->scpu(), it->ucpu(), it->resident_memory(), it->state());
m_procinfos.erase(it2);
flag = 1;
break;
}
} else { // process in m_procinfos has died and a new one has its ID
qDebug() << "The process ID has been reused for another process";
emit updateProcess(it->cmd(), it->tid(), it->euid(), it->scpu(), it->ucpu(), it->resident_memory(), it->state());
m_procinfos.erase(it2);
flag = 1;
}
}
}
if(flag == 0){ // if no ID was found in old vector; that means it's a new process
qDebug() << it->cmd() << " process has been added";
emit addProcess(it->cmd(), it->tid(), it->euid(), it->scpu(), it->ucpu(), it->resident_memory(), it->state());
}
}
for(auto it2 = m_procinfos.begin(); it2 != m_procinfos.end(); ++it2){ // remaining processes in m_procinfos are all dead
qDebug() << "Dead processes are removed";
qDebug() << it2->cmd();
emit removeProcess(it2->tid());
}
}
m_procinfos = procs;
}
void TaskManager::ProcessResponseExtraInfo(QJsonObject &info)
{
QString infoString;
if (info.size() == 0) {
// this is not a valid process list response
QTextStream(&infoString) << "procces is not available";
} else {
infoString = "Task : " + info["cmd"].toString() + "\n"
+ "Exec start : " + info["exec_start"].toString() + "\n"
+ "Exec runtime : " + info["vruntime"].toString() + "\n"
+ "Prio : " + info["prio"].toString();
}
emit showProcessInfo(infoString);
}
void TaskManager::ProcessResponseLoadAvg(QJsonObject &loadInfo)
{
if (loadInfo.size() == 0) {
return;
}
emit updateLoadAverage(loadInfo["value"].toDouble());
}
void TaskManager::ProcessResponseNetStat(QJsonObject &netstat)
{
if (netstat.size() == 0) {
return;
}
unsigned int newInOctets = netstat["InOctets"].toInt();
unsigned int newOutOctets = netstat["OutOctets"].toInt();
emit updateNetworkStats(newInOctets - inOctets, newOutOctets - outOctets);
inOctets = newInOctets;
outOctets = newOutOctets;
}
|
