feat: pubsub and replication's type

Author: treeforest

config.go

@@ -44,8 +44,11 @@ type Config struct {
 	// LogPath 日志文件路径
 	LogPath string
 
-	// PullMembershipInterval 同步成员信息的间隔时间
-	PullMembershipInterval time.Duration
+	// SubscribeTTL 订阅事件的超时时间
+	SubscribeTTL time.Duration
+
+	// ReplicationType 日志复制类型
+	ReplicationType ReplicationType
 }
 
 func DefaultConfig() *Config {
@@ -54,7 +57,7 @@ func DefaultConfig() *Config {
 		Address:                 "localhost:4399",
 		MaxLogEntriesPerRequest: 40,
 		HeartbeatInterval:       time.Millisecond * 100,
-		HeartbeatTimeout:        time.Millisecond * 250,
+		HeartbeatTimeout:        time.Millisecond * 300,
 		ElectionTimeout:         time.Millisecond * 300,
 		DialTimeout:             time.Millisecond * 300,
 		DialOptions: []grpc.DialOption{
@@ -63,8 +66,9 @@ func DefaultConfig() *Config {
 		ServerOptions: []grpc.ServerOption{
 			grpc.Creds(insecure.NewCredentials()),
 		},
-		SnapshotPath:           "./snapshot",
-		LogPath:                "./log",
-		PullMembershipInterval: time.Second * 5,
+		SnapshotPath:    "./snapshot",
+		LogPath:         "./log",
+		SubscribeTTL:    time.Second * 3,
+		ReplicationType: Asynchronous,
 	}
 }

example/kv/main.go

@@ -86,13 +86,13 @@ func (s *Server) Serve(addr string) {
 		cmd.Value = c.Query("value")
 		data, _ := json.Marshal(cmd)
 
-		index, err := s.peer.Do(SET, data)
+		entry, err := s.peer.Do(SET, data)
 		if err != nil {
 			c.JSON(http.StatusOK, gin.H{"code": -1, "error": err.Error()})
 			return
 		}
 
-		log.Info("index=", index)
+		log.Info("index=", entry.Index)
 		c.JSON(http.StatusOK, gin.H{"code": 0})
 	})
 	r.POST("/del", func(c *gin.Context) {
@@ -105,13 +105,13 @@ func (s *Server) Serve(addr string) {
 		cmd.Key = c.Query("key")
 		data, _ := json.Marshal(cmd)
 
-		index, err := s.peer.Do(SET, data)
+		entry, err := s.peer.Do(SET, data)
 		if err != nil {
 			c.JSON(http.StatusOK, gin.H{"code": -1, "error": err.Error()})
 			return
 		}
 
-		log.Info("index=", index)
+		log.Info("index=", entry.Index)
 		c.JSON(http.StatusOK, gin.H{"code": 0})
 	})
 	r.GET("/get", func(c *gin.Context) {

log.go

@@ -8,7 +8,9 @@ import (
 	log "github.com/treeforest/logger"
 	"github.com/treeforest/raft/pb"
 	"os"
+	"strconv"
 	"sync"
+	"time"
 )
 
 const (
@@ -19,6 +21,8 @@ const (
 	currentTermKey = "__latest_term_key__"
 )
 
+type notify []<-chan struct{}
+
 // Log a log is a collection of log entries that are persisted to durable storage.
 type Log struct {
 	ApplyFunc   func(commandName string, command []byte)
@@ -31,17 +35,27 @@ type Log struct {
 	startTerm   uint64
 	currentTerm uint64
 	initialized bool
+	pubsub      *PubSub
 }
 
 // newLog creates a new log.
 func newLog(path string, applyFunc func(string, []byte)) *Log {
 	l := &Log{
 		ApplyFunc: applyFunc,
 		entries:   make([]pb.LogEntry, 0),
+		pubsub:    NewPubSub(),
 	}
 	return l
 }
 
+func (l *Log) Subscribe(index uint64, ttl time.Duration) Subscription {
+	return l.pubsub.Subscribe(strconv.FormatUint(index, 10), ttl)
+}
+
+func (l *Log) publish(index uint64) {
+	_ = l.pubsub.Publish(strconv.FormatUint(index, 10), struct{}{})
+}
+
 // CommitIndex the last committed index in the log.
 func (l *Log) CommitIndex() uint64 {
 	l.mutex.RLock()
@@ -310,6 +324,9 @@ func (l *Log) setCommitIndex(index uint64) error {
 		// Update commit index.
 		l.commitIndex = entry.Index
 		l.flushCommitIndex()
+		log.Infof("update commitIndex: %d", l.commitIndex)
+
+		l.publish(entry.Index)
 
 		// Apply the changes to the state machine and store the error code.
 		l.ApplyFunc(entry.CommandName, entry.Command)

pubsub.go

@@ -0,0 +1,153 @@
+package raft
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+type PubSub struct {
+	sync.RWMutex
+	subscriptions map[string]*Set // topic -> subscriptions
+}
+
+type subscription struct {
+	top string
+	ttl time.Duration
+	c   chan interface{}
+}
+
+type Subscription interface {
+	Listen() (interface{}, error)
+}
+
+// Listen 监听订阅的结果，超时返回timed out
+func (s *subscription) Listen() (interface{}, error) {
+	select {
+	case <-time.After(s.ttl):
+		return nil, errors.New("timed out")
+	case item := <-s.c:
+		return item, nil
+	}
+}
+
+func NewPubSub() *PubSub {
+	return &PubSub{
+		subscriptions: make(map[string]*Set),
+	}
+}
+
+const (
+	subscriptionBuffSize = 50
+)
+
+// Publish 发布topic对应的item，监听topic的所有订阅者都将收到item
+func (ps *PubSub) Publish(topic string, item interface{}) error {
+	ps.RLock()
+	defer ps.RUnlock()
+	s, subscribed := ps.subscriptions[topic]
+	if !subscribed {
+		return errors.New("no subscribers")
+	}
+	for _, sub := range s.ToArray() {
+		c := sub.(*subscription).c
+		// Not enough room in buffer, continue in order to not block publisher
+		if len(c) == subscriptionBuffSize {
+			continue
+		}
+		c <- item
+	}
+	return nil
+}
+
+// Subscribe 订阅事件，返回一个Listen接口
+func (ps *PubSub) Subscribe(topic string, ttl time.Duration) Subscription {
+	sub := &subscription{
+		top: topic,
+		ttl: ttl,
+		c:   make(chan interface{}, subscriptionBuffSize),
+	}
+
+	ps.Lock()
+	// Add subscription to subscriptions map
+	s, exists := ps.subscriptions[topic]
+	// If no subscription set for the topic exists, create one
+	if !exists {
+		s = NewSet()
+		ps.subscriptions[topic] = s
+	}
+	ps.Unlock()
+
+	// Add the subscription
+	s.Add(sub)
+
+	// When the timeout expires, remove the subscription
+	time.AfterFunc(ttl, func() {
+		ps.unSubscribe(sub)
+	})
+	return sub
+}
+
+func (ps *PubSub) unSubscribe(sub *subscription) {
+	ps.Lock()
+	defer ps.Unlock()
+	ps.subscriptions[sub.top].Remove(sub)
+	if ps.subscriptions[sub.top].Size() != 0 {
+		return
+	}
+	// Else, this is the last subscription for the topic.
+	// Remove the set from the subscriptions map
+	delete(ps.subscriptions, sub.top)
+}
+
+type Set struct {
+	items map[interface{}]struct{}
+	lock  *sync.RWMutex
+}
+
+func NewSet() *Set {
+	return &Set{lock: &sync.RWMutex{}, items: make(map[interface{}]struct{})}
+}
+
+func (s *Set) Add(item interface{}) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.items[item] = struct{}{}
+}
+
+func (s *Set) Exists(item interface{}) bool {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	_, exists := s.items[item]
+	return exists
+}
+
+func (s *Set) Size() int {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return len(s.items)
+}
+
+func (s *Set) ToArray() []interface{} {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	a := make([]interface{}, len(s.items))
+	i := 0
+	for item := range s.items {
+		a[i] = item
+		i++
+	}
+	return a
+}
+
+func (s *Set) Clear() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.items = make(map[interface{}]struct{})
+}
+
+func (s *Set) Remove(item interface{}) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	delete(s.items, item)
+}

raft.go

@@ -1,10 +1,12 @@
 package raft
 
+import "github.com/treeforest/raft/pb"
+
 type Raft interface {
 	Start() error
 	Stop()
 	Join(existing string)
-	Do(commandName string, command []byte) (uint64, error)
+	Do(commandName string, command []byte) (*pb.LogEntry, error)
 	Running() bool
 	IsLeader() bool
 	LeaderId() uint64

replication_type.go

@@ -0,0 +1,19 @@
+package raft
+
+// ReplicationType 日志复制类型
+type ReplicationType int
+
+const (
+	// Synchronous 同步复制。leader执行完一个写请求后，必须等待大于1/2的follower
+	// 都执行完毕，并收到确认后，才回复客户端写入成功。
+	Synchronous ReplicationType = iota
+
+	// Asynchronous 异步复制。leader执行完写请求后，会立即将结果返回给客户端，
+	// 无须等待其它副本是否写入完成。
+	Asynchronous
+
+	// Semisynchronous 半同步复制。介于同步复制与异步复制之间一种复制机制。leader只
+	// 需要等待一个follower执行完毕并返回确认信息，不需要等待大于1/2的follower都完成。
+	// 保证至少有两个节点拥有最新的数据副本。
+	Semisynchronous
+)