even less false positive on obuf-limits test.

[redis.git] / src / cluster.c

diff --git a/src/cluster.c b/src/cluster.c
index d16427428f3e59b29945dfb55eff6f1fa28def09..85cb119811008fd770a46f2a46118089a6192b64 100644 (file)
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -377,7 +377,7 @@ clusterNode *clusterLookupNode(char *name) {
     de = dictFind(server.cluster.nodes,s);
     sdsfree(s);
     if (de == NULL) return NULL;
     de = dictFind(server.cluster.nodes,s);
     sdsfree(s);
     if (de == NULL) return NULL;

-    return dictGetEntryVal(de);
+    return dictGetVal(de);

 }
 
 /* This is only used after the handshake. When we connect a given IP/PORT
 }
 
 /* This is only used after the handshake. When we connect a given IP/PORT


@@ -439,7 +439,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
              * time PONG figure if it is newer than our figure.
              * Note that it's not a problem if we have a PING already 
              * in progress against this node. */
              * time PONG figure if it is newer than our figure.
              * Note that it's not a problem if we have a PING already 
              * in progress against this node. */

-            if (node->pong_received < ntohl(g->pong_received)) {
+            if (node->pong_received < (signed) ntohl(g->pong_received)) {

                  redisLog(REDIS_DEBUG,"Node pong_received updated by gossip");
                 node->pong_received = ntohl(g->pong_received);
             }
                  redisLog(REDIS_DEBUG,"Node pong_received updated by gossip");
                 node->pong_received = ntohl(g->pong_received);
             }


@@ -613,7 +613,7 @@ int clusterProcessPacket(clusterLink *link) {
             }
         }
         /* Update our info about the node */
             }
         }
         /* Update our info about the node */

-        link->node->pong_received = time(NULL);
+        if (link->node) link->node->pong_received = time(NULL);

 
         /* Update master/slave info */
         if (sender) {
 
         /* Update master/slave info */
         if (sender) {


@@ -793,7 +793,7 @@ void clusterBroadcastMessage(void *buf, size_t len) {
 
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {
 
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {

-        clusterNode *node = dictGetEntryVal(de);
+        clusterNode *node = dictGetVal(de);

 
         if (!node->link) continue;
         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 
         if (!node->link) continue;
         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;


@@ -804,7 +804,7 @@ void clusterBroadcastMessage(void *buf, size_t len) {
 
 /* Build the message header */
 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
 
 /* Build the message header */
 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {

-    int totlen;
+    int totlen = 0;

 
     memset(hdr,0,sizeof(*hdr));
     hdr->type = htons(type);
 
     memset(hdr,0,sizeof(*hdr));
     hdr->type = htons(type);


@@ -849,7 +849,7 @@ void clusterSendPing(clusterLink *link, int type) {
     /* Populate the gossip fields */
     while(freshnodes > 0 && gossipcount < 3) {
         struct dictEntry *de = dictGetRandomKey(server.cluster.nodes);
     /* Populate the gossip fields */
     while(freshnodes > 0 && gossipcount < 3) {
         struct dictEntry *de = dictGetRandomKey(server.cluster.nodes);

-        clusterNode *this = dictGetEntryVal(de);
+        clusterNode *this = dictGetVal(de);

         clusterMsgDataGossip *gossip;
         int j;
 
         clusterMsgDataGossip *gossip;
         int j;
 


@@ -970,7 +970,7 @@ void clusterCron(void) {
     /* Check if we have disconnected nodes and reestablish the connection. */
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {
     /* Check if we have disconnected nodes and reestablish the connection. */
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {

-        clusterNode *node = dictGetEntryVal(de);
+        clusterNode *node = dictGetVal(de);

 
         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
         if (node->link == NULL) {
 
         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
         if (node->link == NULL) {


@@ -1005,7 +1005,7 @@ void clusterCron(void) {
      * the oldest ping_sent time */
     for (j = 0; j < 5; j++) {
         de = dictGetRandomKey(server.cluster.nodes);
      * the oldest ping_sent time */
     for (j = 0; j < 5; j++) {
         de = dictGetRandomKey(server.cluster.nodes);

-        clusterNode *this = dictGetEntryVal(de);
+        clusterNode *this = dictGetVal(de);

 
         if (this->link == NULL) continue;
         if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue;
 
         if (this->link == NULL) continue;
         if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue;


@@ -1022,7 +1022,7 @@ void clusterCron(void) {
     /* Iterate nodes to check if we need to flag something as failing */
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {
     /* Iterate nodes to check if we need to flag something as failing */
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {

-        clusterNode *node = dictGetEntryVal(de);
+        clusterNode *node = dictGetVal(de);

         int delay;
 
         if (node->flags &
         int delay;
 
         if (node->flags &


@@ -1153,7 +1153,7 @@ sds clusterGenNodesDescription(void) {
 
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {
 
     di = dictGetIterator(server.cluster.nodes);
     while((de = dictNext(di)) != NULL) {

-        clusterNode *node = dictGetEntryVal(de);
+        clusterNode *node = dictGetVal(de);

 
         /* Node coordinates */
         ci = sdscatprintf(ci,"%.40s %s:%d ",
 
         /* Node coordinates */
         ci = sdscatprintf(ci,"%.40s %s:%d ",


@@ -1506,6 +1506,7 @@ void restoreCommand(redisClient *c) {
     /* Create the key and set the TTL if any */
     dbAdd(c->db,c->argv[1],obj);
     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
     /* Create the key and set the TTL if any */
     dbAdd(c->db,c->argv[1],obj);
     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);

+    signalModifiedKey(c->db,c->argv[1]);

     addReply(c,shared.ok);
     server.dirty++;
 }
     addReply(c,shared.ok);
     server.dirty++;
 }


@@ -1598,6 +1599,7 @@ void migrateCommand(redisClient *c) {
             robj *aux;
 
             dbDelete(c->db,c->argv[3]);
             robj *aux;
 
             dbDelete(c->db,c->argv[3]);

+            signalModifiedKey(c->db,c->argv[3]);

             addReply(c,shared.ok);
             server.dirty++;
 
             addReply(c,shared.ok);
             server.dirty++;
 


@@ -1657,6 +1659,19 @@ void dumpCommand(redisClient *c) {
     return;
 }
 
     return;
 }
 

+/* The ASKING command is required after a -ASK redirection.
+ * The client should issue ASKING before to actualy send the command to
+ * the target instance. See the Redis Cluster specification for more
+ * information. */
+void askingCommand(redisClient *c) {
+    if (server.cluster_enabled == 0) {
+        addReplyError(c,"This instance has cluster support disabled");
+        return;
+    }
+    c->flags |= REDIS_ASKING;
+    addReply(c,shared.ok);
+}
+

 /* -----------------------------------------------------------------------------
  * Cluster functions related to serving / redirecting clients
  * -------------------------------------------------------------------------- */
 /* -----------------------------------------------------------------------------
  * Cluster functions related to serving / redirecting clients
  * -------------------------------------------------------------------------- */


@@ -1750,9 +1765,12 @@ clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **arg
     }
     /* Handle the case in which we are receiving this hash slot from
      * another instance, so we'll accept the query even if in the table
     }
     /* Handle the case in which we are receiving this hash slot from
      * another instance, so we'll accept the query even if in the table

-     * it is assigned to a different node. */
-    if (server.cluster.importing_slots_from[slot] != NULL)
+     * it is assigned to a different node, but only if the client
+     * issued an ASKING command before. */
+    if (server.cluster.importing_slots_from[slot] != NULL &&
+        c->flags & REDIS_ASKING) {

         return server.cluster.myself;
         return server.cluster.myself;

+    }

     /* It's not a -ASK case. Base case: just return the right node. */
     return n;
 }
     /* It's not a -ASK case. Base case: just return the right node. */
     return n;
 }