A major change specific to OkCupid and DSDC. Add the ability for Redis

to be able to take part in a DSDC cluster. This is accomplished by
successfully implementing the protocol for registering with the
dsdc_master controller, as well as the corresponding heartbeat protocol.

Author: mmaxim

src/Makefile

@@ -113,7 +113,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server
 REDIS_SENTINEL_NAME=redis-sentinel
-REDIS_SERVER_OBJ=adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o
+REDIS_SERVER_OBJ=adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o dsdc.o dsdc_sunrpc.o
 REDIS_CLI_NAME=redis-cli
 REDIS_CLI_OBJ=anet.o sds.o adlist.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o
 REDIS_BENCHMARK_NAME=redis-benchmark

src/config.c

@@ -160,6 +160,16 @@ void loadServerConfigFromString(char *config) {
             } else if (argc == 2 && !strcasecmp(argv[1],"")) {
                 resetServerSaveParams();
             }
+        } else if (!strcasecmp(argv[0], "dsdc_masters") && argc >= 2) {
+            int j, dsdcs = argc-1;
+            if (dsdcs > REDIS_DSDC_MAX_MASTERS) {
+                err = "Too many DSDC masters specified"; goto loaderr;
+            }
+            for (j = 0; j < dsdcs; j++)
+                server.dsdc_masters[j] = zstrdup(argv[j+1]);
+            server.dsdc_master_count = dsdcs;
+        } else if (!strcasecmp(argv[0], "dsdc_num_keys") && argc == 2) {
+            server.dsdc_num_keys = atoi(argv[1]);
         } else if (!strcasecmp(argv[0],"dir") && argc == 2) {
             if (chdir(argv[1]) == -1) {
                 redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",

src/dsdc.c

@@ -0,0 +1,114 @@
+#include "redis.h"
+#include "dsdc_sunrpc.h"
+#include "sha1.h"
+
+//-----------------------------------------------------------------------------
+
+void* dsdcHeartbeatLoop(void* arg);
+CLIENT* dsdcMasterConnection(int index);
+void dsdcMakekey(dsdc_key_t* out, const char* hostname, int port, int id);
+void dsdcSpawnMasterThread(int index);
+
+//-----------------------------------------------------------------------------
+
+void dsdcInit(void) {
+    for (int i = 0; i < server.dsdc_master_count; i++) {
+        dsdcSpawnMasterThread(i);
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+void dsdcSpawnMasterThread(int index) {
+    pthread_attr_t attr;
+    pthread_t thread;
+
+    CLIENT* clnt = dsdcMasterConnection(index);
+
+    pthread_attr_init(&attr);
+    if (pthread_create(&thread,&attr,dsdcHeartbeatLoop,clnt) != 0) {
+        redisLog(REDIS_WARNING,"Fatal: Can't initialize DSDC Heartbeat.");
+        exit(1);
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+CLIENT* dsdcMasterConnection(int index) {
+    char myname[256];
+    char* master_host = server.dsdc_masters[index];
+    int num_keys = server.dsdc_num_keys;
+
+    CLIENT* clnt = clnt_create(master_host, DSDC_PROG, DSDC_VERS, "tcp");
+
+    if (clnt) {
+        printf("dsdc: connected to master: %s\n", master_host);
+    } else {
+        clnt_pcreateerror(master_host);
+        exit(1);
+    }
+
+    dsdc_register2_arg_t reg_arg;
+    reg_arg.primary = TRUE;
+    reg_arg.lock_server = FALSE;
+
+    gethostname(myname, 256); 
+
+    dsdcx_slave2_t slave;
+    slave.hostname = myname;
+    slave.port = server.port;
+    slave.slave_type = DSDC_REDIS_SLAVE;
+    
+    dsdc_keyset_t keys;
+    keys.dsdc_keyset_t_val = zmalloc(sizeof(dsdc_key_t) * num_keys);
+    keys.dsdc_keyset_t_len = num_keys;
+
+    for (int i = 0; i < num_keys; i++) {
+        dsdcMakekey(&keys.dsdc_keyset_t_val[i], myname, server.port, i);
+    }
+
+    slave.keys = keys;
+    reg_arg.slave = slave;  
+
+    dsdc_res_t* res = dsdc_register2_1(&reg_arg, clnt);
+    if (!(res && *res == DSDC_OK)) {
+        clnt_perror(clnt, "dsdc: registration failed");
+        exit(1);
+    }
+
+    return clnt;
+}
+
+//-----------------------------------------------------------------------------
+
+void* dsdcHeartbeatLoop(void* arg) {
+
+    CLIENT* clnt = (CLIENT*) arg;
+
+    while (TRUE) {
+        dsdc_res_t* res = dsdc_heartbeat_1(NULL, clnt);
+        if (!res) {
+            printf("dsdc: heartbeat failed\n");
+        }
+
+        sleep(1);
+    }
+
+    return NULL;
+}
+
+//-----------------------------------------------------------------------------
+
+void dsdcMakekey(dsdc_key_t* out, const char* hostname, int port, 
+                 int index) {
+
+    char buffer[1024];
+    SHA1_CTX ctx;
+    
+    int len = snprintf(buffer, 1024, "%s-%d-%d", hostname, port, index);
+
+    SHA1Init(&ctx);
+    SHA1Update(&ctx, (unsigned char*) buffer, len);
+    SHA1Final( (unsigned char*) out, &ctx);
+
+}

src/dsdc.h

@@ -0,0 +1,2 @@
+
+void dsdcInit(void);