Skip to content

B
brpc
Commit a378c9f1 authored by zhujiashun's avatar zhujiashun
Browse files
Options

redis_server_protocol: remove vector<vector>

parent da3939ca
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 126 additions and 96 deletions
src/brpc/policy/redis_protocol.cpp
View file @ a378c9f1
......@@ -61,7 +61,9 @@ struct InputResponse : public InputMessageBase {
class RedisConnContext : public Destroyable {
public:
RedisConnContext()
: redis_service(NULL) {}
: redis_service(NULL)
, batched_size(0) {}
~RedisConnContext();
// @Destroyable
void Destroy() override;
......@@ -71,9 +73,10 @@ public:
// If user starts a transaction, handler_continue indicates the
// handler pointer that runs the transaction command.
std::unique_ptr<RedisCommandHandler> handler_continue;
// >0 if command handler is run in batched mode.
int batched_size;
RedisCommandParser parser;
std::vector<std::string> command;
};
static std::string ToLowercase(const std::string& command) {
......@@ -84,65 +87,54 @@ static std::string ToLowercase(const std::string& command) {
return res;
}
int ConsumeTask(RedisConnContext* ctx,
const std::vector<std::vector<std::string> >& commands,
butil::IOBuf* sendbuf) {
butil::Arena arena;
int size = commands.size();
std::string next_comm;
RedisReply reply(&arena);
RedisReply* output = NULL;
if (size == 1) {
// Optimize for the most common case
output = &reply;
} else {
output = (RedisReply*)malloc(sizeof(RedisReply) * size);
for (int i = 0; i < size; ++i) {
new (&output[i]) RedisReply(&arena);
int ConsumeCommand(RedisConnContext* ctx,
const std::unique_ptr<const char*[]>& commands,
int len, butil::Arena* arena,
bool is_last,
butil::IOBuf* sendbuf) {
RedisReply output(arena);
RedisCommandHandler::Result result = RedisCommandHandler::OK;
if (ctx->handler_continue) {
result = ctx->handler_continue->Run(len, commands.get(), &output, is_last);
if (result == RedisCommandHandler::OK) {
ctx->handler_continue.reset(NULL);
} else if (result == RedisCommandHandler::BATCHED) {
LOG(ERROR) << "BATCHED should not be returned in redis transaction process.";
return -1;
}
}
for (int i = 0; i < size; ++i) {
if (ctx->handler_continue) {
bool is_last = (i == size - 1);
RedisCommandHandler::Result result =
ctx->handler_continue->Run(commands[i], &output[i], is_last);
if (result == RedisCommandHandler::OK) {
ctx->handler_continue.reset(NULL);
}
} else {
std::string lcname = ToLowercase(commands[0]);
RedisCommandHandler* ch = ctx->redis_service->FindCommandHandler(lcname);
if (!ch) {
char buf[64];
snprintf(buf, sizeof(buf), "ERR unknown command `%s`", lcname.c_str());
output.SetError(buf);
} else {
bool is_last = true;
std::string comm;
if (i == 0) {
comm = ToLowercase(commands[i][0]);
} else {
comm.swap(next_comm);
}
if ((i + 1) < size) {
next_comm = ToLowercase(commands[i + 1][0]);
if (comm == next_comm) {
is_last = false;
}
}
RedisCommandHandler* ch = ctx->redis_service->FindCommandHandler(comm);
if (!ch) {
char buf[64];
snprintf(buf, sizeof(buf), "ERR unknown command `%s`", comm.c_str());
output[i].SetError(buf);
} else {
RedisCommandHandler::Result result =
ch->Run(commands[i], &output[i], is_last);
if (result == RedisCommandHandler::CONTINUE) {
ctx->handler_continue.reset(ch->NewTransactionHandler());
result = ch->Run(len, commands.get(), &output, is_last);
if (result == RedisCommandHandler::CONTINUE) {
if (ctx->batched_size) {
LOG(ERROR) << "CONTINUE should not be returned in redis batched process.";
return -1;
}
ctx->handler_continue.reset(ch->NewTransactionHandler());
} else if (result == RedisCommandHandler::BATCHED) {
ctx->batched_size++;
}
}
}
for (int i = 0; i < size; ++i) {
output[i].SerializeTo(sendbuf);
}
if (size != 1) {
free(output);
}
if (result == RedisCommandHandler::OK && ctx->batched_size) {
if ((int)output.size() != (ctx->batched_size + 1)) {
LOG(ERROR) << "reply array size can't be matched with batched size, "
<< " expected=" << ctx->batched_size + 1 << " actual=" << output.size();
return -1;
}
for (int i = 0; i < (int)output.size(); ++i) {
output[i].SerializeTo(sendbuf);
}
ctx->batched_size = 0;
} else if (result != RedisCommandHandler::BATCHED) {
output.SerializeTo(sendbuf);
} // else result == RedisCommandHandler::BATCHED, do not serialize to buf
return 0;
}
......@@ -174,27 +166,40 @@ ParseResult ParseRedisMessage(butil::IOBuf* source, Socket* socket,
ctx->redis_service = rs;
socket->reset_parsing_context(ctx);
}
std::vector<std::vector<std::string> > commands;
butil::Arena arena;
std::unique_ptr<const char*[]> current_commands;
int current_len = 0;
butil::IOBuf sendbuf;
ParseError err = PARSE_OK;
err = ctx->parser.Consume(*source, &current_commands, &current_len, &arena);
if (err != PARSE_OK) {
return MakeParseError(err);
}
while (true) {
err = ctx->parser.Consume(*source, &ctx->command);
std::unique_ptr<const char*[]> next_commands;
int next_len = 0;
err = ctx->parser.Consume(*source, &next_commands, &next_len, &arena);
if (err != PARSE_OK) {
break;
}
commands.emplace_back(std::move(ctx->command));
CHECK(ctx->command.empty());
}
if (!commands.empty()) {
butil::IOBuf sendbuf;
if (ConsumeTask(ctx, commands, &sendbuf) != 0) {
// safe to read first element.
// current_commands and next_commands both have at least one element(NULL).
bool is_last = (strcasecmp(current_commands[0], next_commands[0]) != 0);
if (ConsumeCommand(ctx, current_commands, current_len, &arena, is_last, &sendbuf) != 0) {
return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
CHECK(sendbuf.size() > 0) << "invalid size=0 of sendbuf";
Socket::WriteOptions wopt;
wopt.ignore_eovercrowded = true;
LOG_IF(WARNING, socket->Write(&sendbuf, &wopt) != 0)
<< "Fail to send redis reply";
current_commands.swap(next_commands);
current_len = next_len;
}
if (ConsumeCommand(ctx, current_commands, current_len, &arena, true, &sendbuf) != 0) {
return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
}
CHECK(!sendbuf.empty());
Socket::WriteOptions wopt;
wopt.ignore_eovercrowded = true;
LOG_IF(WARNING, socket->Write(&sendbuf, &wopt) != 0)
<< "Fail to send redis reply";
return MakeParseError(err);
} else {
// NOTE(gejun): PopPipelinedInfo() is actually more contended than what
......
src/brpc/redis.h
View file @ a378c9f1
......@@ -237,25 +237,30 @@ public:
enum Result {
OK = 0,
CONTINUE = 1,
BATCHED = 2,
};
~RedisCommandHandler() {}
// Once Server receives commands, it will first find the corresponding handlers and
// call them sequentially(one by one) according to the order that requests arrive,
// just like what redis-server does.
// `args' is the array redis of request command. For example, "set somekey somevalue"
// `args_len` is the length of request command.
// `args' is the array of request command. For example, "set somekey somevalue"
// corresponds to args[0]=="set", args[1]=="somekey" and args[2]=="somevalue".
// `output', which should be filled by user, is the content that sent to client side.
// Read brpc/src/redis_reply.h for more usage.
// `is_last' indicates whether the commands is the last command of this batch. If user
// want to do some batch processing, user should buffer the command and output. Once
// `is_last' is true, then run all the command and set the output of each command.
// want to do some batch processing, user should buffer the command and return
// RedisCommandHandler::BATCHED. Once `is_last' is true, run all the commands and
// set `output' to be an array and set all the results to the corresponding element
// of array.
//
// The return value should be RedisCommandHandler::OK for normal cases. If you want
// to implement transaction, return RedisCommandHandler::CONTINUE once server receives
// an start marker and brpc will call MultiTransactionHandler() to new a transaction
// handler that all the following commands are sent to this tranction handler until
// it returns Result::OK. Read the comment below.
virtual RedisCommandHandler::Result Run(const std::vector<std::string>& command,
virtual RedisCommandHandler::Result Run(int args_len, const char* args[],
brpc::RedisReply* output,
bool is_last) = 0;
......
src/brpc/redis_command.cpp
View file @ a378c9f1
......@@ -364,8 +364,9 @@ RedisCommandParser::RedisCommandParser() {
Reset();
}
ParseError RedisCommandParser::Consume(
butil::IOBuf& buf, std::vector<std::string>* command) {
ParseError RedisCommandParser::Consume(butil::IOBuf& buf,
std::unique_ptr<const char*[]>* commands,
int* len_out, butil::Arena* arena) {
const char* pfc = (const char*)buf.fetch1();
if (pfc == NULL) {
return PARSE_ERROR_NOT_ENOUGH_DATA;
......@@ -396,8 +397,8 @@ ParseError RedisCommandParser::Consume(
_parsing_array = true;
_length = value;
_index = 0;
_commands.clear();
return Consume(buf, command);
_commands.reset(new const char*[value + 1/* for ending NULL */]);
return Consume(buf, commands, len_out, arena);
}
CHECK(_index < _length) << "a complete command has been parsed. "
"impl of RedisCommandParser::Parse is buggy";
......@@ -415,8 +416,10 @@ ParseError RedisCommandParser::Consume(
return PARSE_ERROR_NOT_ENOUGH_DATA;
}
buf.pop_front(crlf_pos + 2/*CRLF*/);
_commands.emplace_back();
buf.cutn(&_commands.back(), len);
char* d = (char*)arena->allocate((len/8 + 1) * 8);
buf.cutn(d, len);
d[len] = '\0';
_commands[_index] = d;
char crlf[2];
buf.cutn(crlf, sizeof(crlf));
if (crlf[0] != '\r' || crlf[1] != '\n') {
......@@ -424,10 +427,11 @@ ParseError RedisCommandParser::Consume(
return PARSE_ERROR_ABSOLUTELY_WRONG;
}
if (++_index < _length) {
return Consume(buf, command);
return Consume(buf, commands, len_out, arena);
}
command->clear();
command->swap(_commands);
_commands[_index] = NULL;
commands->swap(_commands);
*len_out = _index;
Reset();
return PARSE_OK;
}
......@@ -436,6 +440,7 @@ void RedisCommandParser::Reset() {
_parsing_array = false;
_length = 0;
_index = 0;
_commands.reset(NULL);
}
} // namespace brpc
src/brpc/redis_command.h
View file @ a378c9f1
......@@ -23,6 +23,7 @@
#include <vector>
#include "butil/iobuf.h"
#include "butil/status.h"
#include "butil/arena.h"
#include "brpc/parse_result.h"
namespace brpc {
......@@ -47,8 +48,10 @@ public:
RedisCommandParser();
// Parse raw message from `buf'. Return PARSE_OK and set the parsed command
// to `command' if successful.
ParseError Consume(butil::IOBuf& buf, std::vector<std::string>* command);
// to `commands' and length to `len' if successful. Memory of commands are
// allocated in `arena'.
ParseError Consume(butil::IOBuf& buf, std::unique_ptr<const char*[]>* commands,
int*len, butil::Arena* arena);
private:
// Reset parser to the initial state.
......@@ -57,7 +60,7 @@ private:
bool _parsing_array; // if the parser has met array indicator '*'
int _length; // array length
int _index; // current parsing array index
std::vector<std::string> _commands; // parsed command string
std::unique_ptr<const char*[]> _commands; // parsed command string
};
} // namespace brpc
......
src/brpc/redis_reply.cpp
View file @ a378c9f1
......@@ -419,8 +419,8 @@ bool RedisReply::SetArray(int size) {
}
_type = REDIS_REPLY_ARRAY;
if (size < 0) {
_length = npos;
return true;
LOG(ERROR) << "negative size=" << size << " when calling SetArray";
return false;
} else if (size == 0) {
_length = 0;
return true;
......
src/brpc/redis_reply.h
View file @ a378c9f1
......@@ -59,15 +59,18 @@ public:
bool is_string() const; // True if the reply is a string.
bool is_array() const; // True if the reply is an array.
// Set the reply to the nil string. Return True if it is set
// Set the reply to the null string. Return True if it is set
// successfully. If the reply has already been set, return false.
bool SetNilString();
bool SetNullString();
// Set the reply to the array with `size' elements. If `size'
// is -1, then it is a nil array. After call SetArray, use
// operator[] to visit sub replies and set their value. Return
// True if it is set successfully. If the reply has already
// been set, return false.
// Set the reply to the null array. Return True if it is set
// successfully. If the reply has already been set, return false.
bool SetNullArray();
// Set the reply to the array with `size' elements. After calling
// SetArray, use operator[] to visit sub replies and set their
// value. Return True if it is set successfully. If the reply has
// already been set, return false.
bool SetArray(int size);
// Set the reply to status message `str'. Return True if it is set
......@@ -210,8 +213,17 @@ inline int64_t RedisReply::integer() const {
return 0;
}
inline bool RedisReply::SetNilString() {
if (!_arena || _type != REDIS_REPLY_NIL) {
inline bool RedisReply::SetNullArray() {
if (_type != REDIS_REPLY_NIL) {
return false;
}
_type = REDIS_REPLY_ARRAY;
_length = npos;
return true;
}
inline bool RedisReply::SetNullString() {
if (_type != REDIS_REPLY_NIL) {
return false;
}
_type = REDIS_REPLY_STRING;
......@@ -228,7 +240,7 @@ inline bool RedisReply::SetError(const std::string& str) {
}
inline bool RedisReply::SetInteger(int64_t value) {
if (!_arena || _type != REDIS_REPLY_NIL) {
if (_type != REDIS_REPLY_NIL) {
return false;
}
_type = REDIS_REPLY_INTEGER;
......
test/brpc_redis_unittest.cpp
View file @ a378c9f1
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment