From 2e4a524d900bc3dfff8595180ddc402f6d629dd1 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Sat, 2 Nov 2013 20:55:31 +0300 Subject: [PATCH] Replaced memcached with redis. --- acs.coffee | 27 ++++++++++++++++----------- api.coffee | 29 ++++++++++++++--------------- config.coffee | 2 +- config/config-sample.json | 2 +- db.coffee | 34 ++++++++++++++-------------------- package.json | 3 ++- 6 files changed, 48 insertions(+), 49 deletions(-) diff --git a/acs.coffee b/acs.coffee index 6c9a786..0b48bb1 100644 --- a/acs.coffee +++ b/acs.coffee @@ -156,7 +156,7 @@ runTask = (currentRequest, task, methodResponse) -> switch status & ~tasks.STATUS_SAVE when tasks.STATUS_OK 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 res = tr069.response(task._id, cwmpResponse) writeResponse(currentRequest, res) @@ -173,7 +173,8 @@ runTask = (currentRequest, task, methodResponse) -> util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})") db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> 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) ) ) @@ -200,9 +201,10 @@ isTaskExpired = (task) -> assertPresets = (currentRequest) -> - db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) -> - presetsHash = results['presets_hash'] - devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"] + db.redisClient.mget("#{currentRequest.deviceId}_presets_hash", 'presets_hash', (err, res) -> + throw err if err + devicePresetsHash = res[0] + presetsHash = res[1] if devicePresetsHash? and devicePresetsHash == presetsHash # no discrepancy, return empty response res = tr069.response(null, {}) @@ -211,12 +213,14 @@ assertPresets = (currentRequest) -> db.getPresets((allPresets, allObjects) -> if not presetsHash? 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.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 @@ -396,7 +400,8 @@ listener = (httpRequest, httpResponse) -> else if cwmpRequest.methodResponse? taskId = cwmpRequest.id - db.getTask(taskId, (task) -> + db.getTask(taskId, (err, task) -> + throw err if err if not task nextTask(currentRequest) else @@ -410,7 +415,8 @@ listener = (httpRequest, httpResponse) -> writeResponse(currentRequest, res) return - db.getTask(taskId, (task) -> + db.getTask(taskId, (err, task) -> + throw err if err if not task nextTask(currentRequest) else @@ -425,8 +431,7 @@ cluster = require 'cluster' numCPUs = require('os').cpus().length if cluster.isMaster - db.memcached.flush(() -> - ) + db.redisClient.flushdb() cluster.on('listening', (worker, address) -> util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}") ) diff --git a/api.coffee b/api.coffee index 5d3c97a..857c08b 100644 --- a/api.coffee +++ b/api.coffee @@ -70,9 +70,8 @@ else preset._id = presetName db.presetsCollection.save(preset, (err) -> - db.memcached.del('presets', (err, res) -> - ) - db.memcached.del('presets_hash', (err, res) -> + db.redisClient.del('presets', 'presets_hash', (err) -> + throw err if err ) if err response.writeHead(500) @@ -83,9 +82,8 @@ else ) else if request.method == 'DELETE' db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) -> - db.memcached.del('presets', (err, res) -> - ) - db.memcached.del('presets_hash', (err, res) -> + db.redisClient.del('presets', 'presets_hash', (err) -> + throw err if err ) if err response.writeHead(500) @@ -104,9 +102,8 @@ else object._id = objectName db.objectsCollection.save(object, (err) -> - db.memcached.del('objects', (err, res) -> - ) - db.memcached.del('presets_hash', (err, res) -> + db.redisClient.del('objects', 'presets_hash', (err) -> + throw err if err ) if err response.writeHead(500) @@ -117,9 +114,8 @@ else ) else if request.method == 'DELETE' db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) -> - db.memcached.del('objects', (err, res) -> - ) - db.memcached.del('presets_hash', (err, res) -> + db.redisClient.del('objects', 'presets_hash', (err) -> + throw err if err ) if err response.writeHead(500) @@ -137,7 +133,8 @@ else tag = querystring.unescape(r[2]) if request.method == 'POST' 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 response.writeHead(500) @@ -148,7 +145,8 @@ else ) else if request.method == 'DELETE' 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 response.writeHead(500) @@ -167,7 +165,8 @@ else task = JSON.parse(body) task.device = deviceId 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 response.writeHead(500) diff --git a/config.coffee b/config.coffee index 4b2781c..f181f76 100644 --- a/config.coffee +++ b/config.coffee @@ -13,7 +13,7 @@ exports.CACHE_DURATION = 60 # in seconds exports.PRESETS_CACHE_DURATION = 86400 exports.PRESETS_TIME_PADDING = 1 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.LOG_INFORMS = false exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true} diff --git a/config/config-sample.json b/config/config-sample.json index 6367ab3..c6223a0 100644 --- a/config/config-sample.json +++ b/config/config-sample.json @@ -1,7 +1,7 @@ { "DATABASE_NAME" : "genieacs", "MONGODB_SOCKET" : "/tmp/mongodb-27017.sock", - "MEMCACHED_SOCKET" : "/tmp/memcached.sock", + "REDIS_SOCKET" : "/tmp/redis.sock", "ACS_INTERFACE" : "0.0.0.0", "ACS_PORT" : 7547, "ACS_HTTPS_INTERFACE" : "0.0.0.0", diff --git a/db.coffee b/db.coffee index 9663691..e069bce 100644 --- a/db.coffee +++ b/db.coffee @@ -1,7 +1,6 @@ config = require './config' mongodb = require 'mongodb' -Memcached = require 'memcached' -memcached = new Memcached(config.MEMCACHED_SOCKET, {maxExpiration : 86400, retries : 1}) +redisClient = require('redis').createClient(config.REDIS_SOCKET) tasksCollection = null devicesCollection = null @@ -40,27 +39,22 @@ mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABAS 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) - memcached.getMulti([tid], (err, data) -> - task = data[tid] - if not task? - tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) -> - # TODO use err parameter in callback - throw new Error(err) if err? - callback(task) - ) + redisClient.get(tid, (err, res) -> + return callback(err) if err + if res? + callback(err, JSON.parse(res)) else - callback(task) + tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) -> + callback(err, task) + ) ) getPresets = (callback) -> - memcached.get(['presets', 'objects'], (err, res) -> - presets = res.presets - objects = res.objects + redisClient.mget('presets', 'objects', (err, res) -> + presets = JSON.parse(res[0]) + objects = JSON.parse(res[1]) if presets and objects callback(presets, objects) return @@ -69,7 +63,7 @@ getPresets = (callback) -> throw err if err 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 ) @@ -79,12 +73,12 @@ getPresets = (callback) -> for i in o 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 ) ) -exports.memcached = memcached +exports.redisClient = redisClient exports.getTask = getTask exports.getPresets = getPresets diff --git a/package.json b/package.json index ed2648d..b23c854 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,8 @@ }, "dependencies": { "libxmljs": "~0.8.1", - "memcached": "~0.2.5", + "redis": "~0.9.0", + "hiredis": "~0.1.15", "mongodb": "~1.3.18" }, "devDependencies": {