Added infrastructure code for processing commands from clients.

srned · srned · commit c9452aba3621 · 2014-10-10T00:42:23.000-07:00
Added code to handle commands from clients in leader. A logEntry is
created for the command and is appended to the leader log. The log gets
propagated to other nodes in the cluster during its next heartbeat. Each
node upon receipt of the AppendEntry request will update its local log.
diff --git a/src/cluster.c b/src/cluster.c
@@ -162,6 +162,7 @@ void clusterInit(void) {
     server.cluster->election_timeout = PREZ_CLUSTER_ELECTION_TIMEOUT;
     server.cluster->heartbeat_interval = PREZ_CLUSTER_HEARTBEAT_INTERVAL;
     server.cluster->synced_nodes = dictCreate(&clusterNodesDictType,NULL);
+    server.cluster->start_index = 0;
     server.cluster->current_term = 0;
     server.cluster->commit_index = 0;
     server.cluster->votes_granted = 0;
@@ -286,7 +287,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
         anetEnableTcpNoDelay(NULL,cfd);
 
         /* Use non-blocking I/O for cluster messages. */
-        prezLog(PREZ_VERBOSE,"Accepted cluster node %s:%d", cip, cport);
+        prezLog(PREZ_VERBOSE,"Accepted cluster node %s:%d cfd:%d", cip, cport, cfd);
         /* Create a link object we use to handle the connection.
          * It gets passed to the readable handler when data is available.
          * Initiallly the link->node pointer is set to NULL as we don't know
@@ -381,6 +382,28 @@ void clusterRenameNode(clusterNode *node, char *newname) {
     clusterAddNode(node);
 }
 
+void clusterProcessCommand(prezClient *c) {
+    logEntry entry;
+    long long commit_index;
+
+    prezLog(PREZ_DEBUG,"clusterProcessCommand");
+    entry.index = logCurrentIndex()+1;
+    entry.term = server.cluster->current_term;
+    memcpy(entry.commandName,c->cmd->name,strlen(c->cmd->name)+1);
+    memcpy(entry.command,c->cmd->name,strlen(c->cmd->name)+1);
+
+    logWriteEntry(entry);
+
+    dictAdd(server.cluster->synced_nodes,
+            sdsnewlen(myself->name,PREZ_CLUSTER_NAMELEN),&node_synced);
+
+    if(dictSize(server.cluster->nodes) == 1) {
+        commit_index = logCurrentIndex();
+        logCommitIndex(commit_index);
+        prezLog(PREZ_DEBUG,"commit index: %lld", commit_index);
+    }
+}
+
 int clusterProcessPacket(clusterLink *link) {
     clusterMsg *hdr = (clusterMsg*) link->rcvbuf;
     uint32_t totlen = ntohl(hdr->totlen);
@@ -414,7 +437,7 @@ int clusterProcessPacket(clusterLink *link) {
         uint32_t explen;
         explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
         explen += (sizeof(clusterMsgDataAppendEntries) +
-                (hdr->data.appendentries.entries.log_entries_count-1) *
+                (ntohs(hdr->data.appendentries.entries.log_entries_count)-1) *
                 sizeof(logEntry));
         if (totlen != explen) return 1;
 
@@ -442,7 +465,8 @@ int clusterProcessPacket(clusterLink *link) {
         explen += sizeof(clusterMsgDataResponseAppendEntries);
         if (totlen != explen) return 1;
 
-        prezLog(PREZ_DEBUG,"--- Received AppendEntriesResponse term %lld, index: %lld, commit_index: %lld, ok: %d",
+        prezLog(PREZ_DEBUG,"--- Received AppendEntriesResponse port:%d term %lld, index: %lld, commit_index: %lld, ok: %d",
+                link->node->port,
                 hdr->data.responseappendentries.entries.term, 
                 hdr->data.responseappendentries.entries.index,
                 hdr->data.responseappendentries.entries.commit_index,
@@ -767,6 +791,7 @@ void clusterProcessResponseAppendEntries(clusterLink *link,
                     clusterNode *cnode = dictGetVal(de);
                     log_indices[i] = cnode->prev_log_index;
                 }
+                dictReleaseIterator(di);
                 qsort(log_indices,dictSize(server.cluster->nodes),sizeof(long long),
                         compareIndices);
                 reverseIndices(log_indices,dictSize(server.cluster->nodes));
@@ -874,7 +899,7 @@ void clusterSendAppendEntries(clusterLink *link) {
     memcpy(hdr->data.appendentries.entries.leaderid, myself->name,
             PREZ_CLUSTER_NAMELEN);
     hdr->data.appendentries.entries.prev_log_index = node->prev_log_index;
-    hdr->data.appendentries.entries.prev_log_term = 0;
+    hdr->data.appendentries.entries.prev_log_term = node->last_sent_term;
     hdr->data.appendentries.entries.leader_commit_index = 
         server.cluster->commit_index;
     node->last_sent_entry = NULL; 
@@ -889,8 +914,23 @@ void clusterSendAppendEntries(clusterLink *link) {
                 node->prev_log_index-server.cluster->start_index);
         while(ln && logcount <= server.cluster->log_max_entries_per_request) {
             le_node = listNodeValue(ln);
-            memcpy(&(hdr->data.appendentries.entries.log_entries[logcount]),
+            hdr->data.appendentries.entries.log_entries[logcount].term = 
+                le_node->log_entry.term;
+            hdr->data.appendentries.entries.log_entries[logcount].index = 
+                le_node->log_entry.index;
+            memcpy(hdr->data.appendentries.entries.log_entries[logcount].commandName,
+                    le_node->log_entry.commandName, PREZ_COMMAND_NAMELEN);
+            memcpy(hdr->data.appendentries.entries.log_entries[logcount].command,
+                    le_node->log_entry.command, PREZ_COMMAND_NAMELEN);
+            prezLog(PREZ_DEBUG,"AE term:%lld, index:%lld, cmd:%s, cmd:%s",
+                    hdr->data.appendentries.entries.log_entries[logcount].term,
+                    hdr->data.appendentries.entries.log_entries[logcount].index,
+                    hdr->data.appendentries.entries.log_entries[logcount].commandName,
+                    hdr->data.appendentries.entries.log_entries[logcount].command);
+
+            /*memcpy(&(hdr->data.appendentries.entries.log_entries[logcount]),
                     &(le_node->log_entry), sizeof(logEntry));
+            */
             ln_next = listNextNode(ln);
             logcount++;
             if (!ln_next) {
@@ -901,17 +941,19 @@ void clusterSendAppendEntries(clusterLink *link) {
             ln = ln_next;
         } 
     }
-    hdr->data.appendentries.entries.log_entries_count = logcount;
+    hdr->data.appendentries.entries.log_entries_count = htons(logcount);
     node->last_sent_term = server.cluster->current_term;
 
     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
     totlen += (sizeof(clusterMsgDataAppendEntries)-sizeof(logEntry));
     totlen += (sizeof(logEntry)*logcount);
     hdr->totlen = htonl(totlen);
 
-    prezLog(PREZ_DEBUG, "Sending heartbeat buf:%s, sizeof(clustermsg): "
-            "%lu, totlen: %d\n",
-            buf, sizeof(clusterMsg), ntohl(hdr->totlen));
+    prezLog(PREZ_DEBUG, "Sending heartbeat to port:%d, buf:%s, sizeof(clustermsg): "
+            "%lu, totlen: %d logcount: %d\n",
+            node->port, buf, sizeof(clusterMsg), 
+            ntohl(hdr->totlen), 
+            ntohs(hdr->data.appendentries.entries.log_entries_count));
 
     clusterSendMessage(link,buf,ntohl(hdr->totlen));
 }
@@ -1018,6 +1060,7 @@ void clusterCron(void) {
                 clusterSendHeartbeat(node->link);
             }
         }
+        dictReleaseIterator(di);
     }
 
 }
diff --git a/src/cluster.h b/src/cluster.h
@@ -234,10 +234,11 @@ void clusterSendResponseAppendEntries(clusterLink *link, int ok);
 int loadLogFile(void); 
 int logTruncate(long long index, long long term);
 sds catLogEntry(sds dst, int argc, robj **argv);
-int logWriteEntry(logEntry *e);
+int logWriteEntry(logEntry e);
 int logAppendEntries(clusterMsgDataAppendEntries entries);
 int logCommitIndex(long long index);
 int logSync(void);
+long long logCurrentIndex(void);
 
 /* Functions as macros */
 #define quorumSize ((dictSize(server.cluster->nodes) / 2) + 1)
diff --git a/src/log.c b/src/log.c
@@ -29,6 +29,7 @@
 
 #include "prez.h"
 #include "cluster.h"
+#include "endianconv.h"
 
 #include <signal.h>
 #include <fcntl.h>
@@ -81,7 +82,7 @@ int loadLogFile(void) {
         argc = atoi(buf+1);
         if (argc < 1) goto fmterr;
 
-        entry = zmalloc(sizeof(entry));
+        entry = zmalloc(sizeof(*entry));
         for (j = 0; j < argc; j++) {
             if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr;
             if (buf[0] != '$') goto fmterr;
@@ -163,12 +164,12 @@ int logTruncate(long long index, long long term) {
         }
 
         entry = listNodeValue(listIndex(server.cluster->log_entries, 
-                    index - server.cluster->start_index));
+                    index - server.cluster->start_index-1));
         ftruncate(server.cluster->log_fd, entry->position);
         server.cluster->log_current_size = entry->position;
         listRewind(server.cluster->log_entries, &li);
         li.next = listIndex(server.cluster->log_entries,
-                index - server.cluster->start_index - 1);
+                index - server.cluster->start_index-1);
         while ((ln = listNext(&li)) != NULL) {
             logEntryNode *le = listNodeValue(ln);
             listDelNode(server.cluster->log_entries,ln);
@@ -203,31 +204,31 @@ sds catLogEntry(sds dst, int argc, robj **argv) {
     return dst;
 }
 
-int logWriteEntry(logEntry *e) {
+int logWriteEntry(logEntry e) {
     ssize_t nwritten;
     robj *argv[4];
     sds buf = sdsempty();
     logEntryNode *en;
 
     if (listLength(server.cluster->log_entries) > 0) {
         en = listNodeValue(listIndex(server.cluster->log_entries,-1));
-        if (e->term < en->log_entry.term) {
+        if (e.term < en->log_entry.term) {
             prezLog(PREZ_NOTICE, "Cannot append entry with earlier term." 
                     "term:%lld index:%lld, last term:%lld index:%lld",
-                    e->term, e->index, en->log_entry.term, en->log_entry.index);
+                    e.term, e.index, en->log_entry.term, en->log_entry.index);
             return PREZ_ERR;
-        } else if (e->term == en->log_entry.term && e->index <= en->log_entry.index) {
+        } else if (e.term == en->log_entry.term && e.index <= en->log_entry.index) {
             prezLog(PREZ_NOTICE, "Cannot append entry with earlier index." 
                     "term:%lld index:%lld, last term:%lld index:%lld",
-                    e->term, e->index, en->log_entry.term, en->log_entry.index);
+                    e.term, e.index, en->log_entry.term, en->log_entry.index);
             return PREZ_ERR;
         }
     }
 
-    argv[0] = createStringObjectFromLongLong(ntohl(e->index));
-    argv[1] = createStringObjectFromLongLong(ntohl(e->term));
-    argv[2] = createStringObject(e->commandName, PREZ_COMMAND_NAMELEN);
-    argv[3] = createStringObject(e->command, PREZ_COMMAND_NAMELEN);
+    argv[0] = createStringObjectFromLongLong(e.index);
+    argv[1] = createStringObjectFromLongLong(e.term);
+    argv[2] = createStringObject(e.commandName, PREZ_COMMAND_NAMELEN);
+    argv[3] = createStringObject(e.command, PREZ_COMMAND_NAMELEN);
     buf = catLogEntry(buf, 4, argv);
     decrRefCount(argv[0]);
     decrRefCount(argv[1]);
@@ -238,10 +239,18 @@ int logWriteEntry(logEntry *e) {
         prezLog(PREZ_NOTICE,"log write incomplete");
     }
     
-    en = zmalloc(sizeof(en));
-    memcpy(&(en->log_entry),e,sizeof(logEntry));
+    en = zmalloc(sizeof(*en));
+    en->log_entry.term = e.index;
+    en->log_entry.index = e.term;
+    memcpy(en->log_entry.commandName, e.commandName, PREZ_COMMAND_NAMELEN);
+    memcpy(en->log_entry.command, e.command, PREZ_COMMAND_NAMELEN);
+    //memcpy(&en->log_entry,&e,sizeof(logEntry));
     en->position = server.cluster->log_current_size;
     server.cluster->log_current_size += sdslen(buf);
+    prezLog(PREZ_DEBUG,"en term:%lld/%lld index:%lld, %lu", 
+            en->log_entry.term,
+            e.term,
+            en->log_entry.index, sizeof(en));
     listAddNodeTail(server.cluster->log_entries,en);
 
     return PREZ_OK;
@@ -252,8 +261,9 @@ int logAppendEntries(clusterMsgDataAppendEntries entries) {
     logEntry *e;
 
     e = (logEntry*) entries.log_entries;
+    prezLog(PREZ_DEBUG,"count:%d",ntohs(entries.log_entries_count));
     for(i=0;i<ntohs(entries.log_entries_count);i++) {
-        if(logWriteEntry(e)) {
+        if(logWriteEntry(entries.log_entries[i])) {
             prezLog(PREZ_NOTICE, "log write error");
             return PREZ_ERR;
         }
@@ -282,6 +292,7 @@ int logCommitIndex(long long index) {
         server.cluster->commit_index = entry->log_entry.index;
 
         /* Process command which is what commit is really about */
+        prezLog(PREZ_DEBUG,"cmd invoke");
         /* return if join command */
     }
     return PREZ_OK;
@@ -291,3 +302,15 @@ int logSync(void) {
     if(fsync(server.cluster->log_fd) == -1) return PREZ_ERR;
     return PREZ_OK;
 }
+
+long long logCurrentIndex(void) {
+    listNode *ln;
+    logEntryNode *entry;
+
+    ln = listIndex(server.cluster->log_entries, -1);
+    if (ln) {
+        entry = ln->value;
+        return(entry->log_entry.index);
+    }
+    return 0;
+}
diff --git a/src/networking.c b/src/networking.c
@@ -1119,11 +1119,8 @@ void processInputBuffer(prezClient *c) {
             resetClient(c);
         } else {
             /* Only reset the client when the command was executed. */
-            //FIXME: come back
-#if 0
             if (processCommand(c) == PREZ_OK)
                 resetClient(c);
-#endif
         }
     }
 }
diff --git a/src/prez.c b/src/prez.c
diff --git a/src/prez.h b/src/prez.h

Original file line number	Diff line number	Diff line change
`@@ -1119,11 +1119,8 @@ void processInputBuffer(prezClient *c) {`
`1119`	`1119`	`resetClient(c);`
`1120`	`1120`	`} else {`
`1121`	`1121`	`/* Only reset the client when the command was executed. */`
`1122`		`- //FIXME: come back`
`1123`		`-#if 0`
`1124`	`1122`	`if (processCommand(c) == PREZ_OK)`
`1125`	`1123`	`resetClient(c);`
`1126`		`-#endif`
`1127`	`1124`	`}`
`1128`	`1125`	`}`
`1129`	`1126`	`}`