random tested mode for test-redis.tcl, minor other stuff, version switched to 0.8

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index e6606f1822adafbb3801d03c4d75d82d0fea1e11..58b8560603bb8c63cd5caad062f1f94a48e13cea 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
  * POSSIBILITY OF SUCH DAMAGE.
  */
 

-#define REDIS_VERSION "0.07"
+#define REDIS_VERSION "0.08"

 
 #include <stdio.h>
 #include <stdlib.h>
 
 #include <stdio.h>
 #include <stdlib.h>


@@ -89,6 +89,7 @@
 #define REDIS_CLOSE 1       /* This client connection should be closed ASAP */
 #define REDIS_SLAVE 2       /* This client is a slave server */
 #define REDIS_MASTER 4      /* This client is a master server */
 #define REDIS_CLOSE 1       /* This client connection should be closed ASAP */
 #define REDIS_SLAVE 2       /* This client is a slave server */
 #define REDIS_MASTER 4      /* This client is a master server */

+#define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */

 
 /* Server replication state */
 #define REDIS_REPL_NONE 0   /* No active replication */
 
 /* Server replication state */
 #define REDIS_REPL_NONE 0   /* No active replication */


@@ -138,7 +139,7 @@ typedef struct redisClient {
     list *reply;
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */
     list *reply;
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */

-    int flags; /* REDIS_CLOSE | REDIS_SLAVE */
+    int flags; /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */

     int slaveseldb; /* slave selected db, if this client is a slave */
 } redisClient;
 
     int slaveseldb; /* slave selected db, if this client is a slave */
 } redisClient;
 


@@ -154,7 +155,7 @@ struct redisServer {
     dict **dict;
     long long dirty;            /* changes to DB from the last save */
     list *clients;
     dict **dict;
     long long dirty;            /* changes to DB from the last save */
     list *clients;

-    list *slaves;
+    list *slaves, *monitors;

     char neterr[ANET_ERR_LEN];
     aeEventLoop *el;
     int cronloops;              /* number of times the cron function run */
     char neterr[ANET_ERR_LEN];
     aeEventLoop *el;
     int cronloops;              /* number of times the cron function run */


@@ -171,6 +172,7 @@ struct redisServer {
     int maxidletime;
     int dbnum;
     int daemonize;
     int maxidletime;
     int dbnum;
     int daemonize;

+    char *pidfile;

     int bgsaveinprogress;
     struct saveparam *saveparams;
     int saveparamslen;
     int bgsaveinprogress;
     struct saveparam *saveparams;
     int saveparamslen;


@@ -234,7 +236,7 @@ static void addReplySds(redisClient *c, sds s);
 static void incrRefCount(robj *o);
 static int saveDbBackground(char *filename);
 static robj *createStringObject(char *ptr, size_t len);
 static void incrRefCount(robj *o);
 static int saveDbBackground(char *filename);
 static robj *createStringObject(char *ptr, size_t len);

-static void replicationFeedSlaves(struct redisCommand *cmd, int dictid, robj **argv, int argc);
+static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);

 static int syncWithMaster(void);
 
 static void pingCommand(redisClient *c);
 static int syncWithMaster(void);
 
 static void pingCommand(redisClient *c);


@@ -282,6 +284,7 @@ static void sortCommand(redisClient *c);
 static void lremCommand(redisClient *c);
 static void infoCommand(redisClient *c);
 static void mgetCommand(redisClient *c);
 static void lremCommand(redisClient *c);
 static void infoCommand(redisClient *c);
 static void mgetCommand(redisClient *c);

+static void monitorCommand(redisClient *c);

 
 /*================================= Globals ================================= */
 
 
 /*================================= Globals ================================= */
 


@@ -334,6 +337,7 @@ static struct redisCommand cmdTable[] = {
     {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
     {"sort",sortCommand,-2,REDIS_CMD_INLINE},
     {"info",infoCommand,1,REDIS_CMD_INLINE},
     {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
     {"sort",sortCommand,-2,REDIS_CMD_INLINE},
     {"info",infoCommand,1,REDIS_CMD_INLINE},

+    {"monitor",monitorCommand,1,REDIS_CMD_INLINE},

     {NULL,NULL,0,0}
 };
 
     {NULL,NULL,0,0}
 };
 


@@ -715,6 +719,7 @@ static void initServerConfig() {
     server.bindaddr = NULL;
     server.glueoutputbuf = 1;
     server.daemonize = 0;
     server.bindaddr = NULL;
     server.glueoutputbuf = 1;
     server.daemonize = 0;

+    server.pidfile = "/var/run/redis.pid";

     server.dbfilename = "dump.rdb";
     ResetServerSaveParams();
 
     server.dbfilename = "dump.rdb";
     ResetServerSaveParams();
 


@@ -737,11 +742,12 @@ static void initServer() {
 
     server.clients = listCreate();
     server.slaves = listCreate();
 
     server.clients = listCreate();
     server.slaves = listCreate();

+    server.monitors = listCreate();

     server.objfreelist = listCreate();
     createSharedObjects();
     server.el = aeCreateEventLoop();
     server.dict = zmalloc(sizeof(dict*)*server.dbnum);
     server.objfreelist = listCreate();
     createSharedObjects();
     server.el = aeCreateEventLoop();
     server.dict = zmalloc(sizeof(dict*)*server.dbnum);

-    if (!server.dict || !server.clients || !server.slaves || !server.el || !server.objfreelist)
+    if (!server.dict || !server.clients || !server.slaves || !server.monitors || !server.el || !server.objfreelist)

         oom("server initialization"); /* Fatal OOM */
     server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
     if (server.fd == -1) {
         oom("server initialization"); /* Fatal OOM */
     server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
     if (server.fd == -1) {


@@ -878,6 +884,8 @@ static void loadServerConfig(char *filename) {
             else {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
             else {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }

+        } else if (!strcmp(argv[0],"pidfile") && argc == 2) {
+          server.pidfile = zstrdup(argv[1]);

         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }


@@ -918,9 +926,10 @@ static void freeClient(redisClient *c) {
     assert(ln != NULL);
     listDelNode(server.clients,ln);
     if (c->flags & REDIS_SLAVE) {
     assert(ln != NULL);
     listDelNode(server.clients,ln);
     if (c->flags & REDIS_SLAVE) {

-        ln = listSearchKey(server.slaves,c);
+        list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
+        ln = listSearchKey(l,c);

         assert(ln != NULL);
         assert(ln != NULL);

-        listDelNode(server.slaves,ln);
+        listDelNode(l,ln);

     }
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
     }
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;


@@ -1079,7 +1088,9 @@ static int processCommand(redisClient *c) {
     dirty = server.dirty;
     cmd->proc(c);
     if (server.dirty-dirty != 0 && listLength(server.slaves))
     dirty = server.dirty;
     cmd->proc(c);
     if (server.dirty-dirty != 0 && listLength(server.slaves))

-        replicationFeedSlaves(cmd,c->dictid,c->argv,c->argc);
+        replicationFeedSlaves(server.slaves,cmd,c->dictid,c->argv,c->argc);
+    if (listLength(server.monitors))
+        replicationFeedSlaves(server.monitors,cmd,c->dictid,c->argv,c->argc);

     server.stat_numcommands++;
 
     /* Prepare the client for the next command */
     server.stat_numcommands++;
 
     /* Prepare the client for the next command */


@@ -1091,8 +1102,8 @@ static int processCommand(redisClient *c) {
     return 1;
 }
 
     return 1;
 }
 

-static void replicationFeedSlaves(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
-    listNode *ln = server.slaves->head;
+static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
+    listNode *ln = slaves->head;

     robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */
     int outc = 0, j;
     
     robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */
     int outc = 0, j;
     


@@ -1899,6 +1910,9 @@ static void bgsaveCommand(redisClient *c) {
 static void shutdownCommand(redisClient *c) {
     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
     if (saveDb(server.dbfilename) == REDIS_OK) {
 static void shutdownCommand(redisClient *c) {
     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
     if (saveDb(server.dbfilename) == REDIS_OK) {

+        if (server.daemonize) {
+          unlink(server.pidfile);
+        }

         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
         exit(1);
     } else {
         redisLog(REDIS_WARNING,"Server exit now, bye bye...");
         exit(1);
     } else {


@@ -2905,6 +2919,9 @@ static void syncCommand(redisClient *c) {
     time_t start = time(NULL);
     char sizebuf[32];
 
     time_t start = time(NULL);
     char sizebuf[32];
 

+    /* ignore SYNC if aleady slave or in monitor mode */
+    if (c->flags & REDIS_SLAVE) return;
+

     redisLog(REDIS_NOTICE,"Slave ask for syncronization");
     if (flushClientOutput(c) == REDIS_ERR || saveDb(server.dbfilename) != REDIS_OK)
         goto closeconn;
     redisLog(REDIS_NOTICE,"Slave ask for syncronization");
     if (flushClientOutput(c) == REDIS_ERR || saveDb(server.dbfilename) != REDIS_OK)
         goto closeconn;


@@ -3014,6 +3031,16 @@ static int syncWithMaster(void) {
     return REDIS_OK;
 }
 
     return REDIS_OK;
 }
 

+static void monitorCommand(redisClient *c) {
+    /* ignore MONITOR if aleady slave or in monitor mode */
+    if (c->flags & REDIS_SLAVE) return;
+
+    c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
+    c->slaveseldb = 0;
+    if (!listAddNodeTail(server.monitors,c)) oom("listAddNodeTail");
+    addReply(c,shared.ok);
+}
+

 /* =================================== Main! ================================ */
 
 static void daemonize(void) {
 /* =================================== Main! ================================ */
 
 static void daemonize(void) {


@@ -3033,7 +3060,7 @@ static void daemonize(void) {
         if (fd > STDERR_FILENO) close(fd);
     }
     /* Try to write the pid file */
         if (fd > STDERR_FILENO) close(fd);
     }
     /* Try to write the pid file */

-    fp = fopen("/var/run/redis.pid","w");
+    fp = fopen(server.pidfile,"w");

     if (fp) {
         fprintf(fp,"%d\n",getpid());
         fclose(fp);
     if (fp) {
         fprintf(fp,"%d\n",getpid());
         fclose(fp);


@@ -3056,7 +3083,7 @@ int main(int argc, char **argv) {
         redisLog(REDIS_NOTICE,"DB loaded from disk");
     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
         redisLog(REDIS_NOTICE,"DB loaded from disk");
     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");

-    redisLog(REDIS_NOTICE,"The server is now ready to accept connections");
+    redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);

     aeMain(server.el);
     aeDeleteEventLoop(server.el);
     return 0;
     aeMain(server.el);
     aeDeleteEventLoop(server.el);
     return 0;