Added task limiter on mesh agent binary updates and legacy swarm server.
Ylian Saint-Hilaire committed
Jan 28, 2019 at 16:53 UTC
50fa7d760acf25520a947821eb583c500a1b84f9
2 files changed
+55
-38
meshagent.js
+43
-33
@@ -62,7 +62,11 @@ module.exports.CreateMeshAgent = function (parent, db, ws, req, args, domain) {
62
63
// Other clean up may be needed here
64
if (obj.unauth) { delete obj.unauth; }
65
- if (obj.agentUpdate != null) { obj.fs.close(obj.agentUpdate.fd); obj.agentUpdate = null; }
65
+ if (obj.agentUpdate != null) {
66
+ obj.fs.close(obj.agentUpdate.fd);
67
+ obj.parent.parent.taskLimiter.completed(obj.agentUpdate.taskid); // Indicate this task complete
68
+ obj.agentUpdate = null;
69
+ }
70
if (((obj.agentInfo) && (obj.agentInfo.capabilities) && (obj.agentInfo.capabilities & 0x20)) || ((mesh) && (mesh.flags) && (mesh.flags & 1))) { // This is a temporary agent, remote it
71
// Delete this node including network interface information and events
72
obj.db.Remove(obj.dbNodeKey); // Remove node with that id
@@ -163,38 +167,42 @@ module.exports.CreateMeshAgent = function (parent, db, ws, req, args, domain) {
167
if ((msg.length == 52) && (obj.agentExeInfo != null) && (obj.agentExeInfo.update == true)) {
168
var agenthash = obj.common.rstr2hex(msg.substring(4)).toLowerCase();
169
if ((agenthash != obj.agentExeInfo.hash) && (agenthash != '000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000')) {
166
- // Mesh agent update required
167
- if (obj.nodeid != null) { obj.parent.parent.debug(1, 'Agent update required, NodeID=0x' + obj.nodeid.substring(0, 16) + ', ' + obj.agentExeInfo.desc); }
168
- obj.fs.open(obj.agentExeInfo.path, 'r', function (err, fd) {
169
- if (err) { return console.error(err); }
170
- obj.agentUpdate = { oldHash: agenthash, ptr: 0, buf: Buffer.alloc(agentUpdateBlockSize + 4), fd: fd };
171
-
172
- // MeshCommand_CoreModule, ask mesh agent to clear the core.
173
- // The new core will only be sent after the agent updates.
174
- obj.send(obj.common.ShortToStr(10) + obj.common.ShortToStr(0));
175
-
176
- // We got the agent file open on the server side, tell the agent we are sending an update starting with the SHA384 hash of the result
177
- //console.log("Agent update file open.");
178
- obj.send(obj.common.ShortToStr(13) + obj.common.ShortToStr(0)); // Command 13, start mesh agent download
179
-
180
- // Send the first mesh agent update data block
181
- obj.agentUpdate.buf[0] = 0;
182
- obj.agentUpdate.buf[1] = 14;
183
- obj.agentUpdate.buf[2] = 0;
184
- obj.agentUpdate.buf[3] = 1;
185
- var len = -1;
186
- try { len = obj.fs.readSync(obj.agentUpdate.fd, obj.agentUpdate.buf, 4, agentUpdateBlockSize, obj.agentUpdate.ptr); } catch (e) { }
187
- if (len == -1) {
188
- // Error reading the agent file, stop here.
189
- obj.fs.close(obj.agentUpdate.fd);
190
- obj.agentUpdate = null;
191
- } else {
192
- // Send the first block to the agent
193
- obj.agentUpdate.ptr += len;
194
- //console.log("Agent update send first block: " + len);
195
- obj.send(obj.agentUpdate.buf); // Command 14, mesh agent first data block
196
- }
197
- });
170
+ // Mesh agent update required, do it using task limiter so not to flood the network.
171
+ obj.parent.parent.taskLimiter.launch(function (argument, taskid, taskLimiterQueue) {
172
+ if (obj.nodeid != null) { obj.parent.parent.debug(1, 'Agent update required, NodeID=0x' + obj.nodeid.substring(0, 16) + ', ' + obj.agentExeInfo.desc); }
173
+ obj.fs.open(obj.agentExeInfo.path, 'r', function (err, fd) {
174
+ if (err) { return console.error(err); }
175
+ obj.agentUpdate = { oldHash: agenthash, ptr: 0, buf: Buffer.alloc(agentUpdateBlockSize + 4), fd: fd, taskid: taskid };
176
+
177
+ // MeshCommand_CoreModule, ask mesh agent to clear the core.
178
+ // The new core will only be sent after the agent updates.
179
+ obj.send(obj.common.ShortToStr(10) + obj.common.ShortToStr(0));
180
+
181
+ // We got the agent file open on the server side, tell the agent we are sending an update starting with the SHA384 hash of the result
182
+ //console.log("Agent update file open.");
183
+ obj.send(obj.common.ShortToStr(13) + obj.common.ShortToStr(0)); // Command 13, start mesh agent download
184
+
185
+ // Send the first mesh agent update data block
186
+ obj.agentUpdate.buf[0] = 0;
187
+ obj.agentUpdate.buf[1] = 14;
188
+ obj.agentUpdate.buf[2] = 0;
189
+ obj.agentUpdate.buf[3] = 1;
190
+ var len = -1;
191
+ try { len = obj.fs.readSync(obj.agentUpdate.fd, obj.agentUpdate.buf, 4, agentUpdateBlockSize, obj.agentUpdate.ptr); } catch (e) { }
192
+ if (len == -1) {
193
+ // Error reading the agent file, stop here.
194
+ obj.fs.close(obj.agentUpdate.fd);
195
+ obj.parent.parent.taskLimiter.completed(obj.agentUpdate.taskid); // Indicate this task complete
196
+ obj.agentUpdate = null;
197
+ } else {
198
+ // Send the first block to the agent
199
+ obj.agentUpdate.ptr += len;
200
+ //console.log("Agent update send first block: " + len);
201
+ obj.send(obj.agentUpdate.buf); // Command 14, mesh agent first data block
202
+ }
203
+ });
204
+ }, null);
205
+
206
} else {
207
// Check the mesh core, if the agent is capable of running one
208
if (((obj.agentInfo.capabilities & 16) != 0) && (obj.parent.parent.meshAgentsArchitectureNumbers[obj.agentInfo.agentId].core != null)) {
@@ -212,6 +220,7 @@ module.exports.CreateMeshAgent = function (parent, db, ws, req, args, domain) {
220
if (len == -1) {
221
// Error reading the agent file, stop here.
222
obj.fs.close(obj.agentUpdate.fd);
223
+ obj.parent.parent.taskLimiter.completed(obj.agentUpdate.taskid); // Indicate this task complete
224
obj.agentUpdate = null;
225
} else {
226
// Send the next block to the agent
@@ -223,6 +232,7 @@ module.exports.CreateMeshAgent = function (parent, db, ws, req, args, domain) {
232
//console.log("Agent update sent");
233
obj.send(obj.common.ShortToStr(13) + obj.common.ShortToStr(0) + obj.common.hex2rstr(obj.agentExeInfo.hash)); // Command 13, end mesh agent download, send agent SHA384 hash
234
obj.fs.close(obj.agentUpdate.fd);
235
+ obj.parent.parent.taskLimiter.completed(obj.agentUpdate.taskid); // Indicate this task complete
236
obj.agentUpdate = null;
237
}
238
}
swarmserver.js
+12
-5
@@ -214,7 +214,12 @@ module.exports.CreateSwarmServer = function (parent, db, args, certificates) {
214
socket.tag.update = obj.migrationAgents[nodeblock.agenttype][nextAgentVersion];
215
socket.tag.updatePtr = 0;
216
//console.log('Performing legacy agent update from ' + nodeblock.agentversion + '.' + nodeblock.agenttype + ' to ' + socket.tag.update.ver + '.' + socket.tag.update.arch + ' on ' + nodeblock.agentname + '.');
217
- obj.SendCommand(socket, LegacyMeshProtocol.GETSTATE, common.IntToStr(5) + common.IntToStr(0)); // agent.SendQuery(5, 0); // Start the agent download
217
+
218
+ // Start the agent download using the task limiter so not to flood the server.
219
+ obj.parent.taskLimiter.launch(function (socket, taskid, taskLimiterQueue) {
220
+ socket.tag.taskid = taskid;
221
+ obj.SendCommand(socket, LegacyMeshProtocol.GETSTATE, common.IntToStr(5) + common.IntToStr(0)); // agent.SendQuery(5, 0); // Start the agent download
222
+ }, socket);
223
} else {
224
//console.log('No legacy agent update for ' + nodeblock.agentversion + '.' + nodeblock.agenttype + ' on ' + nodeblock.agentname + '.');
225
}
@@ -248,6 +253,8 @@ module.exports.CreateSwarmServer = function (parent, db, args, certificates) {
253
// Send end-of-transfer
254
obj.SendCommand(socket, LegacyMeshProtocol.GETSTATE, common.IntToStr(7) + common.IntToStr(socket.tag.update.binary.length)); //agent.SendQuery(7, AgentFileLen);
255
Debug(3, 'Swarm:Sending end of agent, ptr = ' + socket.tag.updatePtr);
256
+ obj.parent.taskLimiter.completed(socket.tag.taskid); // Indicate this task complete
257
+ delete socket.tag.taskid;
258
delete socket.tag.update;
259
delete socket.tag.updatePtr;
260
}
@@ -274,9 +281,11 @@ module.exports.CreateSwarmServer = function (parent, db, args, certificates) {
281
282
socket.addListener("close", function () {
283
Debug(1, 'Swarm:Connection closed');
277
- try { delete obj.ciraConnections[socket.tag.nodeid]; } catch (e) { }
278
- obj.parent.ClearConnectivityState(socket.tag.meshid, socket.tag.nodeid, 2);
284
if (socket.pingTimer != null) { clearInterval(socket.pingTimer); delete socket.pingTimer; }
285
+ if (socket.tag && (typeof socket.tag.taskid == 'number')) {
286
+ obj.parent.taskLimiter.completed(socket.tag.taskid); // Indicate this task complete
287
+ delete socket.tag.taskid;
288
+ }
289
});
290
291
socket.addListener("error", function () {
@@ -344,8 +353,6 @@ module.exports.CreateSwarmServer = function (parent, db, args, certificates) {
353
// Disconnect legacy agent connection
354
obj.close = function (socket) {
355
try { socket.close(); } catch (e) { }
347
- try { delete obj.ciraConnections[socket.tag.nodeid]; } catch (e) { }
348
- obj.parent.ClearConnectivityState(socket.tag.meshid, socket.tag.nodeid, 2);
356
};
357
358
obj.SendCommand = function (socket, cmdid, data) {