Performance improvements

acs.coffee

@@ -20,8 +20,18 @@ arrayToHash = (arr) ->
   return hash
 
 dbserver = new mongo.Server('localhost', 27017, {auto_reconnect: true})
+tasksCollection = null
+devicesCollection = null
 db = new mongo.Db(DATABASE_NAME, dbserver, {native_parser:true})
 db.open( (err, db) ->
+  db.collection('tasks', (err, collection) ->
+    tasksCollection = collection
+    tasksCollection.ensureIndex({device: 1, timestamp: 1})
+  )
+
+  db.collection('devices', (err, collection) ->
+    devicesCollection = collection
+  )
 )
 
 
@@ -29,40 +39,34 @@ updateDevice = (deviceId, actions) ->
   return if not actions?
   if actions.parameterValues?
     params = arrayToHash(actions.parameterValues)
-    db.collection('devices', (err, collection) ->
+    devicesCollection.update({'_id' : deviceId}, {'$set' : params}, {upsert: true})
-      collection.update({'_id' : deviceId}, {'$set' : params}, {upsert: true})
-    )
 
 
 nextTask = (sessionId, deviceId, response, cookies) ->
   if err?
     throw 'ERROR: Cannot open database connection'
 
-  db.collection('tasks', (err, collection) ->
+  cur = tasksCollection.find({'device' : deviceId}).sort(['timestamp']).limit(1)
-    cur = collection.find({'device' : deviceId}).sort(['timestamp']).limit(1)
+  cur.nextObject( (err, task) ->
-    cur.nextObject( (err, task) ->
+    reqParams = {}
-      reqParams = {}
+    resParams = {}
-      resParams = {}
+    if not task
-      if not task
+      # add init task and end of queue
-        # add init task and end of queue
+      tasksCollection.insert({'name' : 'endOfQueue', 'device': deviceId, 'timestamp' : mongo.Timestamp(new Date(2020, 1, 1))})
-        db.collection('tasks', (err, collection) ->
+      task = {'name' : 'init', 'timestamp' : mongo.Timestamp()}
-          collection.insert({'name' : 'endOfQueue', 'device': deviceId, 'timestamp' : mongo.Timestamp(new Date(2020, 1, 1))})
+    else if task.name == 'endOfQueue'
-        )
+      tr069.response(sessionId, response, resParams, cookies)
-        task = {'name' : 'init', 'timestamp' : mongo.Timestamp()}
-      else if task.name == 'endOfQueue'
-        tr069.response(sessionId, response, resParams, cookies)
-        return
-
-      actions = tasks.task(task, reqParams, resParams)
-      updateDevice(deviceId, actions)
-      # TODO in rare cases a task could finish without sending response
-      collection.save(task, (err) ->
-        console.log("Device #{deviceId}: Running task #{task.name}(#{task._id})")
-        cookies.task = String(task._id)
-        tr069.response(sessionId, response, resParams, cookies)
-      )
       return
+
+    actions = tasks.task(task, reqParams, resParams)
+    updateDevice(deviceId, actions)
+    # TODO in rare cases a task could finish without sending response
+    tasksCollection.save(task, (err) ->
+      console.log("Device #{deviceId}: Running task #{task.name}(#{task._id})")
+      cookies.task = String(task._id)
+      tr069.response(sessionId, response, resParams, cookies)
     )
+    return
   )
 
 
@@ -94,10 +98,8 @@ else
         resParams.inform = true
         cookies.ID = sessionId = reqParams.sessionId
         cookies.DeviceId = deviceId = reqParams.deviceId.SerialNumber
-        db.collection('devices', (err, collection) ->
+        devicesCollection.update({'_id' : deviceId}, {'$set' : arrayToHash(reqParams.informParameterValues)}, {upsert: true}, (err) ->
-          collection.update({'_id' : deviceId}, {'$set' : arrayToHash(reqParams.informParameterValues)}, {upsert: true}, (err) ->
+          tr069.response(reqParams.sessionId, response, resParams, cookies)
-            tr069.response(reqParams.sessionId, response, resParams, cookies)
-          )
         )
         return
 
@@ -109,22 +111,19 @@ else
         nextTask(sessionId, deviceId, response, cookies)
         return
 
-      db.collection('tasks', (err, collection) ->
+      tasksCollection.findOne({'_id' : mongo.ObjectID(taskId)}, (err, task) ->
-        cur = collection.find({'_id' : mongo.ObjectID(taskId)})
+        actions = tasks.task(task, reqParams, resParams)
-        cur.nextObject( (err, task) ->
+        updateDevice(deviceId, actions)
-          actions = tasks.task(task, reqParams, resParams)
-          updateDevice(deviceId, actions)
 
-          if Object.keys(resParams).length > 0
+        if Object.keys(resParams).length > 0
-            collection.save(task)
+          tasksCollection.save(task)
-            tr069.response(reqParams.sessionId, response, resParams, cookies)
+          tr069.response(reqParams.sessionId, response, resParams, cookies)
-          else
+        else
-            console.log("Device #{deviceId}: Completed task #{task.name}(#{task._id})")
+          console.log("Device #{deviceId}: Completed task #{task.name}(#{task._id})")
-            collection.remove({'_id' : task._id}, (err, removed) ->
+          tasksCollection.remove({'_id' : task._id}, (err, removed) ->
-              cookies.task = undefined
+            cookies.task = undefined
-              nextTask(sessionId, deviceId, response, cookies)
+            nextTask(sessionId, deviceId, response, cookies)
-            )
+          )
-        )
       )
 
   server.listen PORT

test.coffee

@@ -1,6 +1,9 @@
 http = require 'http'
 fs = require 'fs'
 
+cluster = require 'cluster'
+numCPUs = require('os').cpus().length
+
 TEMPLATE_SERIAL = '8KA8WA1151100043'
 TEMPLATE_TASKID = '50731c1bfb2b97e764000001'
 
@@ -21,7 +24,8 @@ sendRequest = (serial, taskId, seq = 0) ->
   headers['Content-Length'] = body.length
 
   options = {
-    host: '',
+    #host: '172.17.32.59',
+    host: '127.0.0.1',
     port: 1337,
     method: 'POST',
     headers: headers,
@@ -42,5 +46,15 @@ sendRequest = (serial, taskId, seq = 0) ->
 t = new Date()
 console.log("Start at #{t}")
 
-for i in [0 .. 500]
+if cluster.isMaster
-  sendRequest(TEMPLATE_SERIAL + String(i), TEMPLATE_TASKID, 0)
+  for i in [1 .. numCPUs]
+    cluster.fork()
+  cluster.on('exit', (worker, code, signal) ->
+    console.log('worker ' + worker.process.pid + ' died')
+  )
+else
+  workerid = Math.floor(Math.random() * 1000)
+  
+  for i in [1 .. 1000]
+    sendRequest("device" + String(workerid) + String(i), TEMPLATE_TASKID, 0)
+