partially completed integration

assignment4/src/clean.sh

+#! /bin/bash
+rm {1..5} currentTerm* votedFor* log*

assignment4/src/connhandler/connhandler.go

+package connhandler
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/gob"
+	"log"
+	"net"
+	"raft"
+	"strconv"
+	"strings"
+	"time"
+	"utils"
+)
+
+/*
+ *Helper function to read value or cause timeout after READ_TIMEOUT seconds
+ *parameters: channel to read data from, threshold number of bytes to read, log pointer to write into
+ *returns: the value string and error state
+ */
+func readValue(ch chan []byte, n uint64, logger *log.Logger) ([]byte, bool) {
+	//now we need to read the value which should have been sent
+	valReadLength := uint64(0)
+	var v []byte
+	err := false
+	up := make(chan bool, 1)
+	//after 5 seconds passed reading value, we'll just send err to client
+	go func() {
+		time.Sleep(5 * time.Second)
+		up <- true
+	}()
+
+	//use select for the data channel and the timeout channel
+	for valReadLength < n+2 {
+		select {
+		case temp := <-ch:
+			valReadLength += uint64(len(temp))
+			if valReadLength > n+2 {
+				err = true
+				break
+			}
+			v = append(v, temp...)
+
+		case <-up:
+			err = true
+			break
+		}
+
+		//will be true if timeout occurs
+		if err {
+			logger.Println("Timeout")
+			break
+		}
+	}
+
+	if err {
+		return []byte{0}, err
+	}
+	return v[:n], err
+}
+
+/*Copied from the bufio.Scanner (originally ScanLines).
+ *By default it splits by '\n' but now we want it to split by '\r\n'
+ *arguments: data in bytes, is eof reached
+ *return: next sequence of bytes, chunk of data found, err state
+ */
+func CustomSplitter(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	omega := 0
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+	for {
+		if i := bytes.IndexByte(data[omega:], '\n'); i >= 0 {
+			//here we add omega as we are using the complete data array instead of the slice where we found '\n'
+			if i > 0 && data[omega+i-1] == '\r' {
+				//next byte begins at i+1 and data[0:i+1] returned
+				return omega + i + 1, data[:omega+i+1], nil
+			} else {
+				//move the omega index to the byte after \n
+				omega += i + 1
+			}
+		} else {
+			//need to break free the chains
+			break
+		}
+	}
+	// If we're at EOF, we have a final, non-terminated line. Return it.
+	if atEOF {
+		return len(data), data, nil
+	}
+	// Request more data.
+	return 0, nil, nil
+}
+
+/*Function to read data from the connection and put it on the channel so it could be read in a systematic fashion.
+ *arguments: channel shared between this go routine and other functions performing actions based on the commands given,
+ *client connection
+ *return: none
+ */
+func MyRead(ch chan []byte, conn net.Conn) {
+	scanner := bufio.NewScanner(conn)
+	scanner.Split(CustomSplitter)
+	for {
+		if ok := scanner.Scan(); !ok {
+			break
+		} else {
+			temp := scanner.Bytes()
+			ch <- temp
+		}
+	}
+}
+
+/*Simple write function to send information to the client
+ *arguments: client connection, msg to send to the client
+ *return: none
+ */
+func Write(conn net.Conn, msg string) {
+	buf := []byte(msg)
+	buf = append(buf, []byte("\r\n")...)
+	conn.Write(buf)
+}
+
+/*Will be invoked as go routine by server to every client connection. Will take care of all communication with the
+ *client and the raft/kvstore
+ *arguments: connection to client, pointer to raft, pointer to logger
+ *return: none
+ */
+func HandleClient(conn net.Conn, rft *raft.Raft, logger *log.Logger) {
+	defer conn.Close()
+	//channel for every connection for every client
+	ch := make(chan []byte)
+	go MyRead(ch, conn)
+
+	for {
+		command := new(utils.Command)
+		msg := <-ch
+		logger.Println("got:", msg, string(msg))
+
+		if len(msg) == 0 {
+			continue
+		}
+		command.Cmd = msg
+		flag := false
+		nr := uint64(0)
+		tokens := strings.Fields(string(msg))
+		if tokens[0] == "cas" {
+			n, _ := strconv.ParseUint(tokens[4], 10, 64)
+			nr = n
+			flag = true
+		} else if tokens[0] == "set" {
+			n, _ := strconv.ParseUint(tokens[3], 10, 64)
+			nr = n
+			flag = true
+		}
+		if flag {
+			logger.Println("numbytes", nr)
+			if v, err := readValue(ch, nr, logger); err {
+				logger.Println("error reading value")
+				Write(conn, "ERR_CMD_ERR")
+				continue
+			} else {
+				command.Val = v
+				//command.isVal = true
+			}
+		}
+
+		buffer := new(bytes.Buffer)
+		// writing
+		enc := gob.NewEncoder(buffer)
+		err := enc.Encode(command)
+		if err != nil {
+			//log.Fatal("encode error:", err)
+		}
+
+		if _, err := rft.Append(buffer.Bytes(), conn); err != nil {
+			Write(conn, "ERR_REDIRECT 127.0.0.1 "+strconv.Itoa(raft.CLIENT_PORT+1))
+			conn.Close()
+			break
+		}
+	}
+}

assignment4/src/main.go

+package main
+
+import (
+	"fmt"
+	"raft"
+)
+
+const (
+	SERVERS = 5
+)
+
+func main() {
+	dummyCh := make(chan bool, 1)
+	fmt.Println("Started")
+	for i := 1; i <= 5; i++ {
+		go raft.Start(i, true)
+	}
+	if <-dummyCh {
+		fmt.Println("khattam")
+	}
+}

assignment4/src/raft/server.go

+// server.go
+package raft
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"strconv"
+)
+
+var rafts map[int]*Raft
+
+func getLogger(serverId int, toDebug bool) (l *log.Logger) {
+	if !toDebug {
+		l = log.New(ioutil.Discard, "INFO: ", log.Ltime|log.Lshortfile)
+	} else {
+		logf, _ := os.OpenFile(strconv.Itoa(serverId), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
+		l = log.New(logf, "INFO: ", log.Ltime|log.Lmicroseconds|log.Lshortfile)
+	}
+
+	l.Println("Initialized server.")
+	return l
+}
+
+func Start(serverId int, toDebug bool) {
+	eventCh := make(chan RaftEvent)
+	commitCh := make(chan LogEntry)
+	monitorVotesCh := make(chan bool)
+	clusterConfig, _ := NewClusterConfig(5)
+	rft, _ := NewRaft(clusterConfig, serverId, commitCh, eventCh, monitorVotesCh, true)
+	if rafts == nil {
+		rafts = make(map[int]*Raft)
+	}
+	rafts[serverId] = rft
+	fmt.Println(len(rafts))
+	rft.loop()
+}

assignment4/src/server.go

+// server.go
+package main
+
+import (
+	"bytes"
+	"connhandler"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/rpc"
+	"os"
+	"raft"
+	"strconv"
+)
+
+// Logger
+var Info *log.Logger
+
+//global raft object for each server instance
+var rft *raft.Raft
+
+//Receiver for RPC
+type AppendEntries struct{}
+
+//Receiver for voting related RPC
+type Voting struct{}
+
+//receiver for testing RPC
+//only for testing purpose
+type Tester struct{}
+
+//RPC argument for testing the replication of keys value version across key value stores
+type TestArgs struct {
+	key     string
+	value   []byte
+	version uint64
+}
+
+// RPC argument with boolean value in the reply to confirm that indeed the replication went through across servers
+type TestReply struct {
+	replica_updated bool
+}
+
+//only for testing purpose
+//this function checks for the key value in its kvstore and sets reply.replica_updated true if present and false if absent
+//arguments: args contains the key, value, version to be matched
+//reply is the reply to be sent
+func (t *Tester) testerRPC(args *TestArgs, reply *TestReply) error {
+	table := raft.GetKeyValStr()
+	table.RLock()
+	defer table.RUnlock()
+	dic := table.GetDicKVstr()
+	if v, ok := dic[args.key]; ok {
+		the_val := v.GetVal()
+		the_vers := v.GetVers()
+		if bytes.Equal(the_val, args.value) && the_vers == args.version {
+			reply.replica_updated = true
+			return nil
+		} else {
+			return nil
+		}
+	} else {
+		return nil
+	}
+}
+
+type Reply struct {
+	X int
+}
+
+//RPC for follower server. To let followers know that they can append their logs
+//arguments: pointer to argument struct (has LogEntryData), pointer to reply struct
+//returns: error
+//receiver: pointer to AppendEntries
+func (t *AppendEntries) AppendRPC(args *raft.AppendRPC, reply *Reply) error {
+	Info.Println("append RPC invoked")
+	rft.AddToEventChannel(args)
+	reply.X = 1
+	return nil
+	/*Info.Println("Append Entries RPC invoked", (*args).GetLsn(), (*args).GetData(), (*args).GetCommitted())
+	rft.LogArray = append(rft.LogArray, raft.NewLogEntry((*args).GetData(), (*args).GetCommitted(), nil))
+	reply.X = 1
+	return nil*/
+}
+
+func (t *AppendEntries) AppendReplyRPC(args *raft.AppendReplyRPC, reply *Reply) error {
+	Info.Println("append reply to leader RPC invoked")
+	rft.AddToEventChannel(args)
+	reply.X = 1
+	return nil
+	/*Info.Println("Append Entries RPC invoked", (*args).GetLsn(), (*args).GetData(), (*args).GetCommitted())
+	rft.LogArray = append(rft.LogArray, raft.NewLogEntry((*args).GetData(), (*args).GetCommitted(), nil))
+	reply.X = 1
+	return nil*/
+}
+
+//RPC for follower server. To let followers know that and entry can be committed.
+//arguments: pointer to argument struct (has LogEntry), pointer to reply struct
+//returns: error
+//receiver: pointer to AppendEntries
+func (t *AppendEntries) CommitRPC(args *raft.CommitData, reply *Reply) error {
+	Info.Println("Commit RPC invoked")
+	rft.LogArray[(*args).Id].SetCommitted(true)
+	rft.AddToChannel(rft.LogArray[(*args).Id])
+	reply.X = 1
+	return nil
+}
+
+func (t *Voting) VoteRequestRPC(args *raft.VoteRequest, reply *Reply) {
+	Info.Println("Request Vote RPC received from server", id)
+	rft.AddToEventChannel(args)
+	reply.X = 1
+	return nil
+}
+
+func (t *Voting) CastVoteRPC(args *raft.VoteRequestReply, reply *Reply) {
+	Info.Println("Request Vote RPC received from server", id)
+	rft.AddToMonitorVotesChannel(args)
+	reply.X = 1
+	return nil
+}
+
+//Initialize all the things necessary for start the server for inter raft communication.
+//The servers are running on ports 20000+serverId. {1..5}
+//arguments: pointer to current server config, pointer to raft object, a bool channel to set to true to let
+//the invoker know that the proc ended.
+//returns: none
+//receiver: none
+func initInterServerCommunication(server *raft.ServerConfig, rft *raft.Raft, ch chan bool) {
+	appendRpc := new(AppendEntries)
+	rpc.Register(appendRpc)
+	listener, e := net.Listen("tcp", ":"+strconv.Itoa(server.LogPort))
+	if e != nil {
+		Info.Fatal("listen error:", e)
+	}
+	for {
+		if conn, err := listener.Accept(); err != nil {
+			Info.Fatal("accept error: " + err.Error())
+		} else {
+			Info.Printf("new connection established\n")
+			go rpc.ServeConn(conn)
+		}
+	}
+	ch <- true
+}
+
+//Simple logger that is enabled or disabled according to the command line arguments. In test cases
+//it is redirected to a file per server {1..5}.
+//arguments: current server id, toggle enable/disable
+//return: none
+//receiver: none
+func initLogger(serverId int, toDebug bool) {
+	// Logger Initializaion
+	if !toDebug {
+		Info = log.New(ioutil.Discard, "INFO: ", log.Ldate|log.Ltime|log.Lshortfile)
+	} else {
+		Info = log.New(os.Stdout, "INFO: ", log.Ldate|log.Ltime|log.Lshortfile)
+	}
+
+	Info.Println("Initialized server.")
+}
+
+//Initialize all the things necessary for start the server for communication with client.
+//The servers are running on ports 9000+serverId {1..5}.
+//arguments: pointer to current server config, pointer to raft object, a bool channel to set to true to let
+//the invoker know that the proc ended.
+//returns: none
+//receiver: none
+func initClientCommunication(server *raft.ServerConfig, rft *raft.Raft, ch chan bool) {
+	listener, e := net.Listen("tcp", ":"+strconv.Itoa(server.ClientPort))
+	if e != nil {
+		Info.Fatal("client listen error:", e)
+	}
+	for {
+		if conn, err := listener.Accept(); err != nil {
+			Info.Fatal("client accept error: " + err.Error())
+		} else {
+			Info.Printf("client new connection established\n")
+			go connhandler.HandleClient(conn, rft, Info)
+		}
+	}
+	ch <- true
+}
+
+//Entry point for application. Starts all major server go routines and then waits for ever
+func main() {
+	sid, err := strconv.Atoi(os.Args[1])
+	ch1 := make(chan bool)
+	ch2 := make(chan bool)
+	if err != nil {
+		Info.Println("argument ", os.Args[1], "is not string")
+	}
+
+	if len(os.Args) > 3 {
+		initLogger(sid, true)
+	} else {
+		initLogger(sid, false)
+	}
+	Info.Println("Starting")
+
+	serverCount, err2 := strconv.Atoi((os.Args[2]))
+	if err2 != nil {
+		Info.Println("argument ", os.Args[2], "is not string")
+	}
+
+	server, _ := raft.NewServerConfig(sid)
+	clusterConfig, _ := raft.NewClusterConfig(serverCount)
+	commitCh := make(chan raft.LogEntry)
+
+	rft, _ = raft.NewRaft(clusterConfig, sid, commitCh, Info)
+	raft.InitKVStore(Info, sid)
+
+	go raft.MonitorCommitChannel(commitCh) //for kvstore
+	go initClientCommunication(server, rft, ch1)
+	go initInterServerCommunication(server, rft, ch2)
+
+	for <-ch1 && <-ch2 {
+
+	}
+}

assignment4/src/server_test.go

+//testing
+
+package main
+
+import (
+	"bytes"
+	//"fmt"
+	"net"
+	"net/rpc"
+	"os"
+	"os/exec"
+	"raft"
+	"strconv"
+	"testing"
+	"time"
+)
+
+//constant values used
+const (
+	NUM_SERVERS int = 5
+)
+
+type Testpair struct {
+	to_server   []byte
+	from_server []byte
+}
+
+//
+func TestAll(t *testing.T) {
+	//start the servers
+	for i := 1; i <= NUM_SERVERS; i++ {
+		go startServers(i, t)
+	}
+	//wait for some time so that servers are ready
+	time.Sleep(4 * time.Second)
+}
+
+//run servers
+func startServers(i int, t *testing.T) {
+	cmd := exec.Command("go", "run", "server.go", strconv.Itoa(i), strconv.Itoa(NUM_SERVERS), "x")
+	f, err := os.OpenFile(strconv.Itoa(i), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
+	if err != nil {
+		t.Errorf("error opening file: %v", err)
+	}
+
+	defer f.Close()
+	cmd.Stdout = f
+	cmd.Stderr = f
+	cmd.Run()
+}

assignment4/src/utils/utils.go

+// utils
+package utils
+
+import (
+	"bytes"
+	"encoding/gob"
+)
+
+//Struct to help extraction of command and value from the Data field of raft.LogEntryData
+type Command struct {
+	Cmd []byte //the command like set .s..
+	Val []byte //the value the user wants to send
+}
+
+//Custom encoder to encode the Command struct into a byte array. gob encoder will call it
+//arguments: none
+//returns: the byte array for encoded data, error
+//receiver: pointer to Command struct
+func (d *Command) GobEncode() ([]byte, error) {
+	w := new(bytes.Buffer)
+	encoder := gob.NewEncoder(w)
+	err := encoder.Encode(d.Cmd)
+	if err != nil {
+		return nil, err
+	}
+	err = encoder.Encode(d.Val)
+	if err != nil {
+		return nil, err
+	}
+	return w.Bytes(), nil
+}
+
+//Custom decoder to decode a byte array with appr data to Command struct. gob decoder will call it.
+//arguments: byte array with data to be decoded
+//returns: error if any
+//receiver: pointer to Command struct
+func (d *Command) GobDecode(buf []byte) error {
+	r := bytes.NewBuffer(buf)
+	decoder := gob.NewDecoder(r)
+	err := decoder.Decode(&d.Cmd)
+	if err != nil {
+		return err
+	}
+	return decoder.Decode(&d.Val)
+}