合并上游代码

Author: huangzworks

src/cluster.c

@@ -535,6 +535,8 @@ clusterNode *createClusterNode(char *nodename, int flags) {
     node->port = 0;
     node->fail_reports = listCreate();
     node->voted_time = 0;
+    node->repl_offset_time = 0;
+    node->repl_offset = 0;
     listSetFreeMethod(node->fail_reports,zfree);
 
     return node;
@@ -2325,25 +2327,24 @@ void clusterBroadcastMessage(void *buf, size_t len) {
 // 构建信息
 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
     int totlen = 0;
-    clusterNode *master;
+    uint64_t offset;
+    clusterNode *master, *myself = server.cluster->myself;
 
     /* If this node is a master, we send its slots bitmap and configEpoch.
      *
      * 如果这是一个主节点，那么发送该节点的槽 bitmap 和配置纪元。
      *
      * If this node is a slave we send the master's information instead (the
      * node is flagged as slave so the receiver knows that it is NOT really
-     * in charge for this slots. 
-     *
+     * in charge for this slots.
      * 如果这是一个从节点，
      * 那么发送这个节点的主节点的槽 bitmap 和配置纪元。
      *
      * 因为接收信息的节点通过标识可以知道这个节点是一个从节点，
      * 所以接收信息的节点不会将从节点错认作是主节点。
      */
-    master = (server.cluster->myself->flags & REDIS_NODE_SLAVE &&
-              server.cluster->myself->slaveof) ?
-              server.cluster->myself->slaveof : server.cluster->myself;
+    master = (myself->flags & REDIS_NODE_SLAVE && myself->slaveof) ?
+              myself->slaveof : myself;
 
     // 清零信息头
     memset(hdr,0,sizeof(*hdr));
@@ -2352,24 +2353,23 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
     hdr->type = htons(type);
 
     // 设置信息发送者
-    memcpy(hdr->sender,server.cluster->myself->name,REDIS_CLUSTER_NAMELEN);
+    memcpy(hdr->sender,myself->name,REDIS_CLUSTER_NAMELEN);
 
     // 设置当前节点负责的槽
     memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));
 
     // 清零 slaveof 域
     memset(hdr->slaveof,0,REDIS_CLUSTER_NAMELEN);
+
     // 如果节点是从节点的话，那么设置 slaveof 域
-    if (server.cluster->myself->slaveof != NULL) {
-        memcpy(hdr->slaveof,server.cluster->myself->slaveof->name,
-                                    REDIS_CLUSTER_NAMELEN);
-    }
+    if (myself->slaveof != NULL)
+        memcpy(hdr->slaveof,myself->slaveof->name, REDIS_CLUSTER_NAMELEN);
 
     // 设置端口号
     hdr->port = htons(server.port);
 
     // 设置标识
-    hdr->flags = htons(server.cluster->myself->flags);
+    hdr->flags = htons(myself->flags);
 
     // 设置状态
     hdr->state = server.cluster->state;
@@ -2380,6 +2380,22 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
     // 设置主节点当前配置纪元
     hdr->configEpoch = htonu64(master->configEpoch);
 
+    /* Set the replication offset. */
+    // 设置复制偏移量
+    if (myself->flags & REDIS_NODE_SLAVE) {
+        if (server.master)
+            offset = server.master->reploff;
+        else if (server.cached_master)
+            offset = server.cached_master->reploff;
+        else
+            offset = 0;
+    } else {
+        offset = server.master_repl_offset;
+    }
+    hdr->offset = htonu64(offset);
+
+    /* Compute the message length for certain messages. For other messages
+     * this is up to the caller. */
     // 计算信息的长度
     if (type == CLUSTERMSG_TYPE_FAIL) {
         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);

src/cluster.h

@@ -143,16 +143,22 @@ struct clusterNode {
     struct clusterNode *slaveof; /* pointer to the master node */
 
     // 最后一次发送 PING 命令的时间
-    mstime_t ping_sent;       /* Unix time we sent latest ping */
+    mstime_t ping_sent;      /* Unix time we sent latest ping */
 
     // 最后一次接收 PONG 回复的时间戳
-    mstime_t pong_received;   /* Unix time we received the pong */
+    mstime_t pong_received;  /* Unix time we received the pong */
 
     // 最后一次被设置为 FAIL 状态的时间
-    mstime_t fail_time;       /* Unix time when FAIL flag was set */
+    mstime_t fail_time;      /* Unix time when FAIL flag was set */
 
     // 最后一次给某个从节点投票的时间
-    mstime_t voted_time;      /* Last time we voted for a slave of this master */
+    mstime_t voted_time;     /* Last time we voted for a slave of this master */
+
+    // 最后一次从这个节点接收到复制偏移量的时间
+    mstime_t repl_offset_time;  /* Unix time we received offset for this node */
+
+    // 这个节点的复制偏移量
+    long long repl_offset;      /* Last known repl offset for this node. */
 
     // 节点的 IP 地址
     char ip[REDIS_IP_STR_LEN];  /* Latest known IP address of this node */
@@ -391,8 +397,13 @@ typedef struct {
 
     // 如果消息发送者是一个主节点，那么这里记录的是消息发送者的配置纪元
     // 如果消息发送者是一个从节点，那么这里记录的是消息发送者正在复制的主节点的配置纪元
-    uint64_t configEpoch;   /* The config epoch if it's a master, or the last epoch
-                               advertised by its master if it is a slave. */
+    uint64_t configEpoch;   /* The config epoch if it's a master, or the last
+                               epoch advertised by its master if it is a
+                               slave. */
+
+    // 节点的复制偏移量
+    uint64_t offset;    /* Master replication offset if node is a master or
+                           processed replication offset if node is a slave. */
 
     // 消息发送者的名字（ID）
     char sender[REDIS_CLUSTER_NAMELEN]; /* Name of the sender node */
@@ -425,7 +436,7 @@ typedef struct {
 
 #define CLUSTERMSG_MIN_LEN (sizeof(clusterMsg)-sizeof(union clusterMsgData))
 
-/* ----------------------- API exported outside cluster.c ------------------------- */
+/* ---------------------- API exported outside cluster.c -------------------- */
 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *ask);
 
 #endif /* __REDIS_CLUSTER_H */

src/networking.c

@@ -890,7 +890,7 @@ void freeClient(redisClient *c) {
     }
 
     /* Log link disconnection with slave */
-    if (c->flags & REDIS_SLAVE) {
+    if ((c->flags & REDIS_SLAVE) && !(c->flags & REDIS_MONITOR)) {
         char ip[REDIS_IP_STR_LEN];
 
         if (anetPeerToString(c->fd,ip,sizeof(ip),NULL) != -1) {

src/redis-trib.rb

@@ -912,6 +912,34 @@ def delnode_cluster_cmd(argv,opt)
         node.r.shutdown
     end
 
+    def set_timeout_cluster_cmd(argv,opt)
+        timeout = argv[1].to_i
+        if timeout < 100
+            puts "Setting a node timeout of less than 100 milliseconds is a bad idea."
+            exit 1
+        end
+
+        # Load cluster information
+        load_cluster_info_from_node(argv[0])
+        ok_count = 0
+        err_count = 0
+
+        # Send CLUSTER FORGET to all the nodes but the node to remove
+        xputs ">>> Reconfiguring node timeout in every cluster node..."
+        @nodes.each{|n|
+            begin
+                n.r.config("set","cluster-node-timeout",timeout)
+                n.r.config("rewrite")
+                ok_count += 1
+                xputs "*** New timeout set for #{n}"
+            rescue => e
+                puts "ERR setting node-timeot for #{n}: #{e}"
+                err_count += 1
+            end
+        }
+        xputs ">>> New node timeout set. #{ok_count} OK, #{err_count} ERR."
+    end
+
     def help_cluster_cmd(argv,opt)
         show_help
         exit 0
@@ -952,8 +980,9 @@ def parse_options(cmd)
     "check"   => ["check_cluster_cmd", 2, "host:port"],
     "fix"     => ["fix_cluster_cmd", 2, "host:port"],
     "reshard" => ["reshard_cluster_cmd", 2, "host:port"],
-    "addnode" => ["addnode_cluster_cmd", 3, "new_host:new_port existing_host:existing_port"],
-    "delnode" => ["delnode_cluster_cmd", 3, "host:port node_id"],
+    "add-node" => ["addnode_cluster_cmd", 3, "new_host:new_port existing_host:existing_port"],
+    "del-node" => ["delnode_cluster_cmd", 3, "host:port node_id"],
+    "set-timeout" => ["set_timeout_cluster_cmd", 3, "host:port milliseconds"],
     "help"    => ["help_cluster_cmd", 1, "(show this help)"]
 }
 
@@ -966,14 +995,14 @@ def show_help
     puts "Usage: redis-trib <command> <options> <arguments ...>\n\n"
     COMMANDS.each{|k,v|
         o = ""
-        puts "  #{k.ljust(10)} #{v[2]}"
+        puts "  #{k.ljust(15)} #{v[2]}"
         if ALLOWED_OPTIONS[k]
             ALLOWED_OPTIONS[k].each{|optname,has_arg|
-                puts "             --#{optname}" + (has_arg ? " <arg>" : "")
+                puts "                  --#{optname}" + (has_arg ? " <arg>" : "")
             }
         end
     }
-    puts "\nFor check, fix, reshard, delnode, you can specify host:port of any working node.\n"
+    puts "\nFor check, fix, reshard, del-node, set-timeout you can specify the host and port of any working node in the cluster.\n"
 end
 
 # Sanity check

src/redis.c

@@ -2557,7 +2557,7 @@ int processCommand(redisClient *c) {
     if (server.stop_writes_on_bgsave_err &&
         server.saveparamslen > 0
         && server.lastbgsave_status == REDIS_ERR &&
-        server.masterhost != NULL &&
+        server.masterhost == NULL &&
         (c->cmd->flags & REDIS_CMD_WRITE ||
          c->cmd->proc == pingCommand))
     {