redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-08-06 02:15:48 +03:00
Code Activity

Performance improvements and backpressure controls.

Browse Source 
Simply and speed up command argument processing logic.
Commands now return the true/false value from the underlying socket write(s).
Implement command_queue high water and low water for more better control of queueing.
This commit is contained in:
Matt Ranney
2011-06-30 14:03:36 -06:00
parent 13914295a6
commit f9e17556d2
7 changed files with 230 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/monitor.js
View File

@@ -1,4 +1,4 @@
var client = require("redis").createClient(), var client = require("../index").createClient(),
util = require("util"); util = require("util");
client.monitor(function (err, res) { client.monitor(function (err, res) {

275
index.js
View File

@@ -32,6 +32,9 @@ function RedisClient(stream, options) {
this.ready = false; this.ready = false;
this.connections = 0; this.connections = 0;
this.attempts = 1; this.attempts = 1;
this.should_buffer = false;
this.command_queue_high_water = this.options.command_queue_high_water || 1000;
this.command_queue_low_water = this.options.command_queue_low_water || 0;
this.command_queue = new Queue(); // holds sent commands to de-pipeline them this.command_queue = new Queue(); // holds sent commands to de-pipeline them
this.offline_queue = new Queue(); // holds commands issued but not able to be sent this.offline_queue = new Queue(); // holds commands issued but not able to be sent
this.commands_sent = 0; this.commands_sent = 0;
@@ -132,6 +135,7 @@ function RedisClient(stream, options) {
}); });
this.stream.on("drain", function () { this.stream.on("drain", function () {
self.should_buffer = false;
self.emit("drain"); self.emit("drain");
}); });
@@ -157,26 +161,21 @@ RedisClient.prototype.on_connect = function () {
this.stream.setNoDelay(); this.stream.setNoDelay();
this.stream.setTimeout(0); this.stream.setTimeout(0);
if (this.auth_pass) { // if redis is still loading the db, it will not authenticate and everything else will fail
var self = this; function cmd_do_auth() {
// if redis is still loading the db, it will not authenticate and everything else will fail
function cmd_do_auth() {
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("Sending auth to " + self.host + ":" + self.port + " fd " + self.stream.fd); console.log("Sending auth to " + self.host + ":" + self.port + " fd " + self.stream.fd);
} }
self.send_anyway = true; self.send_anyway = true;
self.send_command("auth", self.auth_pass, function (err, res) { self.send_command("auth", [this.auth_pass], function (err, res) {
if (err) { if (err) {
if (err.toString().match("LOADING")) { if (err.toString().match("LOADING")) {
// still loading, try to authenticate later console.log("Redis still loading, trying to authenticate later");
setTimeout(cmd_do_auth, 2000); // TODO - magic number alert
console.log("Redis still loading, trying to authenticate later"); return;
setTimeout(cmd_do_auth,2000); } else {
return; return self.emit("error", "Auth error: " + err);
} }
else return self.emit("error", "Auth error: " + err);
} }
if (res.toString() !== "OK") { if (res.toString() !== "OK") {
return self.emit("error", "Auth failed: " + res.toString()); return self.emit("error", "Auth failed: " + res.toString());
@@ -191,31 +190,28 @@ RedisClient.prototype.on_connect = function () {
// now we are really connected // now we are really connected
self.emit("connect"); self.emit("connect");
if (self.options.no_ready_check) {
if (self.options.no_ready_check) { self.ready = true;
self.ready = true; self.send_offline_queue();
self.send_offline_queue();
} else { } else {
self.ready_check(); self.ready_check();
} }
}); });
self.send_anyway = false; self.send_anyway = false;
}
cmd_do_auth();
} else {
this.emit("connect");
if (this.options.no_ready_check) {
this.ready = true;
this.send_offline_queue();
} else {
this.ready_check();
}
} }
if (this.auth_pass) {
cmd_do_auth();
} else {
this.emit("connect");
if (this.options.no_ready_check) {
this.ready = true;
this.send_offline_queue();
} else {
this.ready_check();
}
}
}; };
RedisClient.prototype.ready_check = function () { RedisClient.prototype.ready_check = function () {
@@ -275,16 +271,21 @@ RedisClient.prototype.ready_check = function () {
}; };
RedisClient.prototype.send_offline_queue = function () { RedisClient.prototype.send_offline_queue = function () {
var command_obj; var command_obj, buffered_writes = 0;
while (this.offline_queue.length > 0) { while (this.offline_queue.length > 0) {
command_obj = this.offline_queue.shift(); command_obj = this.offline_queue.shift();
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("Sending offline command: " + command_obj.command); console.log("Sending offline command: " + command_obj.command);
} }
this.send_command(command_obj.command, command_obj.args, command_obj.callback); buffered_writes += !this.send_command(command_obj.command, command_obj.args, command_obj.callback);
} }
this.offline_queue = new Queue(); this.offline_queue = new Queue();
// Even though items were shifted off, Queue backing store still uses memory until next add // Even though items were shifted off, Queue backing store still uses memory until next add, so just get a new Queue
if (buffered_writes === 0) {
this.should_buffer = false;
this.emit("drain");
}
}; };
RedisClient.prototype.connection_gone = function (why) { RedisClient.prototype.connection_gone = function (why) {
@@ -361,12 +362,16 @@ RedisClient.prototype.on_data = function (data) {
}; };
RedisClient.prototype.return_error = function (err) { RedisClient.prototype.return_error = function (err) {
var command_obj = this.command_queue.shift(); var command_obj = this.command_queue.shift(), queue_len = this.command_queue.getLength();
if (this.subscriptions === false && this.command_queue.length === 0) { if (this.subscriptions === false && queue_len === 0) {
this.emit("idle"); this.emit("idle");
this.command_queue = new Queue(); this.command_queue = new Queue();
} }
if (this.should_buffer && queue_len <= this.command_queue_low_water) {
this.emit("drain");
this.should_buffer = false;
}
if (command_obj && typeof command_obj.callback === "function") { if (command_obj && typeof command_obj.callback === "function") {
try { try {
@@ -388,11 +393,15 @@ RedisClient.prototype.return_error = function (err) {
RedisClient.prototype.return_reply = function (reply) { RedisClient.prototype.return_reply = function (reply) {
var command_obj = this.command_queue.shift(), var command_obj = this.command_queue.shift(),
obj, i, len, key, val, type, timestamp, args; obj, i, len, key, val, type, timestamp, args, queue_len = this.command_queue.getLength();
if (this.subscriptions === false && this.command_queue.length === 0) { if (this.subscriptions === false && queue_len === 0) {
this.emit("idle"); this.emit("idle");
this.command_queue = new Queue(); this.command_queue = new Queue(); // explicitly reclaim storage from old Queue
}
if (this.should_buffer && queue_len <= this.command_queue_low_water) {
this.emit("drain");
this.should_buffer = false;
} }
if (command_obj && !command_obj.sub_command) { if (command_obj && !command_obj.sub_command) {
@@ -453,52 +462,56 @@ RedisClient.prototype.return_reply = function (reply) {
} }
}; };
RedisClient.prototype.send_command = function () { // This Command constructor is ever so slightly faster than using an object literal
var command, callback, arg, args, this_args, command_obj, i, il, function Command(command, args, sub_command, callback) {
elem_count, stream = this.stream, buffer_args, command_str = ""; this.command = command;
this.args = args;
this.sub_command = sub_command;
this.callback = callback;
}
this_args = to_array(arguments); RedisClient.prototype.send_command = function (command, args, callback) {
var arg, this_args, command_obj, i, il, elem_count, stream = this.stream, buffer_args, command_str = "", buffered_writes = 0;
if (this_args.length === 0) { if (typeof command !== "string") {
throw new Error("send_command: not enough arguments"); throw new Error("First argument to send_command must be the command name string, not " + typeof command);
} }
if (typeof this_args[0] !== "string") { if (Array.isArray(args)) {
throw new Error("First argument of send_command must be the command name"); if (typeof callback === "function") {
} // probably the fastest way:
command = this_args[0].toLowerCase(); // client.command([arg1, arg2], cb); (straight passthrough)
// send_command(command, [arg1, arg2], cb);
if (this_args[1] && Array.isArray(this_args[1])) { } else if (typeof callback === "undefined") {
args = this_args[1]; // most people find this variable argument length form more convenient, but it uses arguments, which is slower
if (typeof this_args[2] === "function") { // client.command(arg1, arg2, cb); (wraps up arguments into an array)
callback = this_args[2]; // send_command(command, [arg1, arg2, cb]);
// client.command(arg1, arg2); (callback is optional)
// send_command(command, [arg1, arg2]);
if (typeof args[args.length - 1] === "function") {
callback = args[args.length - 1];
args.length -= 1;
}
} else {
throw new Error("send_command: last argument must be a callback or undefined");
} }
} else { } else {
if (typeof this_args[this_args.length - 1] === "function") { throw new Error("send_command: second argument must be an array");
callback = this_args[this_args.length - 1];
args = this_args.slice(1, this_args.length - 1);
} else {
args = this_args.slice(1, this_args.length);
}
} }
if (args.length === 2 && Array.isArray(args[1])) { command_obj = new Command(command, args, false, callback);
args = [args[0]].concat(args[1]);
}
command_obj = { if ((!this.ready && !this.send_anyway) || !stream.writable) {
command: command,
args: args,
callback: callback,
sub_command: false
};
if (!this.ready && !this.send_anyway) {
if (exports.debug_mode) { if (exports.debug_mode) {
if (!stream.writable) {
console.log("send command: stream is not writeable.");
}
console.log("Queueing " + command + " for next server connection."); console.log("Queueing " + command + " for next server connection.");
} }
this.offline_queue.push(command_obj); this.offline_queue.push(command_obj);
return; this.should_buffer = true;
return false;
} }
if (command === "subscribe" || command === "psubscribe" || command === "unsubscribe" || command === "punsubscribe") { if (command === "subscribe" || command === "psubscribe" || command === "unsubscribe" || command === "punsubscribe") {
@@ -521,10 +534,6 @@ RedisClient.prototype.send_command = function () {
buffer_args = false; buffer_args = false;
elem_count += args.length; elem_count += args.length;
// Probably should just scan this like a normal person. This is clever, but might be slow.
buffer_args = args.some(function (arg) {
return arg instanceof Buffer;
});
// Always use "Multi bulk commands", but if passed any Buffer args, then do multiple writes, one for each arg // Always use "Multi bulk commands", but if passed any Buffer args, then do multiple writes, one for each arg
// This means that using Buffers in commands is going to be slower, so use Strings if you don't already have a Buffer. // This means that using Buffers in commands is going to be slower, so use Strings if you don't already have a Buffer.
@@ -532,9 +541,10 @@ RedisClient.prototype.send_command = function () {
command_str = "*" + elem_count + "\r\n$" + command.length + "\r\n" + command + "\r\n"; command_str = "*" + elem_count + "\r\n$" + command.length + "\r\n" + command + "\r\n";
if (! stream.writable && exports.debug_mode) { for (i = 0, il = args.length, arg; i < il; i += 1) {
console.log("send command: stream is not writeable, should get a close event next tick."); if (args[i] instanceof Buffer) {
return; buffer_args = true;
}
} }
if (! buffer_args) { // Build up a string and send entire command in one write if (! buffer_args) { // Build up a string and send entire command in one write
@@ -548,13 +558,12 @@ RedisClient.prototype.send_command = function () {
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("send " + this.host + ":" + this.port + " fd " + this.stream.fd + ": " + command_str); console.log("send " + this.host + ":" + this.port + " fd " + this.stream.fd + ": " + command_str);
} }
stream.write(command_str); buffered_writes += !stream.write(command_str);
} else { } else {
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("send command: " + command_str); console.log("send command (" + command_str + ") has Buffer arguments");
console.log("send command has Buffer arguments");
} }
stream.write(command_str); buffered_writes += !stream.write(command_str);
for (i = 0, il = args.length, arg; i < il; i += 1) { for (i = 0, il = args.length, arg; i < il; i += 1) {
arg = args[i]; arg = args[i];
@@ -565,19 +574,32 @@ RedisClient.prototype.send_command = function () {
if (arg instanceof Buffer) { if (arg instanceof Buffer) {
if (arg.length === 0) { if (arg.length === 0) {
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("Using empty string for 0 length buffer"); console.log("send_command: using empty string for 0 length buffer");
} }
stream.write("$0\r\n\r\n"); buffered_writes += !stream.write("$0\r\n\r\n");
} else { } else {
stream.write("$" + arg.length + "\r\n"); buffered_writes += !stream.write("$" + arg.length + "\r\n");
stream.write(arg); buffered_writes += !stream.write(arg);
stream.write("\r\n"); buffered_writes += !stream.write("\r\n");
if (exports.debug_mode) {
console.log("send_command: buffer send " + arg.length + " bytes");
}
} }
} else { } else {
stream.write("$" + Buffer.byteLength(arg) + "\r\n" + arg + "\r\n"); if (exports.debug_mode) {
console.log("send_command: string send " + Buffer.byteLength(arg) + " bytes: " + arg);
}
buffered_writes += !stream.write("$" + Buffer.byteLength(arg) + "\r\n" + arg + "\r\n");
} }
} }
} }
if (exports.debug_mode) {
console.log("send_command buffered_writes: " + buffered_writes, " should_buffer: " + this.should_buffer);
}
if (buffered_writes || this.command_queue.getLength() >= this.command_queue_high_water) {
this.should_buffer = true;
}
return !this.should_buffer;
}; };
RedisClient.prototype.end = function () { RedisClient.prototype.end = function () {
@@ -587,7 +609,6 @@ RedisClient.prototype.end = function () {
return this.stream.end(); return this.stream.end();
}; };
function Multi(client, args) { function Multi(client, args) {
this.client = client; this.client = client;
this.queue = [["MULTI"]]; this.queue = [["MULTI"]];
@@ -622,17 +643,17 @@ commands = set_union(["get", "set", "setnx", "setex", "append", "strlen", "del",
"restore", "migrate", "dump", "object", "client", "eval", "evalsha"], require("./lib/commands")); "restore", "migrate", "dump", "object", "client", "eval", "evalsha"], require("./lib/commands"));
commands.forEach(function (command) { commands.forEach(function (command) {
RedisClient.prototype[command] = function () { RedisClient.prototype[command] = function (args, callback) {
var args = to_array(arguments); if (Array.isArray(args) && typeof callback === "function") {
args.unshift(command); // put command at the beginning return this.send_command(command, args, callback);
this.send_command.apply(this, args); } else {
return this.send_command(command, to_array(arguments));
}
}; };
RedisClient.prototype[command.toUpperCase()] = RedisClient.prototype[command]; RedisClient.prototype[command.toUpperCase()] = RedisClient.prototype[command];
Multi.prototype[command] = function () { Multi.prototype[command] = function () {
var args = to_array(arguments); this.queue.push([command].concat(to_array(arguments)));
args.unshift(command);
this.queue.push(args);
return this; return this;
}; };
Multi.prototype[command.toUpperCase()] = Multi.prototype[command]; Multi.prototype[command.toUpperCase()] = Multi.prototype[command];
@@ -648,29 +669,50 @@ RedisClient.prototype.auth = function () {
} }
if (this.connected) { if (this.connected) {
args.unshift("auth"); this.send_command("auth", args);
this.send_command.apply(this, args);
} }
}; };
RedisClient.prototype.AUTH = RedisClient.prototype.auth; RedisClient.prototype.AUTH = RedisClient.prototype.auth;
RedisClient.prototype.hmset = function () { RedisClient.prototype.hmget = function (arg1, arg2, arg3) {
var args = to_array(arguments), tmp_args; if (Array.isArray(arg2) && typeof arg3 === "function") {
if (args.length >= 2 && typeof args[0] === "string" && typeof args[1] === "object") { return this.send_command("hmget", [arg1].concat(arg2), arg3);
tmp_args = [ "hmset", args[0] ]; } else if (Array.isArray(arg1) && typeof arg2 === "function") {
Object.keys(args[1]).map(function (key) { return this.send_command("hmget", arg1, arg2);
tmp_args.push(key);
tmp_args.push(args[1][key]);
});
if (args[2]) {
tmp_args.push(args[2]);
}
args = tmp_args;
} else { } else {
args.unshift("hmset"); return this.send_command("hmget", to_array(arguments));
}
};
RedisClient.prototype.HMGET = RedisClient.prototype.hmget;
RedisClient.prototype.hmset = function (args, callback) {
var tmp_args, tmp_keys, i, il, key;
if (Array.isArray(args) && typeof callback === "function") {
return this.send_command("hmset", args, callback);
} }
this.send_command.apply(this, args); args = to_array(arguments);
if (typeof args[args.length - 1] === "function") {
callback = args[args.length - 1];
args.length -= 1;
} else {
callback = null;
}
if (args.length === 2 && typeof args[0] === "string" && typeof args[1] === "object") {
// User does: client.hmset(key, {key1: val1, key2: val2})
tmp_args = [ args[0] ];
tmp_keys = Object.keys(args[1]);
for (i = 0, il = tmp_keys.length; i < il ; i++) {
key = tmp_keys[i];
tmp_args.push(key);
tmp_args.push(args[1][key]);
}
args = tmp_args;
}
return this.send_command("hmset", args, callback);
}; };
RedisClient.prototype.HMSET = RedisClient.prototype.hmset; RedisClient.prototype.HMSET = RedisClient.prototype.hmset;
@@ -699,7 +741,7 @@ Multi.prototype.exec = function (callback) {
var self = this; var self = this;
// drain queue, callback will catch "QUEUED" or error // drain queue, callback will catch "QUEUED" or error
// Can't use a for loop here, as we need closure around the index. // TODO - get rid of all of these anonymous functions which are elegant but slow
this.queue.forEach(function (args, index) { this.queue.forEach(function (args, index) {
var command = args[0], obj; var command = args[0], obj;
if (typeof args[args.length - 1] === "function") { if (typeof args[args.length - 1] === "function") {
@@ -730,7 +772,8 @@ Multi.prototype.exec = function (callback) {
}); });
}, this); }, this);
this.client.send_command("EXEC", function (err, replies) { // TODO - make this callback part of Multi.prototype instead of creating it each time
return this.client.send_command("EXEC", [], function (err, replies) {
if (err) { if (err) {
if (callback) { if (callback) {
callback(new Error(err)); callback(new Error(err));

78
lib/parser/javascript.js
View File

@@ -35,8 +35,6 @@ function small_toString(buf, len) {
// Reset parser to it's original state. // Reset parser to it's original state.
RedisReplyParser.prototype.reset = function () { RedisReplyParser.prototype.reset = function () {
this.state = "type";
this.return_buffer = new Buffer(16384); // for holding replies, might grow this.return_buffer = new Buffer(16384); // for holding replies, might grow
this.return_string = ""; this.return_string = "";
this.tmp_string = ""; // for holding size fields this.tmp_string = ""; // for holding size fields
@@ -46,6 +44,22 @@ RedisReplyParser.prototype.reset = function () {
this.multi_bulk_pos = 0; this.multi_bulk_pos = 0;
this.multi_bulk_nested_length = 0; this.multi_bulk_nested_length = 0;
this.multi_bulk_nested_replies = null; this.multi_bulk_nested_replies = null;
this.states = {
TYPE: 1,
SINGLE_LINE: 2,
MULTI_BULK_COUNT: 3,
INTEGER_LINE: 4,
BULK_LENGTH: 5,
ERROR_LINE: 6,
BULK_DATA: 7,
UNKNOWN_TYPE: 8,
FINAL_CR: 9,
FINAL_LF: 10,
MULTI_BULK_COUNT_LF: 11
};
this.state = this.states.TYPE;
}; };
RedisReplyParser.prototype.parser_error = function (message) { RedisReplyParser.prototype.parser_error = function (message) {
@@ -54,7 +68,7 @@ RedisReplyParser.prototype.parser_error = function (message) {
}; };
RedisReplyParser.prototype.execute = function (incoming_buf) { RedisReplyParser.prototype.execute = function (incoming_buf) {
var pos = 0, bd_tmp, bd_str, i, il; var pos = 0, bd_tmp, bd_str, i, il, states = this.states;
//, state_times = {}, start_execute = new Date(), start_switch, end_switch, old_state; //, state_times = {}, start_execute = new Date(), start_switch, end_switch, old_state;
//start_switch = new Date(); //start_switch = new Date();
@@ -63,76 +77,76 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
// console.log("execute: " + this.state + ", " + pos + "/" + incoming_buf.length + ", " + String.fromCharCode(incoming_buf[pos])); // console.log("execute: " + this.state + ", " + pos + "/" + incoming_buf.length + ", " + String.fromCharCode(incoming_buf[pos]));
switch (this.state) { switch (this.state) {
case "type": case states.TYPE:
this.type = incoming_buf[pos]; this.type = incoming_buf[pos];
pos += 1; pos += 1;
switch (this.type) { switch (this.type) {
case 43: // + case 43: // +
this.state = "single line"; this.state = states.SINGLE_LINE;
this.return_buffer.end = 0; this.return_buffer.end = 0;
this.return_string = ""; this.return_string = "";
break; break;
case 42: // * case 42: // *
this.state = "multi bulk count"; this.state = states.MULTI_BULK_COUNT;
this.tmp_string = ""; this.tmp_string = "";
break; break;
case 58: // : case 58: // :
this.state = "integer line"; this.state = states.INTEGER_LINE;
this.return_buffer.end = 0; this.return_buffer.end = 0;
this.return_string = ""; this.return_string = "";
break; break;
case 36: // $ case 36: // $
this.state = "bulk length"; this.state = states.BULK_LENGTH;
this.tmp_string = ""; this.tmp_string = "";
break; break;
case 45: // - case 45: // -
this.state = "error line"; this.state = states.ERROR_LINE;
this.return_buffer.end = 0; this.return_buffer.end = 0;
this.return_string = ""; this.return_string = "";
break; break;
default: default:
this.state = "unknown type"; this.state = states.UNKNOWN_TYPE;
} }
break; break;
case "integer line": case states.INTEGER_LINE:
if (incoming_buf[pos] === 13) { if (incoming_buf[pos] === 13) {
this.send_reply(+small_toString(this.return_buffer, this.return_buffer.end)); this.send_reply(+small_toString(this.return_buffer, this.return_buffer.end));
this.state = "final lf"; this.state = states.FINAL_LF;
} else { } else {
this.return_buffer[this.return_buffer.end] = incoming_buf[pos]; this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
this.return_buffer.end += 1; this.return_buffer.end += 1;
} }
pos += 1; pos += 1;
break; break;
case "error line": case states.ERROR_LINE:
if (incoming_buf[pos] === 13) { if (incoming_buf[pos] === 13) {
this.send_error(this.return_buffer.toString("ascii", 0, this.return_buffer.end)); this.send_error(this.return_buffer.toString("ascii", 0, this.return_buffer.end));
this.state = "final lf"; this.state = states.FINAL_LF;
} else { } else {
this.return_buffer[this.return_buffer.end] = incoming_buf[pos]; this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
this.return_buffer.end += 1; this.return_buffer.end += 1;
} }
pos += 1; pos += 1;
break; break;
case "single line": case states.SINGLE_LINE:
if (incoming_buf[pos] === 13) { if (incoming_buf[pos] === 13) {
this.send_reply(this.return_string); this.send_reply(this.return_string);
this.state = "final lf"; this.state = states.FINAL_LF;
} else { } else {
this.return_string += String.fromCharCode(incoming_buf[pos]); this.return_string += String.fromCharCode(incoming_buf[pos]);
} }
pos += 1; pos += 1;
break; break;
case "multi bulk count": case states.MULTI_BULK_COUNT:
if (incoming_buf[pos] === 13) { // \r if (incoming_buf[pos] === 13) { // \r
this.state = "multi bulk count lf"; this.state = states.MULTI_BULK_COUNT_LF;
} else { } else {
this.tmp_string += String.fromCharCode(incoming_buf[pos]); this.tmp_string += String.fromCharCode(incoming_buf[pos]);
} }
pos += 1; pos += 1;
break; break;
case "multi bulk count lf": case states.MULTI_BULK_COUNT_LF:
if (incoming_buf[pos] === 10) { // \n if (incoming_buf[pos] === 10) { // \n
if (this.multi_bulk_length) { // nested multi-bulk if (this.multi_bulk_length) { // nested multi-bulk
this.multi_bulk_nested_length = this.multi_bulk_length; this.multi_bulk_nested_length = this.multi_bulk_length;
@@ -141,7 +155,7 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
} }
this.multi_bulk_length = +this.tmp_string; this.multi_bulk_length = +this.tmp_string;
this.multi_bulk_pos = 0; this.multi_bulk_pos = 0;
this.state = "type"; this.state = states.TYPE;
if (this.multi_bulk_length < 0) { if (this.multi_bulk_length < 0) {
this.send_reply(null); this.send_reply(null);
this.multi_bulk_length = 0; this.multi_bulk_length = 0;
@@ -158,25 +172,25 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
} }
pos += 1; pos += 1;
break; break;
case "bulk length": case states.BULK_LENGTH:
if (incoming_buf[pos] === 13) { // \r if (incoming_buf[pos] === 13) { // \r
this.state = "bulk lf"; this.state = states.BULK_LF;
} else { } else {
this.tmp_string += String.fromCharCode(incoming_buf[pos]); this.tmp_string += String.fromCharCode(incoming_buf[pos]);
} }
pos += 1; pos += 1;
break; break;
case "bulk lf": case states.BULK_LF:
if (incoming_buf[pos] === 10) { // \n if (incoming_buf[pos] === 10) { // \n
this.bulk_length = +this.tmp_string; this.bulk_length = +this.tmp_string;
if (this.bulk_length === -1) { if (this.bulk_length === -1) {
this.send_reply(null); this.send_reply(null);
this.state = "type"; this.state = states.TYPE;
} else if (this.bulk_length === 0) { } else if (this.bulk_length === 0) {
this.send_reply(new Buffer("")); this.send_reply(new Buffer(""));
this.state = "final cr"; this.state = states.FINAL_CR;
} else { } else {
this.state = "bulk data"; this.state = states.BULK_DATA;
if (this.bulk_length > this.return_buffer.length) { if (this.bulk_length > this.return_buffer.length) {
if (exports.debug_mode) { if (exports.debug_mode) {
console.log("Growing return_buffer from " + this.return_buffer.length + " to " + this.bulk_length); console.log("Growing return_buffer from " + this.return_buffer.length + " to " + this.bulk_length);
@@ -191,7 +205,7 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
} }
pos += 1; pos += 1;
break; break;
case "bulk data": case states.BULK_DATA:
this.return_buffer[this.return_buffer.end] = incoming_buf[pos]; this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
this.return_buffer.end += 1; this.return_buffer.end += 1;
pos += 1; pos += 1;
@@ -206,21 +220,21 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
} }
} }
this.send_reply(bd_tmp); this.send_reply(bd_tmp);
this.state = "final cr"; this.state = states.FINAL_CR;
} }
break; break;
case "final cr": case states.FINAL_CR:
if (incoming_buf[pos] === 13) { // \r if (incoming_buf[pos] === 13) { // \r
this.state = "final lf"; this.state = states.FINAL_LF;
pos += 1; pos += 1;
} else { } else {
this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final CR")); this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final CR"));
return; return;
} }
break; break;
case "final lf": case states.FINAL_LF:
if (incoming_buf[pos] === 10) { // \n if (incoming_buf[pos] === 10) { // \n
this.state = "type"; this.state = states.TYPE;
pos += 1; pos += 1;
} else { } else {
this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final LF")); this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final LF"));

6
lib/queue.js
View File

@@ -45,9 +45,13 @@ Queue.prototype.forEach = function (fn, thisv) {
return array; return array;
}; };
Queue.prototype.getLength = function () {
return this.head.length - this.offset + this.tail.length;
};
Object.defineProperty(Queue.prototype, 'length', { Object.defineProperty(Queue.prototype, 'length', {
get: function () { get: function () {
return this.head.length - this.offset + this.tail.length; return this.getLength();
} }
}); });

3
lib/to_array.js
View File

@@ -1,4 +1,3 @@
// the "new Array(len)" syntax is legal and optimized by V8, but JSHint is utterly confused by it.
function to_array(args) { function to_array(args) {
var len = args.length, var len = args.length,
arr = new Array(len), i; arr = new Array(len), i;
@@ -8,6 +7,6 @@ function to_array(args) {
} }
return arr; return arr;
}; }
module.exports = to_array; module.exports = to_array;

12
multi_bench.js
View File

@@ -1,12 +1,12 @@
var redis = require("./index"), var redis = require("./index"),
num_clients = parseInt(process.argv[2]) || 50, num_clients = parseInt(process.argv[2], 10) || 50,
active_clients = 0, active_clients = 0,
clients = new Array(num_clients), clients = new Array(num_clients),
num_requests = 20000, num_requests = 20000,
issued_requests = 0, issued_requests = 0,
latency = new Array(num_requests), latency = new Array(num_requests),
tests = [], tests = [],
test_start, test_start, parser_logged = false,
client_options = { client_options = {
return_buffers: false return_buffers: false
}; };
@@ -35,7 +35,7 @@ tests.push({
tests.push({ tests.push({
descr: "LPUSH", descr: "LPUSH",
command: ["lpush", "mylist", Array(8).join("-")] command: ["lpush", "mylist", new Array(8).join("-")]
}); });
tests.push({ tests.push({
@@ -58,7 +58,11 @@ function create_clients(callback) {
while (active_clients < num_clients) { while (active_clients < num_clients) {
client = clients[active_clients++] = redis.createClient(6379, "127.0.0.1", client_options); client = clients[active_clients++] = redis.createClient(6379, "127.0.0.1", client_options);
client.on("connect", function() { if (! parser_logged) {
console.log("Using reply parser " + client.reply_parser.name);
parser_logged = true;
}
client.on("connect", function () {
// Fire callback when all clients are connected // Fire callback when all clients are connected
connected += 1; connected += 1;
if (connected === num_clients) { if (connected === num_clients) {

13
test.js
View File

@@ -384,7 +384,7 @@ tests.HSET = function () {
}; };
tests.HMSET_BUFFER_AND_ARRAY = function () { tests.HMSET_BUFFER_AND_ARRAY = function () {
// Saving a buffer and an array to the same document should not error // Saving a buffer and an array to the same key should not error
var key = "test hash", var key = "test hash",
field1 = "buffer", field1 = "buffer",
value1 = new Buffer("abcdefghij"), value1 = new Buffer("abcdefghij"),
@@ -596,6 +596,13 @@ tests.GETSET = function () {
tests.MGET = function () { tests.MGET = function () {
var name = "MGET"; var name = "MGET";
client.mset(["mget keys 1", "mget val 1", "mget keys 2", "mget val 2", "mget keys 3", "mget val 3"], require_string("OK", name)); client.mset(["mget keys 1", "mget val 1", "mget keys 2", "mget val 2", "mget keys 3", "mget val 3"], require_string("OK", name));
client.MGET("mget keys 1", "mget keys 2", "mget keys 3", function (err, results) {
assert.strictEqual(null, err, "result sent back unexpected error: " + err);
assert.strictEqual(3, results.length, name);
assert.strictEqual("mget val 1", results[0].toString(), name);
assert.strictEqual("mget val 2", results[1].toString(), name);
assert.strictEqual("mget val 3", results[2].toString(), name);
});
client.MGET(["mget keys 1", "mget keys 2", "mget keys 3"], function (err, results) { client.MGET(["mget keys 1", "mget keys 2", "mget keys 3"], function (err, results) {
assert.strictEqual(null, err, "result sent back unexpected error: " + err); assert.strictEqual(null, err, "result sent back unexpected error: " + err);
assert.strictEqual(3, results.length, name); assert.strictEqual(3, results.length, name);
@@ -647,7 +654,6 @@ tests.HGETALL = function () {
client.HGETALL(["hosts"], function (err, obj) { client.HGETALL(["hosts"], function (err, obj) {
assert.strictEqual(null, err, name + " result sent back unexpected error: " + err); assert.strictEqual(null, err, name + " result sent back unexpected error: " + err);
assert.strictEqual(3, Object.keys(obj).length, name); assert.strictEqual(3, Object.keys(obj).length, name);
// assert.ok(Buffer.isBuffer(obj.mjr), name);
assert.strictEqual("1", obj.mjr.toString(), name); assert.strictEqual("1", obj.mjr.toString(), name);
assert.strictEqual("23", obj.another.toString(), name); assert.strictEqual("23", obj.another.toString(), name);
assert.strictEqual("1234", obj.home.toString(), name); assert.strictEqual("1234", obj.home.toString(), name);
@@ -1216,7 +1222,8 @@ client.on("reconnecting", function (params) {
}); });
process.on('uncaughtException', function (err) { process.on('uncaughtException', function (err) {
console.log("Uncaught exception: " + err.stack); console.error("Uncaught exception: " + err.stack);
process.exit(1);
}); });
process.on('exit', function (code) { process.on('exit', function (code) {