Skip to content

C
cs733
Commit 93e41dec authored by Bharath Radhakrishnan's avatar Bharath Radhakrishnan
Browse files
Options

deleted src folder

parent de3eae7c
Expand all Show whitespace changes
Inline Side-by-side
Showing with 0 additions and 1148 deletions
assignment2/src/connhandler.go deleted 100644 → 0
View file @ de3eae7c
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/kvstore.go deleted 100644 → 0
View file @ de3eae7c
This diff is collapsed.
assignment2/src/raft.go deleted 100644 → 0
View file @ de3eae7c
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 deleted 100644 → 0
View file @ de3eae7c
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 deleted 100644 → 0
View file @ de3eae7c
// 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 deleted 100644 → 0
View file @ de3eae7c
// 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)
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment