-
Notifications
You must be signed in to change notification settings - Fork 0
/
client.go
161 lines (143 loc) · 4 KB
/
client.go
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
package kvraft
import (
"crypto/rand"
"math/big"
"time"
"../labrpc"
"../labutil"
)
const (
TryNextLeaderInterval = time.Millisecond * 20
)
type Clerk struct {
servers []*labrpc.ClientEnd
// You will have to modify this struct.
leaderId int
clientId TypeClientId
nextMsgId ClerkMsgId
}
func nrand() int64 {
max := big.NewInt(int64(1) << 62)
bigx, _ := rand.Int(rand.Reader, max)
x := bigx.Int64()
return x
}
func MakeClerk(servers []*labrpc.ClientEnd) *Clerk {
ck := new(Clerk)
ck.servers = servers
// You'll have to add code here.
if len(servers) == 0 {
labutil.PrintException("MakeClerk: No servers")
labutil.PanicSystem()
return nil
}
ck.leaderId = 0
ck.clientId = TypeClientId(nrand()) //assume no duplicate
ck.nextMsgId = 0
return ck
}
func (ck *Clerk) getNextMsgId() ClerkMsgId {
return ck.nextMsgId + 1
}
//
// fetch the current value for a key.
// returns "" if the key does not exist.
// keeps trying forever in the face of all other errors.
//
// you can send an RPC with code like this:
// ok := ck.servers[i].Call("KVServer.Get", &args, &reply)
//
// the types of args and reply (including whether they are pointers)
// must match the declared types of the RPC handler function's
// arguments. and reply must be passed as a pointer.
//
func (ck *Clerk) Get(key string) string {
// You will have to modify this function.
ck.nextMsgId = ck.getNextMsgId()
args := GetArgs{key, ck.clientId, ck.nextMsgId}
//args.msgId is the same for multiple RPC retries
//labutil.PrintMessage("Clerk[" + fmt.Sprint(ck.clientId) + "] Get" + " <" + key + ">" + ", msgId = " + fmt.Sprint(ck.nextMsgId))
for {
reply := GetReply{}
//labutil.PrintMessage("Try Leader = Server[" + fmt.Sprint(ck.leaderId) + "]")
ok := ck.servers[ck.leaderId].Call("KVServer.Get", &args, &reply)
if !ok {
//try next server to be leader
time.Sleep(TryNextLeaderInterval)
ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
continue
}
switch reply.Err {
case OK:
return reply.Value
case ErrNoKey:
return ""
case ErrWrongLeader:
//try next server to be leader
time.Sleep(TryNextLeaderInterval)
ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
continue
case ErrTimeout:
continue
default:
labutil.PrintException("Get: Unknown GetReply.Err")
labutil.PanicSystem()
return ""
}
}
return ""
}
//
// shared by Put and Append.
//
// you can send an RPC with code like this:
// ok := ck.servers[i].Call("KVServer.PutAppend", &args, &reply)
//
// the types of args and reply (including whether they are pointers)
// must match the declared types of the RPC handler function's
// arguments. and reply must be passed as a pointer.
//
func (ck *Clerk) PutAppend(key string, value string, op string) {
// You will have to modify this function.
ck.nextMsgId = ck.getNextMsgId()
args := PutAppendArgs{key, value, op, ck.clientId, ck.nextMsgId}
//args.msgId is the same for multiple RPC retries
//labutil.PrintMessage("Clerk[" + fmt.Sprint(ck.clientId) + "] " + op + " <" + key + ", " + value + ">" + ", msgId = " + fmt.Sprint(ck.nextMsgId))
for {
reply := PutAppendReply{}
ok := ck.servers[ck.leaderId].Call("KVServer.PutAppend", &args, &reply)
if !ok {
//try next server to be leader
time.Sleep(TryNextLeaderInterval)
ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
continue
}
switch reply.Err {
case OK:
return
case ErrNoKey:
//PutAppend should not return ErrNoKey in reply
labutil.PrintException("PutAppend return ErrNoKey in reply")
labutil.PanicSystem()
return
case ErrWrongLeader:
//try next server to be leader
time.Sleep(TryNextLeaderInterval)
ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
continue
case ErrTimeout:
continue
default:
labutil.PrintException("Get: Unknown GetReply.Err")
labutil.PanicSystem()
return
}
}
return //should never reach here
}
func (ck *Clerk) Put(key string, value string) {
ck.PutAppend(key, value, "Put")
}
func (ck *Clerk) Append(key string, value string) {
ck.PutAppend(key, value, "Append")
}