update: a little

Author: treeforest

.gitignore

@@ -1,3 +1,4 @@
 .idea/
 vendor/
-example/kv/log
+example/kv/log
+example/kv/snapshot

example/kv/main.go

@@ -28,26 +28,24 @@ type Command struct {
 }
 
 type Server struct {
-	raft.StateMachine
-	peer   raft.Raft
-	r      *gin.Engine
-	state  map[string]string
-	locker sync.RWMutex
-	pool   *sync.Pool
-	ch     chan *Command
+	peer    raft.Raft
+	r       *gin.Engine
+	state   map[string]string
+	locker  sync.RWMutex
+	cmdPool sync.Pool
 }
 
 func New() *Server {
-	return &Server{
+	s := &Server{
 		state:  map[string]string{},
 		locker: sync.RWMutex{},
-		pool: &sync.Pool{
+		cmdPool: sync.Pool{
 			New: func() interface{} {
 				return new(Command)
 			},
 		},
-		ch: make(chan *Command, 1024),
 	}
+	return s
 }
 
 // Save 读取状态机快照
@@ -66,27 +64,24 @@ func (s *Server) Recovery(state []byte) error {
 
 // Apply 状态机执行命令的回调函数
 func (s *Server) Apply(commandName string, command []byte) {
-	//
-}
-
-func (s *Server) handle(commandName string, command []byte) {
-	cmd := s.pool.Get().(*Command)
-	_ = json.Unmarshal(command, cmd)
+	cmd := s.cmdPool.Get().(*Command)
+	_ = json.Unmarshal(command, &cmd)
 
-	// log.Debugf("commandName:%s key:%s value:%s", commandName, cmd.Key, cmd.Value)
 	switch commandName {
 	case SET:
 		s.state[cmd.Key] = cmd.Value
 	case DEL:
 		delete(s.state, cmd.Key)
 	}
-	s.pool.Put(cmd)
+
+	*cmd = Command{}
+	s.cmdPool.Put(cmd)
 }
 
 func (s *Server) Serve(addr string) {
 	r := gin.Default()
 	r.POST("/test", func(c *gin.Context) {
-		cmd := s.pool.Get().(*Command)
+		cmd := &Command{}
 		cmd.Key = "hello"
 		cmd.Value = "world"
 		data, _ := json.Marshal(cmd)
@@ -124,6 +119,8 @@ func (s *Server) Serve(addr string) {
 		used := time.Now().Sub(since).Milliseconds()
 		done <- struct{}{}
 
+		s.peer.TakeSnapshot()
+
 		c.JSON(http.StatusOK, gin.H{
 			"code": 0,
 			"detail": fmt.Sprintf("count:%d success:%d used:%dms average:%dms",
@@ -136,7 +133,7 @@ func (s *Server) Serve(addr string) {
 			return
 		}
 
-		cmd := s.pool.Get().(*Command)
+		cmd := &Command{}
 		cmd.Key = c.Query("key")
 		cmd.Value = c.Query("value")
 		data, _ := json.Marshal(cmd)
@@ -155,7 +152,7 @@ func (s *Server) Serve(addr string) {
 			return
 		}
 
-		cmd := s.pool.Get().(*Command)
+		cmd := &Command{}
 		cmd.Key = c.Query("key")
 		data, _ := json.Marshal(cmd)
 

log.go

@@ -513,7 +513,7 @@ func (l *Log) encodeLogEntry(w io.Writer, e *pb.LogEntry) (int, error) {
 		return -1, err
 	}
 
-	if _, err = fmt.Fprintf(w, "%8x\n", len(b)); err != nil {
+	if _, err = fmt.Fprintf(w, "%08x", len(b)); err != nil {
 		return -1, err
 	}
 
@@ -523,7 +523,7 @@ func (l *Log) encodeLogEntry(w io.Writer, e *pb.LogEntry) (int, error) {
 func (l *Log) decodeLogEntry(r io.Reader, e *pb.LogEntry) (int, error) {
 	var length int = 0
 
-	_, err := fmt.Fscanf(r, "%8x\n", &length)
+	_, err := fmt.Fscanf(r, "%08x", &length)
 	if err != nil {
 		return -1, err
 	}

server.go

@@ -12,6 +12,7 @@ import (
 	"net"
 	"os"
 	"path"
+	"path/filepath"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -944,7 +945,7 @@ func (s *server) CurrentTerm() uint64 {
 
 // snapshotPath 快照路径
 func (s *server) snapshotPath(lastIndex, lastTerm uint64) string {
-	return path.Join(s.config.SnapshotPath, "snapshot", fmt.Sprintf("%d_%d.ss", lastTerm, lastIndex))
+	return filepath.Join(s.config.SnapshotPath, "snapshot", fmt.Sprintf("%d_%d.ss", lastTerm, lastIndex))
 }
 
 // TakeSnapshot 创建快照