1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
type PutAppendArgs struct {
	Key   string
	Value string
	Op    string // "Put" or "Append"
	// You'll have to add definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.
	SequenceId int64
	ClientId   int64
}

type GetArgs struct {
	Key string
	// You'll have to add definitions here.
	SequenceId int64
	ClientId   int64
}

1
2
3
4
5
6
7
8
9
type Clerk struct {
	servers []*labrpc.ClientEnd
	// You will have to modify this struct.
	mu         sync.Mutex
	leaderId   int
	clientId   int64 
	sequenceId int64
}

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
func (ck *Clerk) Get(key string) string {
	// You will have to modify this function.
	args := GetArgs{Key: key, ClientId: ck.clientId, SequenceId: atomic.AddInt64(&ck.sequenceId, 1)}
	leaderId := ck.currentLeader()
	for {
		reply := GetReply{}
		if ck.servers[leaderId].Call("KVServer.Get", &args, &reply) {
			if reply.Err == OK {
				return reply.Value
			} else if reply.Err == ErrNoKey {
				return ""
			}
		}
		leaderId = ck.changeLeader()
		time.Sleep(1 * time.Millisecond)
	}
}

func (ck *Clerk) PutAppend(key string, value string, op string) {
	args := PutAppendArgs{
		// You will have to modify this function.
		Key:        key,
		Value:      value,
		Op:         op,
		SequenceId: atomic.AddInt64(&ck.sequenceId, 1),
		ClientId:   ck.clientId,
	}
	leaderId := ck.currentLeader()
	for {
		reply := PutAppendReply{}
		if ck.servers[leaderId].Call("KVServer.PutAppend", &args, &reply) {
			if reply.Err == OK { 
				break
			}
		}
		leaderId = ck.changeLeader()
		time.Sleep(1 * time.Millisecond)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
type Op struct {
	// Your definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.
	OpId     int64
	ClientId int64
	Index    int
	Term     int
	OpType   string
	OpKey    string
	OpValue  string
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	maxraftstate int // snapshot if log grows this big

	// Your definitions here.
	sequenceMapper map[int64]int64
	requestMapper  map[int]chan Op
	kvStore        map[string]string
}

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
func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.

	var isLeader bool
	clientOp := Op{OpType: "Get", OpKey: args.Key, OpId: args.SequenceId, ClientId: args.ClientId}
	clientOp.Index, clientOp.Term, isLeader = kv.rf.Start(clientOp)

	if !isLeader {
		reply.Err = ErrWrongLeader
		return
	}

	ch := kv.getChannel(clientOp.Index)

	defer func() {
		kv.mu.Lock()
		delete(kv.requestMapper, clientOp.Index)
		kv.mu.Unlock()
	}()

	timer := time.NewTicker(500 * time.Millisecond)
	defer timer.Stop()
	select {
	case op := <-ch:
		kv.mu.Lock()
		opTerm := op.Term
		kv.mu.Unlock()
		if clientOp.Term != opTerm {
			reply.Err = ErrWrongLeader
			return
		} else {
			reply.Err = OK
			kv.mu.Lock()
			reply.Value = kv.kvStore[args.Key]
			kv.mu.Unlock()
			return
		}
	case <-timer.C:
		reply.Err = ErrWrongLeader
	}
}

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
func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	var isLeader bool

	clientOp := Op{OpType: args.Op, OpKey: args.Key, OpValue: args.Value, OpId: args.SequenceId, ClientId: args.ClientId}
	clientOp.Index, clientOp.Term, isLeader = kv.rf.Start(clientOp)

	if !isLeader {
		reply.Err = ErrWrongLeader
		return
	}

	// leader is found
	ch := kv.getChannel(clientOp.Index)

	defer func() {
		kv.mu.Lock()
		delete(kv.requestMapper, clientOp.Index)
		kv.mu.Unlock()
	}()

	timer := time.NewTicker(500 * time.Millisecond)
	defer timer.Stop()
	select {
	case op := <-ch:
		kv.mu.Lock()
		opTerm := op.Term
		kv.mu.Unlock()
		if clientOp.Term != opTerm {
			reply.Err = ErrWrongLeader
		} else {
			reply.Err = OK
		}
	case <-timer.C:
		reply.Err = ErrWrongLeader
	}
}

1
2
3
4
5
6
7
8
9
10
11
func (kv *KVServer) getChannel(index int) chan Op {
	kv.mu.Lock()
	defer kv.mu.Unlock()
	ch, ok := kv.requestMapper[index]
	if !ok {
		ch = make(chan Op, 1)
		kv.requestMapper[index] = ch
	}
	return ch
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {

	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)
	kv.sequenceMapper = make(map[int64]int64)
	kv.requestMapper = make(map[int]chan Op)
	kv.kvStore = make(map[string]string)

	go kv.serverMonitor()

	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
func (kv *KVServer) serverMonitor() {
	for {
		if kv.killed() {
			return
		}
		select {
		case msg := <-kv.applyCh:
			index := msg.CommandIndex
			term := msg.CommandTerm
			op := msg.Command.(Op)
			kv.mu.Lock()
			sequenceInMapper, hasSequence := kv.sequenceMapper[op.ClientId]
			op.Term = term
			if !hasSequence || op.OpId > sequenceInMapper {
				switch op.OpType {
				case "Put":
					kv.kvStore[op.OpKey] = op.OpValue
				case "Append":
					kv.kvStore[op.OpKey] += op.OpValue
				}
				kv.sequenceMapper[op.ClientId] = op.OpId
			}
			kv.mu.Unlock()
			// send message to op chan
			kv.getChannel(index) <- op
		}
	}
}