add cache to redis (#6917)

* add cache to redis

move out cache module from mongo

* fix redis tests

* add callback noop

* fix typo

* del cache on field delete

* make redis/mongo caches separate

src/database/cache.js

@@ -0,0 +1,56 @@
+'use strict';
+
+module.exports.create = function (name) {
+	var LRU = require('lru-cache');
+	var pubsub = require('../pubsub');
+
+	var cache = LRU({
+		max: 20000,
+		length: function () { return 1; },
+		maxAge: 0,
+	});
+
+	cache.misses = 0;
+	cache.hits = 0;
+
+	pubsub.on(name + ':hash:cache:del', function (keys) {
+		keys.forEach(key => cache.del(key));
+	});
+
+	pubsub.on(name + ':hash:cache:reset', function () {
+		cache.reset();
+	});
+
+	cache.delObjectCache = function (keys) {
+		if (!Array.isArray(keys)) {
+			keys = [keys];
+		}
+		pubsub.publish(name + ':hash:cache:del', keys);
+		keys.forEach(key => cache.del(key));
+	};
+
+	cache.resetObjectCache = function () {
+		pubsub.publish(name + ':hash:cache:reset');
+		cache.reset();
+	};
+
+	cache.getUnCachedKeys = function (keys, cachedData) {
+		let data;
+		let isCached;
+		const unCachedKeys = keys.filter(function (key) {
+			data = cache.get(key);
+			isCached = data !== undefined;
+			if (isCached) {
+				cachedData[key] = data;
+			}
+			return !isCached;
+		});
+
+		var hits = keys.length - unCachedKeys.length;
+		var misses = keys.length - hits;
+		cache.hits += hits;
+		cache.misses += misses;
+		return unCachedKeys;
+	};
+	return cache;
+};

src/database/mongo/hash.js

@@ -1,43 +1,16 @@
 'use strict';
 
 var async = require('async');
-var pubsub = require('../../pubsub');
+
 
 module.exports = function (db, module) {
 	var helpers = module.helpers.mongo;
 
-	var LRU = require('lru-cache');
 	var _ = require('lodash');
+	const cache = require('../cache').create('mongo');
 
-	var cache = LRU({
-		max: 10000,
-		length: function () { return 1; },
-		maxAge: 0,
-	});
-
-	cache.misses = 0;
-	cache.hits = 0;
 	module.objectCache = cache;
 
-	pubsub.on('mongo:hash:cache:del', function (key) {
-		cache.del(key);
-	});
-
-	pubsub.on('mongo:hash:cache:reset', function () {
-		cache.reset();
-	});
-
-	module.delObjectCache = function (key) {
-		pubsub.publish('mongo:hash:cache:del', key);
-		cache.del(key);
-	};
-
-	module.resetObjectCache = function () {
-		pubsub.publish('mongo:hash:cache:reset');
-		cache.reset();
-	};
-
-
 	module.setObject = function (key, data, callback) {
 		callback = callback || helpers.noop;
 		if (!key || !data) {
@@ -50,7 +23,7 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.delObjectCache(key);
+			cache.delObjectCache(key);
 			callback();
 		});
 	};
@@ -82,35 +55,22 @@ module.exports = function (db, module) {
 	module.getObjects = function (keys, callback) {
 		var cachedData = {};
 		function getFromCache() {
-			process.nextTick(callback, null, keys.map(function (key) {
+			process.nextTick(callback, null, keys.map(key => _.clone(cachedData[key])));
-				return _.clone(cachedData[key]);
-			}));
 		}
 
 		if (!Array.isArray(keys) || !keys.length) {
 			return callback(null, []);
 		}
 
-		var nonCachedKeys = keys.filter(function (key) {
+		const unCachedKeys = cache.getUnCachedKeys(keys, cachedData);
-			var data = cache.get(key);
-			if (data !== undefined) {
-				cachedData[key] = data;
-			}
-			return data === undefined;
-		});
 
-		var hits = keys.length - nonCachedKeys.length;
+		if (!unCachedKeys.length) {
-		var misses = keys.length - hits;
-		cache.hits += hits;
-		cache.misses += misses;
-
-		if (!nonCachedKeys.length) {
 			return getFromCache();
 		}
 
-		var query = { _key: { $in: nonCachedKeys } };
+		var query = { _key: { $in: unCachedKeys } };
-		if (nonCachedKeys.length === 1) {
+		if (unCachedKeys.length === 1) {
-			query._key = nonCachedKeys[0];
+			query._key = unCachedKeys[0];
 		}
 		db.collection('objects').find(query, { projection: { _id: 0 } }).toArray(function (err, data) {
 			if (err) {
@@ -118,7 +78,7 @@ module.exports = function (db, module) {
 			}
 
 			var map = helpers.toMap(data);
-			nonCachedKeys.forEach(function (key) {
+			unCachedKeys.forEach(function (key) {
 				cachedData[key] = map[key] || null;
 				cache.set(key, cachedData[key]);
 			});
@@ -143,16 +103,8 @@ module.exports = function (db, module) {
 		if (!key) {
 			return callback();
 		}
-		module.getObject(key, function (err, item) {
+		module.getObjectsFields([key], fields, function (err, data) {
-			if (err) {
+			callback(err, data ? data[0] : null);
-				return callback(err);
-			}
-			item = item || {};
-			var result = {};
-			for (var i = 0; i < fields.length; i += 1) {
-				result[fields[i]] = item[fields[i]] !== undefined ? item[fields[i]] : null;
-			}
-			callback(null, result);
 		});
 	};
 
@@ -168,17 +120,14 @@ module.exports = function (db, module) {
 				items = [];
 			}
 
-			var returnData = [];
+			const returnData = items.map((item) => {
-			var item;
+				item = item || {};
-			var result;
+				const result = {};
-			for (var i = 0; i < keys.length; i += 1) {
+				fields.forEach((field) => {
-				item = items[i] || {};
+					result[field] = item[field] !== undefined ? item[field] : null;
-				result = {};
+				});
-				for (var k = 0; k < fields.length; k += 1) {
+				return result;
-					result[fields[k]] = item[fields[k]] !== undefined ? item[fields[k]] : null;
+			});
-				}
-				returnData.push(result);
-			}
 
 			callback(null, returnData);
 		});
@@ -267,7 +216,7 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.delObjectCache(key);
+			cache.delObjectCache(key);
 			callback();
 		});
 	};
@@ -304,9 +253,7 @@ module.exports = function (db, module) {
 					});
 				},
 				function (next) {
-					key.forEach(function (key) {
+					cache.delObjectCache(key);
-						module.delObjectCache(key);
-					});
 
 					module.getObjectsFields(key, [field], next);
 				},
@@ -325,7 +272,7 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.delObjectCache(key);
+			cache.delObjectCache(key);
 			callback(null, result && result.value ? result.value[field] : null);
 		});
 	};

src/database/mongo/main.js

@@ -16,7 +16,7 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.resetObjectCache();
+			module.objectCache.resetObjectCache();
 			callback();
 		});
 	};
@@ -39,7 +39,7 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.delObjectCache(key);
+			module.objectCache.delObjectCache(key);
 			callback();
 		});
 	};
@@ -54,9 +54,7 @@ module.exports = function (db, module) {
 				return callback(err);
 			}
 
-			keys.forEach(function (key) {
+			module.objectCache.delObjectCache(keys);
-				module.delObjectCache(key);
-			});
 
 			callback(null);
 		});
@@ -66,7 +64,8 @@ module.exports = function (db, module) {
 		if (!key) {
 			return callback();
 		}
-		module.getObject(key, function (err, objectData) {
+
+		db.collection('objects').findOne({ _key: key }, { projection: { _id: 0 } }, function (err, objectData) {
 			if (err) {
 				return callback(err);
 			}
@@ -108,8 +107,8 @@ module.exports = function (db, module) {
 			if (err) {
 				return callback(err);
 			}
-			module.delObjectCache(oldKey);
+			module.objectCache.delObjectCache(oldKey);
-			module.delObjectCache(newKey);
+			module.objectCache.delObjectCache(newKey);
 			callback();
 		});
 	};

src/database/redis/hash.js

@@ -3,6 +3,13 @@
 module.exports = function (redisClient, module) {
 	var helpers = module.helpers.redis;
 
+	const async = require('async');
+	const _ = require('lodash');
+
+	const cache = require('../cache').create('redis');
+
+	module.objectCache = cache;
+
 	module.setObject = function (key, data, callback) {
 		callback = callback || function () {};
 		if (!key || !data) {
@@ -23,7 +30,11 @@ module.exports = function (redisClient, module) {
 			return callback();
 		}
 		redisClient.hmset(key, data, function (err) {
-			callback(err);
+			if (err) {
+				return callback(err);
+			}
+			cache.delObjectCache(key);
+			callback();
 		});
 	};
 
@@ -33,16 +44,51 @@ module.exports = function (redisClient, module) {
 			return callback();
 		}
 		redisClient.hset(key, field, value, function (err) {
-			callback(err);
+			if (err) {
+				return callback(err);
+			}
+			cache.delObjectCache(key);
+			callback();
 		});
 	};
 
 	module.getObject = function (key, callback) {
-		redisClient.hgetall(key, callback);
+		module.getObjects([key], function (err, data) {
+			if (err) {
+				return callback(err);
+			}
+			callback(null, data && data.length ? data[0] : null);
+		});
 	};
 
 	module.getObjects = function (keys, callback) {
-		helpers.multiKeys(redisClient, 'hgetall', keys, callback);
+		var cachedData = {};
+		function getFromCache() {
+			process.nextTick(callback, null, keys.map(key => _.clone(cachedData[key])));
+		}
+
+		const unCachedKeys = cache.getUnCachedKeys(keys, cachedData);
+		if (!unCachedKeys.length) {
+			return getFromCache();
+		}
+
+		async.waterfall([
+			function (next) {
+				if (unCachedKeys.length > 1) {
+					helpers.multiKeys(redisClient, 'hgetall', unCachedKeys, next);
+				} else {
+					redisClient.hgetall(unCachedKeys[0], (err, data) => next(err, [data]));
+				}
+			},
+			function (data) {
+				unCachedKeys.forEach(function (key, i) {
+					cachedData[key] = data[i] || null;
+					cache.set(key, cachedData[key]);
+				});
+
+				getFromCache();
+			},
+		], callback);
 	};
 
 	module.getObjectField = function (key, field, callback) {
@@ -61,28 +107,20 @@ module.exports = function (redisClient, module) {
 		if (!Array.isArray(fields) || !fields.length) {
 			return callback(null, keys.map(function () { return {}; }));
 		}
-		var multi = redisClient.multi();
+		module.getObjects(keys, function (err, items) {
-
-		for (var x = 0; x < keys.length; x += 1) {
-			multi.hmget.apply(multi, [keys[x]].concat(fields));
-		}
-
-		function makeObject(array) {
-			var obj = {};
-
-			for (var i = 0, ii = fields.length; i < ii; i += 1) {
-				obj[fields[i]] = array[i];
-			}
-			return obj;
-		}
-
-		multi.exec(function (err, results) {
 			if (err) {
 				return callback(err);
 			}
+			const returnData = items.map((item) => {
+				item = item || {};
+				const result = {};
+				fields.forEach((field) => {
+					result[field] = item[field] !== undefined ? item[field] : null;
+				});
+				return result;
+			});
 
-			results = results.map(makeObject);
+			callback(null, returnData);
-			callback(null, results);
 		});
 	};
 
@@ -112,12 +150,14 @@ module.exports = function (redisClient, module) {
 			return setImmediate(callback);
 		}
 		redisClient.hdel(key, field, function (err) {
+			cache.delObjectCache(key);
 			callback(err);
 		});
 	};
 
 	module.deleteObjectFields = function (key, fields, callback) {
 		helpers.multiKeyValues(redisClient, 'hdel', key, fields, function (err) {
+			cache.delObjectCache(key);
 			callback(err);
 		});
 	};
@@ -131,6 +171,14 @@ module.exports = function (redisClient, module) {
 	};
 
 	module.incrObjectFieldBy = function (key, field, value, callback) {
+		callback = callback || helpers.noop;
+		function done(err, result) {
+			if (err) {
+				return callback(err);
+			}
+			cache.delObjectCache(key);
+			callback(null, Array.isArray(result) ? result.map(value => parseInt(value, 10)) : parseInt(result, 10));
+		}
 		value = parseInt(value, 10);
 		if (!key || isNaN(value)) {
 			return callback(null, null);
@@ -140,9 +188,9 @@ module.exports = function (redisClient, module) {
 			key.forEach(function (key) {
 				multi.hincrby(key, field, value);
 			});
-			multi.exec(callback);
+			multi.exec(done);
 		} else {
-			redisClient.hincrby(key, field, value, callback);
+			redisClient.hincrby(key, field, value, done);
 		}
 	};
 };

src/database/redis/main.js

@@ -10,7 +10,13 @@ module.exports = function (redisClient, module) {
 	};
 
 	module.emptydb = function (callback) {
-		module.flushdb(callback);
+		module.flushdb(function (err) {
+			if (err) {
+				return callback(err);
+			}
+			module.objectCache.resetObjectCache();
+			callback();
+		});
 	};
 
 	module.exists = function (key, callback) {
@@ -22,6 +28,7 @@ module.exports = function (redisClient, module) {
 	module.delete = function (key, callback) {
 		callback = callback || function () {};
 		redisClient.del(key, function (err) {
+			module.objectCache.delObjectCache(key);
 			callback(err);
 		});
 	};
@@ -33,6 +40,7 @@ module.exports = function (redisClient, module) {
 			multi.del(keys[i]);
 		}
 		multi.exec(function (err) {
+			module.objectCache.delObjectCache(keys);
 			callback(err);
 		});
 	};
@@ -56,7 +64,12 @@ module.exports = function (redisClient, module) {
 	module.rename = function (oldKey, newKey, callback) {
 		callback = callback || function () {};
 		redisClient.rename(oldKey, newKey, function (err) {
-			callback(err && err.message !== 'ERR no such key' ? err : null);
+			if (err && err.message !== 'ERR no such key') {
+				return callback(err);
+			}
+			module.objectCache.delObjectCache(oldKey);
+			module.objectCache.delObjectCache(newKey);
+			callback();
 		});
 	};
 

test/database/hash.js

@@ -1,7 +1,7 @@
 'use strict';
 
 
-var	async = require('async');
+var async = require('async');
 var assert = require('assert');
 var db = require('../mocks/databasemock');
 

test/database/keys.js

@@ -172,6 +172,24 @@ describe('Key methods', function () {
 				});
 			});
 		});
+
+		it('should return the correct value', function (done) {
+			db.increment('testingCache', function (err) {
+				assert.ifError(err);
+				db.get('testingCache', function (err, value) {
+					assert.ifError(err);
+					assert.equal(value, 1);
+					db.increment('testingCache', function (err) {
+						assert.ifError(err);
+						db.get('testingCache', function (err, value) {
+							assert.ifError(err);
+							assert.equal(value, 2);
+							done();
+						});
+					});
+				});
+			});
+		});
 	});
 
 	describe('rename', function () {