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
|
#include <QJsonArray>
#include <QJsonObject>
#include <QtDebug>
#include <QString>
#include <unistd.h>
#include <iostream>
#include <vector>
#include <algorithm>
#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 verb = rmsg->requestVerb();
if (rmsg->replyStatus() == "failed") {
qWarning() << "Reply Failed received for verb:" << verb;
return;
}
QJsonObject data = rmsg->replyData();
QString msgType = data.value("msgType").toString();
if (msgType.isNull())
return; // no type supplied, ignoring
if (QString::compare(msgType, "processList") == 0) {
QJsonArray processes = data.value("processes").toArray();
ProcessResponseTasklist(processes);
}
else if (QString::compare(msgType, "extraInfo") == 0) {
QJsonObject info = data.value("info").toObject();
ProcessResponseExtraInfo(info);
}
else if (QString::compare(msgType, "loadAvgInfo") == 0) {
QJsonObject loadInfo = data.value("loadInfo").toObject();
ProcessResponseLoadAvg(loadInfo);
}
else if (QString::compare(msgType, "netStatInfo") == 0) {
QJsonObject netstat = data.value("netstat").toObject();
ProcessResponseNetStat(netstat);
}
// more response types to follow
}
void TaskManager::ProcessResponseTasklist(QJsonArray& processes)
{
std::vector<ProcInfo> procs;
if (processes.size() == 0)
return;
for(auto it = processes.constBegin(); it != processes.constEnd(); ++it) {
if (!it->toObject().isEmpty())
procs.emplace_back(it->toObject());
}
if (procs.empty())
return;
for(auto &item : procs) {
if(m_procinfos.empty()) {
// if old vector is empty, no need to update or remove elements, just add them
emit addProcess(item.cmd(), item.tid(), item.euid(), item.scpu(), item.ucpu(), item.resident_memory(), item.state());
} else {
std::vector<ProcInfo>::iterator it = std::find_if(m_procinfos.begin(), m_procinfos.end(),
[item](const ProcInfo& proc) {
return item.tid() == proc.tid(); });
if (it != m_procinfos.end()) { // the same ID exists in both vectors
if (item.cmd() != it->cmd()) { // but different names:
// the 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(item.cmd(), item.tid(), item.euid(), item.scpu(), item.ucpu(), item.resident_memory(), item.state());
m_procinfos.erase(it);
} else { // if the ID was not found in old vector, that means it's a new process
qDebug() << item.cmd() << " new process has been added";
emit addProcess(item.cmd(), item.tid(), item.euid(), item.scpu(), item.ucpu(), item.resident_memory(), item.state());
}
}
}
// any remaining processes in old vector are all dead, remove them:
for(auto &olditem : m_procinfos) {
emit removeProcess(olditem.tid());
}
m_procinfos.swap(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;
}
|
