lab3-kvraft

src/kvraft/client.go

@@ -2,12 +2,16 @@ package kvraft
 
 import "6.5840/labrpc"
 import "crypto/rand"
+import rand2 "math/rand"
 import "math/big"
-
+import "time"
+import "sync/atomic"
+import "log"
 
 type Clerk struct {
 	servers []*labrpc.ClientEnd
-	// You will have to modify this struct.
+	cid     int32
+	leader  int32
 }
 
 func nrand() int64 {
@@ -20,10 +24,17 @@ func nrand() int64 {
 func MakeClerk(servers []*labrpc.ClientEnd) *Clerk {
 	ck := new(Clerk)
 	ck.servers = servers
-	// You'll have to add code here.
+	ck.cid = rand2.Int31()
 	return ck
 }
 
+func (ck *Clerk) args() Args {
+	return Args{ClientId: ck.cid, RequestId: nrand()}
+}
+
+const RetryInterval = 300 * time.Millisecond
+
+//
 // fetch the current value for a key.
 // returns "" if the key does not exist.
 // keeps trying forever in the face of all other errors.
@@ -34,12 +45,27 @@ func MakeClerk(servers []*labrpc.ClientEnd) *Clerk {
 // 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.
-	return ""
+	args := GetArgs{Key: key}
+	leader := atomic.LoadInt32(&ck.leader)
+	for {
+		var reply GetReply
+		log.Printf("CID%d Client call Get leader=%d", ck.cid, leader)
+		ok := ck.servers[leader].Call("KVServer.Get", &args, &reply)
+		if ok {
+			if reply.Err == OK {
+				return reply.Value
+			} else if reply.Err == ErrTimeout {
+				continue
+			}
+		}
+		leader = ck.nextLeader(leader)
+		time.Sleep(RetryInterval)
+	}
 }
 
+//
 // shared by Put and Append.
 //
 // you can send an RPC with code like this:
@@ -48,13 +74,40 @@ func (ck *Clerk) Get(key string) string {
 // 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.
+	args := PutAppendArgs{Key: key, Value: value, Args: ck.args()}
+	if op == "Put" {
+		args.Type = PutOp
+	} else {
+		args.Type = AppendOp
+	}
+	leader := atomic.LoadInt32(&ck.leader)
+	for {
+		var reply PutAppendReply
+		log.Printf("CID%d Client call PutAppend leader=%d", ck.cid, leader)
+		ok := ck.servers[leader].Call("KVServer.PutAppend", &args, &reply)
+		if ok {
+			if reply.Err == OK {
+				return
+			} else if reply.Err == ErrTimeout {
+				continue
+			}
+		}
+		leader = ck.nextLeader(leader)
+		time.Sleep(RetryInterval)
+	}
+}
+
+func (ck *Clerk) nextLeader(current int32) int32 {
+	next := (current + 1) % int32(len(ck.servers))
+	atomic.StoreInt32(&ck.leader, next)
+	return next
 }
 
 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")
-}
+}

src/kvraft/common.go

@@ -4,18 +4,29 @@ const (
 	OK             = "OK"
 	ErrNoKey       = "ErrNoKey"
 	ErrWrongLeader = "ErrWrongLeader"
+	ErrTimeout     = "ErrTimeout"
 )
 
 type Err string
 
+type Args struct {
+	ClientId  int32
+	RequestId int64
+}
+
+type PutOrAppend int
+
+const (
+	PutOp PutOrAppend = iota
+	AppendOp
+)
+
 // Put or Append
 type PutAppendArgs struct {
+	Type  PutOrAppend
 	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.
+	Args
 }
 
 type PutAppendReply struct {
@@ -24,10 +35,9 @@ type PutAppendReply struct {
 
 type GetArgs struct {
 	Key string
-	// You'll have to add definitions here.
 }
 
 type GetReply struct {
 	Err   Err
 	Value string
-}
+}