Merge pull request #185 from joyieldInc/fix/issue124

avoid to use leader's connection for follow request

conf/command.conf

@@ -0,0 +1,94 @@
+## Custom Command
+## CustomCommand {
+##    command {                                      #command string, must be lowercase
+##        [Mode read|write|admin[|[keyAt2|keyAt3]]   #default write, default key position is 1
+##        [MinArgs [2-]]                             #default 2, including command itself
+##        [MaxArgs [2-]]                             #default 2, must be MaxArgs >= MinArgs
+##    }...
+## }
+
+## Currently support maximum 16 custom commands
+
+## Example:
+#CustomCommand {
+##------------------------------------------------------------------------
+#     custom.ttl {
+#         Mode keyAt2
+#         MinArgs 3
+#         MaxArgs 3
+#     }
+####  custom.ttl miliseconds key
+####  Mode = write|keyAt2, MinArgs/MaxArgs = 3 = command + miliseconds + key
+##------------------------------------------------------------------------
+## from redis source src/modules/hello.c
+#     hello.push.native {
+#         MinArgs 3
+#         MaxArgs 3
+#     }
+####  hello.push.native key value
+####  Mode = write, MinArgs/MaxArgs = 3 = command) + key + value
+##------------------------------------------------------------------------
+#     hello.repl2 {
+#     }
+####  hello.repl2 <list-key>
+####  Mode = write, MinArgs/MaxArgs = 2 = command + list-key
+##------------------------------------------------------------------------
+#     hello.toggle.case {
+#     }
+####  hello.toggle.case key
+####  Mode = write, MinArgs/MaxArgs = 2 = command + key
+##------------------------------------------------------------------------
+#     hello.more.expire {
+#         MinArgs 3
+#         MaxArgs 3
+#     }
+####  hello.more.expire key milliseconds
+####  Mode = write, MinArgs/MaxArgs = 3 = command + key + milliseconds
+##------------------------------------------------------------------------
+#     hello.zsumrange {
+#         MinArgs 4
+#         MaxArgs 4
+#         Mode read
+#     }
+####  hello.zsumrange key startscore endscore
+####  Mode = read, MinArgs/MaxArgs = 4 = command + key + startscore + endscore
+##------------------------------------------------------------------------
+#     hello.lexrange {
+#         MinArgs 6
+#         MaxArgs 6
+#         Mode read
+#     }
+####  hello.lexrange key min_lex max_lex min_age max_age
+####  Mode = read, MinArgs/MaxArgs = 6 = command + key + min_lex + max_lex + min_age + max_age
+##------------------------------------------------------------------------
+#     hello.hcopy {
+#         MinArgs 4
+#         MaxArgs 4
+#     }
+#### hello.hcopy key srcfield dstfield
+#### Mode = write, MinArgs/MaxArgs = 4 = command + key + srcfield) + dstfield
+##------------------------------------------------------------------------
+## from redis source src/modules/hellotype.c
+#     hellotype.insert {
+#         MinArgs 3
+#         MaxArgs 3
+#     }
+####  hellotype.insert key value
+####  Mode = write, MinArgs/MaxArgs = 3 = command + key + value
+##------------------------------------------------------------------------
+#     hellotype.range {
+#         MinArgs 4
+#         MaxArgs 4
+#         Mode read
+#     }
+####  hellotype.range key first count
+####  Mode = read, MinArgs/MaxArgs = 4 = command + key + first + count
+##------------------------------------------------------------------------
+#     hellotype.len {
+#         Mode read
+#     }
+####  hellotype.len key
+####  Mode = read, MinArgs/MaxArgs = 2 = command + key
+##------------------------------------------------------------------------
+#}
+

conf/predixy.conf

@@ -93,6 +93,10 @@ Include try.conf
 # Include dc.conf
 
 
+################################### COMMAND ####################################
+## Custom command define, see command.conf
+#Include command.conf
+
 ################################### LATENCY ####################################
 ## Latency monitor define, see latency.conf
 Include latency.conf

conf/standalone.conf

@@ -0,0 +1,71 @@
+## redis standalone server pool define
+
+##StandaloneServerPool {
+##    [Password xxx]                        #default no
+##    [Databases number]                    #default 1
+##    Hash atol|crc16
+##    [HashTag "xx"]                        #default no
+##    Distribution modula|random
+##    [MasterReadPriority [0-100]]          #default 50
+##    [StaticSlaveReadPriority [0-100]]     #default 0
+##    [DynamicSlaveReadPriority [0-100]]    #default 0
+##    RefreshMethod fixed|sentinel          #
+##    [RefreshInterval number[s|ms|us]]     #default 1, means 1 second
+##    [ServerTimeout number[s|ms|us]]       #default 0, server connection socket read/write timeout
+##    [ServerFailureLimit number]           #default 10
+##    [ServerRetryTimeout number[s|ms|us]]  #default 1
+##    [KeepAlive seconds]                   #default 0, server connection tcp keepalive
+##    Sentinels [sentinel-password] {
+##        + addr
+##        ...
+##    }
+##    Group xxx {
+##        [+ addr]                          #if RefreshMethod==fixed: the first addr is master in a group, then all addrs is slaves in this group
+##        ...
+##    }
+##}
+
+
+## Examples:
+#StandaloneServerPool {
+#    Databases 16
+#    Hash crc16
+#    HashTag "{}"
+#    Distribution modula
+#    MasterReadPriority 60
+#    StaticSlaveReadPriority 50
+#    DynamicSlaveReadPriority 50
+#    RefreshMethod sentinel
+#    RefreshInterval 1
+#    ServerTimeout 1
+#    ServerFailureLimit 10
+#    ServerRetryTimeout 1
+#    KeepAlive 120
+#    Sentinels {
+#        + 10.2.2.2:7500
+#        + 10.2.2.3:7500
+#        + 10.2.2.4:7500
+#    }
+#    Group shard001 {
+#    }
+#    Group shard002 {
+#    }
+#}
+
+#StandaloneServerPool {
+#    Databases 16
+#    Hash crc16
+#    HashTag "{}"
+#    Distribution modula
+#    MasterReadPriority 60
+#    StaticSlaveReadPriority 50
+#    DynamicSlaveReadPriority 50
+#    RefreshMethod fixed
+#    ServerTimeout 1
+#    ServerFailureLimit 10
+#    ServerRetryTimeout 1
+#    KeepAlive 120
+#    Group shard001 {
+#       + 10.2.3.2:6379
+#    }
+#}

doc/config_CN.md

@@ -151,9 +151,9 @@ predixy扩展了redis中AUTH命令的功能，支持定义多个认证密码，
     Authority {
         Auth [password] {
             Mode read|write|admin
-            [KeyPredix Predix...]
-            [ReadKeyPredix Predix...]
-            [WriteKeyPredix Predix...]
+            [KeyPrefix Prefix...]
+            [ReadKeyPrefix Prefix...]
+            [WriteKeyPrefix Prefix...]
         }...
     }
 

src/AcceptConnection.h

@@ -29,6 +29,7 @@ public:
     typedef AcceptConnection Value;
     typedef ListNode<AcceptConnection, SharePtr<AcceptConnection>> ListNodeType;
     typedef DequeNode<AcceptConnection, SharePtr<AcceptConnection>> DequeNodeType;
+    typedef Alloc<AcceptConnection, Const::AcceptConnectionAllocCacheSize> Allocator;
 public:
     AcceptConnection(int fd, sockaddr* addr, socklen_t len);
     ~AcceptConnection();
@@ -97,6 +98,6 @@ private:
 
 typedef List<AcceptConnection> AcceptConnectionList;
 typedef Deque<AcceptConnection> AcceptConnectionDeque;
-typedef Alloc<AcceptConnection, Const::AcceptConnectionAllocCacheSize> AcceptConnectionAlloc;
+typedef AcceptConnection::Allocator AcceptConnectionAlloc;
 
 #endif

src/Alloc.h

@@ -76,7 +76,7 @@ public:
         }
         UsedMemory += allocSize<T>();
         if (MaxMemory == 0 || UsedMemory <= MaxMemory) {
-            void* p = ::operator new(allocSize<T>());
+            void* p = ::operator new(allocSize<T>(), std::nothrow);
             if (p) {
                 try {
                     obj = new (p) T(args...);
@@ -145,7 +145,7 @@ public:
     {
         int n = --mCnt;
         if (n == 0) {
-            Alloc<T>::destroy(static_cast<T*>(this));
+            T::Allocator::destroy(static_cast<T*>(this));
         } else if (n < 0) {
             logError("unref object %p with cnt %d", this, n);
             abort();

src/Buffer.h

@@ -23,6 +23,7 @@ class Buffer :
     public RefCntObj<Buffer>
 {
 public:
+    typedef Alloc<Buffer, Const::BufferAllocCacheSize> Allocator;
     static const int MaxBufFmtAppendLen = 8192;
 public:
     Buffer& operator=(const Buffer&);
@@ -92,12 +93,13 @@ private:
 };
 
 typedef List<Buffer> BufferList;
+typedef Buffer::Allocator BufferAlloc;
+
 template<>
 inline int allocSize<Buffer>()
 {
     return Buffer::getSize() + sizeof(Buffer);
 }
-typedef Alloc<Buffer, Const::BufferAllocCacheSize> BufferAlloc;
 
 struct BufferPos
 {

src/Command.cpp

@@ -9,8 +9,9 @@
 #include <map>
 #include "String.h"
 #include "Command.h"
+#include "Conf.h"
 
-const Command Command::CmdPool[Sentinel] = {
+Command Command::CmdPool[AvailableCommands] = {
     {None,              "",                 0,  MaxArgs,   Read},
     {Ping,              "ping",             1,  2,         Read},
     {PingServ,          "ping",             1,  2,         Inner},
@@ -86,8 +87,8 @@ const Command Command::CmdPool[Sentinel] = {
     {Setrange,          "setrange",         4,  4,         Write},
     {Strlen,            "strlen",           2,  2,         Read},
     {Hdel,              "hdel",             3,  MaxArgs,   Write},
-    {Hexists,           "hexists",          3,  3,         Write},
-    {Hget,              "hget",             3,  3,         Write},
+    {Hexists,           "hexists",          3,  3,         Read},
+    {Hget,              "hget",             3,  3,         Read},
     {Hgetall,           "hgetall",          2,  2,         Read},
     {Hincrby,           "hincrby",          4,  4,         Write},
     {Hincrbyfloat,      "hincrbyfloat",     4,  4,         Write},
@@ -96,7 +97,7 @@ const Command Command::CmdPool[Sentinel] = {
     {Hmget,             "hmget",            3,  MaxArgs,   Read},
     {Hmset,             "hmset",            4,  MaxArgs,   Write},
     {Hscan,             "hscan",            3,  7,         Read},
-    {Hset,              "hset",             4,  4,         Write},
+    {Hset,              "hset",             4,  MaxArgs,   Write},
     {Hsetnx,            "hsetnx",           4,  4,         Write},
     {Hstrlen,           "hstrlen",          3,  3,         Read},
     {Hvals,             "hvals",            2,  2,         Read},
@@ -109,7 +110,7 @@ const Command Command::CmdPool[Sentinel] = {
     {Lpop,              "lpop",             2,  2,         Write},
     {Lpush,             "lpush",            3,  MaxArgs,   Write},
     {Lpushx,            "lpushx",           3,  3,         Write},
-    {Lrange,            "lrange",           4,  4,         Write},
+    {Lrange,            "lrange",           4,  4,         Read},
     {Lrem,              "lrem",             4,  4,         Write},
     {Lset,              "lset",             4,  4,         Write},
     {Ltrim,             "ltrim",            4,  4,         Write},
@@ -138,6 +139,8 @@ const Command Command::CmdPool[Sentinel] = {
     {Zincrby,           "zincrby",          4,  4,         Write},
     {Zinterstore,       "zinterstore",      4,  MaxArgs,   Write},
     {Zlexcount,         "zlexcount",        4,  4,         Read},
+    {Zpopmax,           "zpopmax",          2,  3,         Write},
+    {Zpopmin,           "zpopmin",          2,  3,         Write},
     {Zrange,            "zrange",           4,  5,         Read},
     {Zrangebylex,       "zrangebylex",      4,  7,         Read},
     {Zrangebyscore,     "zrangebyscore",    4,  8,         Read},
@@ -171,11 +174,14 @@ const Command Command::CmdPool[Sentinel] = {
     {SubMsg,            "\000SubMsg",       0,  0,         Admin}
 };
 
+int Command::Sentinel = Command::MaxCommands;
 Command::CommandMap Command::CmdMap;
+
 void Command::init()
 {
     int type = 0;
-    for (auto& i : CmdPool) {
+    for (auto j = 0; j < MaxCommands; j++) {
+        const auto& i = CmdPool[j];
         if (i.type != type) {
             Throw(InitFail, "command %s unmatch the index in commands table", i.name);
         }
@@ -187,3 +193,19 @@ void Command::init()
     }
 }
 
+void Command::addCustomCommand(const CustomCommandConf& ccc) {
+    if (Sentinel >= AvailableCommands) {
+        Throw(InitFail, "too many custom commands(>%d)", MaxCustomCommands);
+    }
+    if (nullptr != find(ccc.name)) {
+        Throw(InitFail, "custom command %s is duplicated", ccc.name.c_str()); 
+    }
+    auto* p = &CmdPool[Sentinel];
+    p->name = ccc.name.c_str();
+    p->minArgs = ccc.minArgs;
+    p->maxArgs = ccc.maxArgs;
+    p->mode = ccc.mode;
+    p->type = (Command::Type)Sentinel++;
+    CmdMap[ccc.name] = p;
+}
+

src/Command.h

@@ -11,6 +11,8 @@
 #include "Exception.h"
 #include "HashFunc.h"
 
+struct CustomCommandConf;
+
 class Command
 {
 public:
@@ -154,6 +156,8 @@ public:
         Zincrby,
         Zinterstore,
         Zlexcount,
+        Zpopmax,
+        Zpopmin,
         Zrange,
         Zrangebylex,
         Zrangebyscore,
@@ -189,7 +193,9 @@ public:
         Unsubscribe,
         SubMsg,
 
-        Sentinel
+        MaxCommands,
+        MaxCustomCommands = 16,
+        AvailableCommands = MaxCommands + MaxCustomCommands,
     };
     enum Mode
     {
@@ -249,9 +255,11 @@ public:
         auto it = CmdMap.find(cmd);
         return it == CmdMap.end() ? nullptr : it->second;
     }
+    static void addCustomCommand(const CustomCommandConf& pc);
+    static int Sentinel;
 private:
     static const int MaxArgs = 100000000;
-    static const Command CmdPool[Sentinel];
+    static Command CmdPool[];
     class H
     {
     public:

src/Common.h

@@ -10,7 +10,7 @@
 #include <limits.h>
 
 #define _PREDIXY_NAME_      "predixy"
-#define _PREDIXY_VERSION_   "1.0.4"
+#define _PREDIXY_VERSION_   "1.0.5"
 
 namespace Const
 {
@@ -32,8 +32,8 @@ namespace Const
     static const int MaxCmdLen = 32;
     static const int MaxKeyLen = 512;
     static const int BufferAllocCacheSize = 64;
-    static const int RequestAllocCacheSize = 32;
-    static const int ResponseAllocCacheSize = 32;
+    static const int RequestAllocCacheSize = 128;
+    static const int ResponseAllocCacheSize = 128;
     static const int AcceptConnectionAllocCacheSize = 32;
     static const int ConnectConnectionAllocCacheSize = 4;
 };

src/Conf.cpp

@@ -6,6 +6,7 @@
 
 #include <ctype.h>
 #include <iostream>
+#include <sstream>
 #include <fstream>
 #include "LogFileSink.h"
 #include "ServerPool.h"
@@ -32,6 +33,13 @@ bool ServerConf::parse(ServerConf& s, const char* str)
     return !s.addr.empty();
 }
 
+void CustomCommandConf::init(CustomCommandConf&c, const char* name) {
+    c.name = name;
+    c.minArgs = 2;
+    c.maxArgs = 2;
+    c.mode = Command::Write;
+}
+
 Conf::Conf():
     mBind("0.0.0.0:7617"),
     mWorkerThreads(1),
@@ -117,8 +125,9 @@ void Conf::setGlobal(const ConfParser::Node* node)
 {
     const ConfParser::Node* authority = nullptr;
     const ConfParser::Node* clusterServerPool = nullptr;
-    const ConfParser::Node* sentinelServerPool = nullptr;
+    const ConfParser::Node* standaloneServerPool = nullptr;
     const ConfParser::Node* dataCenter = nullptr;
+    std::vector<const ConfParser::Node*> latencyMonitors;
     for (auto p = node; p; p = p->next) {
         if (setStr(mName, "Name", p)) {
         } else if (setStr(mBind, "Bind", p)) {
@@ -147,16 +156,19 @@ void Conf::setGlobal(const ConfParser::Node* node)
         } else if (setInt(mLogSample[LogLevel::Warn], "LogWarnSample", p)) {
         } else if (setInt(mLogSample[LogLevel::Error], "LogErrorSample", p)) {
         } else if (strcasecmp(p->key.c_str(), "LatencyMonitor") == 0) {
-            mLatencyMonitors.push_back(LatencyMonitorConf{});
-            setLatencyMonitor(mLatencyMonitors.back(), p);
+            latencyMonitors.push_back(p);
         } else if (strcasecmp(p->key.c_str(), "Authority") == 0) {
             authority = p;
         } else if (strcasecmp(p->key.c_str(), "ClusterServerPool") == 0) {
             clusterServerPool = p;
         } else if (strcasecmp(p->key.c_str(), "SentinelServerPool") == 0) {
-            sentinelServerPool = p;
+            standaloneServerPool = p;
+        } else if (strcasecmp(p->key.c_str(), "StandaloneServerPool") == 0) {
+            standaloneServerPool = p;
         } else if (strcasecmp(p->key.c_str(), "DataCenter") == 0) {
             dataCenter = p;
+        } else if (strcasecmp(p->key.c_str(), "CustomCommand") == 0) {
+            setCustomCommand(p);
         } else {
             Throw(UnknownKey, "%s:%d unknown key %s", p->file, p->line, p->key.c_str());
         }
@@ -164,20 +176,27 @@ void Conf::setGlobal(const ConfParser::Node* node)
     if (authority) {
         setAuthority(authority);
     }
-    if (clusterServerPool && sentinelServerPool) {
-        Throw(LogicError, "Can't define ClusterServerPool and SentinelServerPool at the same time");
+    if (clusterServerPool && standaloneServerPool) {
+        Throw(LogicError, "Can't define ClusterServerPool/StandaloneServerPool at the same time");
     } else if (clusterServerPool) {
         setClusterServerPool(clusterServerPool);
         mServerPoolType = ServerPool::Cluster;
-    } else if (sentinelServerPool) {
-        setSentinelServerPool(sentinelServerPool);
-        mServerPoolType = ServerPool::Sentinel;
+    } else if (standaloneServerPool) {
+        if (strcasecmp(standaloneServerPool->key.c_str(), "SentinelServerPool") == 0) {
+            mStandaloneServerPool.refreshMethod = ServerPoolRefreshMethod::Sentinel;
+        }
+        setStandaloneServerPool(standaloneServerPool);
+        mServerPoolType = ServerPool::Standalone;
     } else {
         Throw(LogicError, "Must define a server pool");
     }
     if (dataCenter) {
         setDataCenter(dataCenter);
     }
+    for (auto& latencyMonitor : latencyMonitors) {
+        mLatencyMonitors.push_back(LatencyMonitorConf{});
+        setLatencyMonitor(mLatencyMonitors.back(), latencyMonitor);
+    }
 }
 
 static void setKeyPrefix(std::vector<std::string>& dat, const std::string& v)
@@ -239,28 +258,21 @@ void Conf::setAuthority(const ConfParser::Node* node)
 
 bool Conf::setServerPool(ServerPoolConf& sp, const ConfParser::Node* p)
 {
+    bool ret = true;
     if (setStr(sp.password, "Password", p)) {
-        return true;
     } else if (setInt(sp.masterReadPriority, "MasterReadPriority", p, 0, 100)) {
-        return true;
     } else if (setInt(sp.staticSlaveReadPriority, "StaticSlaveReadPriority", p, 0, 100)) {
-        return true;
     } else if (setInt(sp.dynamicSlaveReadPriority, "DynamicSlaveReadPriority", p, 0, 100)) {
-        return true;
     } else if (setDuration(sp.refreshInterval, "RefreshInterval", p)) {
-        return true;
     } else if (setDuration(sp.serverTimeout, "ServerTimeout", p)) {
-        return true;
     } else if (setInt(sp.serverFailureLimit, "ServerFailureLimit", p, 1)) {
-        return true;
     } else if (setDuration(sp.serverRetryTimeout, "ServerRetryTimeout", p)) {
-        return true;
     } else if (setInt(sp.keepalive, "KeepAlive", p, 0)) {
-        return true;
     } else if (setInt(sp.databases, "Databases", p, 1, 128)) {
-        return true;
+    } else {
+        ret = false;
     }
-    return false;
+    return ret;
 }
 
 void Conf::setClusterServerPool(const ConfParser::Node* node)
@@ -284,41 +296,47 @@ void Conf::setClusterServerPool(const ConfParser::Node* node)
     }
 }
 
-void Conf::setSentinelServerPool(const ConfParser::Node* node)
+void Conf::setStandaloneServerPool(const ConfParser::Node* node)
 {
     if (!node->sub) {
-        Throw(InvalidValue, "%s:%d SentinelServerPool require scope value", node->file, node->line);
+        Throw(InvalidValue, "%s:%d StandaloneServerPool require scope value", node->file, node->line);
     }
-    mSentinelServerPool.hashTag[0] = '\0';
-    mSentinelServerPool.hashTag[1] = '\0';
+    mStandaloneServerPool.hashTag[0] = '\0';
+    mStandaloneServerPool.hashTag[1] = '\0';
     for (auto p = node->sub; p; p = p->next) {
-        if (setServerPool(mSentinelServerPool, p)) {
+        if (setServerPool(mStandaloneServerPool, p)) {
+        } else if (strcasecmp(p->key.c_str(), "RefreshMethod") == 0) {
+            try {
+                mStandaloneServerPool.refreshMethod = ServerPoolRefreshMethod::parse(p->val.c_str());
+            } catch (ServerPoolRefreshMethod::InvalidEnumValue& excp) {
+                Throw(InvalidValue, "%s:%d unknown RefreshMethod:%s", p->file, p->line, p->val.c_str());
+            }
         } else if (strcasecmp(p->key.c_str(), "Distribution") == 0) {
-            mSentinelServerPool.dist = Distribution::parse(p->val.c_str());
-            if (mSentinelServerPool.dist == Distribution::None) {
+            mStandaloneServerPool.dist = Distribution::parse(p->val.c_str());
+            if (mStandaloneServerPool.dist == Distribution::None) {
                 Throw(InvalidValue, "%s:%d unknown Distribution", p->file, p->line);
             }
         } else if (strcasecmp(p->key.c_str(), "Hash") == 0) {
-            mSentinelServerPool.hash = Hash::parse(p->val.c_str());
-            if (mSentinelServerPool.hash == Hash::None) {
+            mStandaloneServerPool.hash = Hash::parse(p->val.c_str());
+            if (mStandaloneServerPool.hash == Hash::None) {
                 Throw(InvalidValue, "%s:%d unknown Hash", p->file, p->line);
             }
         } else if (strcasecmp(p->key.c_str(), "HashTag") == 0) {
             if (p->val.empty()) {
-                mSentinelServerPool.hashTag[0] = '\0';
-                mSentinelServerPool.hashTag[1] = '\0';
+                mStandaloneServerPool.hashTag[0] = '\0';
+                mStandaloneServerPool.hashTag[1] = '\0';
             } else if (p->val.size() == 2) {
-                mSentinelServerPool.hashTag[0] = p->val[0];
-                mSentinelServerPool.hashTag[1] = p->val[1];
+                mStandaloneServerPool.hashTag[0] = p->val[0];
+                mStandaloneServerPool.hashTag[1] = p->val[1];
             } else {
                 Throw(InvalidValue, "%s:%d HashTag invalid", p->file, p->line);
             }
-        } else if (setServers(mSentinelServerPool.sentinels, "Sentinels", p)) {
-            mSentinelServerPool.sentinelPassword = p->val;
+        } else if (setServers(mStandaloneServerPool.sentinels, "Sentinels", p)) {
+            mStandaloneServerPool.sentinelPassword = p->val;
         } else if (strcasecmp(p->key.c_str(), "Group") == 0) {
-            mSentinelServerPool.groups.push_back(ServerGroupConf{p->val});
+            mStandaloneServerPool.groups.push_back(ServerGroupConf{p->val});
             if (p->sub) {
-                auto& g = mSentinelServerPool.groups.back();
+                auto& g = mStandaloneServerPool.groups.back();
                 setServers(g.servers, "Group", p);
             }
         } else {
@@ -326,18 +344,31 @@ void Conf::setSentinelServerPool(const ConfParser::Node* node)
                     p->file, p->line, p->key.c_str());
         }
     }
-    if (mSentinelServerPool.sentinels.empty()) {
-        Throw(LogicError, "SentinelServerPool no sentinel server");
+    if (mStandaloneServerPool.groups.empty()) {
+        Throw(LogicError, "StandaloneServerPool no server group");
     }
-    if (mSentinelServerPool.groups.empty()) {
-        Throw(LogicError, "SentinelServerPool no server group");
+    if (mStandaloneServerPool.refreshMethod == ServerPoolRefreshMethod::None) {
+        Throw(LogicError, "StandaloneServerPool must define RefreshMethod");
+    } else if (mStandaloneServerPool.refreshMethod == ServerPoolRefreshMethod::Sentinel) {
+        if (mStandaloneServerPool.sentinels.empty()) {
+            Throw(LogicError, "StandaloneServerPool with RefreshMethod(sentinel) but no sentinel servers");
         }
-    if (mSentinelServerPool.groups.size() > 1) {
-        if (mSentinelServerPool.dist == Distribution::None) {
-            Throw(LogicError, "SentinelServerPool must define Dsitribution in multi groups");
+    } else {
+        if (!mStandaloneServerPool.sentinels.empty()) {
+            Throw(LogicError, "StandaloneServerPool with Sentinels but RefreshMethod is not sentinel");
         }
-        if (mSentinelServerPool.hash == Hash::None) {
-            Throw(LogicError, "SentinelServerPool must define Hash in multi groups");
+        for (auto& g : mStandaloneServerPool.groups) {
+            if (g.servers.empty()) {
+                Throw(LogicError, "Group(%s) must add servers", g.name.c_str());
+            }
+        }
+    }
+    if (mStandaloneServerPool.groups.size() > 1) {
+        if (mStandaloneServerPool.dist == Distribution::None) {
+            Throw(LogicError, "StandaloneServerPool must define Dsitribution in multi groups");
+        }
+        if (mStandaloneServerPool.hash == Hash::None) {
+            Throw(LogicError, "StandaloneServerPool must define Hash in multi groups");
         }
     }
 }
@@ -358,6 +389,83 @@ void Conf::setDataCenter(const ConfParser::Node* node)
     }
 }
 
+void Conf::setCustomCommand(const ConfParser::Node* node)
+{
+    if (!node->sub) {
+        Throw(InvalidValue, "%s:%d CustomCommand require scope value", node->file, node->line);
+    }
+    for (auto p = node->sub; p; p = p->next) {
+        mCustomCommands.push_back(CustomCommandConf{});
+        auto& cc = mCustomCommands.back();
+        CustomCommandConf::init(cc, p->key.c_str());
+        auto s = p->sub;
+        for (;s ; s = s->next) {
+            if (setInt(cc.minArgs, "MinArgs", s, 2)) {
+            } else if (setInt(cc.maxArgs, "MaxArgs", s, 2, 9999)) {
+            } else if (setCommandMode(cc.mode, "Mode", s)) {
+            } else {
+                Throw(UnknownKey, "%s:%d unknown key %s", s->file, s->line, s->key.c_str());
+            }
+        }
+        if (cc.maxArgs < cc.minArgs) {
+           Throw(InvalidValue, "%s:%d must be MaxArgs >= MinArgs", p->file, p->line);
+        }
+    }
+    for (const auto& cc : mCustomCommands) {
+        Command::addCustomCommand(cc);
+    }
+}
+
+bool Conf::setCommandMode(int& mode, const char* name, const ConfParser::Node* n, const int defaultMode)
+{
+    if (strcasecmp(name, n->key.c_str()) != 0) {
+        return false;
+    }
+
+    if (n->val.size() == 0) {
+        mode = defaultMode;
+    } else {
+        mode = 0;
+        std::string mask;
+        std::istringstream is(n->val);
+        while (std::getline(is, mask, '|')) {
+            if ((strcasecmp(mask.c_str(), "write") == 0)) {
+                mode |= Command::Write;
+            } else if ((strcasecmp(mask.c_str(), "read") == 0)) {
+                mode |= Command::Read;
+            } else if ((strcasecmp(mask.c_str(), "admin") == 0)) {
+                mode |= Command::Admin;
+            } else if ((strcasecmp(mask.c_str(), "keyAt2") == 0)) {
+                mode |= Command::KeyAt2;
+            } else if ((strcasecmp(mask.c_str(), "keyAt3") == 0)) {
+                mode |= Command::KeyAt3;
+            } else {
+                Throw(InvalidValue, "%s:%d unknown mode %s", n->file, n->line, mask.c_str());
+            }
+        }
+        switch (mode & Command::KeyMask) {
+        case 0:
+        case Command::KeyAt2:
+        case Command::KeyAt3:
+             break;
+        default:
+             Throw(InvalidValue, "%s:%d %s require exclusive key pos", n->file, n->line, name);
+        }
+        switch (mode & Command::AuthMask) {
+        case 0:
+            mode |= Command::Write;
+            break;
+        case Command::Read:
+        case Command::Write:
+        case Command::Admin:
+            break;
+        default:
+            Throw(InvalidValue, "%s:%d %s require exclusive mode", n->file, n->line, name);
+        }
+    }
+    return true;
+}
+
 void Conf::setDC(DCConf& dc, const ConfParser::Node* node)
 {
     if (!node->sub) {

src/Conf.h

@@ -19,6 +19,8 @@
 #include "Distribution.h"
 #include "ConfParser.h"
 #include "Auth.h"
+#include "Command.h"
+#include "Enums.h"
 
 struct AuthConf
 {
@@ -62,8 +64,9 @@ struct ClusterServerPoolConf : public ServerPoolConf
     std::vector<ServerConf> servers;
 };
 
-struct SentinelServerPoolConf : public ServerPoolConf
+struct StandaloneServerPoolConf : public ServerPoolConf
 {
+    ServerPoolRefreshMethod refreshMethod = ServerPoolRefreshMethod::None;
     Distribution dist = Distribution::None;
     Hash hash = Hash::None;
     char hashTag[2];
@@ -89,10 +92,20 @@ struct DCConf
 struct LatencyMonitorConf
 {
     std::string name;
-    std::bitset<Command::Sentinel> cmds;
+    std::bitset<Command::AvailableCommands> cmds;
     std::vector<long> timeSpan;//us
 };
 
+struct CustomCommandConf
+{
+    std::string name;
+    int minArgs;
+    int maxArgs;
+    int mode;
+
+    static void init(CustomCommandConf &c, const char* name);
+};
+
 class Conf
 {
 public:
@@ -164,9 +177,9 @@ public:
     {
         return mClusterServerPool;
     }
-    const SentinelServerPoolConf& sentinelServerPool() const
+    const StandaloneServerPoolConf& standaloneServerPool() const
     {
-        return mSentinelServerPool;
+        return mStandaloneServerPool;
     }
     const std::string& localDC() const
     {
@@ -187,7 +200,7 @@ private:
     void setGlobal(const ConfParser::Node* node);
     void setAuthority(const ConfParser::Node* node);
     void setClusterServerPool(const ConfParser::Node* node);
-    void setSentinelServerPool(const ConfParser::Node* node);
+    void setStandaloneServerPool(const ConfParser::Node* node);
     void setDataCenter(const ConfParser::Node* node);
     void check();
     bool setServerPool(ServerPoolConf& sp, const ConfParser::Node* n);
@@ -201,6 +214,8 @@ private:
     void setDC(DCConf& dc, const ConfParser::Node* n);
     void setReadPolicy(ReadPolicyConf& c, const ConfParser::Node* n);
     void setLatencyMonitor(LatencyMonitorConf& m, const ConfParser::Node* n);
+    void setCustomCommand(const ConfParser::Node* n);
+    bool setCommandMode(int& mode, const char* name, const ConfParser::Node* n, const int defaultMode = Command::Write);
 private:
     std::string mName;
     std::string mBind;
@@ -216,10 +231,11 @@ private:
     std::vector<AuthConf> mAuthConfs;
     int mServerPoolType;
     ClusterServerPoolConf mClusterServerPool;
-    SentinelServerPoolConf mSentinelServerPool;
+    StandaloneServerPoolConf mStandaloneServerPool;
     std::vector<DCConf> mDCConfs;
     std::string mLocalDC;
     std::vector<LatencyMonitorConf> mLatencyMonitors;
+    std::vector<CustomCommandConf> mCustomCommands;
 };
 
 

src/ConfParser.cpp

@@ -283,9 +283,13 @@ Done:
     case SValBody:
         return KeyVal;
     case VValBody:
+    {
+        auto ret = KeyVal;
         val.assign(line, pos, line.size() - pos);
         if (val.back() == '{') {
             val.resize(val.size() - 1);
+            ret = BeginScope;
+        }
         int vsp = 0;
         for (auto it = val.rbegin(); it != val.rend(); ++it) {
             if (isspace(*it)) {
@@ -293,9 +297,7 @@ Done:
             }
         }
         val.resize(val.size() - vsp);
-            return BeginScope;
-        } else {
-            return KeyVal;
+        return ret;
     }
     case ScopeReady:
         return KeyVal;

src/ConnectConnection.h

@@ -23,6 +23,7 @@ public:
     typedef ConnectConnection Value;
     typedef ListNode<ConnectConnection> ListNodeType;
     typedef DequeNode<ConnectConnection> DequeNodeType;
+    typedef Alloc<ConnectConnection, Const::ConnectConnectionAllocCacheSize> Allocator;
 public:
     ConnectConnection(Server* s, bool shared);
     ~ConnectConnection();
@@ -97,6 +98,6 @@ private:
 
 typedef List<ConnectConnection> ConnectConnectionList;
 typedef Deque<ConnectConnection> ConnectConnectionDeque;
-typedef Alloc<ConnectConnection, Const::ConnectConnectionAllocCacheSize> ConnectConnectionAlloc;
+typedef ConnectConnection::Allocator ConnectConnectionAlloc;
 
 #endif

src/Connection.cpp

@@ -27,7 +27,7 @@ BufferPtr Connection::getBuffer(Handler* h, bool allowNew)
         }
     }
     if (!mBuf || mBuf->full()) {
-        BufferPtr buf = Alloc<Buffer>::create();
+        BufferPtr buf = BufferAlloc::create();
         if (mBuf) {
             mBuf->concat(buf);
         }

src/Enums.cpp

@@ -0,0 +1,15 @@
+/*
+ * predixy - A high performance and full features proxy for redis.
+ * Copyright (C) 2017 Joyield, Inc. <joyield.com@gmail.com>
+ * All rights reserved.
+ */
+
+#include "Enums.h"
+
+const ServerPoolRefreshMethod::TypeName
+ServerPoolRefreshMethod::sPairs[3] = {
+    {ServerPoolRefreshMethod::None, "none"},
+    {ServerPoolRefreshMethod::Fixed, "fixed"},
+    {ServerPoolRefreshMethod::Sentinel, "sentinel"},
+};
+

src/Enums.h

@@ -0,0 +1,74 @@
+/*
+ * predixy - A high performance and full features proxy for redis.
+ * Copyright (C) 2017 Joyield, Inc. <joyield.com@gmail.com>
+ * All rights reserved.
+ */
+
+#ifndef _PREDIXY_ENUMS_H_
+#define _PREDIXY_ENUMS_H_
+
+#include <string.h>
+#include <strings.h>
+#include "Exception.h"
+
+template<class T>
+class EnumBase
+{
+public:
+    DefException(InvalidEnumValue);
+    struct TypeName
+    {
+        int type;
+        const char* name;
+    };
+public:
+    EnumBase(int t):
+        mType(t)
+    {
+    }
+    int value() const
+    {
+        return mType;
+    }
+    bool operator==(const T& t) const
+    {
+        return t.value() == mType;
+    }
+    bool operator!=(const T& t) const
+    {
+        return t.value() != mType;
+    }
+    const char* name() const
+    {
+        return T::sPairs[mType].name;
+    }
+    static T parse(const char* str)
+    {
+        for (auto& i : T::sPairs) {
+            if (strcasecmp(i.name, str) == 0) {
+                return T(typename T::Type(i.type));
+            }
+        }
+        Throw(InvalidEnumValue, "invalid enum value:%s", str);
+    }
+protected:
+    int mType;
+};
+
+class ServerPoolRefreshMethod : public EnumBase<ServerPoolRefreshMethod>
+{
+public:
+    enum Type
+    {
+        None,
+        Fixed,
+        Sentinel
+    };
+    static const TypeName sPairs[3];
+    ServerPoolRefreshMethod(Type t = None):
+        EnumBase<ServerPoolRefreshMethod>(t)
+    {
+    }
+};
+
+#endif

src/EpollMultiplexor.cpp

@@ -31,12 +31,9 @@ bool EpollMultiplexor::addSocket(Socket* s, int evts)
     event.events |= (evts & ReadEvent) ? EPOLLIN : 0;
     event.events |= (evts & WriteEvent) ? EPOLLOUT : 0;
     event.events |= EPOLLET;
-    //event.events |= EPOLLONESHOT;
     event.data.ptr = s;
-    int ret = epoll_ctl(mFd, EPOLL_CTL_ADD, s->fd(), &event);
-    if (ret == 0) {
     s->setEvent(evts);
-    }
+    int ret = epoll_ctl(mFd, EPOLL_CTL_ADD, s->fd(), &event);
     return ret == 0;
 }
 
@@ -61,7 +58,6 @@ bool EpollMultiplexor::addEvent(Socket* s, int evts)
     }
     if ((s->getEvent() | evts) != s->getEvent()) {
         event.events |= EPOLLET;
-        //event.events |= EPOLLONESHOT;
         int ret = epoll_ctl(mFd, EPOLL_CTL_MOD, s->fd(), &event);
         if (ret == 0) {
             s->setEvent(s->getEvent() | evts);

src/Handler.cpp

@@ -215,7 +215,6 @@ void Handler::postAcceptConnectionEvent()
                 auto cp = mConnPool[s->server()->id()];
                 s->setStatus(Connection::LogicError);
                 addPostEvent(s, Multiplexor::ErrorEvent);
-                cp->putPrivateConnection(s);
                 c->detachConnectConnection();
                 s->detachAcceptConnection();
             }
@@ -276,6 +275,9 @@ void Handler::postConnectConnectionEvent()
                     s->status(), s->statusStr());
             mEventLoop->delSocket(s);
             s->close(this);
+            if (!s->isShared()) {
+                mConnPool[s->server()->id()]->putPrivateConnection(s);
+            }
             if (c) {
                 addPostEvent(c, Multiplexor::ErrorEvent);
                 s->detachAcceptConnection();
@@ -315,7 +317,6 @@ void Handler::addAcceptSocket(int fd, sockaddr* addr, socklen_t len)
     AcceptConnection* c = nullptr;
     try {
         c = AcceptConnectionAlloc::create(fd, addr, len);
-        logNotice("h %d accept c %s %d", id(), c->peer(), fd);
     } catch (ExceptionBase& e) {
         logWarn("h %d create connection for client %d fail %s",
                 id(), fd, e.what());
@@ -368,6 +369,8 @@ void Handler::addAcceptSocket(int fd, sockaddr* addr, socklen_t len)
         logWarn("h %d destroy c %s %d with add to event loop fail:%s",
                 id(), c->peer(), c->fd(), StrError());
         AcceptConnectionAlloc::destroy(c);
+    } else {
+        logNotice("h %d accept c %s %d assign to h %d", id(), c->peer(), fd, dst->id());
     }
 }
 

src/KqueueMultiplexor.h

@@ -67,6 +67,5 @@ int KqueueMultiplexor::wait(long usec, T* handler)
 
 
 typedef KqueueMultiplexor Multiplexor;
-#define _MULTIPLEXOR_ASYNC_ASSIGN_
 
 #endif

src/LatencyMonitor.h

@@ -98,7 +98,7 @@ public:
     Buffer* output(Buffer* buf) const;
 private:
     String mName;
-    const std::bitset<Command::Sentinel>* mCmds;
+    const std::bitset<Command::AvailableCommands>* mCmds;
     std::vector<TimeSpan> mTimeSpan;
     TimeSpan mLast;
 };

src/Makefile

@@ -72,6 +72,7 @@ objs = \
 	 Buffer.o \
 	 Command.o \
 	 Distribution.o \
+	 Enums.o \
 	 Reply.o \
 	 ConfParser.o \
 	 Conf.o \
@@ -87,7 +88,7 @@ objs = \
 	 ServerPool.o \
 	 ClusterNodesParser.o \
 	 ClusterServerPool.o \
-	 SentinelServerPool.o \
+	 StandaloneServerPool.o \
 	 ConnectConnectionPool.o \
 	 Handler.o \
 	 Proxy.o \

src/Proxy.cpp

@@ -118,10 +118,10 @@ bool Proxy::init(int argc, char* argv[])
             mServPool = p;
         }
         break;
-    case ServerPool::Sentinel:
+    case ServerPool::Standalone:
         {
-            SentinelServerPool* p = new SentinelServerPool(this);
-            p->init(mConf->sentinelServerPool());
+            StandaloneServerPool* p = new StandaloneServerPool(this);
+            p->init(mConf->standaloneServerPool());
             mServPool = p;
         }
         break;

src/Proxy.h

@@ -13,7 +13,7 @@
 #include "DC.h"
 #include "ServerPool.h"
 #include "ClusterServerPool.h"
-#include "SentinelServerPool.h"
+#include "StandaloneServerPool.h"
 #include "LatencyMonitor.h"
 
 class Proxy
@@ -51,15 +51,15 @@ public:
     }
     bool isSplitMultiKey() const
     {
-        return mConf->sentinelServerPool().groups.size() != 1;
+        return mConf->standaloneServerPool().groups.size() != 1;
     }
     bool supportTransaction() const
     {
-        return mConf->sentinelServerPool().groups.size() == 1;
+        return mConf->standaloneServerPool().groups.size() == 1;
     }
     bool supportSubscribe() const
     {
-        return mConf->sentinelServerPool().groups.size() == 1 ||
+        return mConf->standaloneServerPool().groups.size() == 1 ||
                mConf->clusterServerPool().servers.size() > 0;
     }
     const std::vector<Handler*>& handlers() const

src/Request.cpp

@@ -98,10 +98,12 @@ Request::Request(GenericCode code):
 
 Request::~Request()
 {
+    clear();
 }
 
 void Request::clear()
 {
+    mConn = nullptr;
     mRes = nullptr;
     mHead.clear();
     mReq.clear();
@@ -155,13 +157,14 @@ void Request::set(const RequestParser& p, Request* leader)
         }
         mHead = r->mReq;
         mReq = p.request();
-        mLeader = leader;
         if (leader == this) {
             if (mType == Command::Mset || mType == Command::Msetnx) {
                 mFollowers = (p.argNum() - 1) >> 1;
             } else {
                 mFollowers = p.argNum() - 1;
             }
+        } else {
+            mLeader = leader;
         }
     } else {
         mReq = p.request();
@@ -287,10 +290,11 @@ void Request::adjustScanCursor(long cursor)
 
 void Request::follow(Request* leader)
 {
-    ++mFollowers;
+    leader->mFollowers += 1;
     if (leader == this) {
         return;
     }
+    mConn = leader->mConn;
     mType = leader->mType;
     mHead = leader->mHead;
     mReq = leader->mReq;
@@ -338,49 +342,49 @@ int Request::fill(IOVec* vecs, int len)
 void Request::setResponse(Response* res)
 {
     mDone = true;
-    if (mLeader) {
-        mLeader->mFollowersDone += 1;
+    if (Request* ld = leader()) {
+        ld->mFollowersDone += 1;
         switch (mType) {
         case Command::Mget:
             mRes = res;
             break;
         case Command::Mset:
-            if (Response* leaderRes = mLeader->getResponse()) {
+            if (Response* leaderRes = ld->getResponse()) {
                 if (res->isError() && !leaderRes->isError()) {
-                    mLeader->mRes = res;
+                    ld->mRes = res;
                 }
             } else {
-                mLeader->mRes = res;
+                ld->mRes = res;
             }
             break;
         case Command::Msetnx:
-            if (Response* leaderRes = mLeader->getResponse()) {
+            if (Response* leaderRes = ld->getResponse()) {
                 if (!leaderRes->isError() &&
                      (res->isError() || res->integer() == 0)) {
-                    mLeader->mRes = res;
+                    ld->mRes = res;
                 }
             } else {
-                mLeader->mRes = res;
+                ld->mRes = res;
             }
             break;
         case Command::Touch:
         case Command::Exists:
         case Command::Del:
         case Command::Unlink:
-            if (!mLeader->mRes) {
-                mLeader->mRes = res;
+            if (!ld->mRes) {
+                ld->mRes = res;
             }
-            if (mLeader->isDone()) {
-                mLeader->mRes->set(mLeader->mRes->integer());
+            if (ld->isDone()) {
+                ld->mRes->set(ld->mRes->integer());
             }
             break;
         case Command::ScriptLoad:
-            if (Response* leaderRes = mLeader->getResponse()) {
+            if (Response* leaderRes = ld->getResponse()) {
                 if (leaderRes->isString() && !res->isString()) {
-                    mLeader->mRes = res;
+                    ld->mRes = res;
                 }
             } else {
-                mLeader->mRes = res;
+                ld->mRes = res;
             }
             break;
         default:
@@ -395,7 +399,7 @@ void Request::setResponse(Response* res)
 
 bool Request::isDone() const
 {
-    if (mLeader == this) {
+    if (isLeader()) {
         switch (mType) {
         case Command::Mget:
         case Command::Psubscribe:

src/Request.h

@@ -25,6 +25,7 @@ class Request :
 public:
     typedef Request Value;
     typedef ListNode<Request, SharePtr<Request>, RequestListIndex::Size> ListNodeType;
+    typedef Alloc<Request, Const::RequestAllocCacheSize> Allocator;
     static const int MaxRedirectLimit = 3;
     enum GenericCode
     {
@@ -119,11 +120,11 @@ public:
     }
     Request* leader() const
     {
-        return mLeader;
+        return isLeader() ? const_cast<Request*>(this) : (Request*)mLeader;
     }
     bool isLeader() const
     {
-        return mLeader == this;
+        return mFollowers > 0;
     }
     bool isDelivered() const
     {
@@ -181,6 +182,6 @@ private:
 
 typedef List<Request, RequestListIndex::Recv> RecvRequestList;
 typedef List<Request, RequestListIndex::Send> SendRequestList;
-typedef Alloc<Request, Const::RequestAllocCacheSize> RequestAlloc;
+typedef Request::Allocator RequestAlloc;
 
 #endif

src/Response.h

@@ -18,6 +18,7 @@ class Response :
 public:
     typedef Response Value;
     typedef ListNode<Response, SharePtr<Response>> ListNodeType;
+    typedef Alloc<Response, Const::ResponseAllocCacheSize> Allocator;
     enum GenericCode
     {
         Pong,
@@ -137,6 +138,6 @@ private:
 };
 
 typedef List<Response> ResponseList;
-typedef Alloc<Response, Const::ResponseAllocCacheSize> ResponseAlloc;
+typedef Response::Allocator ResponseAlloc;
 
 #endif

src/ServerGroup.cpp

@@ -162,6 +162,9 @@ Server* ServerGroup::getReadServer(Handler* h, DC* localDC) const
             continue;
         }
         DC* dc = s->dc();
+        if (!dc) {
+            continue;
+        }
         int dcrp = localDC->getReadPriority(dc);
         if (dcrp <= 0) {
             continue;
@@ -221,7 +224,7 @@ Server* ServerGroup::getReadServer(Handler* h, DC* localDC) const
         dc = sdc[0];
         found = true;
     }
-    if (!found) {//dc maybe nullptr even we found
+    if (!found) {
         return nullptr;
     }
     Server* deadServs[Const::MaxServInGroup];

src/ServerPool.h

@@ -20,7 +20,7 @@ public:
     {
         Unknown,
         Cluster,
-        Sentinel
+        Standalone
     };
     static const int DefaultServerRetryTimeout = 10000000;
     static const int DefaultRefreshInterval = 1000000;

src/Socket.cpp

@@ -102,7 +102,7 @@ void Socket::getFirstAddr(const char* addr, int type, int protocol, sockaddr* re
     } else {
         std::string tmp;
         const char* host = addr;
-        const char* port = strchr(addr, ':');
+        const char* port = strrchr(addr, ':');
         if (port) {
             tmp.append(addr, port - addr);
             host = tmp.c_str();
@@ -162,6 +162,7 @@ bool Socket::setTcpKeepAlive(int interval)
     if (ret != 0) {
         return false;
     }
+#ifdef __linux__
     val = interval;
     ret = setsockopt(mFd, IPPROTO_TCP, TCP_KEEPIDLE, &val, sizeof(val));
     if (ret != 0) {
@@ -177,6 +178,9 @@ bool Socket::setTcpKeepAlive(int interval)
     if (ret != 0) {
         return false;
     }
+#else
+    ((void)interval); //Avoid unused var warning for non Linux systems
+#endif
     return true;
 }
 

src/StandaloneServerPool.cpp

@@ -0,0 +1,482 @@
+/*
+ * predixy - A high performance and full features proxy for redis.
+ * Copyright (C) 2017 Joyield, Inc. <joyield.com@gmail.com>
+ * All rights reserved.
+ */
+
+#include <algorithm>
+#include "Logger.h"
+#include "ServerGroup.h"
+#include "Handler.h"
+#include "StandaloneServerPool.h"
+
+StandaloneServerPool::StandaloneServerPool(Proxy* p):
+    ServerPoolTmpl(p, Standalone),
+    mDist(Distribution::Modula)
+{
+    mSentinels.reserve(MaxSentinelNum);
+    mServPool.reserve(Const::MaxServNum);
+    mHashTag[0] = mHashTag[1] = '\0';
+}
+
+StandaloneServerPool::~StandaloneServerPool()
+{
+}
+
+void StandaloneServerPool::init(const StandaloneServerPoolConf& conf)
+{
+    ServerPool::init(conf);
+    mRefreshMethod = conf.refreshMethod;
+    mDist = conf.dist;
+    mHash = conf.hash;
+    mHashTag[0] = conf.hashTag[0];
+    mHashTag[1] = conf.hashTag[1];
+    int i = 0;
+    if (conf.refreshMethod == ServerPoolRefreshMethod::Sentinel) {
+        mSentinels.resize(conf.sentinels.size());
+        for (auto& sc : conf.sentinels) {
+            Server* s = new Server(this, sc.addr, true);
+            s->setRole(Server::Sentinel);
+            s->setPassword(sc.password.empty() ? conf.sentinelPassword:sc.password);
+            mSentinels[i++] = s;
+            mServs[s->addr()] = s;
+        }
+    }
+    mGroupPool.resize(conf.groups.size());
+    i = 0;
+    for (auto& gc : conf.groups) {
+        ServerGroup* g = new ServerGroup(this, gc.name);
+        mGroupPool[i++] = g;
+        auto role = Server::Master;
+        for (auto& sc : gc.servers) {
+            Server* s = new Server(this, sc.addr, true);
+            s->setPassword(sc.password.empty() ? conf.password : sc.password);
+            mServPool.push_back(s);
+            mServs[s->addr()] = s;
+            g->add(s);
+            s->setGroup(g);
+            switch (mRefreshMethod.value()) {
+            case ServerPoolRefreshMethod::Fixed:
+                s->setOnline(true);
+                s->setRole(role);
+                role = Server::Slave;
+                break;
+            default:
+                s->setOnline(false);
+                break;
+            }
+        }
+    }
+}
+
+Server* StandaloneServerPool::getServer(Handler* h, Request* req, const String& key) const
+{
+    FuncCallTimer();
+    switch (req->type()) {
+    case Command::SentinelGetMaster:
+    case Command::SentinelSlaves:
+    case Command::SentinelSentinels:
+        if (mSentinels.empty()) {
+            return nullptr;
+        } else  {
+            Server* s = randServer(h, mSentinels);
+            logDebug("sentinel server pool get server %s for sentinel command",
+                     s->addr().data());
+            return s;
+        }
+        break;
+    case Command::Randomkey:
+        return randServer(h, mServPool);
+    default:
+        break;
+    }
+    if (mGroupPool.size() == 1) {
+        return mGroupPool[0]->getServer(h, req);
+    } else if (mGroupPool.size() > 1) {
+        switch (mDist) {
+        case Distribution::Modula:
+            {
+                long idx = mHash.hash(key.data(), key.length(), mHashTag);
+                idx %= mGroupPool.size();
+                return mGroupPool[idx]->getServer(h, req);
+            }
+            break;
+        case Distribution::Random:
+            {
+                int idx = h->rand() % mGroupPool.size();
+                return mGroupPool[idx]->getServer(h, req);
+            }
+            break;
+        default:
+            break;
+        }
+    }
+    return nullptr;
+}
+
+void StandaloneServerPool::refreshRequest(Handler* h)
+{
+    logDebug("h %d update standalone server pool", h->id());
+    switch (mRefreshMethod.value()) {
+    case ServerPoolRefreshMethod::Sentinel:
+        for (auto g : mGroupPool) {
+            RequestPtr req = RequestAlloc::create();
+            req->setSentinels(g->name());
+            req->setData(g);
+            h->handleRequest(req);
+            req = RequestAlloc::create();
+            req->setSentinelGetMaster(g->name());
+            req->setData(g);
+            h->handleRequest(req);
+            req = RequestAlloc::create();
+            req->setSentinelSlaves(g->name());
+            req->setData(g);
+            h->handleRequest(req);
+        }
+        break;
+    default:
+        break;
+    }
+}
+
+void StandaloneServerPool::handleResponse(Handler* h, ConnectConnection* s, Request* req, Response* res)
+{
+    switch (req->type()) {
+    case Command::SentinelSentinels:
+        handleSentinels(h, s, req, res);
+        break;
+    case Command::SentinelGetMaster:
+        handleGetMaster(h, s, req, res);
+        break;
+    case Command::SentinelSlaves:
+        handleSlaves(h, s, req, res);
+        break;
+    default:
+        break;
+    }
+}
+
+class AddrParser
+{
+public:
+    enum Status {
+        Ok,
+        Error,
+        Done
+    };
+public:
+    AddrParser(const Segment& res):
+        mRes(res),
+        mState(Idle),
+        mCnt(0),
+        mArgLen(0),
+        mIp(false),
+        mPort(false)
+    {
+        mRes.rewind();
+    }
+    int count() const {return mCnt;}
+    Status parse(SString<Const::MaxAddrLen>& addr);
+private:
+    enum State {
+        Idle,
+        Count,
+        CountLF,
+        Arg,
+        ArgLen,
+        ArgLenLF,
+        SubArrayLen,
+        Body,
+        BodyLF,
+        Invalid,
+        Finished
+    };
+private:
+    Segment mRes;
+    State mState;
+    int mCnt;
+    int mArgLen;
+    bool mIp;
+    bool mPort;
+    SString<4> mKey;
+};
+
+AddrParser::Status AddrParser::parse(SString<Const::MaxAddrLen>& addr)
+{
+    const char* dat;
+    int len;
+    addr.clear();
+    while (mRes.get(dat, len) && mState != Invalid) {
+        for (int i = 0; i < len && mState != Invalid; ++i) {
+            char ch = dat[i];
+            switch (mState) {
+            case Idle:
+                mState = ch == '*' ? Count : Invalid;
+                break;
+            case Count:
+                if (ch >= '0' && ch <= '9') {
+                    mCnt = mCnt * 10 + (ch - '0');
+                } else if (ch == '\r') {
+                    if (mCnt == 0) {
+                        mState = Finished;
+                        return Done;
+                    } else if (mCnt < 0) {
+                        mState = Invalid;
+                        return Error;
+                    }
+                    mState = CountLF;
+                } else {
+                    mState = Invalid;
+                }
+                break;
+            case CountLF:
+                mState = ch == '\n' ? Arg : Invalid;
+                break;
+            case Arg:
+                if (ch == '$') {
+                    mState = ArgLen;
+                    mArgLen = 0;
+                } else if (ch == '*') {
+                    mState = SubArrayLen;
+                } else {
+                    mState = Invalid;
+                }
+                break;
+            case ArgLen:
+                if (ch >= '0' && ch <= '9') {
+                    mArgLen = mArgLen * 10 + (ch - '0');
+                } else if (ch == '\r') {
+                    mState = ArgLenLF;
+                } else {
+                    mState = Invalid;
+                }
+                break;
+            case ArgLenLF:
+                mState = ch == '\n' ? Body : Invalid;
+                break;
+            case SubArrayLen:
+                if (ch == '\n') {
+                    mState = Arg;
+                }
+                break;
+            case Body:
+                if (ch == '\r') {
+                    mState = BodyLF;
+                    if (mPort) {
+                        mPort = false;
+                        mRes.use(i + 1);
+                        return Ok;
+                    } else if (mIp) {
+                        mIp = false;
+                        addr.append(':');
+                    } else if (mArgLen == 2 && strcmp(mKey.data(), "ip") == 0) {
+                        mIp = true;
+                    } else if (mArgLen == 4 && strcmp(mKey.data(), "port") == 0) {
+                        mPort = true;
+                    }
+                    break;
+                }
+                if (mIp || mPort) {
+                    addr.append(ch);
+                } else if (mArgLen == 2 || mArgLen == 4) {
+                    mKey.append(ch);
+                }
+                break;
+            case BodyLF:
+                mKey.clear();
+                mState = ch == '\n' ? Arg : Invalid;
+                break;
+            default:
+                break;
+            }
+        }
+        mRes.use(len);
+    }
+    return mState != Invalid ? Done : Error;
+}
+
+static bool hasValidPort(const String& addr)
+{
+    const char* p = addr.data() + addr.length();
+    for (int i = 0; i < addr.length(); ++i) {
+        if (*(--p) == ':') {
+            int port = atoi(p + 1);
+            return port > 0 && port < 65536;
+        }
+    }
+    return false;
+}
+
+void StandaloneServerPool::handleSentinels(Handler* h, ConnectConnection* s, Request* req, Response* res)
+{
+    if (!res || !res->isArray()) {
+        return;
+    }
+    AddrParser parser(res->body());
+    SString<Const::MaxAddrLen> addr;
+    while (true) {
+        auto st = parser.parse(addr);
+        if (st == AddrParser::Ok) {
+            logDebug("sentinel server pool parse sentinel %s", addr.data());
+            if (!hasValidPort(addr)) {
+                logNotice("sentinel server pool parse sentienl %s invalid",
+                        addr.data());
+                continue;
+            }
+            auto it = mServs.find(addr);
+            Server* serv = it == mServs.end() ? nullptr : it->second;
+            if (!serv) {
+                if (mSentinels.size() == mSentinels.capacity()) {
+                    logWarn("too many sentinels %d, will ignore new sentinel %s",
+                            (int)mSentinels.size(), addr.data());
+                    continue;
+                }
+                serv = new Server(this, addr, false);
+                serv->setRole(Server::Sentinel);
+                serv->setPassword(password());
+                mSentinels.push_back(serv);
+                mServs[serv->addr()] = serv;
+                logNotice("h %d create new sentinel %s",
+                        h->id(), addr.data());
+            }
+            serv->setOnline(true);
+        } else if (st == AddrParser::Done) {
+            break;
+        } else {
+            logError("sentinel server pool parse sentinel sentinels error");
+            break;
+        }
+    }
+}
+
+void StandaloneServerPool::handleGetMaster(Handler* h, ConnectConnection* s, Request* req, Response* res)
+{
+    if (!res || !res->isArray()) {
+        return;
+    }
+    ServerGroup* g = (ServerGroup*)req->data();
+    if (!g) {
+        return;
+    }
+    SegmentStr<Const::MaxAddrLen + 32> str(res->body());
+    if (!str.complete()) {
+        return;
+    }
+    if (strncmp(str.data(), "*2\r\n$", 5) != 0) {
+        return;
+    }
+    SString<Const::MaxAddrLen> addr;
+    const char* p = str.data() + 5;
+    int len = atoi(p);
+    if (len <= 0) {
+        return;
+    }
+    p = strchr(p, '\r') + 2;
+    if (!addr.append(p, len)) {
+        return;
+    }
+    if (!addr.append(':')) {
+        return;
+    }
+    p += len + 3;
+    len = atoi(p);
+    if (len <= 0) {
+        return;
+    }
+    p = strchr(p, '\r') + 2;
+    if (!addr.append(p, len)) {
+        return;
+    }
+    logDebug("sentinel server pool group %s get master %s",
+             g->name().data(), addr.data());
+    auto it = mServs.find(addr);
+    Server* serv = it == mServs.end() ? nullptr : it->second;
+    if (serv) {
+        serv->setOnline(true);
+        serv->setRole(Server::Master);
+        auto old = serv->group();
+        if (old) {
+            if (old != g) {
+                old->remove(serv);
+                g->add(serv);
+                serv->setGroup(g);
+            }
+        } else {
+            g->add(serv);
+            serv->setGroup(g);
+        }
+    } else {
+        if (mServPool.size() == mServPool.capacity()) {
+            logWarn("too many servers %d, will ignore new master server %s",
+                     (int)mServPool.size(), addr.data());
+            return;
+        }
+        serv = new Server(this, addr, false);
+        serv->setRole(Server::Master);
+        serv->setPassword(password());
+        mServPool.push_back(serv);
+        g->add(serv);
+        serv->setGroup(g);
+        mServs[serv->addr()] = serv;
+        logNotice("sentinel server pool group %s create master server %s %s",
+                  g->name().data(), addr.data(), serv->dcName().data());
+    }
+}
+
+void StandaloneServerPool::handleSlaves(Handler* h, ConnectConnection* s, Request* req, Response* res)
+{
+    if (!res || !res->isArray()) {
+        return;
+    }
+    ServerGroup* g = (ServerGroup*)req->data();
+    if (!g) {
+        return;
+    }
+    AddrParser parser(res->body());
+    SString<Const::MaxAddrLen> addr;
+    while (true) {
+        auto st = parser.parse(addr);
+        if (st == AddrParser::Ok) {
+            logDebug("sentinel server pool group %s parse slave %s",
+                     g->name().data(), addr.data());
+            auto it = mServs.find(addr);
+            Server* serv = it == mServs.end() ? nullptr : it->second;
+            if (serv) {
+                serv->setOnline(true);
+                serv->setRole(Server::Slave);
+                auto old = serv->group();
+                if (old) {
+                    if (old != g) {
+                        old->remove(serv);
+                        g->add(serv);
+                        serv->setGroup(g);
+                    }
+                } else {
+                    g->add(serv);
+                    serv->setGroup(g);
+                }
+            } else {
+                if (mServPool.size() == mServPool.capacity()) {
+                    logWarn("too many servers %d, will ignore new slave server %s",
+                            (int)mServPool.size(), addr.data());
+                    return;
+                }
+                serv = new Server(this, addr, false);
+                serv->setRole(Server::Slave);
+                serv->setPassword(password());
+                mServPool.push_back(serv);
+                g->add(serv);
+                serv->setGroup(g);
+                mServs[serv->addr()] = serv;
+                logNotice("sentinel server pool group %s create slave server %s %s",
+                          g->name().data(), addr.data(), serv->dcName().data());
+            }
+        } else if (st == AddrParser::Done) {
+            break;
+        } else {
+            logError("sentinel server pool group %s parse sentinel sentinels error",
+                    g->name().data());
+            break;
+        }
+    }
+}

src/StandaloneServerPool.h

@@ -0,0 +1,43 @@
+/*
+ * predixy - A high performance and full features proxy for redis.
+ * Copyright (C) 2017 Joyield, Inc. <joyield.com@gmail.com>
+ * All rights reserved.
+ */
+
+#ifndef _PREDIXY_STANDALONE_SERVER_POOL_H_
+#define _PREDIXY_STANDALONE_SERVER_POOL_H_
+
+#include <map>
+#include "Predixy.h"
+#include "ServerPool.h"
+
+class StandaloneServerPool : public ServerPoolTmpl<StandaloneServerPool>
+{
+public:
+    static const int MaxSentinelNum = 64;
+public:
+    StandaloneServerPool(Proxy* p);
+    ~StandaloneServerPool();
+    void init(const StandaloneServerPoolConf& conf);
+    Server* getServer(Handler* h, Request* req, const String& key) const;
+    Server* iter(int& cursor) const
+    {
+        return ServerPool::iter(mServPool, cursor);
+    }
+    void refreshRequest(Handler* h);
+    void handleResponse(Handler* h, ConnectConnection* s, Request* req, Response* res);
+private:
+    void handleSentinels(Handler* h, ConnectConnection* s, Request* req, Response* res);
+    void handleGetMaster(Handler* h, ConnectConnection* s, Request* req, Response* res);
+    void handleSlaves(Handler* h, ConnectConnection* s, Request* req, Response* res);
+    friend class ServerPoolTmpl<StandaloneServerPool>;
+private:
+    ServerPoolRefreshMethod mRefreshMethod;
+    std::vector<Server*> mSentinels;
+    std::vector<Server*> mServPool;
+    Distribution mDist;
+    Hash mHash;
+    char mHashTag[2];
+};
+
+#endif

test/basic.py

@@ -391,7 +391,7 @@ Cases = [
         [('scard', '{k}2'), 3],
     ]),
     ('zset', [
-        [('del', 'k', '{k}2', '{k}3', '{k}4'), ],
+        [('del', 'k', '{k}2', '{k}3', '{k}4', '{k}5', '{k}6'), ],
         [('zadd', 'k', 10, 'apple'), 1],
         [('zcard', 'k'), 1],
         [('zincrby', 'k', 2, 'apple'), '12'],
@@ -438,6 +438,12 @@ Cases = [
         [('zunionstore', '{k}3', 2, 'k', '{k}2'), 4],
         [('zunionstore', '{k}3', 2, 'k', '{k}2', 'AGGREGATE', 'MAX'), 4],
         [('zunionstore', '{k}3', 2, 'k', '{k}2', 'WEIGHTS', 0.5, 1.2, 'AGGREGATE', 'MAX'), 4],
+        [('zadd', '{k}5', 0, 'apple', 9, 'banana', 1, 'pear', 3, 'orange', 4, 'cat'), 5],
+        [('zpopmax', '{k}5'), ['banana', '9']],
+        [('zpopmax', '{k}5', 3), ['cat', '4', 'orange', '3', 'pear', '1']],
+        [('zadd', '{k}6', 0, 'apple', 9, 'banana', 1, 'pear', 3, 'orange', 4, 'cat'), 5],
+        [('zpopmin', '{k}6'), ['apple', '0']],
+        [('zpopmin', '{k}6', 3), ['pear', '1', 'orange', '3', 'cat', '4']],
     ]),
     ('hyperloglog', [
         [('del', 'k', '{k}2', '{k}3'), ],