Adding GC commands from witnesses

Author: seanschoi

src/Makefile

@@ -53,9 +53,9 @@ endif
 # Override default settings if possible
 -include .make-settings
 
-FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS) -I../deps/geohash-int
-FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG)
-FINAL_LIBS=-lm
+FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS) -I../deps/geohash-int -I/home/p4/Desktop/witnesscmd -L/home/p4/Desktop/witnesscmd
+FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG) -L/home/p4/Desktop/witnesscmd
+FINAL_LIBS=-ldl -lwitnesscmd -lm
 DEBUG=-g -ggdb
 
 ifeq ($(uname_S),SunOS)
@@ -105,6 +105,7 @@ endif
 ifeq ($(MALLOC),jemalloc)
 	DEPENDENCY_TARGETS+= jemalloc
 	FINAL_CFLAGS+= -DUSE_JEMALLOC -I../deps/jemalloc/include
+	FINAL_LDFLAGS+= -L../deps/jemalloc/lib/
 	FINAL_LIBS+= ../deps/jemalloc/lib/libjemalloc.a
 endif
 

src/aof.c

@@ -200,7 +200,8 @@ ssize_t aofRewriteBufferWrite(int fd) {
 /* Starts a background task that performs fsync() against the specified
  * file descriptor (the one of the AOF file) in another thread. */
 void aof_background_fsync(int fd) {
-    bioCreateBackgroundJob(BIO_AOF_FSYNC,(void*)(long)fd,NULL,server.currentOpNum);
+    bioCreateBackgroundJob(BIO_AOF_FSYNC,
+        (void*)(long)fd,NULL,server.currentOpNum, 0);
 }
 
 /* Called when the user switches from "appendonly yes" to "appendonly no"
@@ -1504,7 +1505,9 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
             server.aof_state = AOF_ON;
 
         /* Asynchronously close the overwritten AOF. */
-        if (oldfd != -1) bioCreateBackgroundJob(BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,0);
+        if (oldfd != -1) { 
+            bioCreateBackgroundJob(BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,0,0);
+        }
 
         serverLog(LL_VERBOSE,
             "Background AOF rewrite signal handler took %lldus", ustime()-now);

src/bio.c

@@ -60,6 +60,11 @@
 
 #include "server.h"
 #include "bio.h"
+#include "udp.h"
+#include "witnesscmd.h"
+
+#define SRC_ADDR "10.10.10.104"
+#define WITNESS_PORT 1111
 
 static pthread_t bio_threads[BIO_NUM_OPS];
 static pthread_mutex_t bio_mutex[BIO_NUM_OPS];
@@ -81,6 +86,7 @@ struct bio_job {
      * arguments we can just pass a pointer to a structure or alike. */
     void *arg1, *arg2;
     long long arg3;
+    uint32_t arg4;
 };
 
 void *bioProcessBackgroundJobs(void *arg);
@@ -124,13 +130,15 @@ void bioInit(void) {
     }
 }
 
-void bioCreateBackgroundJob(int type, void *arg1, void *arg2, long long arg3) {
+void bioCreateBackgroundJob(int type,
+        void *arg1, void *arg2, long long arg3, uint32_t arg4) {
     struct bio_job *job = zmalloc(sizeof(*job));
 
     job->time = time(NULL);
     job->arg1 = arg1;
     job->arg2 = arg2;
     job->arg3 = arg3;
+    job->arg4 = arg4;
     pthread_mutex_lock(&bio_mutex[type]);
     listAddNodeTail(bio_jobs[type],job);
     bio_pending[type]++;
@@ -186,17 +194,26 @@ void *bioProcessBackgroundJobs(void *arg) {
             aof_fsync((long)job->arg1);
             server.aof_last_fsync_opNum = job->arg3;
         } else if (type == BIO_FSYNC_AND_GC_WITNESS) {
-            sds gcCmd = (sds)job->arg1;
+            /* Create a UDP socket */
+            int s = createSocket();
+            /* Get the GC commands and the number of commands */
+            witnesscmd_t* cmds = (witnesscmd_t*) job->arg1;
+            uint32_t num_cmds = job->arg4;
+            /* lastOpNum used for FSYNC */
             long long lastOpNum = job->arg3;
             if (lastOpNum > server.aof_last_fsync_opNum) {
                 aof_fsync((long)job->arg2);
                 server.aof_last_fsync_opNum = lastOpNum;
             }
+            /* Iterate through the witnesses and send GC cmds */
             for (int i = 0; i < server.numWitness; ++i) {
-                if (anetWrite(server.fdToWitness[i], gcCmd, sdslen(gcCmd)) == -1) {
-                    fprintf(stderr, "Error while sending witness GC. %s", strerror(errno));
+                for (uint32_t j = 0; j < num_cmds; j++) {
+                    udpWrite(s, SRC_ADDR, server.addrToWitness[i], WITNESS_PORT, WITNESS_PORT, witness_data(&cmds[j]), false);
                 }
             }
+            /* Free cmds and close the socket */
+            zfree(cmds);
+            close(s);
         } else {
             serverPanic("Wrong job type in bioProcessBackgroundJobs().");
         }

src/bio.h

@@ -27,9 +27,12 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include <stdlib.h>
+
 /* Exported API */
 void bioInit(void);
-void bioCreateBackgroundJob(int type, void *arg1, void *arg2, long long arg3);
+void bioCreateBackgroundJob(int type,
+    void *arg1, void *arg2, long long arg3, uint32_t arg4);
 unsigned long long bioPendingJobsOfType(int type);
 void bioWaitPendingJobsLE(int type, unsigned long long num);
 time_t bioOlderJobOfType(int type);

src/debug.c

@@ -32,7 +32,6 @@
 #include "crc64.h"
 
 #include <arpa/inet.h>
-#include <signal.h>
 
 #ifdef HAVE_BACKTRACE
 #include <execinfo.h>

src/networking.c

@@ -29,6 +29,7 @@
 
 #include "server.h"
 #include "timeTrace.h"
+#include "udp.h"
 #include <sys/uio.h>
 #include <sys/time.h>
 #include <math.h>
@@ -2083,17 +2084,3 @@ int processEventsWhileBlocked(void) {
     }
     return count;
 }
-
-void connectToWitness() {
-    for (int i = 0; i < server.numWitness; ++i) {
-        if (server.fdToWitness[i] > 0) {
-            continue;
-        }
-        char err[ANET_ERR_LEN];
-        server.fdToWitness[i] = anetTcpConnect(err, server.addrToWitness[i], server.port);
-        if (server.fdToWitness[i] == ANET_ERR) {
-            serverLog(LL_WARNING, "Error connecting to witness:%s", server.addrToWitness[i]);
-            continue;
-        }
-    }
-}

src/server.c

@@ -2041,9 +2041,6 @@ void initServer(void) {
     slowlogInit();
     latencyMonitorInit();
     bioInit();
-
-    /* Connect to witness servers */
-    connectToWitness();
 }
 
 /* Populates the Redis Command Table starting from the hard coded list

src/server.h

@@ -1175,7 +1175,6 @@ int handleClientsWithPendingWrites(void);
 int clientHasPendingReplies(client *c);
 void unlinkClient(client *c);
 int writeToClient(int fd, client *c, int handler_installed);
-void connectToWitness();
 
 #ifdef __GNUC__
 void addReplyErrorFormat(client *c, const char *fmt, ...)

src/witnessTracker.c

@@ -22,6 +22,7 @@
 #include "rifl.h"
 #include "timeTrace.h"
 #include "udp.h"
+#include "witnesscmd.h"
 
 /* functions from aof.c */
 struct client *createFakeClient();
@@ -41,7 +42,7 @@ struct WitnessGcInfo {
 
 /* 0th index is not used. */
 struct WitnessGcInfo unsyncedRpcs[WITNESS_BATCH_SIZE] = {{0,0,0}, };
-int unsyncedRpcsSize = 0;
+uint32_t unsyncedRpcsSize = 0;
 
 struct WitnessGcBioContext {
     long long maxOpNum;
@@ -51,37 +52,23 @@ struct WitnessGcBioContext {
 /*================================= Functions =============================== */
 void scheduleFsyncAndWitnessGc() {
     record("start constructing gc RPC.", 0, 0, 0, 0);
-    // TODO: Construct GC packet
-    char* masterIdxStr = "1";
-    int s = createSocket();
-    sds cmdstr = sdscatprintf(sdsempty(), "*%d\r\n$3\r\nWGC\r\n$%d\r\n%s\r\n",
-            2 + 3 * unsyncedRpcsSize, (int)strlen(masterIdxStr), masterIdxStr);
-    for (int i = 0; i < unsyncedRpcsSize; ++i) {
-        int hashIndex_len, clientId_len, requestId_len;
-        char hashIndex_str[LONG_STR_SIZE];
-        char clientId_str[LONG_STR_SIZE];
-        char requestId_str[LONG_STR_SIZE];
-
-        hashIndex_len = ulltoa64(hashIndex_str, sizeof(hashIndex_str),
-                unsyncedRpcs[i].hashIndex);
-        clientId_len = ulltoa64(clientId_str, sizeof(clientId_str),
-                unsyncedRpcs[i].clientId);
-        requestId_len = ulltoa64(requestId_str, sizeof(requestId_str),
-                unsyncedRpcs[i].requestId);
-//        hashIndex_len = ll2string(hashIndex_str, sizeof(hashIndex_str),
-//                unsyncedRpcs[i].hashIndex);
-//        clientId_len = ll2string(clientId_str, sizeof(clientId_str),
-//                unsyncedRpcs[i].clientId);
-//        requestId_len = ll2string(requestId_str, sizeof(requestId_str),
-//                unsyncedRpcs[i].requestId);
-        cmdstr = sdscatprintf(cmdstr, "$%d\r\n%s\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n",
-                hashIndex_len, hashIndex_str, clientId_len, clientId_str, requestId_len, requestId_str);
+    /* Create a set of GC Cmds */
+    witnesscmd_t* cmds =
+        (witnesscmd_t *) zmalloc(unsyncedRpcsSize * sizeof(witnesscmd_t));
+    for (uint32_t i = 0; i < unsyncedRpcsSize; ++i) {
+        init_witnesscmd(cmds, "D", unsyncedRpcs[i].clientId,
+            unsyncedRpcs[i].requestId,
+            unsyncedRpcs[i].hashIndex, "", 0);
+        cmds++;
     }
     unsyncedRpcsSize = 0;
-    close(s);
     record("constructed gc RPC.", 0, 0, 0, 0);
-    bioCreateBackgroundJob(BIO_FSYNC_AND_GC_WITNESS, (void*)cmdstr,
-            (void*)(long)server.aof_fd, server.currentOpNum);
+
+    /* Submit the GC job to the background thread */
+    bioCreateBackgroundJob(BIO_FSYNC_AND_GC_WITNESS,
+        cmds,
+        (void*)(long)server.aof_fd, server.currentOpNum,
+        unsyncedRpcsSize);
     record("bioBackgroundJob Created.", 0, 0, 0, 0);
 }