1
2
3
4
5

type Config struct {
	Num    int              // config number, version also
	Shards [NShards]int     // shard -> gid
	Groups map[int][]string // gid -> servers[]
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

// Join according to new Group(gid -> servers) to change the Config
func (cf *MemoryConfigStateMachine) Join(groups map[int][]string) Err {
	lastConfig := cf.Configs[len(cf.Configs)-1]
	newConfig := Config{
		Num:    len(cf.Configs),
		Shards: lastConfig.Shards,
		Groups: deepCopy(lastConfig.Groups),
	}

	for gid, servers := range groups {
		if _, ok := newConfig.Groups[gid]; !ok {
			newServers := make([]string, len(servers))
			copy(newServers, servers)
			newConfig.Groups[gid] = newServers
		}
	}
                                       
	// 找到group 中shard 最大和最小的组，将数据进行move => reblance
	g2s := Group2Shards(newConfig)
	for {
		src, dst := GetGIDWIthMaxShards(g2s), GetGIDWithMinShards(g2s)
		if src != 0 && len(g2s[src])-len(g2s[dst]) <= 1 {
			break
		}

		g2s[dst] = append(g2s[dst], g2s[src][0])
		g2s[src] = g2s[src][1:]
	}

	var newShards [NShards]int
	for gid, shards := range g2s {
		for _, shard := range shards {
			newShards[shard] = gid
		}
	}
	newConfig.Shards = newShards
	cf.Configs = append(cf.Configs, newConfig)
	return OK
}

// Leave some group leave the cluster
func (cf *MemoryConfigStateMachine) Leave(gids []int) Err {
	lastConfig := cf.Configs[len(cf.Configs)-1]
	newConfig := Config{
		Num:    len(cf.Configs),
		Shards: lastConfig.Shards,
		Groups: deepCopy(lastConfig.Groups),
	}

	g2s := Group2Shards(newConfig)
	orphanShards := make([]int, 0)

	for _, gid := range gids {
		delete(newConfig.Groups, gid)
		if shards, ok := g2s[gid]; ok {
			orphanShards = append(orphanShards, shards...)
			delete(g2s, gid)
		}
	}

	var newShards [NShards]int
	if len(newConfig.Groups) != 0 {
		// reblance
		for _, shard := range orphanShards {
			target := GetGIDWithMinShards(g2s)
			g2s[target] = append(g2s[target], shard)
		}

		// update Shards: share -> gid
		for gid, shards := range g2s {
			for _, shard := range shards {
				newShards[shard] = gid
			}
		}
	}

	newConfig.Shards = newShards
	cf.Configs = append(cf.Configs, newConfig)
	return OK
}

// Move move No.shard to No.gid
func (cf *MemoryConfigStateMachine) Move(shard, gid int) Err {
	lastConfig := cf.Configs[len(cf.Configs)-1]
	newConfig := Config{
		Num:    len(cf.Configs),
		Shards: lastConfig.Shards,
		Groups: lastConfig.Groups,
	}

	newConfig.Shards[shard] = gid
	cf.Configs = append(cf.Configs, newConfig)
	return OK
}

// Query return the version of num config
func (cf *MemoryConfigStateMachine) Query(num int) (Config, Err) {
	if num < 0 || num >= len(cf.Configs) {
		return cf.Configs[len(cf.Configs)-1], OK
	}
	return cf.Configs[num], OK
}

1
2
3
4
5
6
7
8
9

2023/03/03 19:58:49 [StartServer]-{Node: 0}-{Group: 100} has started
2023/03/03 19:58:49 [StartServer]-{Node: 1}-{Group: 100} has started
2023/03/03 19:58:49 [StartServer]-{Node: 2}-{Group: 100} has started
2023/03/03 19:58:49 [StartServer]-{Node: 0}-{Group: 101} has started
2023/03/03 19:58:49 [StartServer]-{Node: 1}-{Group: 101} has started
2023/03/03 19:58:49 [StartServer]-{Node: 2}-{Group: 101} has started
2023/03/03 19:58:49 [StartServer]-{Node: 0}-{Group: 102} has started
2023/03/03 19:58:49 [StartServer]-{Node: 1}-{Group: 102} has started
2023/03/03 19:58:49 [StartServer]-{Node: 2}-{Group: 102} has started

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

type Clerk struct {
	sc      *shardctrler.Clerk
	config  shardctrler.Config
	makeEnd func(string) *labrpc.ClientEnd

	// You will have to modify this struct.
	leaderIds map[int]int // {groupid: leader if hardid of this groupid}
	clientId  int64
	commandId int64 //clientId + commandId define unique operation
}

// 省略一些方法

func (ck *Clerk) Command(req *CommandRequest) string {
	req.ClientId, req.CommandId = ck.clientId, ck.commandId

	for {
		shard := key2shard(req.Key)
		gid := ck.config.Shards[shard]

		if servers, ok := ck.config.Groups[gid]; ok {
			// 找到Group 对应的LeaderId, 如果没有从Id 0 开始轮询
			if _, ok := ck.leaderIds[gid]; !ok {
				ck.leaderIds[gid] = 0
			}

			oldLeaderId := ck.leaderIds[gid]
			newLeaderId := oldLeaderId

			for {
				var resp CommandResponse
				ok := ck.makeEnd(servers[newLeaderId]).Call("ShardKV.Command", req, &resp)

				if ok && (resp.Err == OK || resp.Err == ErrNoKey) {
					ck.commandId++
					return resp.Value
				} else if ok && resp.Err == ErrWrongGroup {
					break
				} else {
					// Err is 	ErrWrongLeader ErrOutDated ErrTimeout ErrNotReady
					newLeaderId = (newLeaderId + 1) % len(servers)
					if newLeaderId == oldLeaderId {
						// 所有server 轮询一遍之后退出，避免raft 集群处于无leader 状态中一直重试
						break
					}
					continue
				}
			}
		}

		time.Sleep(100 * time.Millisecond)

		ck.config = ck.sc.Query(-1)

	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

type Shard struct {
	KV     map[string]string
	Status ShardStatus
}

type ShardKV struct {
	mu   sync.RWMutex
	me   int
	rf   *raft.Raft
	dead int32

	applyCh chan raft.ApplyMsg

	makeEnd func(string) *labrpc.ClientEnd
	gid     int
	sc      *shardctrler.Clerk

	maxRaftState int // snapshot if log grows this big
	lastApplied  int // 记录applied Index 防止状态机apply 小的index

	lastConfig    shardctrler.Config
	currentConfig shardctrler.Config

	stateMachine   map[int]*Shard                // {shardId: shard of KV}
	lastOperations map[int64]OperationContext    // {clientId: ctx}
	notifyChans    map[int]chan *CommandResponse // {commitIndex: commandResp}
}

func StartServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister,
	maxraftstate int, gid int, ctrlers []*labrpc.ClientEnd,
	make_end func(string) *labrpc.ClientEnd) *ShardKV {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Command{})
	labgob.Register(CommandRequest{})
	labgob.Register(shardctrler.Config{})
	labgob.Register(ShardOperationRequest{})
	labgob.Register(ShardOperationResponse{})

	applyCh := make(chan raft.ApplyMsg)

	kv := &ShardKV{
		me:             me,
		rf:             raft.Make(servers, me, persister, applyCh),
		dead:           0,
		applyCh:        applyCh,
		makeEnd:        make_end,
		gid:            gid,
		sc:             shardctrler.MakeClerk(ctrlers),
		maxRaftState:   maxraftstate,
		lastApplied:    0,
		lastConfig:     shardctrler.DefaultConfig(),
		currentConfig:  shardctrler.DefaultConfig(),
		stateMachine:   make(map[int]*Shard),
		lastOperations: make(map[int64]OperationContext),
		notifyChans:    make(map[int]chan *CommandResponse),
	}

	kv.restoreSnapshot(persister.ReadSnapshot())
	// start applier goroutine to apply committed logs to stateMachine
	go kv.applier()

	// start configuration monitor goroutine to fetch latest configuration
	go kv.Monitor(kv.configureAction, ConfigureMonitorTimeout)
	// start migration monitor goroutine to pull related shards
	go kv.Monitor(kv.migrationAction, MigrationMonitorTimeout)
	// start gc monitor goroutine to delete useless shards in remote groups
	go kv.Monitor(kv.gcAction, GCMonitorTimeout)

	// start entry-in-currentTerm monitor goroutine to advance commitIndex by
	// appending empty entries in current term periodically to avoid live locks
	go kv.Monitor(kv.checkEntryIncurrentTermAction, EmptyEntryDetectorTimeout)

	DPrintf("[StartServer]-{Node: %v}-{Group: %v} has started", kv.me, kv.gid)
	return kv
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

type Command struct {
	Op   CommandType
	Data interface{}
}

func (cmd Command) String() string {
	return fmt.Sprintf("{Op: %v, Data: %v}", cmd.Op, cmd.Data)
}

func NewOperationCommand(req *CommandRequest) Command {
	return Command{
		Op:   Operation,
		Data: *req,
	}
}

func NewConfigurationCommand(config *shardctrler.Config) Command {
	return Command{
		Op:   Configuration,
		Data: *config,
	}
}

func NewInsertShardsCommand(response *ShardOperationResponse) Command {
	return Command{InsertShards, *response}
}

func NewDeleteShardsCommand(request *ShardOperationRequest) Command {
	return Command{DeleteShards, *request}
}

func NewEmptyEntryCommand() Command {
	return Command{EmptyEntry, nil}
}

// -------------------------------------------------------------

type CommandType uint8

const (
	Operation CommandType = iota
	Configuration
	InsertShards
	DeleteShards
	EmptyEntry
)

var ctmap = [...]string{
	"Operation", "Configuration", "InsertShards", "DeleteShards", "EmptyEntry",
}

func (ct CommandType) String() string {
	return ctmap[ct]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

// canServe 判断shard 的状态是否可以对外服务
// Serving 默认初始状态, GCing 表示该shard 的数据刚刚拉取完毕，但是需要清除
// 远端 该shardId 数据
func (kv *ShardKV) canServe(ShardId int) bool {
	return kv.currentConfig.Shards[ShardId] == kv.gid &&
		(kv.stateMachine[ShardId].Status == Serving ||
			kv.stateMachine[ShardId].Status == GCing)
}


func (kv *ShardKV) Command(req *CommandRequest, resp *CommandResponse) {
	kv.mu.RLock()

	if req.Op != OpGet && kv.isDuplicateRequest(req.ClientId, req.CommandId) {
		lastResp := kv.lastOperations[req.ClientId].LastResponse
		resp.Err = lastResp.Err
		resp.Value = lastResp.Value
		kv.mu.RUnlock()
		return
	}

	// return ErrWrongGroup directly to let client fetch latest configuration
	// and perform a retry if this key can't be served by this shard at present
	if !kv.canServe(key2shard(req.Key)) {
		resp.Err = ErrWrongGroup
		resp.Value = ""
		kv.mu.RUnlock()
		return
	}

	kv.mu.RUnlock()
	kv.Execute(NewOperationCommand(req), resp)
}

// Execute shardKV 执行相关的RPC req
func (kv *ShardKV) Execute(command Command, resp *CommandResponse) {
	// do not hold lock to improve throughput
	// when KVServer holds the lock to take snapshot, underlying raft can still commit raft logs
	index, _, isLeader := kv.rf.Start(command)
	if !isLeader {
		resp.Err = ErrWrongLeader
		return
	}

	defer DPrintf("[Execute]-{Node: %v}-{Group: %v} process Command %v with CommandResponse %v",
		kv.me, kv.gid, command, resp)

	kv.mu.Lock()
	ch := kv.getNotifyChan(index)
	kv.mu.Unlock()

	select {
	case res := <-ch:
		resp.Value, resp.Err = res.Value, res.Err
	case <-time.After(ExecuteTimeout):
		resp.Err = ErrTimeout
	}

	go func() {
		kv.mu.Lock()
		kv.deleteOutdatedNotifyChan(index)
		kv.mu.Unlock()
	}()
}

// applyOperation 对状态机的操作, Get, Put, Append
func (kv *ShardKV) applyOperation(msg *raft.ApplyMsg, req *CommandRequest) *CommandResponse {
	var resp *CommandResponse
	shardId := key2shard(req.Key)

	if kv.canServe(shardId) {
		if req.Op != OpGet && kv.isDuplicateRequest(req.ClientId, req.CommandId) {
			DPrintf("[applyOperation]-{Node: %v}-{Group: %v} doesn't apply duplicated message %v to stateMachine because maxAppliedCommandId is %v for client %v",
				kv.me, kv.gid, kv.lastOperations[req.ClientId], kv.lastApplied, req.ClientId)

			lastResp := kv.lastOperations[req.ClientId].LastResponse
			return lastResp
		}
		resp = kv.applyLogToStateMachines(req, shardId)
		if req.Op != OpGet {
			// save max command resp
			kv.lastOperations[req.ClientId] = OperationContext{
				MaxAppliedCommandId: req.CommandId,
				LastResponse:        resp,
			}
		}
		return resp
	}
	return &CommandResponse{ErrWrongGroup, ""}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// configureAction kvctrller execute apply configuration
func (kv *ShardKV) configureAction() {
	canPerformNextConfig := true

	kv.mu.RLock()
	for _, shard := range kv.stateMachine {
		if shard.Status != Serving {
			canPerformNextConfig = false
			DPrintf("[configureAction]-{Node: %v}-{Group: %v} will not try to fetch latest configuration because shards status are %v when currentConfig is %v",
				kv.me, kv.gid, kv.getShardStatus(), kv.currentConfig)
			break
		}
	}

	currentConfigNum := kv.currentConfig.Num
	kv.mu.RUnlock()

	if canPerformNextConfig {
		nextConfig := kv.sc.Query(currentConfigNum + 1)
		if nextConfig.Num == currentConfigNum+1 {
			DPrintf("[configureAction]-{Node: %v}-{Group: %v} fetches latest configuration %v when currentConfigNum is %v",
				kv.me, kv.gid, nextConfig, currentConfigNum)

			kv.Execute(NewConfigurationCommand(&nextConfig), &CommandResponse{})
		}
	}
}

// applyConfiguration 对kv controller 的配置进行更新
func (kv *ShardKV) applyConfiguration(conf *shardctrler.Config) *CommandResponse {
	if conf.Num == kv.currentConfig.Num+1 {
		DPrintf("[applyConfiguration]-{Node: %v}-{Group: %v} updates currentConfig from %v to %v",
			kv.me, kv.gid, kv.currentConfig, conf)

		kv.updateShardStatus(conf)
		kv.lastConfig = kv.currentConfig
		kv.currentConfig = *conf
		return &CommandResponse{OK, ""}
	}

	DPrintf("[applyConfiguration]-{Node: %v}-{Group: %v} rejects outdated config %v when currentConfig is %v",
		kv.me, kv.gid, conf, kv.currentConfig)

	return &CommandResponse{ErrOutDated, ""}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// migrationAction shard migration data when in Pulling status
func (kv *ShardKV) migrationAction() {
	kv.mu.RLock()
	gid2shardIds := kv.getShardIdsByStatus(Pulling)

	var wg sync.WaitGroup
	for gid, shardIds := range gid2shardIds {
		DPrintf("[migrationAction]-{Node: %v}-{Group: %v} starts a PullTask to get shards %v from group %v when config is %v",
			kv.me, kv.gid, shardIds, gid, kv.currentConfig)

		wg.Add(1)
		go func(servers []string, configNum int, shardIds []int) {
			defer wg.Done()
			PullTaskRequest := ShardOperationRequest{
				ConfigNum: configNum,
				ShardIDs:  shardIds,
			}

			// 本Node 向Pulling Status 的Shard 所在Group 的全部Server
			// 发送RPC 但是只有Leader 有响应，其他忽略
			for _, server := range servers {
				var pullTaskResp ShardOperationResponse
				srv := kv.makeEnd(server)
				DPrintf("[migrationAction]-{Node: %v}-{Group: %v} server call %v", kv.me, kv.gid, server)
				if srv.Call("ShardKV.GetShardsData", &PullTaskRequest, &pullTaskResp) && pullTaskResp.Err == OK {
					DPrintf("[migrationAction]-{Node: %v}-{Group: %v} gets a PullTaskResponse %v and tries to commit it when currentConfigNum is %v",
						kv.me, kv.gid, pullTaskResp, configNum)
					kv.Execute(NewInsertShardsCommand(&pullTaskResp), &CommandResponse{})
				}
			}
		}(kv.lastConfig.Groups[gid], kv.currentConfig.Num, shardIds)
	}
	kv.mu.RUnlock()
	wg.Wait()
}


// RPC GetShardsData ConfigOperation 生效之后数据迁移， 将request 中shardId的数据，迁移到resp中 返回给调用方
func (kv *ShardKV) GetShardsData(req *ShardOperationRequest, resp *ShardOperationResponse) {
	// ...
}

func (kv *ShardKV) applyInsertShards(shardsInfo *ShardOperationResponse) *CommandResponse {
	// ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

// gcAction Pulling 数据之后把状态改为GCing，调用RPC 删除远端的该ShardId 的数据
func (kv *ShardKV) gcAction() {
	kv.mu.RLock()
	gid2shardIds := kv.getShardIdsByStatus(GCing)
	var wg sync.WaitGroup
	for gid, shardIds := range gid2shardIds {
		DPrintf("[gcAction]-{Node: %v}-{Group: %v} starts a GCTask to delete shards %v in group %v when config is %v",
			kv.me, kv.gid, shardIds, gid, kv.currentConfig)

		wg.Add(1)
		go func(servers []string, configNum int, shardIds []int) {
			defer wg.Done()
			gcTaskReq := ShardOperationRequest{ConfigNum: configNum, ShardIDs: shardIds}
			for _, server := range servers {
				var gcTaskResp ShardOperationResponse
				srv := kv.makeEnd(server)
				// 远端执行删除数据的逻辑
				if srv.Call("ShardKV.DeleteShardsData", &gcTaskReq, &gcTaskResp) && gcTaskResp.Err == OK {
					DPrintf("[gcAction]-{Node: %v}-{Group: %v} deletes shards %v in remote group successfully when currentConfigNum is %v",
						kv.me, kv.gid, shardIds, configNum)

					// 远端删除完数据之后，Raft 本节点同样需要 把GCing 状态恢复为Server 状态
					kv.Execute(NewDeleteShardsCommand(&gcTaskReq), &CommandResponse{})
				}
			}
		}(kv.lastConfig.Groups[gid], kv.currentConfig.Num, shardIds)
	}

	kv.mu.RUnlock()
	wg.Wait()
}

// RPC DeleteShardsData 删除迁移之后的 shard 中的数据
func (kv *ShardKV) DeleteShardsData(req *ShardOperationRequest, resp *ShardOperationResponse) {
	if _, isLeader := kv.rf.GetState(); !isLeader {
		resp.Err = ErrWrongLeader
		return
	}

	defer DPrintf("[DeleteShardsData]-{Node: %v}-{Group: %v} processes GCTaskRequest %v with response %v",
		kv.me, kv.gid, req, resp)

	kv.mu.RLock()
	if kv.currentConfig.Num > req.ConfigNum {
		DPrintf("[DeleteShardsData]-{Node: %v}-{Group: %v} encounters duplicated shards deletions %v when currentConfig is %v",
			kv.me, kv.gid, req, kv.currentConfig)
		resp.Err = OK
		kv.mu.RUnlock()
		return
	}
	kv.mu.RUnlock()

	var commandResp CommandResponse
	kv.Execute(NewDeleteShardsCommand(req), &commandResp)
	resp.Err = commandResp.Err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

func (kv *ShardKV) checkEntryIncurrentTermAction() {
	if !kv.rf.HasLogInCurrentTerm() {
		kv.Execute(NewEmptyEntryCommand(), &CommandResponse{})
	}
}

func (kv *ShardKV) applyEmptyEntry() *CommandResponse {
	return &CommandResponse{OK, ""}
}