Get rid of partial param loading mechanism

Simplify device loading for better performance in most case, and the
potential for further optimizations to follow.
This commit is contained in:
Zaid Abdulla committed 2019-06-05 00:17:36 -07:00
1 parent f4c9321d4d
commit c5553d95c5
4 files changed
+82 -419

No files matched your search

+16
View File
@@ -1350,6 +1350,22 @@ async function listenerAsync(
}
}
const parameters = await db.fetchDevice(
_sessionContext.deviceId,
_sessionContext.timestamp
);
if (parameters) {
for (const p of parameters) {
const path = _sessionContext.deviceData.paths.add(p[0]);
_sessionContext.deviceData.timestamps.set(path, p[1], 0);
if (p[2]) _sessionContext.deviceData.attributes.set(path, p[2], 0);
}
} else {
// Device not available in database, mark as new
_sessionContext.new = true;
}
const authenticated = await authenticate(_sessionContext, bodyStr);
if (!authenticated) return responseUnauthorized(_sessionContext);
+15 -252
View File
@@ -17,7 +17,7 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/
import { MongoClient, ObjectID, Collection, FindOneOptions } from "mongodb";
import { MongoClient, ObjectID, Collection } from "mongodb";
import { get } from "./config";
import { escapeRegExp } from "./common";
import { parse } from "./common/expression-parser";
@@ -96,234 +96,27 @@ function optimizeProjection(obj: {}): {} {
export async function fetchDevice(
id: string,
timestamp: number,
patterns: [Path, number][]
): Promise<[[Path, number, Attributes?][], [Path, number][]]> {
const MAX_DEPTH = +get("MAX_DEPTH", id);
if (!patterns || !patterns.length)
patterns = [[Path.parse(""), (1 << MAX_DEPTH) - 1]];
let projection = {};
const projectionTree = {};
function func(
path: string,
pathLength: number,
pats: [Path, number][],
projTree
): void {
const children: { [fragment: string]: [Path, number][] } = {};
for (const pat of pats) {
const fragment = (pat[0].segments[pathLength] as string) || "*";
if (fragment === "*") {
if (pat[1] << pathLength) projection[path.slice(0, -1)] = 1;
return;
}
if (!projTree[fragment]) projTree[fragment] = {};
if (pat[1] & (1 << pathLength)) {
projection[path + "_timestamp"] = 1;
projection[path + "_object"] = 1;
projection[path + "_instance"] = 1;
projection[path + fragment + "._timestamp"] = 1;
projection[path + fragment + "._value"] = 1;
projection[path + fragment + "._type"] = 1;
projection[path + fragment + "._object"] = 1;
projection[path + fragment + "._instance"] = 1;
projection[path + fragment + "._writable"] = 1;
projection[path + fragment + "._orig"] = 1;
}
if (pat[1] >> (pathLength + 1)) {
if (!children[fragment]) children[fragment] = [];
children[fragment].push(pat);
}
}
for (const [k, v] of Object.entries(children))
func(path + k + ".", pathLength + 1, v, projTree[k]);
}
func("", 0, patterns, projectionTree);
delete projectionTree["DeviceID"];
delete projectionTree["Events"];
delete projectionTree["Tags"];
const res: [Path, number, Attributes?][] = [];
const loaded: [Path, number][] = [];
projection = optimizeProjection(projection);
for (const k of Object.keys(projection)) {
if (k === "" || k === "Events" || k.startsWith("Events.")) {
if (k === "" || k === "Events" || k === "Events._writable") {
res.push([
timestamp: number
): Promise<[Path, number, Attributes?][]> {
const res: [Path, number, Attributes?][] = [
[
Path.parse("Events"),
timestamp,
{ object: [timestamp, 1], writable: [timestamp, 0] }
]);
if (k === "Events._writable") loaded.push([Path.parse("Events"), 1]);
}
if (k === "Events") {
projection["_lastInform"] = 1;
projection["_lastBoot"] = 1;
projection["_lastBootstrap"] = 1;
projection["_registered"] = 1;
loaded.push([Path.parse("Events"), (1 << MAX_DEPTH) - 1]);
} else if (k === "Events.Inform._writable" || k === "Events.Inform") {
projection["_lastInform"] = 1;
loaded.push([Path.parse("Events.Inform"), 1 ^ ((1 << MAX_DEPTH) - 1)]);
} else if (k === "Events.1_BOOT._writable" || k === "Events.1_BOOT") {
projection["_lastBoot"] = 1;
loaded.push([Path.parse("Events.1_BOOT"), 1 ^ ((1 << MAX_DEPTH) - 1)]);
} else if (
k === "Events.0_BOOTSTRAP._writable" ||
k === "Events.0_BOOTSTRAP"
) {
projection["_lastBootstrap"] = 1;
loaded.push([
Path.parse("Events.0_BOOTSTRAP"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (
k === "Events.Registered._writable" ||
k === "Events.Registered"
) {
projection["_registered"] = 1;
loaded.push([
Path.parse("Events.Registered"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (k.endsWith("._writable") && k !== "Events._writable") {
loaded.push([
Path.parse(k.split(".", 2).join(".")),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
}
if (k !== "") delete projection[k];
}
if (k === "" || k === "DeviceID" || k.startsWith("DeviceID.")) {
if (k === "" || k === "DeviceID" || k === "DeviceID._writable") {
res.push([
],
[
Path.parse("DeviceID"),
timestamp,
{ object: [timestamp, 1], writable: [timestamp, 0] }
]);
if (k === "DeviceID._writable")
loaded.push([Path.parse("DeviceID"), 1]);
}
if (k === "DeviceID") {
projection["_id"] = 1;
projection["_deviceId"] = 1;
loaded.push([Path.parse("DeviceID"), (1 << MAX_DEPTH) - 1]);
} else if (k === "DeviceID.ID._writable" || k === "DeviceID.ID") {
projection["_id"] = 1;
loaded.push([Path.parse("DeviceID.ID"), 1 ^ ((1 << MAX_DEPTH) - 1)]);
} else if (
k === "DeviceID.Manufacturer._writable" ||
k === "DeviceID.DeManufacturer"
) {
projection["_deviceId._Manufacturer"] = 1;
loaded.push([
Path.parse("DeviceID.Manufacturer"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (
k === "DeviceID.ProductClass._writable" ||
k === "DeviceID.ProductClass"
) {
projection["_deviceId._ProductClass"] = 1;
loaded.push([
Path.parse("DeviceID.ProductClass"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (k === "DeviceID.OUI._writable" || k === "DeviceID.OUI") {
projection["_deviceId._OUI"] = 1;
loaded.push([
Path.parse("DeviceID.ProductClass"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (
k === "DeviceID.SerialNumber._writable" ||
k === "DeviceID.SerialNumber"
) {
projection["_deviceId._SerialNumber"] = 1;
loaded.push([
Path.parse("DeviceID.SerialNumber"),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
} else if (k.endsWith("._writable") && k !== "DeviceID._writable") {
loaded.push([
Path.parse(k.split(".", 2).join(".")),
1 ^ ((1 << MAX_DEPTH) - 1)
]);
}
if (k !== "") delete projection[k];
}
if (k === "Tags" || k.startsWith("Tags.")) {
if (!projection["_tags"]) {
projection["_tags"] = 1;
loaded.push([Path.parse("Tags"), (1 << MAX_DEPTH) - 1]);
}
delete projection[k];
}
}
let options: FindOneOptions;
if (!projection[""]) {
if (!Object.keys(projection).length) options = { projection: { _id: 1 } };
else options = { projection: projection };
}
const device = await devicesCollection.findOne({ _id: id }, options);
if (!device) return [null, null];
function storeParams(
obj,
path: string,
pathLength: number,
ts,
descendantsFetched,
projTree
): void {
let thisFetched = false;
if (descendantsFetched) {
thisFetched = true;
} else if (projection[path.slice(0, -1)]) {
descendantsFetched = true;
if (
pathLength &&
projection[
path.slice(0, path.lastIndexOf(".", path.length - 2) + 1) +
"_timestamp"
]
) {
thisFetched = true;
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << (pathLength - 1)) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
} else {
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
}
} else if (projection[path + "_writable"]) {
loaded.push([Path.parse(path.slice(0, -1)), 1 << (pathLength - 1)]);
thisFetched = true;
}
];
const device = await devicesCollection.findOne({ _id: id });
if (!device) return null;
function storeParams(obj, path: string, pathLength: number, ts): void {
if (obj["_timestamp"]) obj["_timestamp"] = +obj["_timestamp"];
// For compatibility with v1.0 database
if (obj["_instance"] && obj["_object"] == null) obj["_object"] = true;
if (thisFetched) {
const attrs: Attributes = {};
let t = obj["_timestamp"] || 1;
if (ts > t) t = ts;
@@ -341,36 +134,17 @@ export async function fetchDevice(
if (obj["_object"] != null) attrs.object = [t, obj["_object"] ? 1 : 0];
res.push([Path.parse(path.slice(0, -1)), t, attrs]);
}
for (const [k, v] of Object.entries(obj)) {
if (!k.startsWith("_")) {
obj["_object"] = true;
storeParams(
v,
path + k + ".",
pathLength + 1,
obj["_timestamp"],
descendantsFetched,
projTree ? projTree[k] : null
);
if (projTree) delete projTree[k];
storeParams(v, path + k + ".", pathLength + 1, obj["_timestamp"]);
}
}
if (!descendantsFetched) {
if (projTree) {
for (const k of Object.keys(projTree)) {
const p = Path.parse(path + k);
loaded.push([p, ((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)]);
if ((obj["_object"] || !pathLength) && obj["_timestamp"])
res.push([p, obj["_timestamp"]]);
}
}
} else if ((obj["_object"] || !pathLength) && obj["_timestamp"]) {
res.push([Path.parse(path + "*"), obj["_timestamp"]]);
}
}
for (const [k, v] of Object.entries(device)) {
switch (k) {
@@ -384,7 +158,6 @@ export async function fetchDevice(
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBoot":
res.push([
@@ -396,7 +169,6 @@ export async function fetchDevice(
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBootstrap":
res.push([
@@ -408,7 +180,6 @@ export async function fetchDevice(
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_registered":
// Use current timestamp for registered event attribute timestamps
@@ -421,10 +192,8 @@ export async function fetchDevice(
value: [timestamp, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_id":
if (projection[""] || projection["_id"]) {
res.push([
Path.parse("DeviceID.ID"),
timestamp,
@@ -434,9 +203,6 @@ export async function fetchDevice(
value: [timestamp, [v as string, "xsd:string"]]
}
]);
}
delete device[k];
break;
case "_tags":
if ((v as string[]).length) {
@@ -459,7 +225,6 @@ export async function fetchDevice(
}
]);
}
delete device[k];
break;
case "_deviceId":
if (v["_Manufacturer"] != null) {
@@ -509,13 +274,11 @@ export async function fetchDevice(
}
]);
}
delete device[k];
}
}
storeParams(device, "", 0, 0, false, projectionTree);
return [res, loaded];
storeParams(device, "", 0, 0);
return res;
}
export async function saveDevice(
+51 -165
View File
@@ -17,8 +17,6 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/
import * as config from "./config";
import * as db from "./db";
import * as device from "./device";
import * as sandbox from "./sandbox";
import * as localCache from "./local-cache";
@@ -67,7 +65,6 @@ function initDeviceData(): DeviceData {
paths: new PathSet(),
timestamps: new VersionedMap(),
attributes: new VersionedMap(),
loaded: new Map(),
trackers: new Map(),
changes: new Set()
};
@@ -100,45 +97,6 @@ export function init(
return sessionContext;
}
async function loadParameters(sessionContext: SessionContext): Promise<void> {
if (!sessionContext.toLoad || !sessionContext.toLoad.size) return;
const toLoad = Array.from(sessionContext.toLoad.entries());
let [parameters, loaded] = await db.fetchDevice(
sessionContext.deviceId,
sessionContext.timestamp,
toLoad
);
if (!parameters) {
// Device not available in database, mark as new
sessionContext.new = true;
loaded = [
[
Path.parse(""),
(1 << +config.get("MAX_DEPTH", sessionContext.deviceId)) - 1
]
];
parameters = [];
}
for (const p of loaded) {
const path = sessionContext.deviceData.paths.add(p[0]);
if (p[1]) {
const l = sessionContext.deviceData.loaded.get(path) | 0;
sessionContext.deviceData.loaded.set(path, l | p[1]);
}
}
for (const p of parameters) {
const path = sessionContext.deviceData.paths.add(p[0]);
sessionContext.deviceData.timestamps.set(path, p[1], 0);
if (p[2]) sessionContext.deviceData.attributes.set(path, p[2], 0);
}
delete sessionContext.toLoad;
}
function generateRpcId(sessionContext: SessionContext): string {
return (
sessionContext.timestamp.toString(16) +
@@ -228,13 +186,6 @@ export async function inform(
]);
}
// Preload DeviceID params
loadPath(sessionContext, Path.parse("DeviceID.*"));
for (const p of params) loadPath(sessionContext, p[0]);
await loadParameters(sessionContext);
if (sessionContext.new) {
params.push([
Path.parse("DeviceID.ID"),
@@ -275,7 +226,11 @@ export async function inform(
}
}
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return { name: "InformResponse" };
}
@@ -321,10 +276,6 @@ export async function transferComplete(
};
}
loadPath(sessionContext, Path.parse(`Downloads.${instance}.*`));
await loadParameters(sessionContext);
let toClear = null;
const timestamp = sessionContext.timestamp + sessionContext.iteration + 1;
@@ -396,7 +347,11 @@ export async function transferComplete(
toClear
);
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return {
acsResponse: { name: "TransferCompleteResponse" },
operation: operation,
@@ -408,10 +363,6 @@ async function revertDownloadParameters(
sessionContext: SessionContext,
instance
): Promise<void> {
loadPath(sessionContext, Path.parse(`Downloads.${instance}.*`));
await loadParameters(sessionContext);
const timestamp = sessionContext.timestamp + sessionContext.iteration + 1;
let p;
@@ -436,7 +387,10 @@ async function revertDownloadParameters(
]
});
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
}
export async function timeoutOperations(
@@ -1047,13 +1001,13 @@ export async function rpcRequest(
d.attrGet[k] = sessionContext.timestamp;
}
}
for (const ad of device.getAliasDeclarations(d.path, 1))
loadPath(sessionContext, ad.path);
}
await clear(sessionContext, toClear);
await loadParameters(sessionContext);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return rpcRequest(sessionContext, _declarations);
}
@@ -1063,13 +1017,6 @@ export async function rpcRequest(
sessionContext.declarations[0] = sessionContext.declarations[0].concat(
_declarations
);
for (const d of _declarations) {
for (const ad of device.getAliasDeclarations(d.path, 1))
loadPath(sessionContext, ad.path);
}
await loadParameters(sessionContext);
return rpcRequest(sessionContext, null);
}
@@ -1232,7 +1179,11 @@ export async function rpcRequest(
return false;
});
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
sessionContext.syncState.virtualParameterDeclarations[inception] = vpd;
return rpcRequest(sessionContext, null);
}
@@ -1407,7 +1358,10 @@ export async function rpcRequest(
}
if (toClear || sessionContext.deviceData.changes.has("prerequisite")) {
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return rpcRequest(sessionContext, null);
}
@@ -1472,7 +1426,11 @@ export async function rpcRequest(
);
}
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return rpcRequest(sessionContext, null);
}
@@ -2171,46 +2129,6 @@ function processDeclarations(
return virtualParameterDeclarations;
}
function loadPath(
sessionContext: SessionContext,
path: Path,
depth?: number
): boolean {
depth = depth || (1 << path.length) - 1;
if (sessionContext.new || !depth) return true;
if (!sessionContext.toLoad) sessionContext.toLoad = new Map();
// Trim trailing wildcards
let trimWildcard = path.length;
while (trimWildcard && path.segments[trimWildcard - 1] === "*")
--trimWildcard;
if (trimWildcard < path.length) path = path.slice(0, trimWildcard);
for (let i = 0; i <= path.length; ++i) {
const d = i === path.length ? 99 : i;
for (const sup of sessionContext.deviceData.paths.find(
path.slice(0, i),
true,
false,
d
)) {
let v =
sessionContext.deviceData.loaded.get(sup) |
sessionContext.toLoad.get(sup);
if (sup.length > i) v &= (1 << i) - 1;
depth &= depth ^ v;
if (depth === 0) return true;
}
}
path = sessionContext.deviceData.paths.add(path);
depth |= sessionContext.toLoad.get(path);
sessionContext.toLoad.set(path, depth);
return false;
}
function processInstances(
sessionContext: SessionContext,
parent: Path,
@@ -2272,38 +2190,6 @@ function processInstances(
}
}
async function clear(
sessionContext: SessionContext,
toClear: Clear[]
): Promise<void> {
if (!toClear || !toClear.length) return;
const MAX_DEPTH = +config.get("MAX_DEPTH", sessionContext.deviceId);
for (const c of toClear) {
if (c[1]) {
const p = c[0].slice(0, -1); // in order to include superset
loadPath(
sessionContext,
p,
((1 << p.length) - 1) ^ ((1 << MAX_DEPTH) - 1)
);
} else if (c[2] && c[2].object) {
loadPath(
sessionContext,
c[0],
(((1 << c[0].length) - 1) >> 1) ^ ((1 << MAX_DEPTH) - 1)
);
} else {
loadPath(sessionContext, c[0], (1 << c[0].length) >> 1);
}
}
await loadParameters(sessionContext);
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
export async function rpcResponse(
sessionContext: SessionContext,
id: string,
@@ -2456,14 +2342,6 @@ export async function rpcResponse(
return bl - al;
});
if (rpcReq.nextLevel) {
loadPath(sessionContext, root, (1 << (root.length + 1)) - 1);
} else {
const MAX_DEPTH = +config.get("MAX_DEPTH", sessionContext.deviceId);
loadPath(sessionContext, root, (1 << MAX_DEPTH) - 1);
}
await loadParameters(sessionContext);
if (!root.length) {
for (const n of [
"DeviceID",
@@ -2490,7 +2368,11 @@ export async function rpcResponse(
);
}
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return;
case "SetParameterValuesResponse":
if (rpcReq.name !== "SetParameterValues")
@@ -2635,7 +2517,11 @@ export async function rpcResponse(
default:
throw new Error("Response name not recognized");
}
await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
}
export async function rpcFault(
@@ -2677,7 +2563,10 @@ export async function rpcFault(
toClear = [[Path.parse(rpcReq.objectName.replace(/\.$/, "")), timestamp]];
}
if (toClear) await clear(sessionContext, toClear);
if (toClear) {
for (const c of toClear)
device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
}
return null;
}
@@ -2702,13 +2591,11 @@ export async function deserialize(
for (const r of sessionContext.deviceData) {
const path = deviceData.paths.add(Path.parse(r[0]));
if (r[1]) deviceData.loaded.set(path, r[1]);
if (r[1]) deviceData.trackers.set(path, r[1]);
if (r[2]) deviceData.trackers.set(path, r[2]);
if (r[3]) {
deviceData.timestamps.setRevisions(path, r[3]);
if (r[4]) deviceData.attributes.setRevisions(path, r[4]);
if (r[2]) {
deviceData.timestamps.setRevisions(path, r[2]);
if (r[3]) deviceData.attributes.setRevisions(path, r[3]);
}
}
@@ -2732,7 +2619,6 @@ export async function serialize(
)) {
const e = [
path.toString(),
sessionContext.deviceData.loaded.get(path) || 0,
sessionContext.deviceData.trackers.get(path) || null,
sessionContext.deviceData.timestamps.getRevisions(path) || null,
sessionContext.deviceData.attributes.getRevisions(path) || null
-2
View File
@@ -50,7 +50,6 @@ export interface DeviceData {
paths: PathSet;
timestamps: VersionedMap<Path, number>;
attributes: VersionedMap<Path, Attributes>;
loaded: Map<Path, number>;
trackers: Map<Path, { [name: string]: number }>;
changes: Set<string>;
}
@@ -118,7 +117,6 @@ export interface SessionContext {
httpRequest?: IncomingMessage;
faultsTouched?: { [channel: string]: boolean };
presetCycles?: number;
toLoad?: Map<Path, number>;
new?: boolean;
debug?: boolean;
state: number;