moved all files to src

assignment2/src/connhandler.go

+package handler
+
+import (
+	"bytes"
+	"kvstore"
+	"raft"
+	"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
+ *returns: the value string and error state
+ */
+func readValue(ch chan []byte, n uint64) ([]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(READ_TIMEOUT * 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 {
+			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(CRLF)...)
+	conn.Write(buf)
+}
+
+func HandleClient(conn net.Conn, rft *raft.SharedLog) {
+	defer conn.Close()
+	//channel for every connection for every client
+	ch := make(chan []byte)
+	go MyRead(ch, conn)
+
+	for {
+		command := new(Command)
+		msg := <-ch
+		if len(msg) == 0 {
+			continue
+		}
+		command.cmd = msg
+		if kvstore.IsCasOrSet(string(msg)) {
+			if v, err := readValue(ch, n); err {
+				Write(conn, kvstore.ERR_CMD_ERR)
+			} 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)
+		}
+
+		logEntry, err := rft.Append(buffer.Bytes())
+	}
+}

assignment2/src/raft.go

+package raft
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/rpc"
+	"os"
+	"reflect"
+	"strconv"
+	"time"
+)
+
+//constant values used
+const (
+	CLIENT_PORT = 9000
+	ACK_TIMEOUT = 5
+)
+
+type Lsn uint64 //Log sequence number, unique for all time.
+
+type ErrRedirect int // See Log.Append. Implements Error interface.
+
+type LogEntry interface {
+	Lsn() Lsn
+	Data() []byte
+	Committed() bool
+}
+
+type LogEntryData struct {
+	id        Lsn
+	data      []byte
+	committed bool
+}
+
+type ServerConfig struct {
+	Id         int    // Id of server. Must be unique
+	Hostname   string // name or ip of host
+	ClientPort int    // port at which server listens to client messages.
+	LogPort    int    // tcp port for inter-replica protocol messages.
+}
+
+type ClusterConfig struct {
+	Path    string         // Directory for persistent log
+	Servers []ServerConfig // All servers in this cluster
+}
+
+type SharedLog interface {
+	Append(data []byte) (LogEntry, error)
+}
+
+type Raft struct {
+	log_array      []*LogEntryData
+	commitCh       chan LogEntry
+	cluster_config *ClusterConfig //cluster
+	id             int            //this server id
+}
+
+func NewRaft(config *ClusterConfig, thisServerId int, commitCh chan LogEntry) (*Raft, error) {
+	rft := new(Raft)
+	rft.commitCh = commitCh
+	rft.cluster_config = config
+	rft.id = thisServerId
+	return rft, nil
+}
+
+//goroutine that monitors channel for commiting log entry
+func monitorCommitChannel(raft *Raft) { //unidirectional -- can only read from the channel
+	for {
+		//var temp LogEntry
+		temp := <-raft.commitCh                                  //receive from the channel
+		raft.log_array[temp.(*LogEntryData).id].committed = true //commit the value
+		//update the kv store here
+	}
+}
+
+//goroutine that monitors channel to check if the majority of servers have replied
+func monitorAckChannel(rft *Raft, ack_ch <-chan int, log_entry LogEntry, majEventCh chan int) {
+	acks_received := 0
+	num_servers := len(rft.cluster_config.Servers)
+	required_acks := num_servers / 2
+	up := make(chan bool, 1)
+	err := false
+
+	go func() {
+		time.Sleep(ACK_TIMEOUT * time.Second)
+		up <- true
+	}()
+
+	for {
+		select {
+		case temp := <-ack_ch:
+			acks_received += temp
+			if acks_received == required_acks {
+				rft.commitCh <- log_entry
+				majEventCh <- 1
+				err = true
+				break
+			}
+
+		case <-up:
+			err = true
+			break
+		}
+		if err {
+			break
+		}
+	}
+}
+
+//make LogEntryData implement the
+func (entry *LogEntryData) Lsn() Lsn {
+	return entry.id
+}
+
+func (entry *LogEntryData) Data() []byte {
+	return entry.data
+}
+
+func (entry *LogEntryData) Committed() bool {
+	return entry.committed
+}
+
+//make raft implement the append function
+func (raft *Raft) Append(data []byte) (LogEntry, error) {
+	if raft.id != 1 {
+		return nil, ErrRedirect(1)
+	}
+	temp := new(LogEntryData)
+	temp.id = 1
+	temp.committed = false
+	temp.data = data
+	raft.log_array = append(raft.log_array, temp)
+
+	ackChan := make(chan int)
+	majEventCh := make(chan int)
+	go monitorAckChannel(raft, ackChan, temp, majEventCh)
+	go monitorCommitChannel(raft)
+
+	for _, server := range cluster_config.Servers[1:] {
+		go func(ackChan chan int) {
+			client, err := rpc.Dial("tcp", server.Hostname+":"+strconv.Itoa(server.LogPort))
+			if err != nil {
+				Info.Fatal("Dialing:", err)
+			}
+			reply := new(Reply)
+			args := &Args{7}
+			appendCall := client.Go("AppendEntries.AppendEntriesRPC", args, reply, nil) //let go allocate done channel
+			appendCall = <-appendCall.Done
+			ackChan <- reply.X
+		}(ackChan)
+	}
+	//channel will return 1 if majority
+	if <-majEventCh == 1 {
+		raft.commitCh <- temp
+	}
+	return temp, nil
+}
+
+func NewServerConfig(server_id int) (*ServerConfig, error) {
+	server := new(ServerConfig)
+	server.Id = server_id
+	server.Hostname = "127.0.0.1"
+	server.ClientPort = CLIENT_PORT
+	server.LogPort = CLIENT_PORT + server_id
+	return server, nil
+}
+
+func NewClusterConfig(num_servers int) (*ClusterConfig, error) {
+	config := new(ClusterConfig)
+	config.Path = ""
+	config.Servers = make([]ServerConfig, num_servers)
+
+	for i := 1; i <= num_servers; i++ {
+		curr_server, _ := NewServerConfig(i)
+		config.Servers[i-1] = *(curr_server)
+	}
+
+	return config, nil
+}
+
+func (e ErrRedirect) Error() string {
+	return "Redirect to server " + strconv.Itoa(0)
+}

assignment2/src/runner.go

+package main
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+)
+
+//constant values used
+const (
+	NUM_SERVERS int = 5
+)
+
+func TestServersCommunic(i int) {
+	cmd := exec.Command("go", "run", "server.go", strconv.Itoa(i+1), strconv.Itoa(NUM_SERVERS))
+	f, err := os.OpenFile(strconv.Itoa(i), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
+	if err != nil {
+		fmt.Println("error opening file: %v", err)
+	}
+
+	defer f.Close()
+	cmd.Stdout = f
+	cmd.Stderr = f
+	cmd.Run()
+	/*if err != nil {
+		fmt.Println(err)
+	}
+
+	fmt.Println(string(out))*/
+}
+
+func main() {
+	for i := 0; i < NUM_SERVERS; i++ {
+		go TestServersCommunic(i)
+	}
+
+	var dummy_input string
+	fmt.Scanln(&dummy_input)
+}

assignment2/src/server.go

+// server.go
+package main
+
+import (
+	"cs733/assignment2/handler"
+	"cs733/assignment2/kvstore"
+	"cs733/assignment2/raft"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/rpc"
+	"os"
+	"reflect"
+	"strconv"
+	"time"
+)
+
+// Logger
+var Info *log.Logger
+
+// Flag for enabling/disabling logging functionality
+var DEBUG = true
+
+type Args struct {
+	X int
+}
+
+type Reply struct {
+	X int
+}
+
+type AppendEntries struct{}
+
+func (t *AppendEntries) AppendEntriesRPC(args *Args, reply *Reply) error {
+	reply.X = args.X
+	return nil
+}
+
+func initInterServerCommunication(server *raft.ServerConfig, rft *raft.SharedLog) {
+	appendRpc := new(AppendEntries)
+	rpc.Register(appendRpc)
+	listener, e := net.Listen("tcp", ":"+strconv.Itoa(this_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)
+		}
+	}
+}
+
+// Initialize Logger
+func initLogger(serverId int) {
+	// Logger Initializaion
+	if !DEBUG {
+		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")
+}
+
+func initClientCommunication(server *raft.ServerConfig, rft *raft.SharedLog) {
+	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 handler.HandleClient(conn, rft)
+		}
+	}
+}
+
+func main() {
+	sid, err := strconv.Atoi(os.Args[1])
+	if err != nil {
+		Info.Println("argument ", os.Args[1], "is not string")
+	}
+
+	initLogger(sid)
+	Info.Println("Start")
+
+	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)
+
+	initClientCommunication(server, rft)
+	initInterServerCommunication(server, rft)
+
+	var dummy string
+	fmt.Scanln(&dummy)
+}

assignment2/src/utils.go

+// utils
+package utils
+
+import (
+	"encoding/gob"
+	"fmt"
+)
+
+type Command struct {
+	cmd []byte
+	val []byte
+}
+
+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
+}
+
+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)
+}