《游戏服务端编程实践》14.3 状态恢复与副本同步

// file: ch14/replica/replica.go
// tier: L1
// verify: go test ./ch14/replica/

package replica

import (
	"errors"
	"sort"
	"sync"
)

var (
	ErrNotPrimary     = errors.New("replica: not primary")
	ErrFenced         = errors.New("replica: stale fencing token")
	ErrUnknownReplica = errors.New("replica: unknown replica")
	ErrNoCandidate    = errors.New("replica: no failover candidate")
)

// Entry 是一条复制日志，Index 从 1 连续递增，Term 在切换时递增。
type Entry struct {
	Index   uint64
	Term    uint64
	Command string
}

// View 是一次加锁读取出的副本快照。
type View struct {
	Epoch   uint64
	Log     []Entry
	Commit  uint64
	Applied []string
}

type Replica struct {
	primary bool
	term    uint64
	epoch   uint64
	log     []Entry
	commit  uint64
	applied []string
}

// Cluster 用单调递增的 epoch 作为 fencing 令牌：旧主节点持有的令牌
// 与新主节点不同，写入会被拒绝。
type Cluster struct {
	mu      sync.Mutex
	epoch   uint64
	term    uint64
	primary string
	nodes   map[string]*Replica
	order   []string
}

func NewCluster(names ...string) *Cluster {
	c := &Cluster{
		nodes: make(map[string]*Replica, len(names)),
		order: append([]string(nil), names...),
	}
	for _, name := range names {
		c.nodes[name] = &Replica{}
	}
	return c
}

// PrimaryName 返回当前主节点名称，空字符串表示尚未选主。
func (c *Cluster) PrimaryName() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.primary
}

// View 返回指定副本的快照，第二个返回值表示副本是否存在。
func (c *Cluster) View(name string) (View, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	r, ok := c.nodes[name]
	if !ok {
		return View{}, false
	}
	return View{
		Epoch:   r.epoch,
		Log:     append([]Entry(nil), r.log...),
		Commit:  r.commit,
		Applied: append([]string(nil), r.applied...),
	}, true
}

// Elect 把指定副本提升为主节点并签发新令牌。
func (c *Cluster) Elect(name string) (uint64, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	node, ok := c.nodes[name]
	if !ok {
		return 0, ErrUnknownReplica
	}
	c.term++
	c.epoch++
	for _, other := range c.nodes {
		other.term = c.term
		other.primary = other == node
		if other == node {
			other.epoch = c.epoch
		}
	}
	c.primary = name
	return c.epoch, nil
}

// Failover 选出日志最完整的副本接管。旧主节点不会被降级，它仍以为
// 自己是主，但令牌已经过期，写入会被拒绝。
func (c *Cluster) Failover() (string, uint64, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	chosen, best := "", 0
	for _, name := range c.order {
		if name == c.primary {
			continue
		}
		if n := len(c.nodes[name].log); chosen == "" || n > best {
			chosen, best = name, n
		}
	}
	if chosen == "" {
		return "", 0, ErrNoCandidate
	}
	c.term++
	c.epoch++
	node := c.nodes[chosen]
	node.term = c.term
	node.primary = true
	node.epoch = c.epoch
	c.primary = chosen
	return chosen, c.epoch, nil
}

// Write 向主节点写入。可选的 target 用于向指定节点写入，主要用来
// 验证旧主节点在切换后会被 fencing 拒绝。
func (c *Cluster) Write(command string, target ...string) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	name := c.primary
	if len(target) > 0 {
		name = target[0]
	}
	node, ok := c.nodes[name]
	if !ok {
		return ErrUnknownReplica
	}
	if !node.primary {
		return ErrNotPrimary
	}
	if node.epoch != c.epoch {
		return ErrFenced
	}
	node.log = append(node.log, Entry{
		Index:   uint64(len(node.log)) + 1,
		Term:    node.term,
		Command: command,
	})
	return nil
}

func matchedLog(a, b []Entry) int {
	n := 0
	for n < len(a) && n < len(b) && a[n].Index == b[n].Index && a[n].Term == b[n].Term {
		n++
	}
	return n
}

// ReplicateTo 把主节点日志复制到指定副本，最多复制 max 条；
// max <= 0 表示不限制，返回本次复制的条数以便调用方限流。
func (c *Cluster) ReplicateTo(name string, max int) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.primary == "" {
		return 0, ErrNotPrimary
	}
	target, ok := c.nodes[name]
	if !ok {
		return 0, ErrUnknownReplica
	}
	if name == c.primary {
		return 0, nil
	}
	primaryLog := c.nodes[c.primary].log
	common := matchedLog(target.log, primaryLog)
	missing := len(primaryLog) - common
	if max > 0 && missing > max {
		missing = max
	}
	target.log = append([]Entry(nil), primaryLog[:common+missing]...)
	if target.commit > uint64(len(target.log)) {
		target.commit = uint64(len(target.log))
	}
	c.advanceCommitLocked()
	return missing, nil
}

func (c *Cluster) advanceCommitLocked() {
	primaryLog := c.nodes[c.primary].log
	acks := make([]int, 0, len(c.nodes))
	for name, node := range c.nodes {
		if name == c.primary {
			acks = append(acks, len(primaryLog))
		} else {
			acks = append(acks, matchedLog(node.log, primaryLog))
		}
	}
	sort.Sort(sort.Reverse(sort.IntSlice(acks)))
	committed := uint64(acks[len(acks)/2])
	for _, node := range c.nodes {
		limit := committed
		if limit > uint64(len(node.log)) {
			limit = uint64(len(node.log))
		}
		for i := node.commit; i < limit; i++ {
			node.applied = append(node.applied, node.log[i].Command)
		}
		if limit > node.commit {
			node.commit = limit
		}
	}
}

cd .local/checks/gs-practice-lab
gofmt -l ch14/replica
go vet ./ch14/replica/
go test ./ch14/replica/ -count=1