bughz/predixy
1
0
Fork 0
mirror of https://github.com/joyieldInc/predixy.git synced 2025-12-24 22:46:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

improve performance

Browse Source 
fix scan for sentinel groups
This commit is contained in:
joyield 2017-07-30 15:00:51 +08:00
parent 9d86d04374
commit ecc63e5586
28 changed files with 377 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
conf/auth.conf
View File

@ -28,7 +28,7 @@
# Auth bcd {
# Mode admin
# }
#### password is "abc", the client must send command Auth bcd
#### password is "bcd", the client must send command Auth bcd
#### Mode admin, client connection can read and write and admin,
#### the CONFIG command need admin permission
#### No KeyPrefix, ReadKeyPrefix, WriteKeyPrefix define, all key can be visit
@ -47,7 +47,7 @@
# ReadKeyPrefix User Stats
# WriteKeyPrefix User
# }
#### password is "cde", the client must send command Auth cde
#### password is "def", the client must send command Auth def
#### Mode read, client connection can read and write, but read and write
#### keyspace is diffrent, client can GET User.123 and also
#### SET User.123 SomeValue, but SET Stats.123 will be deny

4
conf/latency.conf
View File

@ -10,8 +10,8 @@
## see latency monitor for specify server
## redis> INFO ServerLatency ServAddr [name]
##
## reset all stats info, include latency monitor
## redis> INFO ResetStats
## reset all stats info, include latency monitor, require admin permission
## redis> CONFIG ResetStat
##
## Examples:
## LatencyMonitor name {

5
src/AcceptConnection.cpp
View File

@ -198,11 +198,10 @@ void AcceptConnection::parse(Handler* h, Buffer* buf, int pos)
req->set(mParser);
h->handleRequest(req);
} else {
SegmentStr<RequestParser::MaxCmdLen> cmd(mParser.cmd());
ResponsePtr res = ResponseAlloc::create();
char err[1024];
int len = snprintf(err, sizeof(err), "unknown command '%.*s'",
cmd.length(), cmd.data());
int len = snprintf(err, sizeof(err), "unknown command '%s'",
mParser.cmd());
res->setErr(err, len);
h->handleResponse(nullptr, req, res);
}

2
src/AcceptConnection.h
View File

@ -23,7 +23,7 @@ class AcceptConnection :
public Subscribe,
public ListNode<AcceptConnection, SharePtr<AcceptConnection>>,
public DequeNode<AcceptConnection, SharePtr<AcceptConnection>>,
public RefCntObj<AcceptConnection>
public RefCntObj<AcceptConnection, AtomicInt>
{
public:
typedef AcceptConnection Value;

5
src/Alloc.h
View File

@ -122,7 +122,7 @@ thread_local T* Alloc<T, CacheSize>::Free[CacheSize];
template<class T, int CacheSize>
thread_local int Alloc<T, CacheSize>::Size = 0;
template<class T>
template<class T, class CntType = int>
class RefCntObj
{
public:
@ -138,6 +138,7 @@ public:
}
void ref()
{
FuncCallTimer();
++mCnt;
}
void unref()
@ -156,7 +157,7 @@ protected:
mCnt = 0;
}
private:
AtomicInt mCnt;
CntType mCnt;
};
template<class T>

2
src/Buffer.h
View File

@ -192,7 +192,7 @@ public:
}
bool empty() const
{
return mCur.buf == mEnd.buf && mCur.pos == mEnd.pos;
return mBegin.buf == mEnd.buf && mBegin.pos == mEnd.pos;
}
private:
BufferPos mBegin;

3
src/ClusterServerPool.cpp
View File

@ -33,7 +33,7 @@ ClusterServerPool::~ClusterServerPool()
}
}
Server* ClusterServerPool::getServer(Handler* h, Request* req) const
Server* ClusterServerPool::getServer(Handler* h, Request* req, const String& key) const
{
FuncCallTimer();
switch (req->type()) {
@ -43,7 +43,6 @@ Server* ClusterServerPool::getServer(Handler* h, Request* req) const
default:
break;
}
SegmentStr<Const::MaxKeyLen> key(req->key());
int i = mHash.hash(key.data(), key.length(), HashTag);
i &= Const::RedisClusterSlotsMask;
ServerGroup* g = mSlots[i];

2
src/ClusterServerPool.h
View File

@ -26,7 +26,7 @@ public:
return mServPool;
}
private:
Server* getServer(Handler* h, Request* req) const;
Server* getServer(Handler* h, Request* req, const String& key) const;
void refreshRequest(Handler* h);
void handleResponse(Handler* h, ConnectConnection* s, Request* req, Response* res);
ServerGroup* getGroup(const String& nodeid) const

5
src/Command.cpp
View File

@ -11,7 +11,7 @@
#include "Command.h"
const Command Command::CmdPool[Sentinel] = {
{None, "", 0, 0, Read},
{None, "", 0, MaxArgs, Read},
{Ping, "ping", 1, 2, Read},
{PingServ, "ping", 1, 2, Inner},
{Echo, "echo", 2, 2, Read},
@ -24,7 +24,7 @@ const Command Command::CmdPool[Sentinel] = {
{SentinelSlaves, "sentinel slaves", 3, 3, Inner},
{Cmd, "command", 1, 1, Read},
{Info, "info", 1, 4, Read},
{Config, "config", 3, 4, Admin},
{Config, "config", 2, 4, Admin},
{Cluster, "cluster", 2, 2, Inner},
{ClusterNodes, "cluster nodes", 2, 2, SubCmd|Inner},
{Asking, "asking", 1, 1, Inner},
@ -171,7 +171,6 @@ const Command Command::CmdPool[Sentinel] = {
};
Command::CommandMap Command::CmdMap;
void Command::init()
{
int type = 0;

16
src/Command.h
View File

@ -7,7 +7,9 @@
#ifndef _PREDIXY_COMMAND_H_
#define _PREDIXY_COMMAND_H_
#include <unordered_map>
#include "Exception.h"
#include "HashFunc.h"
class Command
{
@ -224,6 +226,10 @@ public:
{
return mode & MultiKeyVal;
}
bool isAnyMulti() const
{
return mode & (MultiKey|SMultiKey|MultiKeyVal);
}
static void init();
static const Command& get(Type type)
{
@ -244,7 +250,15 @@ public:
private:
static const int MaxArgs = 100000000;
static const Command CmdPool[Sentinel];
typedef std::map<String, const Command*, StringCaseCmp> CommandMap;
class H
{
public:
size_t operator()(const String& s) const
{
return Hash::crc16(s.data(), s.length());
}
};
typedef std::unordered_map<String, const Command*, H> CommandMap;
static CommandMap CmdMap;
};

1
src/Common.h
View File

@ -29,6 +29,7 @@ namespace Const
static const int MaxAddrLen = 128;
static const int MaxDcLen = 32;
static const int MaxIOVecLen = IOV_MAX;
static const int MaxCmdLen = 32;
static const int MaxKeyLen = 512;
static const int BufferAllocCacheSize = 64;
static const int RequestAllocCacheSize = 32;

8
src/ConnectConnection.cpp
View File

@ -218,14 +218,6 @@ void ConnectConnection::handleResponse(Handler* h)
}
}
void ConnectConnection::send(Handler* h, Request* req)
{
FuncCallTimer();
mSendRequests.push_back(req);
logDebug("h %d s %s %d pend req %ld",
h->id(), peer(), fd(), req->id());
}
void ConnectConnection::close(Handler* h)
{
SendRequestList* reqs[2] = {&mSentRequests, &mSendRequests};

5
src/ConnectConnection.h
View File

@ -28,8 +28,11 @@ public:
~ConnectConnection();
bool writeEvent(Handler* h);
void readEvent(Handler* h);
void send(Handler* h, Request* req);
void close(Handler* h);
void send(Handler* h, Request* req)
{
mSendRequests.push_back(req);
}
Server* server() const
{
return mServ;

19
src/Handler.cpp
View File

@ -46,6 +46,8 @@ Handler::~Handler()
void Handler::run()
{
Request::init();
Response::init();
auto conf = mProxy->conf();
refreshServerPool();
while (!mStop) {
@ -134,7 +136,7 @@ void Handler::postEvent()
}
}
void Handler::addPostEvent(AcceptConnection* c, int evts)
inline void Handler::addPostEvent(AcceptConnection* c, int evts)
{
if (!c->getPostEvent()) {
mPostAcceptConns.push_back(c);
@ -143,12 +145,12 @@ void Handler::addPostEvent(AcceptConnection* c, int evts)
c->addPostEvent(evts);
}
void Handler::addPostEvent(ConnectConnection* c, int evts)
inline void Handler::addPostEvent(ConnectConnection* s, int evts)
{
if (!c->getPostEvent()) {
mPostConnectConns.push_back(c);
if (!s->getPostEvent()) {
mPostConnectConns.push_back(s);
}
c->addPostEvent(evts);
s->addPostEvent(evts);
}
void Handler::postAcceptConnectionEvent()
@ -353,7 +355,6 @@ void Handler::handleAcceptConnectionEvent(AcceptConnection* c, int evts)
try {
if (c->good() && (evts & Multiplexor::ReadEvent)) {
c->readEvent(this);
setAcceptConnectionActiveTime(c);
}
if (c->good() && (evts & Multiplexor::WriteEvent)) {
addPostEvent(c, Multiplexor::WriteEvent);
@ -483,7 +484,7 @@ void Handler::handleRequest(Request* req)
return;
}
auto sp = mProxy->serverPool();
Server* serv = sp->getServer(this, req);
Server* serv = sp->getServer(this, req, key);
if (!serv) {
directResponse(req, Response::NoServer);
return;
@ -634,6 +635,7 @@ bool Handler::preHandleRequest(Request* req, const String& key)
void Handler::postHandleRequest(Request* req, ConnectConnection* s)
{
FuncCallTimer();
auto c = req->connection();
if (!c) {
return;
@ -1127,6 +1129,9 @@ void Handler::configRequest(Request* req, const String& key)
configGetRequest(req);
} else if (key.equal("set", true)) {
configSetRequest(req);
} else if (key.equal("resetstat", true)) {
mProxy->incrStatsVer();
directResponse(req, Response::Ok);
} else {
directResponse(req, Response::ConfigSubCmdUnknown);
}

13
src/HashFunc.cpp
View File

@ -45,19 +45,6 @@ const char* Hash::hashTagStr(const char* buf, int& len, const char* tag)
return buf;
}
long Hash::hash(const char* buf, int len) const
{
switch (mType) {
case Atol:
return atol(buf, len);
case Crc16:
return crc16(buf, len);
default:
break;
}
return 0;
}
long Hash::atol(const char* buf, int len)
{
long v = 0;

13
src/HashFunc.h
View File

@ -33,7 +33,18 @@ public:
{
return mType;
}
long hash(const char* buf, int len) const;
long hash(const char* buf, int len) const
{
switch (mType) {
case Atol:
return atol(buf, len);
case Crc16:
return crc16(buf, len);
default:
break;
}
return 0;
}
long hash(const char* buf, int len, const char* tag) const
{
buf = hashTagStr(buf, len, tag);

2
src/Proxy.cpp
View File

@ -102,8 +102,6 @@ bool Proxy::init(int argc, char* argv[])
}
mLatencyMonitorSet.init(mConf->latencyMonitors());
Request::init();
Response::init();
ListenSocket* s = new ListenSocket(mConf->bind(), SOCK_STREAM);
if (!s->setNonBlock()) {
logError("proxy listener set nonblock fail:%s", StrError());

63
src/Request.cpp
View File

@ -13,39 +13,40 @@ struct GenericRequest
Request::GenericCode code;
Command::Type type;
const char* content;
const Request* req;
};
static GenericRequest GenericRequests[] = {
{Request::Ping, Command::Ping, "*1\r\n$4\r\nping\r\n", nullptr},
{Request::PingServ, Command::PingServ, "*1\r\n$4\r\nping\r\n", nullptr},
{Request::ClusterNodes, Command::ClusterNodes, "*2\r\n$7\r\ncluster\r\n$5\r\nnodes\r\n", nullptr},
{Request::Asking, Command::Asking, "*1\r\n$6\r\nasking\r\n", nullptr},
{Request::Readonly, Command::Readonly, "*1\r\n$8\r\nreadonly\r\n", nullptr},
{Request::UnwatchServ, Command::UnwatchServ, "*1\r\n$7\r\nunwatch\r\n", nullptr},
{Request::DiscardServ, Command::DiscardServ, "*1\r\n$7\r\ndiscard\r\n", nullptr},
{Request::MgetHead, Command::Mget, "*2\r\n$4\r\nmget\r\n", nullptr},
{Request::MsetHead, Command::Mset, "*3\r\n$4\r\nmset\r\n", nullptr},
{Request::MsetnxHead, Command::Msetnx, "*3\r\n$6\r\nmsetnx\r\n", nullptr},
{Request::TouchHead, Command::Touch, "*2\r\n$5\r\ntouch\r\n", nullptr},
{Request::ExistsHead, Command::Exists, "*2\r\n$6\r\nexists\r\n", nullptr},
{Request::DelHead, Command::Del, "*2\r\n$3\r\ndel\r\n", nullptr},
{Request::UnlinkHead, Command::Unlink, "*2\r\n$6\r\nunlink\r\n", nullptr},
{Request::PsubscribeHead,Command::Psubscribe, "*2\r\n$10\r\npsubscribe\r\n", nullptr},
{Request::SubscribeHead,Command::Subscribe, "*2\r\n$9\r\nsubscribe\r\n", nullptr}
static const GenericRequest GenericRequestDefs[] = {
{Request::Ping, Command::Ping, "*1\r\n$4\r\nping\r\n"},
{Request::PingServ, Command::PingServ, "*1\r\n$4\r\nping\r\n"},
{Request::ClusterNodes, Command::ClusterNodes, "*2\r\n$7\r\ncluster\r\n$5\r\nnodes\r\n"},
{Request::Asking, Command::Asking, "*1\r\n$6\r\nasking\r\n"},
{Request::Readonly, Command::Readonly, "*1\r\n$8\r\nreadonly\r\n"},
{Request::UnwatchServ, Command::UnwatchServ, "*1\r\n$7\r\nunwatch\r\n"},
{Request::DiscardServ, Command::DiscardServ, "*1\r\n$7\r\ndiscard\r\n"},
{Request::MgetHead, Command::Mget, "*2\r\n$4\r\nmget\r\n"},
{Request::MsetHead, Command::Mset, "*3\r\n$4\r\nmset\r\n"},
{Request::MsetnxHead, Command::Msetnx, "*3\r\n$6\r\nmsetnx\r\n"},
{Request::TouchHead, Command::Touch, "*2\r\n$5\r\ntouch\r\n"},
{Request::ExistsHead, Command::Exists, "*2\r\n$6\r\nexists\r\n"},
{Request::DelHead, Command::Del, "*2\r\n$3\r\ndel\r\n"},
{Request::UnlinkHead, Command::Unlink, "*2\r\n$6\r\nunlink\r\n"},
{Request::PsubscribeHead,Command::Psubscribe, "*2\r\n$10\r\npsubscribe\r\n"},
{Request::SubscribeHead,Command::Subscribe, "*2\r\n$9\r\nsubscribe\r\n"}
};
thread_local static Request* GenericRequests[Request::CodeSentinel];
void Request::init()
{
BufferPtr buf = BufferAlloc::create();
for (auto& r : GenericRequests) {
for (auto& r : GenericRequestDefs) {
Request* req = new Request();
req->mType= r.type;
if (buf->room() < (int)strlen(r.content)) {
buf = BufferAlloc::create();
}
buf = req->mReq.set(buf, r.content);
r.req = req;
GenericRequests[r.code] = req;
}
}
@ -85,7 +86,7 @@ Request::Request(GenericCode code):
mCreateTime(Util::elapsedUSec()),
mData(nullptr)
{
auto r = GenericRequests[code].req;
auto r = GenericRequests[code];
mType = r->mType;
mReq = r->mReq;
}
@ -110,38 +111,38 @@ void Request::set(const RequestParser& p, Request* leader)
const Request* r = nullptr;
switch (mType) {
case Command::Mget:
r = GenericRequests[MgetHead].req;
r = GenericRequests[MgetHead];
break;
case Command::Mset:
r = GenericRequests[MsetHead].req;
r = GenericRequests[MsetHead];
break;
case Command::Msetnx:
r = GenericRequests[MsetnxHead].req;
r = GenericRequests[MsetnxHead];
break;
case Command::Touch:
r = GenericRequests[TouchHead].req;
r = GenericRequests[TouchHead];
break;
case Command::Exists:
r = GenericRequests[ExistsHead].req;
r = GenericRequests[ExistsHead];
break;
case Command::Del:
r = GenericRequests[DelHead].req;
r = GenericRequests[DelHead];
break;
case Command::Unlink:
r = GenericRequests[UnlinkHead].req;
r = GenericRequests[UnlinkHead];
break;
case Command::Psubscribe:
r = GenericRequests[PsubscribeHead].req;
r = GenericRequests[PsubscribeHead];
break;
case Command::Subscribe:
r = GenericRequests[SubscribeHead].req;
r = GenericRequests[SubscribeHead];
break;
default:
//should never reach
break;
}
mHead = r->mReq;
mReq = p.arg();
mReq = p.request();
mLeader = leader;
if (leader == this) {
if (mType == Command::Mset || mType == Command::Msetnx) {

4
src/Request.h
View File

@ -43,7 +43,9 @@ public:
DelHead,
UnlinkHead,
PsubscribeHead,
SubscribeHead
SubscribeHead,
CodeSentinel
};
static void init();
public:

260
src/RequestParser.cpp
View File

@ -21,8 +21,6 @@ void RequestParser::reset()
mType = Command::None;
mCommand = nullptr;
mReq.clear();
mCmd.clear();
mArg.clear();
mKey.clear();
mStatus = Normal;
mState = Idle;
@ -34,9 +32,8 @@ void RequestParser::reset()
mByteCnt = 0;
}
bool RequestParser::isKey(bool split) const
inline bool RequestParser::isKey(bool split) const
{
if (mCommand) {
switch (mCommand->mode & Command::KeyMask) {
case Command::NoKey:
return false;
@ -51,6 +48,15 @@ bool RequestParser::isKey(bool split) const
default:
return mArgCnt == 1;
}
return false;
}
inline bool RequestParser::isSplit(bool split) const
{
if (mCommand->mode & (Command::MultiKey|Command::MultiKeyVal)) {
return split && mStatus == Normal && isKey(true);
} else if (mCommand->mode & Command::SMultiKey) {
return mStatus == Normal;
}
return false;
}
@ -88,16 +94,21 @@ RequestParser::Status RequestParser::parse(Buffer* buf, int& pos, bool split)
} else if (!isspace(ch)) {
mReq.begin().buf = buf;
mReq.begin().pos = pos;
mCmd.begin() = mReq.begin();
mCmd[0] = tolower(ch);
mArgLen = 1;
mState = InlineCmd;
}
break;
case InlineCmd:
if (isspace(ch)) {
mCmd.end().buf = buf;
mCmd.end().pos = pos;
mCmd[mArgLen < Const::MaxCmdLen ? mArgLen : Const::MaxCmdLen - 1] = '\0';
parseCmd();
mState = ch == '\r' ? InlineLF : InlineArg;
} else {
if (mArgLen < Const::MaxCmdLen) {
mCmd[mArgLen] = tolower(ch);
}
++mArgLen;
}
break;
case InlineArg:
@ -117,7 +128,6 @@ RequestParser::Status RequestParser::parse(Buffer* buf, int& pos, bool split)
if (ch >= '0' && ch <= '9') {
mArgNum = mArgNum * 10 + (ch - '0');
} else if (ch == '\r') {
//mState = mArgNum > 0 ? ArgNumLF : Error;
mArgNum > 0 ? mState = ArgNumLF : error = __LINE__;
} else {
error = __LINE__;
@ -125,26 +135,128 @@ RequestParser::Status RequestParser::parse(Buffer* buf, int& pos, bool split)
break;
case ArgNumLF:
mArgCnt = 0;
//mState = ch == '\n' ? ArgTag : Error;
ch == '\n' ? mState = ArgTag : error = __LINE__;
ch == '\n' ? mState = CmdTag : error = __LINE__;
break;
case ArgTag:
mArgLen = 0;
case CmdTag:
if (ch == '$') {
mState = ArgLen;
if (isKey(split)) {
mArg.begin().buf = buf;
mArg.begin().pos = pos;
mArgLen = 0;
mState = CmdLen;
} else {
error = __LINE__;
}
break;
case CmdLen:
if (ch >= '0' && ch <= '9') {
mArgLen = mArgLen * 10 + (ch - '0');
} else if (ch == '\r') {
mArgLen > 0 ? mState = CmdLenLF : error = __LINE__;
} else {
error = __LINE__;
}
break;
case CmdLenLF:
if (ch == '\n') {
mArgBodyCnt = 0;
mState = mArgLen < Const::MaxCmdLen ? CmdBody : CmdBodyTooLong;
} else {
error = __LINE__;
}
break;
case CmdBody:
if (mArgBodyCnt == mArgLen) {
mCmd[mArgLen] = '\0';
ch == '\r' ? mState = CmdBodyLF : error = __LINE__;
} else {
mCmd[mArgBodyCnt++] = tolower(ch);
}
break;
case CmdBodyTooLong:
if (mArgBodyCnt == mArgLen) {
mCmd[Const::MaxCmdLen - 1] = '\0';
ch == '\r' ? mState = CmdBodyLF : error = __LINE__;
} else {
if (mArgBodyCnt < Const::MaxCmdLen) {
mCmd[mArgBodyCnt] = ch;
}
++mArgBodyCnt;
}
break;
case CmdBodyLF:
if (ch == '\n') {
parseCmd();
if (++mArgCnt == mArgNum) {
mState = Finished;
goto Done;
}
if (mCommand->mode & Command::KeyMask) {
mState = SArgTag;
} else {
mState = KeyTag;
}
} else {
error = __LINE__;
}
break;
case KeyTag:
mArgLen = 0;
ch == '$' ? mState = KeyLen : error = __LINE__;
break;
case KeyLen:
if (ch >= '0' && ch <= '9') {
mArgLen = mArgLen * 10 + (ch - '0');
} else if (ch == '\r') {
mArgLen >= 0 ? mState = KeyLenLF : error = __LINE__;
} else {
error = __LINE__;
}
break;
case KeyLenLF:
if (ch == '\n') {
mArgBodyCnt = 0;
mState = KeyBody;
} else {
error = __LINE__;
}
break;
case KeyBody:
if (mArgBodyCnt == 0) {
mKey.begin().buf = buf;
mKey.begin().pos = pos;
}
if (mArgBodyCnt + (end - cursor) > mArgLen) {
pos += mArgLen - mArgBodyCnt;
cursor = buf->data() + pos;
if (*cursor == '\r') {
mState = KeyBodyLF;
mKey.end().buf = buf;
mKey.end().pos = pos;
} else {
error = __LINE__;
}
} else {
mArgBodyCnt += end - cursor;
pos = buf->length() - 1;
}
break;
case KeyBodyLF:
if (ch == '\n') {
if (++mArgCnt == mArgNum) {
mState = Finished;
goto Done;
}
mState = ArgTag;
} else {
error = __LINE__;
}
break;
case ArgTag:
mArgLen = 0;
ch == '$' ? mState = ArgLen : error = __LINE__;
break;
case ArgLen:
if (ch >= '0' && ch <= '9') {
mArgLen = mArgLen * 10 + (ch - '0');
} else if (ch == '\r') {
//mState = mArgLen >= 0 ? ArgLenLF : Error;
mArgLen >= 0 ? mState = ArgLenLF : error = __LINE__;
} else {
error = __LINE__;
@ -155,31 +267,10 @@ RequestParser::Status RequestParser::parse(Buffer* buf, int& pos, bool split)
ch == '\n' ? mState = ArgBody : error = __LINE__;
break;
case ArgBody:
if (mArgBodyCnt == 0) {
if (mArgCnt == 0) {
mCmd.begin().buf = buf;
mCmd.begin().pos = pos;
} else if (isKey(split)) {
mKey.begin().buf = buf;
mKey.begin().pos = pos;
}
}
if (mArgBodyCnt + (end - cursor) > mArgLen) {
pos += mArgLen - mArgBodyCnt;
cursor = buf->data() + pos;
if (*cursor == '\r') {
mState = ArgBodyLF;
if (mArgCnt == 0) {
mCmd.end().buf = buf;
mCmd.end().pos = pos;
parseCmd();
} else if (isKey(split)) {
mKey.end().buf = buf;
mKey.end().pos = pos;
}
} else {
error = __LINE__;
}
*cursor == '\r' ? mState = ArgBodyLF : error = __LINE__;
} else {
mArgBodyCnt += end - cursor;
pos = buf->length() - 1;
@ -192,16 +283,69 @@ RequestParser::Status RequestParser::parse(Buffer* buf, int& pos, bool split)
goto Done;
} else {
mState = ArgTag;
if (mArgCnt > 1 && isKey(split) && mStatus == Normal &&
(mCommand->mode&(Command::MultiKey|Command::SMultiKey|Command::MultiKeyVal))) {
goto Done;
}
if (mArgCnt > 1 && mCommand && mStatus == Normal && split) {
if (mCommand->isMultiKey()) {
goto Done;
} else if (mCommand->isMultiKeyVal() && (mArgCnt & 1)) {
goto Done;
} else {
error = __LINE__;
}
break;
case SArgTag:
mArgLen = 0;
if (ch == '$') {
mState = SArgLen;
if (isSplit(split)) {
mReq.begin().buf = buf;
mReq.begin().pos = pos;
}
} else {
error = __LINE__;
}
break;
case SArgLen:
if (ch >= '0' && ch <= '9') {
mArgLen = mArgLen * 10 + (ch - '0');
} else if (ch == '\r') {
mArgLen >= 0 ? mState = SArgLenLF : error = __LINE__;
} else {
error = __LINE__;
}
break;
case SArgLenLF:
mArgBodyCnt = 0;
ch == '\n' ? mState = SArgBody : error = __LINE__;
break;
case SArgBody:
if (mArgBodyCnt == 0) {
if (isKey(split)) {
mKey.begin().buf = buf;
mKey.begin().pos = pos;
}
}
if (mArgBodyCnt + (end - cursor) > mArgLen) {
pos += mArgLen - mArgBodyCnt;
cursor = buf->data() + pos;
if (*cursor == '\r') {
mState = SArgBodyLF;
if (isKey(split)) {
mKey.end().buf = buf;
mKey.end().pos = pos;
}
} else {
error = __LINE__;
}
} else {
mArgBodyCnt += end - cursor;
pos = buf->length() - 1;
}
break;
case SArgBodyLF:
if (ch == '\n') {
if (++mArgCnt == mArgNum) {
mState = Finished;
goto Done;
} else {
mState = SArgTag;
if (isSplit(split)) {
goto Done;
}
}
} else {
@ -230,8 +374,6 @@ Done:
mReq.end().buf = buf;
mReq.end().pos = ++pos;
mReq.rewind();
mArg.end() = mReq.end();
mArg.rewind();
if (mState == Finished) {
return mStatus == Normal ? Complete : mStatus;
} else {
@ -242,19 +384,18 @@ Done:
void RequestParser::parseCmd()
{
FuncCallTimer();
SegmentStr<MaxCmdLen> cmd(mCmd);
if (!cmd.complete()) {
if (mArgLen >= Const::MaxCmdLen) {
mStatus = CmdError;
mType = Command::None;
logNotice("unknown request cmd too long:%.*s...",
cmd.length(), cmd.data());
logNotice("unknown request cmd too long:%s...", mCmd);
return;
}
auto c = Command::find(cmd);
auto c = Command::find(mCmd);
if (!c) {
mCommand = &Command::get(Command::None);
mStatus = CmdError;
mType = Command::None;
logNotice("unknown request cmd:%.*s", cmd.length(), cmd.data());
logNotice("unknown request cmd:%s", mCmd);
return;
}
mType = c->type;
@ -269,8 +410,8 @@ void RequestParser::parseCmd()
}
if (mArgNum < c->minArgs || mArgNum > c->maxArgs) {
mStatus = ArgError;
logNotice("request argument is invalid cmd %.*s argnum %d",
cmd.length(), cmd.data(), mArgNum);
logNotice("request argument is invalid cmd %s argnum %d",
mCmd, mArgNum);
return;
}
switch (mType) {
@ -278,9 +419,8 @@ void RequestParser::parseCmd()
case Command::Msetnx:
if (!(mArgNum & 1)) {
mStatus = ArgError;
logNotice("request argument is invalid cmd %.*s argnum %d",
cmd.length(), cmd.data(), mArgNum);
return;
logNotice("request argument is invalid cmd %s argnum %d",
mCmd, mArgNum);
}
break;
default:

40
src/RequestParser.h
View File

@ -14,7 +14,6 @@ class RequestParser
{
public:
static const int MaxAllowInvalidByteCount = 1024;
static const int MaxCmdLen = 32;
enum State
{
Idle, // * or inline command
@ -24,11 +23,27 @@ public:
InlineArg,
ArgNum, // 2
ArgNumLF, // \r\n
CmdTag,
CmdLen,
CmdLenLF,
CmdBody,
CmdBodyTooLong,
CmdBodyLF,
KeyTag,
KeyLen,
KeyLenLF,
KeyBody,
KeyBodyLF,
ArgTag, // $ $
ArgLen, // 3 5
ArgLenLF, // \r\n \r\n
ArgBody, // get hello
ArgBodyLF, // \r\n \r\n
SArgTag, // $ $
SArgLen, // 3 5
SArgLenLF, // \r\n \r\n
SArgBody, // get hello
SArgBodyLF, // \r\n \r\n
Finished,
Error
@ -76,19 +91,7 @@ public:
{
return mArgNum;
}
Segment& arg()
{
return mArg;
}
const Segment& arg() const
{
return mArg;
}
Segment& cmd()
{
return mCmd;
}
const Segment& cmd() const
const char* cmd() const
{
return mCmd;
}
@ -103,14 +106,13 @@ public:
private:
void parseCmd();
bool isKey(bool split) const;
bool isSplit(bool split) const;
private:
Command::Type mType;
const Command* mCommand;
// *2\r\n$3\r\nget\r\n$3\r\nkey\r\n
Segment mReq; // |------------------------------|
Segment mCmd; // |-|
Segment mArg; // |-----------|
Segment mKey; // |-|
Segment mReq;
Segment mKey;
char mCmd[Const::MaxCmdLen];
Status mStatus;
State mState;
bool mInline;

49
src/Response.cpp
View File

@ -12,41 +12,42 @@ struct GenericResponse
Response::GenericCode code;
Reply::Type type;
const char* content;
const Response* res;
};
static GenericResponse GenericResponses[] = {
{Response::Pong, Reply::Status, "+PONG\r\n", nullptr},
{Response::Ok, Reply::Status, "+OK\r\n", nullptr},
static const GenericResponse GenericResponseDefs[] = {
{Response::Pong, Reply::Status, "+PONG\r\n"},
{Response::Ok, Reply::Status, "+OK\r\n"},
{Response::Cmd, Reply::Array, "*0\r\n"},
{Response::UnknownCmd, Reply::Error, "-ERR unknown command\r\n", nullptr},
{Response::ArgWrong, Reply::Error, "-ERR argument wrong\r\n", nullptr},
{Response::InvalidDb, Reply::Error, "-ERR invalid DB index\r\n", nullptr},
{Response::NoPasswordSet, Reply::Error, "-ERR Client sent AUTH, but no password is set\r\n", nullptr},
{Response::InvalidPassword, Reply::Error, "-ERR invalid password\r\n", nullptr},
{Response::Unauth, Reply::Error, "-NOAUTH Authentication required.\r\n", nullptr},
{Response::PermissionDeny, Reply::Error, "-ERR auth permission deny\r\n", nullptr},
{Response::NoServer, Reply::Error, "-ERR no server avaliable\r\n", nullptr},
{Response::NoServerConnection, Reply::Error, "-ERR no server connection avaliable\r\n", nullptr},
{Response::ServerConnectionClose, Reply::Error, "-ERR server connection close\r\n", nullptr},
{Response::DeliverRequestFail, Reply::Error, "-ERR deliver request fail\r\n", nullptr},
{Response::ForbidTransaction, Reply::Error, "-ERR forbid transaction in current server pool\r\n", nullptr},
{Response::ConfigSubCmdUnknown, Reply::Error, "-ERR CONFIG subcommand must be one of GET, SET\r\n", nullptr},
{Response::InvalidScanCursor, Reply::Error, "-ERR invalid cursor\r\n", nullptr},
{Response::ScanEnd, Reply::Array, "*2\r\n$1\r\n0\r\n*0\r\n", nullptr}
{Response::UnknownCmd, Reply::Error, "-ERR unknown command\r\n"},
{Response::ArgWrong, Reply::Error, "-ERR argument wrong\r\n"},
{Response::InvalidDb, Reply::Error, "-ERR invalid DB index\r\n"},
{Response::NoPasswordSet, Reply::Error, "-ERR Client sent AUTH, but no password is set\r\n"},
{Response::InvalidPassword, Reply::Error, "-ERR invalid password\r\n"},
{Response::Unauth, Reply::Error, "-NOAUTH Authentication required.\r\n"},
{Response::PermissionDeny, Reply::Error, "-ERR auth permission deny\r\n"},
{Response::NoServer, Reply::Error, "-ERR no server avaliable\r\n"},
{Response::NoServerConnection, Reply::Error, "-ERR no server connection avaliable\r\n"},
{Response::ServerConnectionClose, Reply::Error, "-ERR server connection close\r\n"},
{Response::DeliverRequestFail, Reply::Error, "-ERR deliver request fail\r\n"},
{Response::ForbidTransaction, Reply::Error, "-ERR forbid transaction in current server pool\r\n"},
{Response::ConfigSubCmdUnknown, Reply::Error, "-ERR CONFIG subcommand must be one of GET, SET\r\n"},
{Response::InvalidScanCursor, Reply::Error, "-ERR invalid cursor\r\n"},
{Response::ScanEnd, Reply::Array, "*2\r\n$1\r\n0\r\n*0\r\n"}
};
thread_local static Response* GenericResponses[Response::CodeSentinel];
void Response::init()
{
BufferPtr buf = BufferAlloc::create();
for (auto& r : GenericResponses) {
for (auto& r : GenericResponseDefs) {
Response* res = new Response();
res->mType = r.type;
if (buf->room() < (int)strlen(r.content)) {
buf = BufferAlloc::create();
}
buf = res->mRes.set(buf, r.content);
r.res = res;
GenericResponses[r.code] = res;
}
}
@ -60,7 +61,7 @@ Response::Response(GenericCode code):
mType(Reply::None),
mInteger(0)
{
auto r = GenericResponses[code].res;
auto r = GenericResponses[code];
mType = r->mType;
mRes = r->mRes;
}
@ -74,7 +75,6 @@ void Response::set(const ResponseParser& p)
{
mType = p.type();
mInteger = p.integer();
mHead.clear();
mRes = p.response();
}
@ -83,14 +83,12 @@ void Response::set(int64_t num)
{
mType = Reply::Integer;
mInteger = num;
mHead.clear();
mRes.fset(nullptr, ":%ld\r\n", num);
}
void Response::setStr(const char* str, int len)
{
mType = Reply::String;
mHead.clear();
if (len < 0) {
len = strlen(str);
}
@ -100,7 +98,6 @@ void Response::setStr(const char* str, int len)
void Response::setErr(const char* str, int len)
{
mType = Reply::Error;
mHead.clear();
if (len < 0) {
len = strlen(str);
}

4
src/Response.h
View File

@ -37,7 +37,9 @@ public:
ForbidTransaction,
ConfigSubCmdUnknown,
InvalidScanCursor,
ScanEnd
ScanEnd,
CodeSentinel
};
static void init();
static Response* create(GenericCode code, Request* req = nullptr);

23
src/SentinelServerPool.cpp
View File

@ -39,11 +39,11 @@ void SentinelServerPool::init(const SentinelServerPoolConf& conf)
mSentinels[i++] = s;
mServs[s->addr()] = s;
}
mGroups.resize(conf.groups.size());
mGroupPool.resize(conf.groups.size());
i = 0;
for (auto& gc : conf.groups) {
ServerGroup* g = new ServerGroup(this, gc.name);
mGroups[i++] = g;
mGroupPool[i++] = g;
for (auto& sc : gc.servers) {
Server* s = new Server(this, sc.addr, true);
s->setPassword(sc.password.empty() ? conf.password : sc.password);
@ -55,7 +55,7 @@ void SentinelServerPool::init(const SentinelServerPoolConf& conf)
}
}
Server* SentinelServerPool::getServer(Handler* h, Request* req) const
Server* SentinelServerPool::getServer(Handler* h, Request* req, const String& key) const
{
FuncCallTimer();
switch (req->type()) {
@ -76,22 +76,21 @@ Server* SentinelServerPool::getServer(Handler* h, Request* req) const
default:
break;
}
if (mGroups.size() == 1) {
return mGroups[0]->getServer(h, req);
} else if (mGroups.size() > 1) {
if (mGroupPool.size() == 1) {
return mGroupPool[0]->getServer(h, req);
} else if (mGroupPool.size() > 1) {
switch (mDist) {
case Distribution::Modula:
{
SegmentStr<Const::MaxKeyLen> key(req->key());
long idx = mHash.hash(key.data(), key.length(), mHashTag);
idx %= mGroups.size();
return mGroups[idx]->getServer(h, req);
idx %= mGroupPool.size();
return mGroupPool[idx]->getServer(h, req);
}
break;
case Distribution::Random:
{
int idx = h->rand() % mGroups.size();
return mGroups[idx]->getServer(h, req);
int idx = h->rand() % mGroupPool.size();
return mGroupPool[idx]->getServer(h, req);
}
break;
default:
@ -104,7 +103,7 @@ Server* SentinelServerPool::getServer(Handler* h, Request* req) const
void SentinelServerPool::refreshRequest(Handler* h)
{
logDebug("h %d update sentinel server pool", h->id());
for (auto g : mGroups) {
for (auto g : mGroupPool) {
RequestPtr req = RequestAlloc::create();
req->setSentinels(g->name());
req->setData(g);

3
src/SentinelServerPool.h
View File

@ -19,7 +19,7 @@ public:
SentinelServerPool(Proxy* p);
~SentinelServerPool();
void init(const SentinelServerPoolConf& conf);
Server* getServer(Handler* h, Request* req) const;
Server* getServer(Handler* h, Request* req, const String& key) const;
Server* iter(int& cursor) const
{
return ServerPool::iter(mServPool, cursor);
@ -34,7 +34,6 @@ private:
private:
std::vector<Server*> mSentinels;
std::vector<Server*> mServPool;
std::vector<ServerGroup*> mGroups;
Distribution mDist;
Hash mHash;
char mHashTag[2];

1
src/ServerGroup.cpp
View File

@ -56,6 +56,7 @@ Server* ServerGroup::getServer(Handler* h, Request* req) const
}
if (s->role() == Server::Master) {
serv = s;
break;
}
}
} else if (auto dataCenter = mPool->proxy()->dataCenter()) {

10
src/ServerPool.h
View File

@ -78,9 +78,9 @@ public:
auto it = mServs.find(addr);
return it == mServs.end() ? nullptr : it->second;
}
Server* getServer(Handler* h, Request* req) const
Server* getServer(Handler* h, Request* req, const String& key) const
{
return mGetServerFunc(this, h, req);
return mGetServerFunc(this, h, req, key);
}
Server* iter(int& cursor) const
{
@ -92,7 +92,7 @@ public:
}
void handleResponse(Handler* h, ConnectConnection* s, Request* req, Response* res);
protected:
typedef Server* (*GetServerFunc)(const ServerPool* p, Handler* h, Request* req);
typedef Server* (*GetServerFunc)(const ServerPool* p, Handler* h, Request* req, const String& key);
typedef Server* (*IterFunc)(const ServerPool* p, int& cursor);
typedef void (*RefreshRequestFunc)(ServerPool* p, Handler* h);
typedef void (*HandleResponseFunc)(ServerPool* p, Handler* h, ConnectConnection* s, Request* req, Response* res);
@ -147,9 +147,9 @@ public:
{
}
private:
static Server* getServer(const ServerPool* p, Handler* h, Request* req)
static Server* getServer(const ServerPool* p, Handler* h, Request* req, const String& key)
{
return static_cast<const T*>(p)->getServer(h, req);
return static_cast<const T*>(p)->getServer(h, req, key);
}
static Server* iter(const ServerPool* p, int& cursor)
{

6
src/Timer.cpp
View File

@ -48,12 +48,12 @@ void TimerPoint::report()
std::sort(points, points + cnt,
[](TimerPoint* p1, TimerPoint* p2)
{return p1->elapsed() > p2->elapsed();});
printf("%16s %12s %8s %s\n","Total(us)", "Count", "Avg(us)", "Point" );
printf("%16s %12s %8s %s\n","Total(us)", "Count", "Avg(ns)", "Point" );
for (i = 0; i < cnt; ++i) {
auto p = points[i];
printf("%16ld %12ld %8ld %s\n",
printf("%16ld %12ld %9ld %s\n",
p->elapsed(), p->count(),
p->elapsed()/p->count(), p->key());
p->elapsed()*1000/p->count(), p->key());
}
if (points != points0) {
delete[] points;