From db6eae2aee08101b7627b109391aa990c957ca16 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Thu, 24 Jul 2014 01:49:04 +0300 Subject: [PATCH] Queue device updates and commit them using fewer database queries for better performance. Partial updates caused by faulty tasks will no longer be committed in order to avoid messing the device model. --- lib/cwmp.coffee | 51 ++++---- lib/tasks.coffee | 53 +++++---- lib/update-device.coffee | 251 ++++++++++++++++++++++++--------------- 3 files changed, 213 insertions(+), 142 deletions(-) diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index de165c0..be25bef 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -111,13 +111,11 @@ inform = (currentRequest, cwmpRequest) -> informHash = crypto.createHash('md5').update(JSON.stringify(parameterNames)).update(JSON.stringify(config.CUSTOM_COMMANDS)).digest('hex') - actions = {parameterValues : cwmpRequest.methodRequest.parameterList, set : {}} - actions.set._lastInform = now - actions.set._lastBoot = now if '1 BOOT' in cwmpRequest.methodRequest.event + deviceUpdates = {parameterValues : cwmpRequest.methodRequest.parameterList, informEvents : cwmpRequest.methodRequest.event} + # Delete all pending tasks on bootstrap (e.g. factory reset) if '0 BOOTSTRAP' in cwmpRequest.methodRequest.event - actions.set._lastBootstrap = lastBootstrap = now - # ignore all pending tasks + lastBootstrap = now db.tasksCollection.remove({device : currentRequest.session.deviceId}, (err, removed) -> throw err if err ) @@ -126,17 +124,17 @@ inform = (currentRequest, cwmpRequest) -> throw err if err updateAndRespond = () -> - updateDevice(currentRequest.session.deviceId, actions, (err) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, false, (err) -> throw err if err - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'InformResponse'}, - cwmpVersion : cwmpRequest.cwmpVersion - }) - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}" - writeResponse(currentRequest, res) db.redisClient.setex("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, JSON.stringify(currentRequest.session), (err) -> throw err if err + res = soap.response({ + id : cwmpRequest.id, + methodResponse : {type : 'InformResponse'}, + cwmpVersion : cwmpRequest.cwmpVersion + }) + res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}" + writeResponse(currentRequest, res) ) ) @@ -189,8 +187,8 @@ inform = (currentRequest, cwmpRequest) -> delete deviceCustomCommands[cmd] for cmd of deviceCustomCommands - actions.deletedObjects ?= [] - actions.deletedObjects.push("_customCommands.#{cmd}") + deviceUpdates.deletedObjects ?= [] + deviceUpdates.deletedObjects.push("_customCommands.#{cmd}") # clear presets hash if parameters are potentially modified if _tasks.length > 0 @@ -227,7 +225,7 @@ runTask = (currentRequest, task, methodResponse) -> tasks.task(task, methodResponse, (err, status, methodRequest, deviceUpdates) -> throw err if err # TODO handle error - updateDevice(currentRequest.session.deviceId, deviceUpdates, (err) -> + updateDevice.queueUpdates(currentRequest.session.deviceId, deviceUpdates, (err) -> throw err if err timeDiff = process.hrtime(timeDiff)[0] + 1 @@ -251,19 +249,25 @@ runTask = (currentRequest, task, methodResponse) -> ) if save - db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> throw err if err - f() + db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) -> + throw err if err + f() + ) ) else f() when tasks.STATUS_COMPLETED util.log("#{currentRequest.session.deviceId}: Completed task #{task.name}(#{task._id})") - db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> throw err if err - db.redisClient.del(String(task._id), (err, res) -> + db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> throw err if err - nextTask(currentRequest) + db.redisClient.del(String(task._id), (err, res) -> + throw err if err + nextTask(currentRequest) + ) ) ) when tasks.STATUS_FAULT @@ -358,7 +362,10 @@ nextTask = (currentRequest) -> ) else util.log("#{currentRequest.session.deviceId}: Started task #{task.name}(#{task._id})") - runTask(currentRequest, task, {}) + updateDevice.clearUpdatesQueue(currentRequest.session.deviceId, (err) -> + throw err if err + runTask(currentRequest, task, {}) + ) ) diff --git a/lib/tasks.coffee b/lib/tasks.coffee index 6a3c919..723e494 100644 --- a/lib/tasks.coffee +++ b/lib/tasks.coffee @@ -65,12 +65,16 @@ this.refreshObject = (task, methodResponse, callback) -> if task.session.subtask.name is 'getParameterNames' this.getParameterNames(task.session.subtask, methodResponse, (err, status, methodRequest, deviceUpdates) => - if status & STATUS_COMPLETED - parameterNames = [] + task.session.parameterNames ?= [] + if deviceUpdates?.parameterNames? for p in deviceUpdates.parameterNames if not common.endsWith(p[0], '.') - parameterNames.push(p[0]) - task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames} + task.session.parameterNames.push(p[0]) + + if status & STATUS_COMPLETED + task.session.subtask = {name : 'getParameterValues', parameterNames : task.session.parameterNames} + delete task.session.parameterNames + this.getParameterValues(task.session.subtask, {}, (err, status, methodRequest) -> # ignore deviceUpdates returned by firt call to getParameterValues callback(err, status, methodRequest, deviceUpdates) @@ -78,7 +82,7 @@ this.refreshObject = (task, methodResponse, callback) -> else if status & STATUS_OK callback(err, STATUS_OK, methodRequest, deviceUpdates) else - throw Error('Unexpected subtask status') + throw new Error('Unexpected subtask status') ) else if task.session.subtask.name is 'getParameterValues' if methodResponse.faultcode? @@ -103,15 +107,31 @@ this.getParameterNames = (task, methodResponse, callback) -> return (param[...-1] + '.').split('.').length - 1 if methodResponse.type is 'GetParameterNamesResponse' - task.session.parameterNames = task.session.parameterNames.concat(methodResponse.parameterList) path = task.session.queue.pop() # If parameter depth higher than the threshold, nextLevel was set to false if not task.nextLevel? and getParameterDepth(path) < config.GET_PARAMETER_NAMES_DEPTH_THRESHOLD for p in methodResponse.parameterList task.session.queue.push(p[0]) if p[0][-1..] == '.' + + deviceUpdates = {parameterNames : methodResponse.parameterList} + + # Make sure that for each parameter, all its parents are explicitly included + found = {} + found[path] = 0 if !!path + for p in deviceUpdates.parameterNames + param = p[0] + i = param.length + while (i = param.lastIndexOf('.', i-1)) > path.length + pp = param.slice(0, i + 1) + break if found[pp]? + found[pp] = 0 + found[p[0]] = 1 + + for k, v of found + deviceUpdates.parameterNames.push([k]) if v == 0 else - task.session = {queue : [task.parameterPath], parameterNames : []} + task.session = {queue : [task.parameterPath]} if task.session.queue.length > 0 path = task.session.queue[-1..][0] @@ -120,25 +140,8 @@ this.getParameterNames = (task, methodResponse, callback) -> parameterPath : path, nextLevel : task.nextLevel ? getParameterDepth(path) < config.GET_PARAMETER_NAMES_DEPTH_THRESHOLD } - return callback(null, STATUS_OK, methodRequest) + return callback(null, STATUS_OK, methodRequest, deviceUpdates) else - deviceUpdates = {parameterNames : task.session.parameterNames} - - # Make sure that for each parameter, all its parents are explicitly included - found = {} - found[task.parameterPath] = 0 if !!task.parameterPath - for p in deviceUpdates.parameterNames - param = p[0] - i = param.length - while (i = param.lastIndexOf('.', i-1)) > task.parameterPath.length - pp = param.slice(0, i + 1) - break if found[pp]? - found[pp] = 0 - found[p[0]] = 1 - - for k, v of found - deviceUpdates.parameterNames.push([k]) if v == 0 - return callback(null, STATUS_COMPLETED, null, deviceUpdates) diff --git a/lib/update-device.coffee b/lib/update-device.coffee index f37ddb9..c399e10 100644 --- a/lib/update-device.coffee +++ b/lib/update-device.coffee @@ -25,136 +25,197 @@ OBJECT_REGEX = /\.$/ INSTANCE_REGEX = /\.[\d]+\.$/ -updateDevice = (deviceId, actions, callback) -> - return callback?() if not actions? +queueUpdates = (deviceId, deviceUpdates, callback) -> + return callback() if not deviceUpdates? or Object.keys(deviceUpdates).length == 0 + # TODO consider setting expiry + db.redisClient.rpush("#{deviceId}_updates", JSON.stringify(deviceUpdates), (err, res) -> + return callback(err) + ) + +clearUpdatesQueue = (deviceId, callback) -> + db.redisClient.del("#{deviceId}_updates", (err, res) -> + return callback(err) + ) + + +commitUpdates = (deviceId, deviceUpdates, commitQueue, callback) -> now = new Date(Date.now()) update = {'$set' : {}, '$unset' : {}} - if actions.parameterValues? - for p in actions.parameterValues - origValue = p[1] - v = normalize(p[0], origValue) + # Used to find parameters that no longer exist in device + presentPaths = null + projection = null - path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." - if v == origValue - update['$unset']["#{path}_orig"] = 1 - else - update['$set']["#{path}_orig"] = origValue - update['$set']["#{path}_value"] = v - update['$set']["#{path}_timestamp"] = now - if p[2]? - update['$set']["#{path}_type"] = p[2] + # Keep track of alias changes to clear cache if needed + newAliases = null - if actions.deletedObjects? - for p in actions.deletedObjects - update['$unset'][p] = 1 + processBatch = (updatesBatch, cb) -> + if updatesBatch.informEvents? + update['$set']['_lastInform'] = now + update['$set']['_lastBoot'] if '1 BOOT' in updatesBatch.informEvents + update['$set']['_lastBootstrap'] if '0 BOOTSTRAP' in updatesBatch.informEvents - if actions.instanceName? - for i in actions.instanceName - update['$set']["#{i[0]}._name"] = i[1] + if updatesBatch.parameterValues? + for p in updatesBatch.parameterValues + origValue = p[1] + v = normalize(p[0], origValue) - if actions.parameterNames? - projection = {} # fetch only necessary parameters - presentPaths = {} # to be compared with to find no longer existing parameters - - for p in actions.parameterNames - if OBJECT_REGEX.test(p[0]) - path = p[0].slice(0, -1) - if INSTANCE_REGEX.test(p[0]) - update['$set']["#{path}._instance"] = true + path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." + if v == origValue + update['$unset']["#{path}_orig"] = 1 else - update['$set']["#{path}._object"] = true - else - path = p[0] - update['$set']["#{path}._writable"] = p[1] if p[1]? - update['$set']["#{path}._timestamp"] = now + update['$set']["#{path}_orig"] = origValue + update['$set']["#{path}_value"] = v + update['$set']["#{path}_timestamp"] = now + if p[2]? + update['$set']["#{path}_type"] = p[2] - # Add parameter (sans any dot suffix) if has an alias - if (a = parameters.getAlias(path))? - newAliases = {} if not newAliases? - newAliases[path] = a + if updatesBatch.deletedObjects? + for p in updatesBatch.deletedObjects + update['$unset'][p] = 1 - projection[path] = 1 + if updatesBatch.instanceName? + for i in updatesBatch.instanceName + update['$set']["#{i[0]}._name"] = i[1] - # Store this param path and all its parents for later look up - presentPaths[path] = true - i = p[0].indexOf('.') - while i != -1 - presentPaths[p[0][0...i]] = true - i = p[0].indexOf('.', i + 1) + if updatesBatch.parameterNames? + presentPaths ?= {} + projection ?= {} - mongoQuery.optimizeProjection(projection) + for p in updatesBatch.parameterNames + if OBJECT_REGEX.test(p[0]) + path = p[0].slice(0, -1) + if INSTANCE_REGEX.test(p[0]) + update['$set']["#{path}._instance"] = true + else + update['$set']["#{path}._object"] = true + else + path = p[0] + update['$set']["#{path}._writable"] = p[1] if p[1]? + update['$set']["#{path}._timestamp"] = now - if actions.customCommands? - for p in actions.customCommands - commandName = p[0] - update['$set']["_customCommands.#{commandName}._value"] = p[1] - update['$set']["_customCommands.#{commandName}._timestamp"] = now + # Add parameter (sans any dot suffix) if has an alias + if (a = parameters.getAlias(path))? + newAliases ?= {} + newAliases[path] = a - if actions.set? - common.extend(update['$set'], actions.set) + projection[path] = 1 - f = () -> + # Store this param path and all its parents for later look up + presentPaths[path] = true + i = p[0].indexOf('.') + while i != -1 + presentPaths[p[0][0...i]] = true + i = p[0].indexOf('.', i + 1) + + if updatesBatch.customCommands? + for p in updatesBatch.customCommands + commandName = p[0] + update['$set']["_customCommands.#{commandName}._value"] = p[1] + update['$set']["_customCommands.#{commandName}._timestamp"] = now + + processBatches = (cb) -> + if not commitQueue? + processBatch(deviceUpdates) if deviceUpdates? + return cb() + + db.redisClient.lpop("#{deviceId}_updates", (err, res) -> + callback(err) if err + + if not res? + processBatch(deviceUpdates) if deviceUpdates? + return cb() + + processBatch(JSON.parse(res)) + return processBatches(cb) + ) + + commit = () -> for k of update delete update[k] if Object.keys(update[k]).length == 0 if Object.keys(update).length db.devicesCollection.update({'_id' : deviceId}, update, {}, (err, count) -> - throw err if err + return callback(err) if err + # Clear aliases cache if there's a new aliased parameter that has not been included in aliases cache if newAliases? + newAliasFound = false db.getAliases((aliases) -> for p, a of newAliases if not aliases[a]? or p not in aliases[a] - db.redisClient.del('aliases', (err, res) -> - throw err if err - ) + newAliasFound = true break + + return callback() if not newAliasFound + db.redisClient.del('aliases', (err, res) -> + return callback(err) + ) ) - callback?(err) + else + return callback() ) else - callback?() + return callback() - if projection? - db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> - throw err if err + processBatches(() -> + if projection? + mongoQuery.optimizeProjection(projection) - recursive = (obj, prefix) -> - for k,v of obj - p = prefix + k - if k[0] == '_' - # Remove set operations for unmodified values for optimization - if update['$set'][p] == v - delete update['$set'][p] - continue + # Fetch device from DB to discover and delete parameters that no longer exist + # While we're fetching the object, we might as well use the information + # to optimize our update query + db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + return callback(err) if err - # Remove unset operations for values that don't exist - delete update['$unset']["#{p}._orig"] if not v._orig? + recursive = (obj, prefix) -> + for k, v of obj + p = prefix + k - if presentPaths[p] - recursive(v, "#{p}.") + # Mark unset operations we want to keep + if update['$unset'][p]? + update['$unset'][p] = 2 + continue + + if k[0] == '_' + # Remove set operations for unmodified values for optimization + if update['$set'][p] == v + delete update['$set'][p] + continue + + if presentPaths[p] + recursive(v, "#{p}.") + else + update['$unset'][p] = 1 + + # Scan the DB object and remove parameters that no longer exists in device + for path of projection + root = device + rootPath = '' + for p in path.split('.') + rootPath += "#{p}." + root = root[p] + break if not root? + + if root + recursive(root, rootPath) else - update['$unset'][p] = 1 + if not presentPaths[rootPath.slice(0, -1)]? + update['$unset'][rootPath.slice(0, -1)] = 1 - # Scan the DB object and remove parameters that no longer exists in device - for k, v of projection - root = device - rootPath = '' - for p in path.split('.') - rootPath += "#{p}." - root = root[p] - break if not root? + # Remove unnecessary unset updates + for k, v of update['$unset'] + if common.startsWith(k, path + '.') and v != 2 + delete update['$unset'][k] - if root - recursive(root, rootPath) - else - if not presentPaths[rootPath.slice(0, -1)] - update['$unset'][rootPath.slice(0, -1)] = 1 - f() - ) - else - f() + commit() + ) + else + commit() + ) -module.exports = updateDevice \ No newline at end of file + +exports.queueUpdates = queueUpdates +exports.clearUpdatesQueue = clearUpdatesQueue +exports.commitUpdates = commitUpdates