update

Author: yizhuoliang

Diff for: acceptor/main.go

@@ -113,6 +113,10 @@ func readPortsFile() {
 	}
 }
 
+func commandId2String(cid int64) string {
+	return strconv.FormatInt(cid>>54, 10) + strconv.FormatInt(cid%int64(1)<<54, 10)
+}
+
 // gRPC handlers
 func (s *acceptorServer) Scouting(ctx context.Context, in *pb.Message) (*pb.Message, error) {
 
@@ -178,13 +182,13 @@ func (s *acceptorServer) Commanding(ctx context.Context, in *pb.Message) (*pb.Me
 
 	<-mutexChannel
 	ballotNumber := ballotNumber // concurrency concern, avoid ballot number update during execution
-	fmt.Printf("Commanding received:  ballot number %d, ballot leader %d, comID: %s, slot: %d\n", ballotNumber, ballotLeader, in.Bsc.Command.CommandId, in.Bsc.SlotNumber)
-	acceptCid := ""
+	fmt.Printf("Commanding received:  ballot number %d, ballot leader %d, comID: %s, slot: %d\n", ballotNumber, ballotLeader, commandId2String(in.Bsc.Command.CommandId), in.Bsc.SlotNumber)
+	acceptCid := int64(-1)
 	if in.Bsc.BallotNumber >= ballotNumber && in.LeaderId == ballotLeader {
 		ballotNumber = in.Bsc.BallotNumber
 		acceptCid = in.Bsc.Command.CommandId
 		accepted[ballotNumber] = append(accepted[ballotNumber], in.Bsc)
-		fmt.Printf("Commanding accepted: ballot number %d, ballot leader %d, comID: %s, slot: %d\n", ballotNumber, ballotLeader, in.Bsc.Command.CommandId, in.Bsc.SlotNumber)
+		fmt.Printf("Commanding accepted: ballot number %d, ballot leader %d, comID: %s, slot: %d\n", ballotNumber, ballotLeader, commandId2String(in.Bsc.Command.CommandId), in.Bsc.SlotNumber)
 	}
 	currentBallotNumber := ballotNumber
 	currentBallotLeader := ballotLeader

Diff for: client/main.go

@@ -79,7 +79,7 @@ func main() {
 			fmt.Scanf("%s", &value)
 			// generate a new commandID
 			commandCount += 1
-			cid := "client" + strconv.Itoa(int(clientId)) + "-W" + strconv.Itoa(commandCount)
+			cid := int64(clientId)<<54 + int64(commandCount)
 			// push client commands to command buffers
 			for i := 0; i < replicaNum; i++ {
 				messageBuffers[i] <- &pb.Message{Type: WRITE, Command: &pb.Command{Type: WRITE, CommandId: cid, ClientId: clientId, Key: key, Value: value}, CommandId: cid, ClientId: clientId, Key: key, Value: value}
@@ -93,7 +93,7 @@ func main() {
 			fmt.Scanf("%s", &key)
 			// generate a new commandID
 			commandCount += 1
-			cid := "client" + strconv.Itoa(int(clientId)) + "-R" + strconv.Itoa(commandCount)
+			cid := int64(clientId)<<54 + int64(commandCount)
 			for i := 0; i < replicaNum; i++ {
 				messageBuffers[i] <- &pb.Message{Type: READ, Command: &pb.Command{Type: READ, CommandId: cid, ClientId: clientId, Key: key}}
 			}

Diff for: gopaxos.pb.go

@@ -27,7 +27,7 @@ message Message {
     int32 type = 1;
 
     int32 clientId = 2;
-    string commandId = 3;
+    int64 commandId = 3;
     string key = 4;
     string value = 5;
     Command command = 6;
@@ -50,7 +50,7 @@ message Message {
 message Command {
     int32 type = 1;
     int32 clientId = 2;
-    string commandId = 3;
+    int64 commandId = 3;
     string key = 4;
     string value = 5;
 }

Diff for: gopaxos.proto

@@ -356,7 +356,7 @@ func ScoutMessenger(serial int, scoutCollectChannel chan *pb.P1B, scoutBallotNum
 }
 
 func CommanderRoutine(bsc *pb.BSC) {
-	fmt.Printf("Commander spawned for ballot number %d, slot number %d, command id %s\n", bsc.BallotNumber, bsc.SlotNumber, bsc.Command.CommandId)
+	fmt.Printf("Commander spawned for ballot number %d, slot number %d, command id %s\n", bsc.BallotNumber, bsc.SlotNumber, commandId2String(bsc.Command.CommandId))
 	commanderCollectChannel := make(chan *pb.P2B)
 	// send messages
 	for i := 0; i < acceptorNum; i++ {
@@ -494,6 +494,10 @@ func readPortsFile() {
 	}
 }
 
+func commandId2String(cid int64) string {
+	return strconv.FormatInt(cid>>54, 10) + strconv.FormatInt(cid%int64(1)<<54, 10)
+}
+
 // gRPC HANDLERS
 func (s *leaderServer) Propose(ctx context.Context, in *pb.Message) (*pb.Message, error) {
 
@@ -510,7 +514,7 @@ func (s *leaderServer) Propose(ctx context.Context, in *pb.Message) (*pb.Message
 	}
 
 	leaderStateUpdateChannel <- &leaderStateUpdateRequest{updateType: 1, newProposal: &pb.Proposal{SlotNumber: in.SlotNumber, Command: in.Command}} // Weird? Yes! Things can only be down after chekcing proposals
-	fmt.Printf("Received proposal with commandId %s and slot number %d\n", in.Command.CommandId, in.SlotNumber)
+	fmt.Printf("Received proposal with commandId %s and slot number %d\n", commandId2String(in.Command.CommandId), in.SlotNumber)
 	return &pb.Message{Type: EMPTY, Content: "success"}, nil
 }
 

Diff for: leader/main.go

@@ -3,7 +3,7 @@ package client
 import (
 	"context"
 	"errors"
-	"strconv"
+	"sync/atomic"
 	"time"
 
 	pb "github.com/yizhuoliang/gopaxos"
@@ -32,7 +32,7 @@ const (
 
 type Client struct {
 	clientId     int32
-	commandCount int
+	commandCount uint32
 	replicaPorts []string
 	// commandChannels  map[int][]chan *pb.Message
 	// replyChannels    map[int][]chan *reply
@@ -76,17 +76,12 @@ func NewPaxosClient(clientId int, simon int, replicaPorts []string) *Client {
 	client.incrementCommandNumberChannel = make(chan int, 1)
 	client.commandNumberReplyChannel = make(chan int, 1)
 
-	go client.OperationPreperationAndCleanupRoutine()
-
 	return client
 }
 
 func (client *Client) Store(key string, value string) error {
 
-	client.incrementCommandNumberChannel <- 0
-	cNum := <-client.commandNumberReplyChannel
-
-	cid := "client" + strconv.Itoa(int(client.clientId)) + "-W" + strconv.Itoa(cNum)
+	cid := int64(atomic.AddUint32(&client.commandCount, 1))
 
 	replyChannels := make([]chan *reply, replicaNum)
 	for i := 0; i < replicaNum; i++ {
@@ -114,10 +109,7 @@ func (client *Client) Store(key string, value string) error {
 
 func (client *Client) Read(key string) (string, error) {
 
-	client.incrementCommandNumberChannel <- 0
-	cNum := <-client.commandNumberReplyChannel
-
-	cid := "client" + strconv.Itoa(int(client.clientId)) + "-R" + strconv.Itoa(cNum)
+	cid := int64(atomic.AddUint32(&client.commandCount, 1))
 
 	replyChannels := make([]chan *reply, replicaNum)
 	for i := 0; i < replicaNum; i++ {
@@ -146,14 +138,6 @@ func (client *Client) Read(key string) (string, error) {
 	return value, nil
 }
 
-func (client *Client) OperationPreperationAndCleanupRoutine() {
-	for {
-		<-client.incrementCommandNumberChannel
-		client.commandCount += 1
-		client.commandNumberReplyChannel <- client.commandCount
-	}
-}
-
 func (client *Client) TempMessengerRoutine(msg *pb.Message, replyChannel chan *reply, replicaSerial int) {
 	// reset connection for each message
 	// conn, err := grpc.Dial(client.replicaPorts[replicaSerial], grpc.WithTransportCredentials(insecure.NewCredentials()))

Diff for: paxosClient/client.go

@@ -67,7 +67,7 @@ var (
 	keyValueLog map[string]string
 
 	// this is for replying
-	clientReplyMap map[string]chan string
+	clientReplyMap map[int64]chan string
 	replyMapLock   sync.RWMutex
 
 	// simc *comm.RPCConnection
@@ -106,7 +106,7 @@ func main() {
 	proposals = make(map[int32]*pb.Proposal)
 	decisions = make(map[int32]*pb.Decision)
 	keyValueLog = make(map[string]string)
-	clientReplyMap = make(map[string]chan string)
+	clientReplyMap = make(map[int64]chan string)
 	replicaStateUpdateChannel = make(chan *replicaStateUpdateRequest, 1)
 	newCommandsChannel = make(chan *pb.Command, 1)
 	for i := 0; i < leaderNum; i++ {
@@ -259,9 +259,13 @@ func readPortsFile() {
 	}
 }
 
+func commandId2String(cid int64) string {
+	return strconv.FormatInt(cid>>54, 10) + strconv.FormatInt(cid%int64(1)<<54, 10)
+}
+
 // gRPC handlers for Clients
 func (s *replicaServer) Write(ctx context.Context, in *pb.Message) (*pb.Message, error) {
-	fmt.Printf("Request with command id %s received", in.CommandId)
+	fmt.Printf("Request with command id %s received", commandId2String(in.CommandId))
 	myChan := make(chan string, 1)
 	replyMapLock.Lock()
 	clientReplyMap[in.Command.CommandId] = myChan
@@ -292,7 +296,7 @@ func (s *replicaServer) Read(ctx context.Context, in *pb.Message) (*pb.Message,
 
 // gRPC handlers for Leader
 func (s *replicaServer) Decide(ctx context.Context, in *pb.Message) (*pb.Message, error) {
-	fmt.Printf("Decision with command id %s received", in.Decision.Command.CommandId)
+	fmt.Printf("Decision with command id %s received", commandId2String(in.Decision.Command.CommandId))
 	replicaStateUpdateChannel <- &replicaStateUpdateRequest{updateType: 1, newDecision: in.Decision}
 	return &pb.Message{Type: EMPTY}, nil
 }

Diff for: replica/main.go

@@ -1,11 +1,5 @@
 package main
 
-import (
-	"fmt"
-
-	pb "github.com/yizhuoliang/gopaxos"
-)
-
 type struc struct {
 	str string
 }
@@ -30,14 +24,15 @@ func main() {
 	// 	}
 	// }
 
-	p1 := &pb.Command{CommandId: "client0-1", Key: "ding", Value: "ding"}
-	p2 := &pb.Command{CommandId: "client0-2", Key: "zhang", Value: "han"}
-	// fmt.Printf("%t\n", reflect.DeepEqual(p1, p2))
-
-	sli := make([]*pb.Command, 1)
-	sli[0] = p1
-	sli2 := make([]*pb.Command, 2)
-	sli = append(sli, sli2...)
-	sli[2] = p2
-	fmt.Printf("%v\n", sli)
+	// p1 := &pb.Command{CommandId: "client0-1", Key: "ding", Value: "ding"}
+	// p2 := &pb.Command{CommandId: "client0-2", Key: "zhang", Value: "han"}
+	// // fmt.Printf("%t\n", reflect.DeepEqual(p1, p2))
+
+	// sli := make([]*pb.Command, 1)
+	// sli[0] = p1
+	// sli2 := make([]*pb.Command, 2)
+	// fmt.Printf("%d\n", len(sli2))
+	// sli = append(sli, sli2...)
+	// sli[2] = p2
+	// fmt.Printf("%v\n", sli)
 }