feat:command_test

command.go

@@ -156,6 +156,9 @@ BREAK:
 			break BREAK
 		}
 	}
+	if len(futures) > 1 {
+		r.logger.Debug("batch append", len(futures), batch)
+	}
 	r.applyLog(futures...)
 }
 

command_test.go

@@ -0,0 +1,32 @@
+package papillon
+
+import (
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+func TestCommand(t *testing.T) {
+	Convey("", t, func() {
+		fu := &testFuture{}
+		fu.init()
+		new(Raft).processCommand(&command{
+			enum:     commandTest,
+			callback: fu,
+		})
+		_, err := fu.Response()
+		So(err, ShouldNotBeNil)
+		fu = &testFuture{}
+		fu.init()
+		r := Raft{
+			state: Leader,
+		}
+		r.processCommand(&command{
+			enum:     commandTest,
+			callback: fu,
+		})
+		resp, err := fu.Response()
+		So(resp, ShouldEqual, "succ")
+		So(err, ShouldBeNil)
+	})
+
+}

example/create.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+for i in {200..300} ; do
+  curl '127.0.0.1:8080/set?key=age'${i}'&value='${i}
+done
+
+

main.go

@@ -1,7 +1,6 @@
 package papillon
 
 import (
-	"bytes"
 	"container/list"
 	"errors"
 	"fmt"
@@ -1015,16 +1014,12 @@ func (r *Raft) applyLog(future ...*LogFuture) {
 	}
 	for _, fu := range future {
 		index++
-		logs = append(logs, &LogEntry{
-			Index:     index,
-			Term:      term,
-			Data:      bytes.Clone(fu.log.Data),
-			Type:      fu.log.Type,
-			CreatedAt: now,
-		})
+		fu.log.Index = index
+		fu.log.Term = term
+		fu.log.CreatedAt = now
+		logs = append(logs, fu.log)
 	}
-	err := r.logStore.SetLogs(logs)
-	if err != nil {
+	if err := r.logStore.SetLogs(logs); err != nil {
 		r.logger.Errorf("applyLog|SetLogs err :%s", err)
 		for _, fu := range future {
 			fu.fail(err)
@@ -1054,7 +1049,6 @@ func (r *Raft) processLeaderCommit() {
 		stepDown       bool
 	)
 
-	r.logger.Infof("leader commit ori :%d,cur:%d", oldCommitIndex, newCommitIndex)
 	r.setCommitIndex(newCommitIndex)
 
 	if r.cluster.latestIndex > oldCommitIndex && r.cluster.latestIndex <= newCommitIndex {
@@ -1071,6 +1065,7 @@ func (r *Raft) processLeaderCommit() {
 			readElem = append(readElem, e)
 		}
 	}
+	//r.logger.Infof("leader commit ori :%d,cur:%d", oldCommitIndex, newCommitIndex)
 	r.applyLogToFsm(newCommitIndex, readyFuture)
 	for _, element := range readElem {
 		r.leaderState.inflight.Remove(element)

raft_test.go

@@ -53,7 +53,7 @@ func buildRaft(localID string, rpc RpcInterface, store interface {
 		MaxAppendEntries:        10,
 		SnapshotThreshold:       100,
 		TrailingLogs:            1000,
-		ApplyBatch:              1,
+		ApplyBatch:              10,
 		LeadershipCatchUpRounds: 500,
 		//ShutdownOnRemove: false,
 	}
@@ -143,7 +143,8 @@ func (g *getHandle) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 func (s *setHandle) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	key := request.URL.Query().Get("key")
 	value := request.URL.Query().Get("value")
-	raft := getLeader(s.raftList...)
+	//raft := getLeader(s.raftList...)
+	raft := s.raftList[0]
 	fu := raft.Apply(kvSchema{}.encode(key, value), time.Second)
 	_, err := fu.Response()
 	if err != nil {