Replaced memcached with redis.

This commit is contained in:
Zaid Abdulla committed 2013-11-02 20:55:31 +03:00
1 parent 5a2648c4d0
commit 2e4a524d90
6 files changed
+48 -49

No files matched your search

+16 -11
View File
@@ -156,7 +156,7 @@ runTask = (currentRequest, task, methodResponse) ->
switch status & ~tasks.STATUS_SAVE switch status & ~tasks.STATUS_SAVE
when tasks.STATUS_OK when tasks.STATUS_OK
f = () -> f = () ->
db.memcached.set(String(task._id), task, config.CACHE_DURATION, (err, res) -> db.redisClient.setex(String(task._id), config.CACHE_DURATION, JSON.stringify(task), (err) ->
throw err if err throw err if err
res = tr069.response(task._id, cwmpResponse) res = tr069.response(task._id, cwmpResponse)
writeResponse(currentRequest, res) writeResponse(currentRequest, res)
@@ -173,7 +173,8 @@ runTask = (currentRequest, task, methodResponse) ->
util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})") util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) ->
throw err if err throw err if err
db.memcached.del(String(task._id), (err, res) -> db.redisClient.del(String(task._id), (err, res) ->
throw err if err
nextTask(currentRequest) nextTask(currentRequest)
) )
) )
@@ -200,9 +201,10 @@ isTaskExpired = (task) ->
assertPresets = (currentRequest) -> assertPresets = (currentRequest) ->
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) -> db.redisClient.mget("#{currentRequest.deviceId}_presets_hash", 'presets_hash', (err, res) ->
presetsHash = results['presets_hash'] throw err if err
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"] devicePresetsHash = res[0]
presetsHash = res[1]
if devicePresetsHash? and devicePresetsHash == presetsHash if devicePresetsHash? and devicePresetsHash == presetsHash
# no discrepancy, return empty response # no discrepancy, return empty response
res = tr069.response(null, {}) res = tr069.response(null, {})
@@ -211,12 +213,14 @@ assertPresets = (currentRequest) ->
db.getPresets((allPresets, allObjects) -> db.getPresets((allPresets, allObjects) ->
if not presetsHash? if not presetsHash?
presetsHash = presets.calculatePresetsHash(allPresets, allObjects) presetsHash = presets.calculatePresetsHash(allPresets, allObjects)
db.memcached.set('presets_hash', presetsHash, config.PRESETS_CACHE_DURATION, (err, res) -> db.redisClient.setex('presets_hash', config.PRESETS_CACHE_DURATION, presetsHash, (err, res) ->
throw err if err
) )
presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) -> presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) ->
presets.processDevicePreset(currentRequest.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) -> presets.processDevicePreset(currentRequest.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) ->
db.memcached.set("#{currentRequest.deviceId}_presets_hash", presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) -> db.redisClient.setex("#{currentRequest.deviceId}_presets_hash", expiry - config.PRESETS_TIME_PADDING, presetsHash, (err, res) ->
throw err if err
) )
if addTags.length + deleteTags.length + taskList.length if addTags.length + deleteTags.length + taskList.length
@@ -396,7 +400,8 @@ listener = (httpRequest, httpResponse) ->
else if cwmpRequest.methodResponse? else if cwmpRequest.methodResponse?
taskId = cwmpRequest.id taskId = cwmpRequest.id
db.getTask(taskId, (task) -> db.getTask(taskId, (err, task) ->
throw err if err
if not task if not task
nextTask(currentRequest) nextTask(currentRequest)
else else
@@ -410,7 +415,8 @@ listener = (httpRequest, httpResponse) ->
writeResponse(currentRequest, res) writeResponse(currentRequest, res)
return return
db.getTask(taskId, (task) -> db.getTask(taskId, (err, task) ->
throw err if err
if not task if not task
nextTask(currentRequest) nextTask(currentRequest)
else else
@@ -425,8 +431,7 @@ cluster = require 'cluster'
numCPUs = require('os').cpus().length numCPUs = require('os').cpus().length
if cluster.isMaster if cluster.isMaster
db.memcached.flush(() -> db.redisClient.flushdb()
)
cluster.on('listening', (worker, address) -> cluster.on('listening', (worker, address) ->
util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}") util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
) )
+14 -15
View File
@@ -70,9 +70,8 @@ else
preset._id = presetName preset._id = presetName
db.presetsCollection.save(preset, (err) -> db.presetsCollection.save(preset, (err) ->
db.memcached.del('presets', (err, res) -> db.redisClient.del('presets', 'presets_hash', (err) ->
) throw err if err
db.memcached.del('presets_hash', (err, res) ->
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -83,9 +82,8 @@ else
) )
else if request.method == 'DELETE' else if request.method == 'DELETE'
db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) -> db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) ->
db.memcached.del('presets', (err, res) -> db.redisClient.del('presets', 'presets_hash', (err) ->
) throw err if err
db.memcached.del('presets_hash', (err, res) ->
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -104,9 +102,8 @@ else
object._id = objectName object._id = objectName
db.objectsCollection.save(object, (err) -> db.objectsCollection.save(object, (err) ->
db.memcached.del('objects', (err, res) -> db.redisClient.del('objects', 'presets_hash', (err) ->
) throw err if err
db.memcached.del('presets_hash', (err, res) ->
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -117,9 +114,8 @@ else
) )
else if request.method == 'DELETE' else if request.method == 'DELETE'
db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) -> db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) ->
db.memcached.del('objects', (err, res) -> db.redisClient.del('objects', 'presets_hash', (err) ->
) throw err if err
db.memcached.del('presets_hash', (err, res) ->
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -137,7 +133,8 @@ else
tag = querystring.unescape(r[2]) tag = querystring.unescape(r[2])
if request.method == 'POST' if request.method == 'POST'
db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) -> db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) ->
db.memcached.del("#{deviceId}_presets_hash", (err, res) -> db.redisClient.del("#{deviceId}_presets_hash", (err) ->
throw err if err
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -148,7 +145,8 @@ else
) )
else if request.method == 'DELETE' else if request.method == 'DELETE'
db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) -> db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) ->
db.memcached.del("#{deviceId}_presets_hash", (err, res) -> db.redisClient.del("#{deviceId}_presets_hash", (err) ->
throw err if err
) )
if err if err
response.writeHead(500) response.writeHead(500)
@@ -167,7 +165,8 @@ else
task = JSON.parse(body) task = JSON.parse(body)
task.device = deviceId task.device = deviceId
apiFunctions.insertTasks(task, (err) -> apiFunctions.insertTasks(task, (err) ->
db.memcached.del("#{deviceId}_presets_hash", (err, res) -> db.redisClient.del("#{deviceId}_presets_hash", (err) ->
throw err if err
) )
if err if err
response.writeHead(500) response.writeHead(500)
+1 -1
View File
@@ -13,7 +13,7 @@ exports.CACHE_DURATION = 60 # in seconds
exports.PRESETS_CACHE_DURATION = 86400 exports.PRESETS_CACHE_DURATION = 86400
exports.PRESETS_TIME_PADDING = 1 exports.PRESETS_TIME_PADDING = 1
exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock' exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock'
exports.MEMCACHED_SOCKET = '/tmp/memcached.sock' exports.REDIS_SOCKET = '/tmp/redis.sock'
exports.WORKER_RESPAWN_TIME = 60000 exports.WORKER_RESPAWN_TIME = 60000
exports.LOG_INFORMS = false exports.LOG_INFORMS = false
exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true} exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true}
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"DATABASE_NAME" : "genieacs", "DATABASE_NAME" : "genieacs",
"MONGODB_SOCKET" : "/tmp/mongodb-27017.sock", "MONGODB_SOCKET" : "/tmp/mongodb-27017.sock",
"MEMCACHED_SOCKET" : "/tmp/memcached.sock", "REDIS_SOCKET" : "/tmp/redis.sock",
"ACS_INTERFACE" : "0.0.0.0", "ACS_INTERFACE" : "0.0.0.0",
"ACS_PORT" : 7547, "ACS_PORT" : 7547,
"ACS_HTTPS_INTERFACE" : "0.0.0.0", "ACS_HTTPS_INTERFACE" : "0.0.0.0",
+14 -20
View File
@@ -1,7 +1,6 @@
config = require './config' config = require './config'
mongodb = require 'mongodb' mongodb = require 'mongodb'
Memcached = require 'memcached' redisClient = require('redis').createClient(config.REDIS_SOCKET)
memcached = new Memcached(config.MEMCACHED_SOCKET, {maxExpiration : 86400, retries : 1})
tasksCollection = null tasksCollection = null
devicesCollection = null devicesCollection = null
@@ -40,27 +39,22 @@ mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABAS
getTask = (taskId, callback) -> getTask = (taskId, callback) ->
# TODO using getMulti instead of get because of possible bug
# in node-memcached where it sometimes returns incorrect values
# when under heavy load. getMulti works fine.
tid = String(taskId) tid = String(taskId)
memcached.getMulti([tid], (err, data) -> redisClient.get(tid, (err, res) ->
task = data[tid] return callback(err) if err
if not task? if res?
tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) -> callback(err, JSON.parse(res))
# TODO use err parameter in callback
throw new Error(err) if err?
callback(task)
)
else else
callback(task) tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) ->
callback(err, task)
)
) )
getPresets = (callback) -> getPresets = (callback) ->
memcached.get(['presets', 'objects'], (err, res) -> redisClient.mget('presets', 'objects', (err, res) ->
presets = res.presets presets = JSON.parse(res[0])
objects = res.objects objects = JSON.parse(res[1])
if presets and objects if presets and objects
callback(presets, objects) callback(presets, objects)
return return
@@ -69,7 +63,7 @@ getPresets = (callback) ->
throw err if err throw err if err
presets = p presets = p
memcached.set('presets', presets, config.PRESETS_CACHE_DURATION) redisClient.setex('presets', config.PRESETS_CACHE_DURATION, JSON.stringify(presets))
callback(presets, objects) if objects callback(presets, objects) if objects
) )
@@ -79,12 +73,12 @@ getPresets = (callback) ->
for i in o for i in o
objects[i._id] = i objects[i._id] = i
memcached.set('objects', objects, config.PRESETS_CACHE_DURATION) redisClient.setex('objects', config.PRESETS_CACHE_DURATION, JSON.stringify(objects))
callback(presets, objects) if presets callback(presets, objects) if presets
) )
) )
exports.memcached = memcached exports.redisClient = redisClient
exports.getTask = getTask exports.getTask = getTask
exports.getPresets = getPresets exports.getPresets = getPresets
+2 -1
View File
@@ -7,7 +7,8 @@
}, },
"dependencies": { "dependencies": {
"libxmljs": "~0.8.1", "libxmljs": "~0.8.1",
"memcached": "~0.2.5", "redis": "~0.9.0",
"hiredis": "~0.1.15",
"mongodb": "~1.3.18" "mongodb": "~1.3.18"
}, },
"devDependencies": { "devDependencies": {