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

@@ -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,7 +124,9 @@ 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
+        db.redisClient.setex("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, JSON.stringify(currentRequest.session), (err) ->
           throw err if err
           res = soap.response({
             id : cwmpRequest.id,
@@ -135,8 +135,6 @@ inform = (currentRequest, cwmpRequest) ->
           })
           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
         )
       )
 
@@ -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,14 +249,19 @@ runTask = (currentRequest, task, methodResponse) ->
             )
 
           if save
+            updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) ->
+              throw err if err
               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})")
+          updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) ->
+            throw err if err
             db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) ->
               throw err if err
               db.redisClient.del(String(task._id), (err, res) ->
@@ -266,6 +269,7 @@ runTask = (currentRequest, task, methodResponse) ->
                 nextTask(currentRequest)
               )
             )
+          )
         when tasks.STATUS_FAULT
           retryAfter = config.RETRY_DELAY * Math.pow(2, task.retries ? 0)
           util.log("#{currentRequest.session.deviceId}: Fault response for task #{task._id}. Retrying after #{retryAfter} seconds.")
@@ -358,8 +362,11 @@ nextTask = (currentRequest) ->
       )
     else
       util.log("#{currentRequest.session.deviceId}: Started task #{task.name}(#{task._id})")
+      updateDevice.clearUpdatesQueue(currentRequest.session.deviceId, (err) ->
+        throw err if err
         runTask(currentRequest, task, {})
       )
+  )
 
 
 getSession = (httpRequest, callback) ->

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)
 
 

lib/update-device.coffee

@@ -25,14 +25,39 @@ 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
+  # Used to find parameters that no longer exist in device
+  presentPaths = null
+  projection = null
+
+  # Keep track of alias changes to clear cache if needed
+  newAliases = null
+
+  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 updatesBatch.parameterValues?
+      for p in updatesBatch.parameterValues
         origValue = p[1]
         v = normalize(p[0], origValue)
 
@@ -46,19 +71,19 @@ updateDevice = (deviceId, actions, callback) ->
         if p[2]?
           update['$set']["#{path}_type"] = p[2]
 
-  if actions.deletedObjects?
-    for p in actions.deletedObjects
+    if updatesBatch.deletedObjects?
+      for p in updatesBatch.deletedObjects
         update['$unset'][p] = 1
 
-  if actions.instanceName?
-    for i in actions.instanceName
+    if updatesBatch.instanceName?
+      for i in updatesBatch.instanceName
         update['$set']["#{i[0]}._name"] = i[1]
 
-  if actions.parameterNames?
-    projection = {} # fetch only necessary parameters
-    presentPaths = {} # to be compared with to find no longer existing parameters
+    if updatesBatch.parameterNames?
+      presentPaths ?= {}
+      projection ?= {}
 
-    for p in actions.parameterNames
+      for p in updatesBatch.parameterNames
         if OBJECT_REGEX.test(p[0])
           path = p[0].slice(0, -1)
           if INSTANCE_REGEX.test(p[0])
@@ -72,7 +97,7 @@ updateDevice = (deviceId, actions, callback) ->
 
         # Add parameter (sans any dot suffix) if has an alias
         if (a = parameters.getAlias(path))?
-        newAliases = {} if not newAliases?
+          newAliases ?= {}
           newAliases[path] = a
 
         projection[path] = 1
@@ -84,62 +109,88 @@ updateDevice = (deviceId, actions, callback) ->
           presentPaths[p[0][0...i]] = true
           i = p[0].indexOf('.', i + 1)
 
-    mongoQuery.optimizeProjection(projection)
-
-  if actions.customCommands?
-    for p in actions.customCommands
+    if updatesBatch.customCommands?
+      for p in updatesBatch.customCommands
         commandName = p[0]
         update['$set']["_customCommands.#{commandName}._value"] = p[1]
         update['$set']["_customCommands.#{commandName}._timestamp"] = now
 
-  if actions.set?
-    common.extend(update['$set'], actions.set)
+  processBatches = (cb) ->
+    if not commitQueue?
+      processBatch(deviceUpdates) if deviceUpdates?
+      return cb()
 
-  f = () ->
+    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
-      callback?()
+          return callback()
+      )
+    else
+      return callback()
 
+  processBatches(() ->
     if projection?
+      mongoQuery.optimizeProjection(projection)
+
+      # 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) ->
-      throw err if err
+        return callback(err) if err
 
         recursive = (obj, prefix) ->
-        for k,v of obj
+          for k, v of obj
             p = prefix + k
+
+            # 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
 
-          # Remove unset operations for values that don't exist
-          delete update['$unset']["#{p}._orig"] if not v._orig?
-
             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 k, v of projection
+        for path of projection
           root = device
           rootPath = ''
           for p in path.split('.')
@@ -150,11 +201,21 @@ updateDevice = (deviceId, actions, callback) ->
           if root
             recursive(root, rootPath)
           else
-          if not presentPaths[rootPath.slice(0, -1)]
+            if not presentPaths[rootPath.slice(0, -1)]?
               update['$unset'][rootPath.slice(0, -1)] = 1
-      f()
+
+          # Remove unnecessary unset updates
+          for k, v of update['$unset']
+            if common.startsWith(k, path + '.') and v != 2
+              delete update['$unset'][k]
+
+        commit()
       )
     else
-    f()
+      commit()
+  )
 
-module.exports = updateDevice
+
+exports.queueUpdates = queueUpdates
+exports.clearUpdatesQueue = clearUpdatesQueue
+exports.commitUpdates = commitUpdates